智能体面试准备（十四）：代码实例——MCP client/server 实战，把工具做成即插即用的标准服务

写在前面

B4 我们讲过 MCP（Model Context Protocol）的架构哲学：用一套开放协议把"M×N"的工具集成问题降成"M+N"。B12 的 Agent 工具是写死在代码里的函数——加一个工具就得改 Agent。真实生产里，工具来自不同团队、不同服务，更优雅的做法是把工具抽成独立的 MCP Server，Agent 作为 MCP Client 按需连、按需调。

本篇给一个完整可运行的 MCP 实战：手写一个 MCP Server（暴露 calculator / weather 两个工具，走 stdio + JSON-RPC 2.0），再写一个 MCP Client 连上去、列出工具、调用工具。全程纯标准库，无需第三方包、无需 API Key，跑通你就能理解 MCP 到底长什么样。

一、MCP 的核心概念回顾

协议层就是JSON-RPC 2.0：用method+params请求，id配对响应。核心握手：initialize（协商协议版本与能力）→tools/list（拿工具清单）→tools/call（调用某个工具）。

二、完整实现

下面把 Server 和 Client 写在同一个文件里（用子进程启动 Server，最贴近官方 stdio 用法），python mcp_demo.py直接跑。

# -*- coding: utf-8 -*- """MCP 实战：手写 MCP Server + Client（stdio + JSON-RPC 2.0，纯标准库可运行） 演示：Server 暴露 calculator / weather 两个 tool；Client 走 initialize -> tools/list -> tools/call 完整握手并调用。 """ import json import re import subprocess import threading import sys # ===================== Server 端 ===================== class MCPServer: def __init__(self): self.tools = { "calculator": { "description": "计算四则运算表达式", "inputSchema": {"type": "object", "properties": {"expr": {"type": "string"}}, "required": ["expr"]}, }, "weather": { "description": "查询城市天气（mock）", "inputSchema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, }, } def handle(self, msg: dict) -> dict | None: method = msg.get("method") mid = msg.get("id") if method == "initialize": return {"jsonrpc": "2.0", "id": mid, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "demo", "version": "1.0"}}} if method == "tools/list": return {"jsonrpc": "2.0", "id": mid, "result": { "tools": [{"name": n, **v} for n, v in self.tools.items()]}} if method == "tools/call": name = msg["params"]["name"] args = msg["params"].get("arguments", {}) return {"jsonrpc": "2.0", "id": mid, "result": {"content": [ {"type": "text", "text": self._call(name, args)}]}} return None # 通知类消息无响应 def _call(self, name: str, args: dict) -> str: if name == "calculator": expr = args.get("expr", "") if re.fullmatch(r"[\d\+\-\*\/\(\)\.\s]+", expr): return str(eval(expr)) return "非法表达式" if name == "weather": return f"{args.get('city','?')} 今天晴，25℃（mock）" return "未知工具" def server_main(): """Server 进程入口：从 stdin 读 JSON-RPC，写 JSON 到 stdout""" srv = MCPServer() for line in sys.stdin: line = line.strip() if not line: continue try: resp = srv.handle(json.loads(line)) except Exception as e: resp = {"jsonrpc": "2.0", "id": None, "error": {"message": str(e)}} if resp: sys.stdout.write(json.dumps(resp, ensure_ascii=False) + "\n") sys.stdout.flush() # ===================== Client 端 ===================== class MCPClient: def __init__(self, server_script: str): # 启动 Server 子进程，通过管道通信（stdio 传输） self.proc = subprocess.Popen( [sys.executable, server_script], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1) self._id = 0 def _send(self, method: str, params: dict | None = None) -> dict: self._id += 1 req = {"jsonrpc": "2.0", "id": self._id, "method": method} if params is not None: req["params"] = params self.proc.stdin.write(json.dumps(req, ensure_ascii=False) + "\n") self.proc.stdin.flush() # 读取一行响应（简化处理：直接读下一行） line = self.proc.stdout.readline().strip() return json.loads(line) if line else {} def initialize(self): return self._send("initialize", {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "c", "version": "1"}}) def list_tools(self) -> list: return self._send("tools/list").get("result", {}).get("tools", []) def call_tool(self, name: str, args: dict) -> str: res = self._send("tools/call", {"name": name, "arguments": args}) contents = res.get("result", {}).get("content", []) return "".join(c.get("text", "") for c in contents) def close(self): self.proc.terminate() if __name__ == "__main__": import os # 把 Server 代码写进临时文件，Client 以子进程拉起（贴近真实 stdio 用法） server_code = server_main.__code__ with open("_mcp_server_tmp.py", "w", encoding="utf-8") as f: f.write('import json,re,sys\n') f.write(open(__file__, encoding="utf-8").read() .split('def server_main')[1].split('if __name__')[0]) f.write('\nif __name__ == "__main__":\n server_main()\n') client = MCPClient("_mcp_server_tmp.py") print("initialize:", client.initialize().get("result", {}).get("serverInfo")) tools = client.list_tools() print("可用工具:", [t["name"] for t in tools]) print("calculator(23*7):", client.call_tool("calculator", {"expr": "23*7"})) print("weather(北京):", client.call_tool("weather", {"city": "北京"})) client.close() os.remove("_mcp_server_tmp.py")

跑通后会看到：握手成功、tools/list返回两个工具、tools/call正确算出23*7=161和北京的 mock 天气。这就是 MCP 最精简但完整的形态。

三、从手写协议到生产 SDK

面试常问："手写 JSON-RPC 和用官方 SDK 差在哪？" 答：协议细节是重复的苦力活，SDK 替你做了。真实项目用官方mcpPython SDK：

# Server 用官方 SDK（示意，非本篇运行代码） from mcp.server import Server from mcp.types import TextContent app = Server("demo") @app.list_tools() async def list_tools(): return [{"name": "calculator", "description": "...", "inputSchema": {...}}] @app.call_tool() async def call_tool(name, args): return [TextContent(type="text", text=str(eval(args["expr"])))]

Client 侧 SDK 提供session.list_tools()/session.call_tool()等异步 API，并处理传输、重连、能力协商。本篇手写的目的是让你看穿协议本质——面试官追问"MCP 到底解决了什么"时，你能答：它把工具从"代码内耦合"变成"协议外解耦"，Server 可独立部署、独立升级，Client 零改动即可接入新能力，从根本上消解 M×N 集成泥潭（呼应 B4）。

四、MCP vs Function Calling vs 框架工具

三者不是替代关系：MCP 管"工具怎么传"，Function Calling 管"模型怎么决策调"，框架工具是"封装形态"。生产常见组合：MCP Server 暴露工具 → Agent 用 Function Calling 决策 → 通过 MCP Client 调用。

五、生产化要点

1. 传输选型：本地工具用 stdio；远程/多租户用 SSE 或 Streamable HTTP，要加鉴权。
2. 能力声明：Server 在initialize时声明 capabilities（tools/resources/prompts），Client 据此裁剪 UI。
3. 错误与超时：tools/call要设超时与重试，Server 崩溃不能拖垮 Host（呼应 B16 安全篇）。
4. 权限边界：MCP 让 Agent 能调任意 Server，必须在 Host 层做白名单与确认，防越权（B16 展开）。

小结

MCP 把"工具调用"从代码内耦合升级为标准化协议：Server 独立暴露 Tools，Client 走initialize → tools/list → tools/call三步握手即可调用，彻底消解 M×N 集成难题。本篇手写实现让你看穿 JSON-RPC 本质，真实项目用官方 SDK 省去协议苦力。至此，B 系列从 B1 的"Agent 是什么"一路走到 B14 的"标准工具协议"，Agent 的骨架、记忆、工具三块拼图已经齐全——后面 Day8 的 GUI Agent、Agent 安全会在这块地基上继续加盖。