深入拆解 MCP 架构：手把手搭建智能体通信桥梁

// 请求示例{  "jsonrpc": "2.0",  "id": "req-123",  "method": "calculate",  "params": {"a": 5, "b": 3}}// 响应示例{  "jsonrpc": "2.0",  "id": "req-123",  "result": 8}

from fastmcp import FastMCPmcp = FastMCP("Math Server")@mcp.tool(name="calculate")  # 装饰器注册工具def math_tool(operation: str, a: float, b: float) -> float:    """执行数学运算，支持add/subtract/multiply/divide"""    if operation == "add": return a + b    elif operation == "subtract": return a - b    elif operation == "multiply": return a * b    elif operation == "divide": return a / b  # 注意除零校验

npx @modelcontextprotocol/inspector node math_server.py

@app.route('/stream', methods=['GET'])def sse_stream():    def event_stream():        while True:            # 监听工具调用请求            result = get_tool_result()              yield f"data: {json.dumps(result)}\n\n"  # SSE格式要求        return Response(event_stream(), mimetype="text/event-stream")

const eventSource = new EventSource('http://localhost:5000/stream');// 监听计算结果eventSource.addEventListener('calculate', (event) => {    const data = JSON.parse(event.data);    console.log("收到结果：", data.result); });// 错误处理eventSource.ｏｎｅｒｒｏｒ = (err) => {    console.error("SSE连接异常：", err);};

@mcp.tool()def advanced_calc(expression: str) -> float:    """执行复杂数学表达式，支持sin/cos/log等"""    try:        return ｅｖａｌ(expression, {"__builtins__": None}, math.__dict__)    except Exception as e:        raise ValueError(f"表达式错误：{str(e)}")

async def solve_math_problem(question: str):    async with Client(mcp) as client:        # 1. 获取可用工具        tools = await client.list_tools()                  # 2. 构造LLM提示词        prompt = f"问题：{question}。可用工具：{json.dumps(tools)}"                # 3. 解析LLM返回的工具调用指令        tool_call = await llm.generate(prompt)                  # 4. 执行调用        result = await client.call_tool(tool_call.name, tool_call.params)        return result[0].text  # 返回文本结果

mcp = FastMCP("AI Server", endpoint="/mcp")  # 单端点支持POST/GET/SSE

@mcp.tool()def recommend_product(user_id: str):    ctx = get_context()  # 获取会话上下文    history = ctx.get("purchase_history")     return db.query(f"SELECT * FROM products WHERE category IN {history}")

from fastmcp.integrations import FastAPIPluginfastapi_app.include_router(FastAPIPlugin(mcp).router)

# 输入消毒def sanitize_input(sql: str) -> str:    return re.sub(r"[;\"]", "", sql)  # 移除危险字符# 权限分级TOOL_PERMISSIONS = {"query_db": ["user"], "execute_payment": ["admin"]}# 审计日志@app.middleware("http")async def log_requests(request: Request, call_next):    logger.info(f"{request.client} called {request.url.path}")    return await call_next(request)