行代码透彻解析MCP原理

MCP（Model Context Protocol）的核心原理并不复杂，仅需100行代码即可清晰阐述。然而，深入理解其机制后，许多AI应用集成中的棘手问题都将找到更优雅的解法。本文将围绕几个关键要点展开：MCP的原理与代码实现、两种通信方式（STDIO和SSE），以及SSE模式下的具体通信流程。

初次接触Model Context Protocol（MCP）时，我们面临一个常见实际问题：市面上大多数文档仅介绍如何使用 @mcp.tool 注解来注入工具。然而，真实业务场景中往往充斥着异步流程，并非每个环节都有现成的函数体可添加注解。难道为了接入MCP，真的需要将每个异步流程强制转化为同步函数吗？

这确实值得深入探讨。为此，我们动手分析了MCP的通信原理，试图寻找更灵活的接入方式。

MCP通信方式解析

MCP支持两种传输协议：STDIO和SSE。当前许多实验性工具运行在STDIO之上，但若需将MCP以服务形式对外提供，通常选择SSE（Server-Sent Events）。因此，本文将重点探讨SSE模式下的MCP接入。

在查阅SSE相关资料时，发现了阮一峰老师在2017年总结的内容：

SSE与WebSocket的功能类似，皆用于在浏览器与服务器之间建立通信渠道，实现服务器向浏览器的信息推送。区别在于，WebSocket是全双工通道，支持双向通信；而SSE是单向通道，仅允许服务器向浏览器推送数据，本质上相当于一个下载流。若浏览器需要向服务器发送信息，则需额外发起一次HTTP请求。

这一特性激起了更多好奇——在STDIO模式中，可通过stdin输入、stdout输出实现双向；但SSE单向，MCP如何实现双向通信？难道需要建立两根SSE通道？带着这个疑问，我们开始了实践探索。

MCP SSE通信流程详解

借助MCP官方提供的工具 npx @modelcontextprotocol/inspector，可快速启动一个验证MCP的管理页面。通过对该页面进行抓包分析，我们很快就发现了SSE通信的端倪。

从抓包结果来看：

• 端点 /sse 仅负责推送信息，无法发送信息；发送信息需通过另一个URL。
• 客户端连接上 /sse 后，接收到的第一个Event会告知其应向哪个URL发送消息，该URL通常附带一个唯一的会话ID。

双向通信的疑问到这里基本就有了答案：

• 仅通过一根SSE长连接实现服务器向客户端的数据推送；另一条通道则采用普通的HTTP POST请求。
• 客户端的POST请求仅返回2xx状态码表示收到指令，所有真实数据返回均经由最初建立的SSE长连接进行推送。

为验证这一猜想，我们特意进行了实验——使用curl模拟发起一个 /message?sessionId=*** 的POST请求。结果不出所料，SSE事件流中新增了一条事件。

MCP SSE通信实现原理

通过抓包基本摸清了MCP的SSE通信流程：

1. 建立 /sse SSE长连接后，首先返回一个 endpoint（通常为 /message），其数据格式是纯文本的同域名URL字符串。
2. 客户端使用POST方法向该 endpoint(/message) 发送调用请求，请求体遵循JSON-RPC规范，包含 jsonrpc、method、params、id 等字段。
3. 在 /sse 长连接中返回的event同样遵循JSON-RPC规范，包含 jsonrpc、result、id、error（执行错误时）等字段。

看起来并不复杂，接下来尝试用Python实现（无需依赖MCP Python SDK）。

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

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()
    
    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": ..})
        elif payload.method == "tools/list":
            ...

@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"

这段代码里引入了几个设计：

• 采用 asyncio.Queue() 实现业务流与MCP服务流的解耦。消息队列与 EventSourceResponse 的数据流联动，每当队列中加入一条消息，就会自动通过 EventSourceResponse 向客户端推送一条。从服务端角度看，客户端如同一个标准的MQ消费者。
• 在内存中维护一个 client_id 到消息队列的映射字典，接收到消息后即可确定应投递到哪个MQ。在分布式系统中，client_id 可作为消息队列的全局唯一标识，无论请求打到哪台机器，都能定位到正确的队列。
• 服务端处理完成后，将结果消息投回队列，客户端随即能感知到。MCPServer与MCPClient保持长连接后，后端业务系统理论上可无限时长执行——只要客户端不主动超时退出，一切以消息投递返回为准。

还可以查阅官方文档，了解需要支持哪些 method：

MCP订阅模式扩展思考

在MCP的resource相关method里，有个 resources/subcribe 引起了注意。先看看 resource 的定义：

Resources represent any kind of data that an MCP server wants to make a vailable to clients. This can include: File contents、Database records、API responses、Live system data、Screenshots and images、Log files、And more

如果通过 resources/subcribe 订阅一个 Database，那么该数据库的所有变更将会源源不断地推送过来——这已经很接近流计算的常见使用形态了。

SSE本身建立了服务端向客户端的单向数据流，因此如果客户端发起一个订阅，可以创建一个Flink流计算任务向MQ发送消息，这几乎是原生地实现了资源订阅。我们可以进一步扩展这个拓扑结构。

从大模型的视角看流计算：基于MCP协议，大模型能够优雅地接入流计算能力，完成复杂业务逻辑的构建。从流计算的视角看大模型：运用MCP协议后，大模型仿佛变成了一个标准的流计算处理节点，既能接收流式消息，也能向其他MQ投递消息。

这的确是MCP设计上的一大优势。MCP既似RPC，又像MQ，那么它到底是什么？不妨从编程模型的角度来深入思考。

MCP编程模型思考

从编程模型来看，MCP本质上是一种有状态的双向RPC模型，融合了事件驱动与请求-响应的特性。这种混合模式使其在AI应用与外部系统集成方面具有独特优势。

MCP的核心特征包括：

• 有状态会话：与传统无状态REST API不同，MCP维持会话状态，客户端与服务器之间建立长期连接，会话具有明确的生命周期。
• 双向通信：不仅客户端可以调用服务器（传统RPC模式），服务器也能调用客户端（反向RPC）。例如，服务器可请求客户端执行AI采样。
• 基于能力的协商：初始化阶段进行能力协商，动态发现可用功能，以适应不同实现和版本。
• 事件通知机制：支持单向通知、资源变更订阅模式以及异步事件处理。
• 标准化接口：定义了一组标准操作，使用JSON Schema定义参数和返回值，以促进互操作性。

为更好理解MCP的定位，我们可以将其与其他常见编程模型进行对比：

• MCP vs REST API：相比REST API，MCP更注重状态与双向性。
• MCP vs 消息队列（MQ）：相比消息队列，MCP更直接、更轻量。
• MCP vs WebSocket：相比WebSocket，MCP更结构化、更标准化。
• MCP vs gRPC：相比gRPC，MCP更灵活、更易理解。
• MCP vs GraphQL：相比GraphQL，MCP更专注于工具调用。

MCP在这些编程模型中占据了一个独特的位置。正因其独特的功能定位，不应因为当前的一些能力局限性而放弃MCP的原生化适配。异步任务、事件驱动等架构，本身就应当能够原生对接MCP。

MCP服务的简易实现

既然MCP的运行原理并不复杂，我们不妨自己实现一次，以此致敬这一优秀设计。

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

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

    def list_tool(self):
        result = []
        for tool in self.tools:
            toolInfo = {
                "name": tool.__name__,
                "description": tool.__doc__,
                "inputSchema": {"type": "object","properties":{}},
            }
            for name, param in inspect.signature(tool).parameters.items():
                toolInfo["inputSchema"]["properties"][name] = {
                    "title": name,
                    "type": "string",
                }
            result.append(toolInfo)
        return result

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

    @staticmethod
    def response(result, id):
        message = {
            "jsonrpc": "2.0",
            "": result,
        }
        if (id is not None):
            return json.dumps(message)

    async def request(self, req: McpRequest):
        if req.method == "initialize":
            await self.queue.put({"event": "message", "data": self.response(self.info, req.id)})
        elif req.method == "tools/list":await self.self.queue.put({"event": "message": self.response({"tools": tools}, req.id)})
        elif req.method == "tools/call":
            pass

约100行代码即可实现一个简易版MCP服务。与官方的MCP Python SDK相比，有几个重要特性优化：

• tool注册不再依赖 @mcp.tool 注解，支持动态传入，在不同场景下提供不同的MCP URL和Tool。
• 编程模型采用MQ驱动，对异步系统、事件驱动系统或平台更加友好。参考本Python实现，转写为其他语言版本也较为方便。
• 不依赖 /sse、/message 等默认路由地址也能正常运行，表明MCP的URL完全可以自定义。

总结：深入理解MCP的本质

在深入探讨MCP的原理、通信机制和编程模型本质后，可以认识到MCP不仅仅是一个简单的API或SDK，而是一个精心设计的协议：

• 采用client-host-server架构，支持多种服务器连接。
• 实现有状态的双向RPC模型，融合事件驱动特性。
• 提供标准化的工具调用与资源访问机制。
• 支持动态能力协商和功能发现。
• 相比MQ、API、WS，占据独特的功能位置，专为AI应用与外部系统集成进行优化。

正如用100行代码所展示的那样，MCP的核心原理并不复杂，但其设计确实巧妙。这种深入理解将帮助我们超越简单的SDK使用，去创建更强大、更灵活的AI应用集成方案。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：行代码透彻解析MCP原理要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题