MCP Server Architecture: Components, Transport, and Patterns

The Model Context Protocol (MCP) server architecture separates the protocol lifecycle from your business logic using a standardized transport layer, capability negotiation, and a pluggable tool or resource registry. If you are building an MCP server today, you are essentially building a specialized JSON-RPC 2.0 server that speaks stdio, SSE, or Streamable HTTP. I run a few of these in production backing LLM agents, and the protocol details matter more than the framework you pick.

What are the core components and data flow?

An MCP server centers on three runtime components: the transport listener, the protocol handler, and the capability registry. The transport listener accepts connections (stdio pipes, HTTP requests, or WebSocket upgrades). The protocol handler parses JSON-RPC frames, validates method names against the negotiated version, and routes initialize, tools/list, tools/call, resources/read, and prompts/list to the registry. The registry holds your actual Python callables, Pydantic schemas for arguments, and metadata like annotations or pagination cursors.

Data flows one way: wire bytes to parsed JSON, to a validated Pydantic model, to your handler, back to a model, to JSON, to wire bytes. Do not skip the validation step. I learned this when a malformed arguments payload from a downstream client crashed a worker because I trusted the schema inference. Explicit model_config = ConfigDict(extra='forbid') on every request model saves debugging hours.

from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, ConfigDict

class SearchArgs(BaseModel):
    model_config = ConfigDict(extra='forbid')
    query: str
    top_k: int = 5

server = Server("rag-search")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="vector_search",
            description="Search the knowledge base",
            inputSchema=SearchArgs.model_json_schema(),
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name != "vector_search":
        raise ValueError(f"Unknown tool: {name}")
    args = SearchArgs(**arguments)
    results = await search_index(args.query, args.top_k)
    return [TextContent(type="text", text=r) for r in results]

Transport layer options: stdio, SSE, and Streamable HTTP

How do you choose a transport? It depends entirely on your deployment model. Stdio is the default for local CLI tools like claude or cursor. The server spawns as a child process, communicates over stdin/stdout, and dies when the parent exits. Zero network config, zero auth complexity. But you cannot scale it horizontally, and you cannot inspect traffic with standard tooling.

SSE (Server-Sent Events) was the original HTTP transport. The client opens a GET stream for server-to-client messages and POSTs to a separate endpoint for client-to-server messages. It works behind corporate proxies that hate WebSockets, but it burns two connections per session. Connection state lives in server memory. If you run multiple replicas behind a load balancer, you need sticky sessions or a shared Redis backend for the session map. I avoid SSE for anything that needs to scale past a few hundred concurrent sessions.

Streamable HTTP is the modern replacement. It uses a single HTTP endpoint. The client POSTs a request; the server responds with either a direct JSON response (for quick calls) or an text/event-stream response (for streaming). No sticky sessions required because the protocol is stateless at the HTTP layer. The MCP spec now recommends this for all new HTTP deployments. FastAPI handles it cleanly with StreamingResponse.

from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
from mcp.server.streamable_http import StreamableHTTPSessionManager
from server.core.registry import server as mcp_server

app = FastAPI()
session_manager = StreamableHTTPSessionManager(
    app=mcp_server,
    event_store=None,  # plug in Redis later for resume
    json_response=True,
)

@app.post("/mcp")
async def mcp_endpoint(request: Request):
    return await session_manager.handle_request(request)

How does capability negotiation and versioning work?

The initialize handshake is where version mismatches explode. The client sends protocolVersion (currently 2024-11-05) and its capabilities. The server responds with its supported version and capabilities. If the client speaks a newer version than the server, the server must respond with its max supported version. The client then decides whether to proceed or error out.

Always implement a version guard in your initialize handler. Do not assume the latest spec is stable. I pin my servers to a specific protocol version in config and reject initialization requests outside a supported range. This prevents a silent degradation where a new client feature fails mysteriously because the server quietly ignored an unknown capability flag.

SUPPORTED_VERSIONS = ["2024-11-05", "2025-03-26", "2025-06-18"]

@server.initialize()
async def initialize(params: InitializeParams) -> InitializeResult:
    client_version = params.protocolVersion
    if client_version not in SUPPORTED_VERSIONS:
        # Negotiate down to the highest common version
        common = max(v for v in SUPPORTED_VERSIONS if v <= client_version)
    else:
        common = client_version
    return InitializeResult(
        protocolVersion=common,
        capabilities=server.get_capabilities(),
        serverInfo=Implementation(name="rag-search", version="1.2.0"),
    )

Security patterns: auth, OAuth, and mTLS

MCP itself has no auth spec. It delegates to the transport layer. For stdio, the OS process boundary is your trust boundary. For HTTP transports, you own the auth layer.

Bearer tokens are the pragmatic default. Validate a JWT in a FastAPI dependency before the request hits the MCP session manager. Store the token claims in request.state so your tools can enforce row-level security. For enterprise customers, implement the OAuth 2.1 Protected Resource Metadata draft so clients can discover your auth server automatically. I use authlib for the provider side and a simple dependency for the resource server.

from fastapi import Depends, HTTPException, Request
from jose import jwt, JWTError

async def verify_token(request: Request) -> dict:
    auth = request.headers.get("Authorization")
    if not auth or not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing bearer token")
    token = auth.split(" ")[1]
    try:
        payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"], audience="mcp-server")
    except JWTError as e:
        raise HTTPException(401, f"Invalid token: {e}")
    request.state.user = payload
    return payload

# Apply to the MCP endpoint
@app.post("/mcp", dependencies=[Depends(verify_token)])
async def mcp_endpoint(request: Request):
    ...

mTLS is overkill for most LLM agent workloads but required in regulated environments. Terminate it at your ingress (Envoy, Nginx, or AWS ALB) and pass the verified client cert DN down as a header. Your FastAPI app validates the header exists; the ingress did the crypto.

Scaling patterns: stateless vs stateful, connection pooling

The MCP protocol is designed to be stateless, but Streamable HTTP introduces optional session state for features like request resumption and progress notifications. If you enable the event_store in the session manager, you are now stateful. That store (Redis, usually) becomes a single point of failure and a scaling bottleneck.

My rule: keep the MCP layer stateless. Disable event_store unless you have a hard requirement for resumable streams. Push long-running work to a task queue (Celery, Dramatiq, or plain Redis streams) and return a task_id immediately. The client polls a separate /tasks/{id} endpoint or uses a WebSocket for completion events. This keeps your MCP workers CPU-bound and horizontally scalable.

Connection pooling matters on the outbound side. Your tools likely call databases, vector stores, or upstream APIs. Configure asyncpg pool sizes to match your worker count times a safety factor. Set pool_timeout and command_timeout explicitly. I’ve seen connection leaks take down a cluster because a tool handler raised an exception before releasing the connection. Use async with pool.acquire() as conn: everywhere. No exceptions.

Production deployment with FastAPI

Deploying an MCP server on FastAPI is boring in a good way. Containerize it. Run Gunicorn with Uvicorn workers. Put it behind a load balancer that speaks HTTP/2. Configure health checks on a separate /healthz endpoint that verifies DB and Redis connectivity, not just process liveness.

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry install --no-root --only main
COPY . .
EXPOSE 8000
CMD ["gunicorn", "server.transports.streamable_http:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000", "--timeout", "60", "--graceful-timeout", "30"]

Set timeout and graceful-timeout to handle long streaming responses. If you use SSE or Streamable HTTP with streaming, the default 30s Gunicorn timeout kills active streams. Match these to your longest expected tool execution time plus buffer.

Observability: emit structured JSON logs. Correlate request_id across the MCP session, the HTTP request, and your downstream tool calls. Trace the tools/call latency histogram. Alert on p99 > 5s. The LLM client will time out anyway, but you want to know why before the user complains.

For a deeper look at running this stack with FastAPI specifically, see Model Context Protocol Examples: How to Run MCP with FastAPI. If you are weighing whether this protocol fits an event-driven or request-response pattern for your agents, Event Driven Architecture vs Request Response: Trade-offs covers the architectural decision.

When not to use MCP

Don’t build an MCP server if you control both the client and the server and you don’t need LLM interoperability. A plain REST or gRPC API is simpler, faster, and easier to debug. MCP adds serialization overhead, schema negotiation latency, and a constrained type system. Use it when you need to plug into Claude, Cursor, or a local agent framework that speaks MCP. Otherwise, it is abstraction for its own sake.

FAQ

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

SSE uses two HTTP connections per session (one GET stream, one POST endpoint) and requires sticky sessions for horizontal scaling. Streamable HTTP uses a single POST endpoint that can return either JSON or an event stream, making it stateless and easier to scale behind a standard load balancer.

Do I need an event store for Streamable HTTP?

No. The event store enables request resumption and progress notifications but introduces statefulness. Disable it (event_store=None) for stateless horizontal scaling unless you specifically need clients to reconnect and resume interrupted streams.

How do I handle authentication for MCP over stdio?

Stdio transport inherits the trust boundary of the parent process. The OS user running the client owns the server process. No additional auth layer is needed or possible within the protocol; secure the host environment instead.

Can I run MCP tools as background jobs?

Yes. Return a task ID from tools/call immediately and push the actual work to a queue like Celery or Redis Streams. The client polls a separate status endpoint. This keeps MCP workers fast and stateless.

Key Takeaways

• MCP server architecture is a JSON-RPC 2.0 layer over stdio, SSE, or Streamable HTTP with a capability registry for tools, resources, and prompts.
• Prefer Streamable HTTP for production HTTP deployments; it is stateless and avoids sticky sessions.
• Pin your supported protocol versions explicitly and negotiate during initialize to prevent silent capability mismatches.
• Auth happens at the transport layer: bearer tokens for HTTP, OS process isolation for stdio.
• Keep the MCP layer stateless; offload long-running work to async task queues.
• Deploy with Gunicorn/Uvicorn, tune timeouts for streaming, and correlate logs with request IDs.