用Python从零构建MCP Server实战：让AI Agent连接万物
MCP（Model Context Protocol）正在成为AI Agent连接外部世界的标准协议。本文手把手教你用Python从零构建一个生产级MCP Server，让你的Agent能调用任何自定义工具。
前言
2024年底，Anthropic发布了MCP（Model Context Protocol）协议，目标是统一AI模型与外部工具、数据源之间的通信标准。简单来说，MCP就是AI世界的"USB接口"——任何工具只要实现了MCP协议，就能被任何支持MCP的AI模型使用。
我在第一次接触MCP时，觉得官方文档过于理论化，缺少实操指导。于是我决定自己从零实现一个MCP Server，过程中踩了不少坑：stdio通信的缓冲区问题、Schema不兼容、流式响应处理等等。
这篇文章会带你从协议原理开始，一步步实现一个完整的MCP Server，并与Claude和LangChain集成测试。
一、MCP协议核心概念
1.1 架构总览
MCP采用Client-Server架构：
AI应用（Host） ↓ MCP Client（内置在Host中） ↓ ← 传输层（stdio / HTTP SSE） MCP Server（你的代码） ↓ 外部资源（数据库、API、文件系统...）核心概念：
- Host：AI应用（如Claude Desktop、Cursor、你的Agent程序）
- Client：MCP协议客户端，由Host管理，负责与Server通信
- Server：你实现的服务，暴露工具（Tools）、资源（Resources）、提示词（Prompts）
- Transport：通信方式，分stdio（标准输入输出）和HTTP SSE两种
1.2 三大能力
MCP Server可以暴露三种能力：
- Tools（工具）：AI可以调用的函数（如查询数据库、发送邮件）
- Resources（资源）：AI可以读取的数据（如文件内容、API数据）
- Prompts（提示词）：预定义的提示词模板
1.3 传输层对比
| 特性 | stdio | HTTP SSE |
|---|---|---|
| 通信方式 | 标准输入输出 | HTTP长连接 + Server-Sent Events |
| 适用场景 | 本地进程 | 远程服务 |
| 延迟 | 极低 | 较低 |
| 部署方式 | 本地安装 | Web服务 |
| 典型用户 | Claude Desktop、Cursor | 自定义Agent、Web应用 |
二、从零实现stdio MCP Server
2.1 环境准备
# 安装MCP Python SDKpipinstallmcp# 或者用uv（推荐）uvaddmcp2.2 最小MCP Server
# server.py - 最简单的MCP Serverfrommcp.serverimportServerfrommcp.server.stdioimportstdio_serverimportasyncio# 创建Server实例server=Server("my-first-mcp")@server.tool()defadd(a:int,b:int)->int:"""两数相加"""returna+b@server.tool()defgreet(name:str)->str:"""打招呼"""returnf"你好，{name}！我是MCP Server。"asyncdefmain():asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream,server.create_initialization_options())if__name__=="__main__":asyncio.run(main)2.3 Claude Desktop配置
在Claude Desktop的配置文件中注册你的Server：
{"mcpServers":{"my-first-mcp":{"command":"python","args":["/path/to/server.py"]}}}配置文件位置：
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json
三、工具注册与参数验证
3.1 基础工具注册
frommcp.serverimportServerfrompydanticimportBaseModel,FieldfromtypingimportOptional server=Server("tool-server")# 方式1：简单函数（自动从类型注解生成Schema）@server.tool()defcalculate(expression:str)->str:"""计算数学表达式"""try:result=eval(expression)# 生产环境不要用eval！returnstr(result)exceptExceptionase:returnf"计算错误:{e}"# 方式2：使用Pydantic模型（更精确的Schema控制）classSearchInput(BaseModel):query:str=Field(description="搜索关键词")max_results:int=Field(default=10,description="最大结果数",ge=1,le=100)category:Optional[str]=Field(default=None,description="分类筛选")@server.tool()defsearch_products(input:SearchInput)->str:"""搜索商品"""results=[{"name":f"商品{i}","price":i*10}foriinrange(input.max_results)]returnjson.dumps(results,ensure_ascii=False)3.2 复杂参数处理
classDatabaseQueryInput(BaseModel):table:str=Field(description="表名")columns:list[str]=Field(description="查询的列名列表")where:Optional[dict]=Field(default=None,description="查询条件，如 {'age': {'$gt': 18}}")order_by:Optional[str]=Field(default=None,description="排序字段")limit:int=Field(default=100,ge=1,le=10000)@server.tool()defquery_database(input:DatabaseQueryInput)->str:"""查询数据库"""# 构建SQL（简化示例，生产环境用ORM或参数化查询）cols=", ".join(input.columns)sql=f"SELECT{cols}FROM{input.table}"ifinput.where:conditions=[]forkey,valueininput.where.items():ifisinstance(value,dict
