Model Context Protocol Advanced Topics for Production

If you are running MCP in production you already know the protocol spec is only the starting point. The real work lives in authentication, streaming, schema evolution, and observability. This post walks through the model context protocol advanced topics that cause incidents at 2 AM so you can ship reliable servers.

How do I implement OAuth 2.1 authentication for MCP servers?

You implement OAuth 2.1 by treating your MCP server as a standard resource server that validates access tokens issued by an authorization server. The spec mandates PKCE for public clients and encourages token introspection or JWKS validation for resource servers.

I run a FastAPI gateway in front of my MCP endpoints. That gateway handles token validation before requests hit the MCP transport layer. Here is the pattern I use in production:

from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx
import jwt
from jwt import PyJWKClient

security = HTTPBearer(auto_error=False)
jwks_client = PyJWKClient("https://auth.example.com/.well-known/jwks.json")

async def get_current_token(
    credentials: HTTPAuthorizationCredentials = Security(security)
) -> dict:
    if not credentials:
        raise HTTPException(status_code=401, detail="Missing bearer token")

    token = credentials.credentials
    try:
        signing_key = jwks_client.get_signing_key_from_jwt(token)
        payload = jwt.decode(
            token,
            signing_key.key,
            algorithms=["RS256"],
            audience="mcp-server",
            issuer="https://auth.example.com"
        )
        return payload
    except jwt.PyJWTError as exc:
        raise HTTPException(status_code=401, detail="Invalid token") from exc

@app.post("/mcp")
async def mcp_endpoint(
    request: Request,
    token: dict = Depends(get_current_token)
):
    # Forward to MCP transport layer
    ...

Do not roll your own token validation. Use python-jose or PyJWT with a cached JWKS client. Cache the signing keys. Hitting the JWKS endpoint on every request will get you rate limited by your identity provider.

Scope design matters. I use mcp:tools:read and mcp:tools:execute as distinct scopes. That lets me gate destructive tools behind a stricter policy without blocking read-only inspection. If your auth server cannot mint custom scopes, put a proxy layer in front that checks claims and rewrites the request.

How do I implement streaming and progress notifications in FastAPI?

You implement streaming by using the StreamableHttpTransport from the Python SDK and yielding ServerSentEvent objects from your tool handlers. The transport handles the SSE framing. Your job is to yield progress updates as structured notifications.

from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent
from mcp.server.sse import StreamableHttpTransport
import asyncio

mcp = FastMCP("data-pipeline")

@mcp.tool()
async def run_etl_job(dataset_id: str, progress: Progress) -> list[TextContent]:
    total_steps = 5
    for step in range(total_steps):
        await asyncio.sleep(2)  # Simulate work
        await progress.notify(
            progress=step + 1,
            total=total_steps,
            message=f"Completed step {step + 1}"
        )
    return [TextContent(type="text", text=f"ETL complete for {dataset_id}")]

app = mcp.streamable_http_app()

The Progress object is injected by the framework when the client advertises progress capability during initialization. If the client does not support progress, the notify call becomes a no-op. That backward compatibility is built in.

One failure mode I have seen: buffering. If you run behind Nginx or a load balancer, disable proxy buffering for the MCP path.

location /mcp {
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
}

Without that, your progress events sit in a buffer until the response closes. The client sees nothing until the job finishes. That defeats the purpose.

How do I build composable MCP servers with FastMCP?

You build composable servers by mounting multiple FastMCP instances on distinct route prefixes and aggregating them behind a single gateway. Each instance owns a bounded context: one for database tools, one for vector search, one for external APIs.

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

db_mcp = FastMCP("database")
search_mcp = FastMCP("vector-search")
api_mcp = FastMCP("external-api")

@db_mcp.tool()
async def query_users(limit: int = 10) -> list[TextContent]:
    ...

@search_mcp.tool()
async def semantic_search(query: str, top_k: int = 5) -> list[TextContent]:
    ...

@api_mcp.tool()
async def fetch_weather(city: str) -> list[TextContent]:
    ...

app = FastAPI()
app.mount("/mcp/db", db_mcp.streamable_http_app())
app.mount("/mcp/search", search_mcp.streamable_http_app())
app.mount("/mcp/api", api_mcp.streamable_http_app())

Each mounted app exposes its own /mcp endpoint and its own tool namespace. The client connects to the specific mount it needs. This isolates failure domains. If the vector search index crashes, the database tools stay up.

I learned this the hard way. My first production MCP server was a monolith with 40 tools. A bad dependency in one tool brought down the whole process. Splitting by domain let me deploy and scale independently. It also lets me apply different auth scopes per mount. The database mount requires mcp:db:read. The API mount requires mcp:api:external.

If you need a single logical endpoint for clients, put a router in front that forwards based on tool name prefix. But I prefer explicit mounts. It makes the contract obvious.

How do I handle schema evolution and versioning in production?

You handle schema evolution by versioning your tool namespaces and never mutating existing tool signatures. Add new tools with a version suffix. Deprecate old ones with a sunset header.

@db_mcp.tool()
async def query_users_v1(limit: int = 10) -> list[TextContent]:
    ...

@db_mcp.tool()
async def query_users_v2(
    limit: int = 10,
    offset: int = 0,
    sort_by: str = "created_at"
) -> list[TextContent]:
    ...

The client discovers tools via tools/list. It chooses the version it supports. This is simpler than trying to negotiate versions at the protocol level. MCP does not have a built-in version negotiation for tools. Do not wait for one.

For resource templates, use URI templates with explicit version segments: resource://v1/users/{id} and resource://v2/users/{id}. Clients that hardcode URIs break when you change the schema. Versioned URIs let you run both indefinitely.

I keep a changelog in the tool description field. The description parameter on @mcp.tool() supports markdown. I write:

@db_mcp.tool(
    description="""Query users.

**Version 2** adds pagination and sorting.
**Version 1** is deprecated and will be removed 2025-06-01.
"""
)
async def query_users_v2(...):
    ...

LLMs read that description. They self-correct. It works better than any external documentation.

How do I debug MCP connections with MCP Inspector and structured logs?

You debug MCP connections by running the MCP Inspector against your local server and correlating its trace with structured JSON logs from your application.

Start the inspector:

npx @modelcontextprotocol/inspector

Point it at http://localhost:8000/mcp. The inspector shows the initialize handshake, capability negotiation, every tools/call request, and the raw response. It renders progress notifications in a timeline. That alone has saved me hours.

But the inspector only shows the protocol layer. You need application context. I configure structlog in FastAPI to emit JSON with a request_id that flows through the MCP transport.

import structlog
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("x-request-id", uuid4().hex)
        structlog.contextvars.bind_contextvars(request_id=request_id)
        response = await call_next(request)
        response.headers["x-request-id"] = request_id
        return response

app.add_middleware(RequestIDMiddleware)

Now every log line inside a tool handler carries the same request_id as the HTTP request. When the inspector shows a failed tools/call at timestamp T, I grep my log aggregator for that request_id and see the full stack trace, database queries, and external API calls.

Do not rely on print statements. They do not survive container restarts and they do not correlate. Structured logging is not optional for production MCP servers.

When should I not use these patterns?

Do not add OAuth 2.1 if your MCP server runs strictly on a private network with mutual TLS. The overhead is not worth it. Use mTLS and call it a day.

Do not use streaming for tools that complete in under 200ms. The SSE framing overhead exceeds the latency budget. Return a direct response instead.

Do not split into composable mounts if you have fewer than 10 tools and a single team owns all of them. A monolith is simpler to deploy and debug. Split when the blast radius of a failure justifies the operational complexity.

Do not version every tool from day one. Start with unversioned names. Add version suffixes when you have a breaking change. Premature versioning clutters the namespace.

FAQ

What is the difference between MCP SSE and Streamable HTTP transport?

Streamable HTTP is the modern transport. It uses a single endpoint for both requests and SSE responses. The legacy SSE transport uses separate /sse and /message endpoints. Use Streamable HTTP for new servers.

How do I test MCP tools without an LLM client?

Use the MCP Inspector or write a test client with the ClientSession from the Python SDK. It speaks the same protocol and lets you assert on tool outputs in pytest. I covered a working example in my Model Context Protocol Examples: How to Run MCP with FastAPI post.

Can I run MCP over WebSockets instead of SSE?

The spec defines WebSocket as an alternative transport. The Python SDK has WebSocketTransport. It works well for long-lived connections but adds complexity for load balancing. Stick with Streamable HTTP unless you have a specific need for bidirectional streaming.

How do I handle rate limiting on MCP tools?

Apply rate limiting at the gateway layer (Nginx, Envoy, or FastAPI middleware) keyed on the authenticated principal. Do not implement it inside tool handlers. That couples business logic to infrastructure concerns.

Key Takeaways

• Validate OAuth 2.1 tokens at the gateway with cached JWKS; never roll your own crypto.
• Disable proxy buffering for SSE streams or progress events will stall.
• Mount composable FastMCP instances on distinct prefixes to isolate failure domains.
• Version tool names with suffixes (_v2) and put deprecation notices in descriptions.
• Correlate MCP Inspector traces with structured JSON logs using a shared request_id.
• Start simple. Add auth, streaming, composition, and versioning only when the pain is real.