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

智能体面试准备十四代码实例——MCP client/server 实战把工具做成即插即用的标准服务写在前面B4 我们讲过 MCPModel Context Protocol的架构哲学用一套开放协议把M×N的工具集成问题降成MN。B12 的 Agent 工具是写死在代码里的函数——加一个工具就得改 Agent。真实生产里工具来自不同团队、不同服务更优雅的做法是把工具抽成独立的 MCP ServerAgent 作为 MCP Client 按需连、按需调。本篇给一个完整可运行的 MCP 实战手写一个 MCP Server暴露 calculator / weather 两个工具走 stdio JSON-RPC 2.0再写一个 MCP Client 连上去、列出工具、调用工具。全程纯标准库无需第三方包、无需 API Key跑通你就能理解 MCP 到底长什么样。一、MCP 的核心概念回顾概念含义本篇对应Host承载 Agent 的应用如 IDE、聊天端我们的 Client 主程序Client与 Server 建立 1:1 连接的会话端MCPClientServer暴露 Tools/Resources/Prompts 的服务MCPServer原语tools可调用的动作/ resources可读上下文/ prompts模板本篇实现 tools传输stdio本地子进程/ SSE远程 HTTPstdio协议层就是JSON-RPC 2.0用methodparams请求id配对响应。核心握手initialize协商协议版本与能力→tools/list拿工具清单→tools/call调用某个工具。二、完整实现下面把 Server 和 Client 写在同一个文件里用子进程启动 Server最贴近官方 stdio 用法python mcp_demo.py直接跑。# -*- coding: utf-8 -*- MCP 实战手写 MCP Server Clientstdio JSON-RPC 2.0纯标准库可运行 演示Server 暴露 calculator / weather 两个 toolClient 走 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_asciiFalse) \n) sys.stdout.flush() # Client 端 class MCPClient: def __init__(self, server_script: str): # 启动 Server 子进程通过管道通信stdio 传输 self.proc subprocess.Popen( [sys.executable, server_script], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, textTrue, bufsize1) 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_asciiFalse) \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, encodingutf-8) as f: f.write(import json,re,sys\n) f.write(open(__file__, encodingutf-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*7161和北京的 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(typetext, textstr(eval(args[expr])))]Client 侧 SDK 提供session.list_tools()/session.call_tool()等异步 API并处理传输、重连、能力协商。本篇手写的目的是让你看穿协议本质——面试官追问MCP 到底解决了什么时你能答它把工具从代码内耦合变成协议外解耦Server 可独立部署、独立升级Client 零改动即可接入新能力从根本上消解 M×N 集成泥潭呼应 B4。四、MCP vs Function Calling vs 框架工具维度Function CallingMCP框架内置工具LangChain Tool定位模型决定调哪个函数工具的标准化传输协议框架内工具封装解耦度工具定义和调用紧耦合在请求里Server 独立部署彻底解耦绑定框架迁移成本高跨语言/跨进程否同一进程是stdio/SSE 跨进程否适用单模型单次调用多 Agent 共享工具生态快速原型三者不是替代关系MCP 管工具怎么传Function Calling 管模型怎么决策调框架工具是封装形态。生产常见组合MCP Server 暴露工具 → Agent 用 Function Calling 决策 → 通过 MCP Client 调用。五、生产化要点传输选型本地工具用 stdio远程/多租户用 SSE 或 Streamable HTTP要加鉴权。能力声明Server 在initialize时声明 capabilitiestools/resources/promptsClient 据此裁剪 UI。错误与超时tools/call要设超时与重试Server 崩溃不能拖垮 Host呼应 B16 安全篇。权限边界MCP 让 Agent 能调任意 Server必须在 Host 层做白名单与确认防越权B16 展开。小结MCP 把工具调用从代码内耦合升级为标准化协议Server 独立暴露 ToolsClient 走initialize → tools/list → tools/call三步握手即可调用彻底消解 M×N 集成难题。本篇手写实现让你看穿 JSON-RPC 本质真实项目用官方 SDK 省去协议苦力。至此B 系列从 B1 的Agent 是什么一路走到 B14 的标准工具协议Agent 的骨架、记忆、工具三块拼图已经齐全——后面 Day8 的 GUI Agent、Agent 安全会在这块地基上继续加盖。