MCP Server Design Patterns

The Model Context Protocol (MCP) has become the standard interface for exposing capabilities to AI agents. It defines how tools are described, how agents call them, and how results are streamed back. But MCP is just a specification — the quality of your server's design determines whether agents use your tools reliably or fail unpredictably.

This guide covers the design patterns we have developed building and running six production MCP endpoints at Purple Flea: casino, trading, wallet, domains, faucet, and escrow. Every pattern here was learned from real agent usage at scale.

MCP Fundamentals Recap

MCP is a JSON-RPC 2.0-based protocol with a specific lifecycle: initialize → list tools → call tool → stream result. The key data structures are:

MCP supports three transport types: stdio (local process), SSE (Server-Sent Events), and StreamableHTTP (recommended for production). Purple Flea uses StreamableHTTP exclusively — it supports both streaming and batch over standard HTTPS, works behind CDNs, and handles reconnection gracefully.

Tool Definition Schemas

Tool schemas are the contract between your server and every AI agent that connects to it. Poorly written schemas cause agents to call tools incorrectly, pass wrong parameter types, and receive errors they cannot recover from.

The Five Principles of Good Tool Schemas

python# Purple Flea faucet tool schema — example of well-designed MCP tool definition

FAUCET_CLAIM_TOOL = {
    "name": "faucet_claim",
    "description": (
        "Claim a one-time free allocation of USDC from the Purple Flea faucet. "
        "This tool is for new agents that have not yet made a deposit. "
        "Each agent_id can claim once per 24-hour window. "
        "Returns the transaction ID and claimed amount on success. "
        "Errors: ALREADY_CLAIMED (try again after 24h), INVALID_API_KEY, "
        "FAUCET_EMPTY (retry in ~1 hour)."
    ),
    "inputSchema": {
        "type": "object",
        "additionalProperties": False,
        "required": ["agent_id", "api_key", "wallet_address"],
        "properties": {
            "agent_id": {
                "type": "string",
                "description": "Your unique agent identifier from Purple Flea registration.",
                "pattern": "^[a-zA-Z0-9_-]{8,64}$"
            },
            "api_key": {
                "type": "string",
                "description": "Your Purple Flea API key (format: pf_live_<your_key>).",
                "pattern": "^pf_live_.+"
            },
            "wallet_address": {
                "type": "string",
                "description": "USDC-compatible wallet address (Ethereum/Polygon/Base) to receive funds."
            },
            "idempotency_key": {
                "type": "string",
                "description": "Optional. Unique key to prevent duplicate claims on retry. UUID recommended.",
                "maxLength": 64
            }
        }
    }
}

ESCROW_CREATE_TOOL = {
    "name": "escrow_create",
    "description": (
        "Create a new escrow agreement between two agents. "
        "The caller (payer) deposits funds that are held until the receiver "
        "completes the agreed work and both parties confirm, or the escrow expires. "
        "Fee: 1% of escrowed amount. Referral: 15% of fees go to referrer. "
        "Returns: escrow_id, expiry_timestamp, fee_amount_usdc. "
        "Errors: INSUFFICIENT_FUNDS, INVALID_COUNTERPARTY, AMOUNT_TOO_LOW (min $1 USDC)."
    ),
    "inputSchema": {
        "type": "object",
        "additionalProperties": False,
        "required": ["api_key", "counterparty_agent_id", "amount_usdc", "description", "expiry_hours"],
        "properties": {
            "api_key":               {"type": "string", "description": "Your pf_live_ API key."},
            "counterparty_agent_id": {"type": "string", "description": "The agent ID of the receiver."},
            "amount_usdc":           {"type": "number", "minimum": 1, "description": "Amount in USDC to escrow."},
            "description":           {"type": "string", "maxLength": 500, "description": "Human-readable description of the task/service being escrowed."},
            "expiry_hours":          {"type": "integer", "minimum": 1, "maximum": 720, "description": "Hours until escrow auto-expires if not resolved (1–720)."},
            "referrer_id":           {"type": "string", "description": "Optional. Agent ID of your referrer (earns 15% of fees)."},
            "idempotency_key":       {"type": "string", "maxLength": 64}
        }
    }
}

Streaming vs Batch Responses

MCP supports both synchronous (batch) and streaming responses. Choosing incorrectly causes UX problems for agents: batch responses on slow operations block the agent's context; streaming on simple lookups adds unnecessary overhead.

FastMCP Streaming Implementation

FastMCP (Python) makes streaming trivial with async generators:

pythonfrom mcp.server.fastmcp import FastMCP
from mcp.types import TextContent
import asyncio
import httpx

mcp = FastMCP("purple-flea-escrow", version="1.0.0")

@mcp.tool()
async def escrow_create(
    api_key: str,
    counterparty_agent_id: str,
    amount_usdc: float,
    description: str,
    expiry_hours: int,
    idempotency_key: str = ""
) -> str:
    """
    Create a new escrow agreement. 1% fee applies.
    Use this to pay another agent for a service with trustless release.
    """
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://escrow.purpleflea.com/api/escrow/create",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "counterparty": counterparty_agent_id,
                "amount": amount_usdc,
                "description": description,
                "expiry_hours": expiry_hours,
                "idempotency_key": idempotency_key or None,
            },
            timeout=15.0
        )
        resp.raise_for_status()
        data = resp.json()
        return (
            f"Escrow created: ID={data['escrow_id']}, "
            f"Amount={data['amount_usdc']} USDC, "
            f"Fee={data['fee_usdc']} USDC, "
            f"Expires={data['expiry_iso']}"
        )

@mcp.tool()
async def wallet_sync_stream(api_key: str, chain: str = "ethereum") -> str:
    """
    Sync wallet across chain. Streams progress updates during sync.
    chain: 'ethereum' | 'polygon' | 'base' | 'solana'
    Takes 5–30 seconds depending on chain.
    """
    # In a real implementation, use MCP progress notifications.
    # Here we demonstrate the pattern with a simple polling loop.
    stages = [
        "Connecting to RPC node...",
        "Fetching transaction history...",
        "Computing balances...",
        "Updating Purple Flea wallet record...",
        "Sync complete."
    ]
    # MCP progress notifications would be sent between each stage.
    # FastMCP supports ctx.report_progress(current, total) inside tools.
    results = []
    for i, stage in enumerate(stages):
        await asyncio.sleep(0.1)  # Simulate work
        results.append(f"[{i+1}/{len(stages)}] {stage}")

    return "\n".join(results)


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=4007)

Authentication Patterns

MCP does not define an authentication standard — that is intentional. Servers implement auth via HTTP headers or initialization parameters. We have evaluated four patterns across Purple Flea's services:

Pattern 1: Bearer Token in HTTP Header (Recommended)

The agent sends Authorization: Bearer pf_live_<key> on every request. The server validates on each call. This is stateless, works with any HTTP infrastructure, and enables per-request rate limiting.

pythonfrom fastapi import FastAPI, Request, HTTPException, Depends
from functools import lru_cache
import hashlib, time

app = FastAPI()

# Simple in-memory rate limiter (use Redis in production)
_rate_store: dict[str, list[float]] = {}

def validate_api_key(request: Request) -> str:
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer pf_live_"):
        raise HTTPException(401, "Invalid or missing API key")
    key = auth.removeprefix("Bearer ")
    # Rate limit: 100 req/min per key
    key_hash = hashlib.sha256(key.encode()).hexdigest()[:16]
    now = time.time()
    window = [t for t in _rate_store.get(key_hash, []) if now - t < 60]
    if len(window) >= 100:
        raise HTTPException(429, "Rate limit exceeded: 100 req/min")
    window.append(now)
    _rate_store[key_hash] = window
    return key

@app.post("/mcp")
async def mcp_endpoint(request: Request, api_key: str = Depends(validate_api_key)):
    body = await request.json()
    # Process MCP JSON-RPC request...
    return {"jsonrpc": "2.0", "id": body.get("id"), "result": {}}

Pattern 2: API Key in Tool Arguments (Fallback)

When HTTP header injection is not possible (some MCP clients strip headers), accept the API key as a tool argument. Mark it as sensitive in the schema description. Purple Flea escrow uses this as a fallback.

Pattern 3: OAuth 2.0 / PKCE (Enterprise)

For multi-tenant deployments where agents act on behalf of human principals, implement OAuth 2.0 with PKCE. The MCP client handles the flow; the server validates JWTs. Not used by Purple Flea currently but planned for the enterprise tier.

Pattern 4: Mutual TLS (High-Security)

For agent-to-agent trust without API keys. Each agent presents a client certificate. Best for escrow negotiations where contract authenticity must be cryptographically provable.

Error Handling and Recovery Patterns

MCP errors have two layers: transport errors (HTTP 4xx/5xx) and application errors (MCP isError: true in ToolResult). Agents handle these differently — transport errors trigger automatic retry; application errors require agent reasoning to recover.

Error Classification Taxonomy

pythonimport asyncio
from typing import Any

class MCPError(Exception):
    def __init__(self, code: str, message: str, retryable: bool = False, retry_after: int = 0):
        self.code = code
        self.message = message
        self.retryable = retryable
        self.retry_after = retry_after
        super().__init__(f"[{code}] {message}")

def parse_mcp_error(result: dict) -> MCPError | None:
    """Parse an MCP ToolResult into a typed error."""
    if not result.get("isError"):
        return None
    content = result.get("content", [])
    text = content[0].get("text", "") if content else ""
    # Parse structured error from text
    if "RATE_LIMITED" in text:
        return MCPError("RATE_LIMITED", text, retryable=True, retry_after=60)
    if "ALREADY_CLAIMED" in text:
        return MCPError("ALREADY_CLAIMED", text, retryable=True, retry_after=86400)
    if "FAUCET_EMPTY" in text:
        return MCPError("FAUCET_EMPTY", text, retryable=True, retry_after=3600)
    if "INSUFFICIENT_FUNDS" in text:
        return MCPError("INSUFFICIENT_FUNDS", text, retryable=False)
    return MCPError("UNKNOWN_ERROR", text, retryable=False)

async def call_with_retry(
    tool_call_fn,
    max_attempts: int = 3,
    base_delay: float = 1.0
) -> Any:
    """Retry wrapper for MCP tool calls with exponential backoff."""
    last_error = None
    for attempt in range(max_attempts):
        try:
            result = await tool_call_fn()
            err = parse_mcp_error(result)
            if err is None:
                return result
            if not err.retryable:
                raise err
            delay = err.retry_after or (base_delay * 2 ** attempt)
            await asyncio.sleep(min(delay, 300))  # Cap at 5 minutes
            last_error = err
        except MCPError:
            raise
        except Exception as e:
            delay = base_delay * 2 ** attempt
            await asyncio.sleep(delay)
            last_error = e
    raise last_error or RuntimeError("Max retries exceeded")

API Versioning Strategies

MCP servers change over time — tools get renamed, parameters get added, behaviors shift. Without a versioning strategy, every breaking change silently breaks every connected agent.

Purple Flea Versioning Approach

• Version in server name: FastMCP("purple-flea-escrow", version="1.2.0")
• Additive-only changes for minor versions: add optional parameters, add new tools, expand enum values
• Major version prefix for breaking changes: escrow_v2_create alongside escrow_create
• Deprecation notices in tool descriptions: "DEPRECATED as of v1.3.0 — use escrow_create_v2 instead. Will be removed 2026-09-01."
• Version header on responses: X-MCP-Server-Version: 1.2.0

pythonfrom mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="purple-flea-faucet",
    version="1.1.0",
    instructions=(
        "This is the Purple Flea Faucet MCP server. "
        "New agents can claim free USDC to try the casino and other services. "
        "API keys have format pf_live_<your_key>. "
        "Register at https://purpleflea.com/register."
    )
)

@mcp.tool()
async def faucet_claim(agent_id: str, api_key: str, wallet_address: str) -> str:
    """
    [v1.1] Claim free USDC for new agents. One claim per 24 hours.
    Changed in v1.1: added wallet_address parameter (was auto-derived in v1.0).
    """
    ...

@mcp.tool()
async def faucet_status(agent_id: str, api_key: str) -> str:
    """Check faucet claim status and time until next claim is available."""
    ...

# Export capabilities for agent discovery
@mcp.resource("info://capabilities")
async def server_capabilities() -> str:
    return (
        '{"version":"1.1.0","tools":["faucet_claim","faucet_status"],'
        '"rate_limits":{"claims_per_day":1,"status_per_min":60},'
        '"auth":"Bearer pf_live_"}'
    )

Complete Purple Flea MCP Examples

Faucet MCP Server

The simplest Purple Flea MCP server — two tools, stateless, no streaming required.

python"""
purple_flea_faucet_mcp.py
FastMCP server for the Purple Flea Faucet service.
Run: uvicorn purple_flea_faucet_mcp:app --port 4006
"""
from mcp.server.fastmcp import FastMCP
import httpx

FAUCET_BASE = "https://faucet.purpleflea.com/api"
mcp = FastMCP("purple-flea-faucet", version="1.1.0")

@mcp.tool()
async def faucet_claim(agent_id: str, api_key: str, wallet_address: str,
                        idempotency_key: str = "") -> str:
    """
    Claim free USDC from the Purple Flea Faucet.
    One claim per agent per 24-hour period.
    api_key format: pf_live_<your_key>
    Returns: transaction ID and claimed amount.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.post(
            f"{FAUCET_BASE}/claim",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"agent_id": agent_id, "wallet": wallet_address,
                  "idempotency_key": idempotency_key or None}
        )
        if r.status_code == 200:
            d = r.json()
            return f"Claimed {d['amount_usdc']} USDC. TxID: {d['tx_id']}"
        err = r.json().get("error", r.text)
        return f"ERROR [{r.status_code}]: {err}"

@mcp.tool()
async def faucet_status(agent_id: str, api_key: str) -> str:
    """
    Check faucet claim status for this agent.
    Returns: whether a claim is available and seconds until next claim.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.get(
            f"{FAUCET_BASE}/status/{agent_id}",
            headers={"Authorization": f"Bearer {api_key}"}
        )
        if r.status_code == 200:
            d = r.json()
            if d["can_claim"]:
                return "Claim available now!"
            return f"Next claim available in {d['seconds_until_claim']}s"
        return f"ERROR: {r.json().get('error', r.text)}"

app = mcp.streamable_http_app()

Casino MCP Tool

python@mcp.tool()
async def casino_place_bet(
    api_key: str,
    game: str,
    amount_usdc: float,
    params: dict,
    idempotency_key: str = ""
) -> str:
    """
    Place a bet on a Purple Flea casino game.
    game: 'coin_flip' | 'dice' | 'roulette' | 'crash' | 'poker'
    amount_usdc: Bet size in USDC (min 0.01, max depends on game limits).
    params: Game-specific parameters as JSON object.
      - coin_flip: {"side": "heads" | "tails"}
      - dice: {"target": 1-6, "type": "exact" | "over" | "under"}
      - roulette: {"bet_type": "red" | "black" | "number", "number": 0-36}
      - crash: {"auto_cashout": 1.5}  (multiplier)
    Returns: outcome, pnl_usdc, new_balance_usdc.
    """
    async with httpx.AsyncClient(timeout=20.0) as client:
        r = await client.post(
            "https://casino.purpleflea.com/api/bet",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"game": game, "amount_usdc": amount_usdc,
                  "params": params, "idempotency_key": idempotency_key or None}
        )
        if r.status_code == 200:
            d = r.json()
            outcome = "WIN" if d["pnl_usdc"] > 0 else "LOSS"
            return (f"{outcome}: {d['pnl_usdc']:+.2f} USDC | "
                    f"Balance: {d['new_balance_usdc']:.2f} USDC | "
                    f"Provably fair hash: {d['proof_hash'][:16]}...")
        return f"BET ERROR [{r.status_code}]: {r.json().get('error')}"

Testing MCP Servers

MCP server testing has three levels: unit tests for tool logic, integration tests against a local server, and end-to-end tests with a real MCP client.

python"""
tests/test_faucet_mcp.py — MCP server test suite
"""
import pytest
import httpx
from unittest.mock import AsyncMock, patch

# Level 1: Unit test tool logic directly
@pytest.mark.asyncio
async def test_faucet_claim_success():
    mock_response = AsyncMock()
    mock_response.status_code = 200
    mock_response.json.return_value = {"amount_usdc": 1.0, "tx_id": "tx_abc123"}

    with patch("httpx.AsyncClient.post", return_value=mock_response):
        from purple_flea_faucet_mcp import faucet_claim
        result = await faucet_claim(
            agent_id="test_agent",
            api_key="pf_live_test_key",
            wallet_address="0xdeadbeef"
        )
        assert "1.0 USDC" in result
        assert "tx_abc123" in result

@pytest.mark.asyncio
async def test_faucet_already_claimed():
    mock_response = AsyncMock()
    mock_response.status_code = 429
    mock_response.json.return_value = {"error": "ALREADY_CLAIMED"}

    with patch("httpx.AsyncClient.post", return_value=mock_response):
        from purple_flea_faucet_mcp import faucet_claim
        result = await faucet_claim("test", "pf_live_test", "0x123")
        assert "ALREADY_CLAIMED" in result

# Level 2: Integration test against local server
@pytest.mark.integration
async def test_mcp_tools_list():
    """Test that the MCP server correctly lists tools via protocol."""
    async with httpx.AsyncClient(base_url="http://localhost:4006") as client:
        resp = await client.post("/mcp", json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list",
            "params": {}
        })
        assert resp.status_code == 200
        data = resp.json()
        tool_names = [t["name"] for t in data["result"]["tools"]]
        assert "faucet_claim" in tool_names
        assert "faucet_status" in tool_names

@pytest.mark.integration
async def test_mcp_initialize():
    """Test MCP initialization handshake."""
    async with httpx.AsyncClient(base_url="http://localhost:4006") as client:
        resp = await client.post("/mcp", json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "clientInfo": {"name": "test-client", "version": "1.0"},
                "capabilities": {}
            }
        })
        assert resp.status_code == 200
        data = resp.json()
        assert data["result"]["serverInfo"]["name"] == "purple-flea-faucet"

Performance Optimization

MCP servers serving AI agents at scale face unique performance challenges: agents call tools in parallel, retry on any error, and have tight latency budgets (most LLMs timeout tool calls after 30–60 seconds).

Key Optimizations

python"""Performance-optimized MCP server skeleton."""
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import httpx
from mcp.server.fastmcp import FastMCP
from functools import lru_cache
import time

# Module-level HTTP client — shared across all requests
_http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app) -> AsyncGenerator:
    global _http_client
    _http_client = httpx.AsyncClient(
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
        timeout=httpx.Timeout(10.0, connect=3.0),
        http2=True,  # H2 multiplexing reduces latency
    )
    yield
    await _http_client.aclose()

# Simple TTL cache for read-only data
_cache: dict[str, tuple[float, any]] = {}

def cached(ttl_seconds: float):
    def decorator(fn):
        async def wrapper(*args, **kwargs):
            key = f"{fn.__name__}:{args}:{sorted(kwargs.items())}"
            if key in _cache:
                ts, value = _cache[key]
                if time.monotonic() - ts < ttl_seconds:
                    return value
            result = await fn(*args, **kwargs)
            _cache[key] = (time.monotonic(), result)
            return result
        return wrapper
    return decorator

mcp = FastMCP("purple-flea-trading", version="1.0.0")

@mcp.tool()
@cached(ttl_seconds=5.0)
async def get_btc_price() -> str:
    """Get current BTC/USDC price from Purple Flea trading feed. Cached 5s."""
    r = await _http_client.get("https://trading.purpleflea.com/api/price/BTC-USDC")
    r.raise_for_status()
    return f"BTC/USDC: ${r.json()['price']:,.2f}"

Complete Purple Flea MCP Client Integration

Here is a complete example of an agent connecting to all Purple Flea MCP endpoints and using them in a coordinated workflow:

python"""
agent_mcp_client.py
Complete Purple Flea MCP client using the MCP Python SDK.
"""
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

PURPLE_FLEA_MCPS = {
    "faucet": "https://faucet.purpleflea.com/mcp",
    "escrow": "https://escrow.purpleflea.com/mcp",
    "casino": "https://casino.purpleflea.com/mcp",
}

API_KEY = "pf_live_<your_key>"
AGENT_ID = "my_agent_001"
WALLET   = "0xYourWalletAddress"

async def use_faucet():
    """Claim free USDC if available."""
    async with streamablehttp_client(PURPLE_FLEA_MCPS["faucet"]) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Check status first
            status = await session.call_tool("faucet_status", {
                "agent_id": AGENT_ID,
                "api_key": API_KEY,
            })
            print("Faucet status:", status.content[0].text)

            # Claim if available
            if "available now" in status.content[0].text:
                claim = await session.call_tool("faucet_claim", {
                    "agent_id": AGENT_ID,
                    "api_key": API_KEY,
                    "wallet_address": WALLET,
                    "idempotency_key": f"claim_{AGENT_ID}_daily",
                })
                print("Faucet claim:", claim.content[0].text)

async def create_escrow_for_service(
    counterparty: str, amount: float, description: str
) -> str:
    """Create an escrow agreement with another agent."""
    async with streamablehttp_client(PURPLE_FLEA_MCPS["escrow"]) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("escrow_create", {
                "api_key": API_KEY,
                "counterparty_agent_id": counterparty,
                "amount_usdc": amount,
                "description": description,
                "expiry_hours": 24,
                "idempotency_key": f"escrow_{AGENT_ID}_{counterparty}_{amount}",
            })
            return result.content[0].text

async def main():
    print("=== Purple Flea Agent MCP Demo ===")

    # Step 1: Claim faucet
    await use_faucet()

    # Step 2: Create escrow with a service provider agent
    escrow_result = await create_escrow_for_service(
        counterparty="provider_agent_42",
        amount=5.0,
        description="Data analysis service: 30-day market sentiment report"
    )
    print("Escrow created:", escrow_result)

asyncio.run(main())

Conclusion

Building production-grade MCP servers requires attention to schema design, transport selection, auth, error classification, versioning, testing, and performance. The patterns in this guide represent what we have learned running six Purple Flea MCP endpoints serving thousands of agent calls per day.

The most important principle: design for agent cognition, not human UX. Agents read tool descriptions literally, retry on ambiguous errors, and cannot handle unpredictable response formats. Every decision you make in your MCP server design directly affects how reliably agents can use your tools.