Unlocking Model Context Protocol: Build Powerful AI Integration with SSE and Python

MCP Communication Overview

Model Context Protocol (MCP) offers two transport protocols: STDIO and Server‑Sent Events (SSE). For production services SSE is used, providing a unidirectional server‑to‑client stream while client requests are sent via ordinary HTTP POST.

SSE Communication Flow

When a client connects to /sse the server first sends an endpoint event containing the URL (e.g., /message?sessionId=…) that the client must use for POST requests. The POST request follows JSON‑RPC 2.0 format (fields jsonrpc, method, params, id). Responses and events are pushed back through the same SSE connection.

Key observations from packet capture:

Only one SSE long‑lived connection is used for server‑to‑client pushes.

Client‑to‑server communication uses a normal HTTP 2xx POST; all data is returned via SSE events.

Python Reference Implementation

Using fastapi, sse_starlette and an asyncio.Queue we can decouple business logic from the MCP service stream. The queue feeds events to EventSourceResponse, which the client receives as a standard subscription.

from fastapi import FastAPI, Request
import uuid
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
import json
import asyncio

app = FastAPI()
mcpHub = {}

class McpRequest(BaseModel):
    id: Optional[int] = None
    jsonrpc: str
    method: str
    params: Optional[dict] = None

class MCPServer:
    def __init__(self):
        self.queue = asyncio.Queue()
        self.client_id = str(uuid.uuid4())
        self.info = {
            "protocolVersion": "2024-11-05",
            "capabilities": {"experimental": {}, "tools": {"listChanged": False}},
            "serverInfo": {"name": "mcp-test", "version": "1.6.0"}
        }
        self.tools = []

    async def reader(self):
        while True:
            event = await self.queue.get()
            yield event

    async def request(self, payload: McpRequest):
        if payload.method == "initialize":
            await self.queue.put({"event": "message", "data": json.dumps({"jsonrpc": "2.0", "result": self.info, "id": payload.id})})
        # handle other methods …

@app.get("/sse")
async def sse():
    client_id = str(uuid.uuid4())
    mcp = MCPServer()
    mcpHub[client_id] = mcp
    await mcp.queue.put({"event": "endpoint", "data": f"/message?client_id={client_id}"})
    return EventSourceResponse(mcp.reader())

@app.post("/message")
async def message(request: Request, payload: McpRequest):
    client_id = request.query_params.get("client_id")
    if client_id not in mcpHub:
        return "no client"
    await mcpHub[client_id].request(payload)
    return "ok"

The implementation demonstrates three design choices:

Use of asyncio.Queue to decouple business flow from MCP service flow.

Maintain a client_id ‑to‑queue mapping so any server can locate the correct queue.

After processing, the server pushes a message back to the client, allowing effectively unlimited execution time as long as the client keeps the SSE connection alive.

Subscription Mode and Programming Model

Resources such as databases can be subscribed via resources/subscribe. Changes are streamed to the client, turning the MCP connection into a native source for stream‑processing frameworks like Flink.

From a programming‑model perspective MCP is a stateful bidirectional RPC that blends event‑driven and request‑response semantics, offering:

Stateful sessions with clear lifecycles.

Two‑way communication (server can invoke client).

Capability negotiation during initialization.

Standardized tool registration and invocation.

Comparison with Other Models

Compared with REST, MQ, WebSocket and GraphQL, MCP occupies a unique niche: more stateful and bidirectional than REST, lighter than full‑blown message queues, more structured than raw WebSockets, and focused on tool calls rather than generic RPC.

Simple MCP Service Example

The following ~100‑line script provides a minimal MCP server without relying on the default /sse and /message routes, showing that URLs are fully customizable.

from fastapi import FastAPI, Request
import asyncio, json, uuid
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from typing import Optional

app = FastAPI()
mcpHub = {}

class McpRequest(BaseModel):
    id: Optional[int] = None
    jsonrpc: str
    method: str
    params: Optional[dict] = None

class MCPServer:
    def __init__(self, name, message_path, tools):
        self.queue = asyncio.Queue()
        self.client_id = str(uuid.uuid4())
        self.message_path = message_path
        self.info = {"protocolVersion": "2024-11-05", "capabilities": {"experimental": {}, "tools": {"listChanged": False}}, "serverInfo": {"name": name, "version": "1.6.0"}}
        self.tools = tools

    async def reader(self):
        while True:
            event = await self.queue.get()
            yield event

    async def request(self, req: McpRequest):
        if req.method == "initialize":
            await self.queue.put({"event": "message", "data": json.dumps({"jsonrpc": "2.0", "result": self.info, "id": req.id})})
        # other method handling …

@app.get("/receive_test")
async def receive_test():
    mcp = MCPServer(name="mcp-test", message_path="/send_test", tools=[])
    mcpHub[mcp.client_id] = mcp
    await mcp.queue.put({"event": "endpoint", "data": f"/send_test?client_id={mcp.client_id}"})
    return EventSourceResponse(mcp.reader())

@app.post("/send_test")
async def send_test(request: Request, payload: McpRequest):
    client_id = request.query_params.get("client_id")
    if client_id not in mcpHub:
        return "no client"
    await mcpHub[client_id].request(payload)
    return "ok"

Key advantages over the official SDK include dynamic tool registration without @mcp.tool annotations, easy adaptation to other languages, and flexible routing.

Conclusion

MCP’s client‑host‑server architecture, stateful bidirectional RPC model, standardized tool calls, dynamic capability negotiation, and its position between MQ, API and WebSocket make it especially suitable for AI‑centric integration scenarios.

References

Model Context Protocol detailed tutorial – https://blog.csdn.net/ZYC88888/article/details/146414158

Server‑Sent Events tutorial – https://www.ruanyifeng.com/blog/2017/05/server-sent_events.html

Inspector tool – https://github.com/modelcontextprotocol/inspector

JSON‑RPC 2.0 specification (Chinese) – https://wiki.geekdream.com/Specification/json-rpc_2.0.html