100行代码讲透MCP原理

主要观点总结

关键观点总结

关键观点1: MCP的原理和通信机制

关键观点2: MCP的编程模型本质

关键观点3: 简易MCP服务的实现

关键观点4: MCP的优势和应用场景

关键观点5: 湖仓一体架构的简介

正文

请到「今天看啥」查看全文

2.client使用POST向 endpoint(/message) 发送调用请求，POST中的body满足JSON-RPC规范，包含字段 jsonrpc 、 method 、 params 、 id 。

3.在 /sse 长连接中返回的event满足JSON-RPC规范，包含字段 jsonrpc 、 result 、 id 、 error(执行错误时) 。

看起好像并不复杂，我们尝试用Python来实现一下（不使用MCP Python SDK）。

from fastapi import FastAPI, Requestimport uuidfrom sse_starlette.sse import EventSourceResponsefrom pydantic import BaseModelimport 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"

在这段代码中，我们引入了这样几个设计：

1.我们使用了 asyncio.Queue() 来解耦业务流和MCP服务流。这个消息队列联动 EventSourceResponse 的数据流。每往这个消息队列打一个消息，就会自动通过 EventSourceResponse

请到「今天看啥」查看全文