从零搭建 MCP 服务端 + 客户端：实战踩坑全记录

引言

经过前两篇的铺垫——第一篇拆解了 MCP 协议的三层架构，第二篇剖析了跨语言调用的细节——现在是时候把理论付诸实践了。本文目标是：从零开始，搭建一个完整可用的 MCP 服务端 + 客户端，并集成到 LangGraph AI Agent 中。我们会遇到 stdin 缓冲区问题、异步事件循环冲突、tool 调用参数类型不匹配等真实 bug。

我们的实战项目 "Tool Hub" 涵盖五种实用工具：系统信息查询（OS、CPU 核数、内存使用率）、文件读取（安全路径白名单限制）、Web 搜索（通过外部 API 聚合搜索结果）、时间日期（多时区转换和格式化）、计算器（安全沙箱内的数学表达式求值）。这些工具覆盖了 MCP 工具注册的常见场景：无参数工具、带校验参数工具、外部 HTTP 调用工具和计算密集型工具。

整套代码可以在本地完全运行，不需要外部依赖（Web 搜索除外，可配置 mock 模式）。我们使用 Python 3.10+ 和 MCP Python SDK v2.x（FastMCP），客户端通过 stdio 连接。LangGraph Agent 使用 ReAct 循环模式，让 LLM 自主决定调用哪个工具以及何时输出最终结果。这种模式是当前 AI Agent 中最主流的设计范式。

查看代码 ▸ 查看代码 ▸

项目架构

我们要构建的是 "Tool Hub" 系统工具集合服务端，提供：系统信息查询（OS、CPU、内存）、文件读取（安全受限）、Web 搜索（HTTP 聚合）、时间日期（时区转换）、计算器（安全表达式求值）。然后搭建 MCP 客户端连接服务端，通过 LangGraph Agent 让 AI 模型自主调用这些工具。

项目采用模块化设计：server/ 目录包含完整的 MCP 服务端实现，client/ 目录包含 MCP 客户端封装和 LangGraph Agent 集成，tests/ 目录包含针对每个工具的单元测试。配置文件 config.py 控制文件白名单、Web 搜索 API 密钥、时区列表等参数。这种分离让服务端可以独立测试和部署，客户端只需要引用服务的 stdio 命令即可。

数据流非常清晰：用户消息 → LangGraph Agent（LLM 推理）→ Agent 选择工具 → MCP 客户端转发 → MCP 服务端执行 → 结果返回 Agent → Agent 决定下一步或输出。这个过程中，MCP 协议层完全透明——Agent 和 LLM 不知道底层是 MCP，它们只看到 LangChain 的 StructuredTool 接口。这种抽象是 MCP 的核心价值：Agent 框架无需感知底层协议细节。

查看代码 ▸ 查看代码 ▸

MCP 服务端

使用 FastMCP（MCP Python SDK v2.x 的高层封装）搭建服务端。核心是 mcp/server/mcpserver/server.py 中的 MCPServer 类。工具注册使用装饰器模式：@server.tool("name", description="...")。每个工具接受参数并返回结果，SDK 自动处理 JSON-RPC 序列化和 ToolManager 注册。

下面看具体的工具实现。以 系统信息工具 为例：它不需要参数，返回一个包含 OS、主机名、CPU 核数和内存使用率的字典。计算器 工具需要特别注意安全性——使用 eval 但清空 __builtins__ 以防止任意代码执行。当然生产环境应该使用 ast.literal_eval 或专门的 SafeEval 库。每个工具函数都使用 async def 异步定义，即使内部操作是同步的，也方便未来切换到真正的异步 I/O。

文件读取工具 是展示 MCP 参数校验威力的好例子：它要求 path 参数必须是以 / 开头的绝对路径，并且必须在 ALLOWED_PATHS 白名单内。如果路径不合法，返回清晰的错误消息而非抛出异常——LLM 需要这些错误描述来修正自己的行为。这是 MCP 工具开发的一个重要原则：错误消息要面向 AI，而非面向开发者。

dict:\n import os, platform\n return {\n "os": platform.system(),\n "hostname": platform.node(),\n "cpu_count": os.cpu_count()\n }\n\n@server.tool("calculator", description="安全表达式求值")\nasync def calculator(expression: str) -> str:\n # 安全沙箱求值\n return str(eval(expression, {"__builtins__": {}}, {}))','MCP 服务端')">查看代码 ▸ str:\n # 安全校验\n if not os.path.isabs(path):\n return f"Error: path must be absolute, got: {path}"\n allowed = any(\n path.startswith(p) for p in config.ALLOWED_PATHS\n )\n if not allowed:\n return f"Error: path not in allowed paths: {config.ALLOWED_PATHS}"\n if not os.path.exists(path):\n return f"Error: file not found: {path}"\n try:\n with open(path, "r", encoding=encoding) as f:\n return f.read()\n except Exception as e:\n return f"Error reading file: {str(e)}"\n\n// 用户问 "帮我看看 /tmp/test.txt"\n// → LLM 调用 read_file(path="/tmp/test.txt")\n// → 服务端校验 → 返回内容或错误','安全文件读取')">查看代码 ▸

MCP 客户端

客户端通过 ClientSession（mcp/client/session.py）和服务端建立会话。核心流程：connect_to_server() → list_tools() 发现工具 → 将 MCP 工具封装为 LangChain 兼容的 StructuredTool → 供 Agent 调用。注意：async with 上下文管理器确保资源正确释放，Stdio 传输需要管理子进程生命周期。

StructuredTool 封装是客户端的关键抽象。每个 MCP 工具需要创建一个对应的 StructuredTool 实例，设置 name、description 和 coroutine（实际的工具调用函数）。工具调用函数将参数序列化为 JSON，通过 session.call_tool() 发送给服务器，然后解析返回的 TextContent 列表。注意 MCP 返回的 content 是数组（可能包含文本、图片、资源等），LangChain 工具需要将其展平为纯文本。

一个容易被忽视的细节是 异常处理边界。MCP 服务端可能在 JSON-RPC 层面返回 isError: true 的响应（而非抛出异常），而 Python 的异常可能来自子进程崩溃、超时或参数校验失败。客户端需要区分这些情况：JSON-RPC 错误应传递给 LLM 让它重试，子进程崩溃应触发客户端重连，超时应返回用户友好的消息。生产实现中，建议为每个工具调用添加 asyncio timeout 包装。

查看代码 ▸ 查看代码 ▸

LangGraph 集成

将 MCP 客户端发现的工具传入 LangGraph Agent（ReAct 循环模式）。Agent 状态图：agent 节点（LLM 推理 + 工具选择）→ tools 节点（执行工具调用）→ 循环直到 Agent 决定输出。工具调用经过 MCP 客户端包装层，自动处理 JSON-RPC 序列化和响应解析。

LangGraph 的 StateGraph 是核心抽象。我们的 AgentState 只是一个消息列表，但 LangGraph 会自动追踪每次状态变更。关键在于 条件边（should_continue）：如果 LLM 返回了 tool_calls，就走 "tools" 节点；否则走到 END。这种"LLM 决定何时停止"的循环是 ReAct 模式的精髓。工具执行节点遍历所有 tool_calls，每个调用通过 MCP 客户端转发到服务端。

bind_tools 是 LangChain 提供的重要机制。当你调用 model.bind_tools(tools) 时，LangChain 会将工具的 JSON Schema 描述注入到 LLM 的 system prompt 中，让模型知道有哪些工具可用、每个工具的入参是什么、应该何时调用。工具的模式来源于 MCP 服务端的 inputSchema，通过 StructuredTool 的 args_schema 属性自动映射。这意味着你只需在 MCP 服务端定义一次工具，LangGraph Agent 就能自动理解如何使用它。

AgentState:\n response = model.invoke(state["messages"])\n return {"messages": [response]}\n\n# 工具执行节点\ndef call_tool(state: AgentState) -> AgentState:\n last_msg = state["messages"][-1]\n for tool_call in last_msg.tool_calls:\n tool = tool_map[tool_call["name"]]\n result = tool.invoke(tool_call["args"])\n state["messages"].append(ToolMessage(result, tool_call_id=tool_call["id"]))\n return state\n\n# 条件边\ndef should_continue(state: AgentState) -> Literal["tools", END]:\n if state["messages"][-1].tool_calls:\n return "tools"\n return END\n\n# 构建图\ngraph = StateGraph(AgentState)\ngraph.add_node("agent", call_model)\ngraph.add_node("tools", call_tool)\ngraph.add_edge("tools", "agent")\ngraph.add_conditional_edges("agent", should_continue)\ngraph.set_entry_point("agent")','LangGraph')">查看代码 ▸ str:\n ...\n\n# 2. MCP 自动生成 JSON Schema (给 LLM 看的)\n# {\n# "name": "web_search",\n# "description": "搜索网络信息",\n# "inputSchema": {\n# "type": "object",\n# "properties": {\n# "query": {"type": "string", "description": "搜索关键词"},\n# "num_results": {"type": "number", "default": 5}\n# },\n# "required": ["query"]\n# }\n# }\n\n# 3. LangChain bind_tools 将 Schema 注入 LLM Prompt\n# LLM 知道可以调用 web_search(query="...", num_results=10)\n# → 生成 tool_call → 经过 MCP 客户端 → 服务端执行','bind_tools 机制')">查看代码 ▸

测试与调试

常见踩坑点：1）stdin 缓冲区——子进程的 stdout 默认行缓冲，需设置环境变量；2）异步事件循环冲突——嵌套事件循环导致死锁，使用 nest_asyncio；3）参数类型不匹配——MCP 使用 JSON Schema，Python 类型需要精确对应；4）SSE 重连超时——设置合理的心跳间隔和重试策略。

让我们逐一深挖这些坑：stdin 缓冲区是最常见的"潜伏性 bug"。MCP 使用 \n 作为消息分隔符，但 Python 子进程的 stdout 默认启用缓冲区。当工具调用返回大量数据时，缓冲区可能不会立即刷新，导致客户端读到不完整的 JSON。解决方案是设置环境变量 PYTHONUNBUFFERED=1 或在启动参数中添加 -u 标志。异步事件循环冲突在 Jupyter Notebook 和某些测试框架中尤为常见——解决方案是 nest_asyncio.apply()。

参数类型不匹配则是跨语言调用中的高频痛点。MCP 服务端用 Python 定义参数类型（str、int、float、bool），而 LLM 生成 JSON 时可能给出错误的类型（如数字 5 写成字符串 "5"）。MCP SDK 有内置的 Schema 校验，但如果 Python 的类型注解和 JSON Schema 映射不精确，会导致奇怪的行为。建议：工具函数使用显式的类型注解，避免 Any 和 Optional 不加 default 值。使用 MCP Inspector（npx @modelcontextprotocol/inspector）可以图形化地调试工具调用。

查看代码 ▸ str:\n ...\n\n# ❌ 避免下面这种\n@server.tool("bad_tool", description="有问题")\nasync def bad_tool(\n options: Any = None, # ❌ Any 类型无法生成 JSON Schema\n data: Optional[str], # ❌ 无默认值的 Optional 会导致校验失败\n **kwargs # ❌ **kwargs 完全不可序列化\n) -> dict: # ❌ dict 返回类型不够精确\n ...\n\n# 正确做法\n@server.tool("calculation", description="计算")\nasync def calculation(a: float, b: float, op: str = "+") -> float:\n if op == "+": return a + b\n elif op == "-": return a - b\n elif op == "*": return a * b\n elif op == "/": return a / b if b != 0 else float("inf")\n else: raise ValueError(f"Unknown operator: {op}")','参数类型陷阱')">查看代码 ▸

部署与最佳实践

生产部署清单：Docker 容器化——将 MCP Server 打包为 Docker 镜像；健康检查——暴露 /health 端点；日志聚合——JSON 格式日志输出到 stdout；多服务编排——使用 Supervisor/Docker Compose 管理多个 MCP 服务。建议使用 Streamable HTTP 传输模式以获得更好的网络兼容性。

具体到 Docker 部署，推荐使用 多阶段构建：第一阶段安装系统依赖和 Python pip 包，第二阶段只复制 wheels，最终镜像仅包含运行时所需的最小文件。Dockerfile 的关键配置点：设置 PYTHONUNBUFFERED=1 避免缓冲区问题，使用 exec 形式而非 shell 形式启动 CMD，确保信号能正确传递到 Python 进程。docker-compose 中还可以配置 depends_on 和 healthcheck 确保服务依赖顺序正确。

除了容器化，生产环境还需要考虑：1）速率限制——防止 LLM 循环调用消耗过多资源，每个工具应有独立调用配额；2）审计日志——记录每个工具调用的用户、时间、参数和结果，便于问题排查和安全审计；3）配置热更新——允许在不重启服务端的情况下更新工具注册或变更配置。对于多 MCP 服务器场景，建议使用 API Gateway 统一路由或者像 OpenCode 这样的 统一 MCP 管理层来聚合多个服务端。

查看代码 ▸ 查看代码 ▸