从零构建MCP-Server实战

用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可以暴露三种能力：

1. Tools（工具）：AI可以调用的函数（如查询数据库、发送邮件）
2. Resources（资源）：AI可以读取的数据（如文件内容、API数据）
3. Prompts（提示词）：预定义的提示词模板

1.3 传输层对比

二、从零实现stdio MCP Server

2.1 环境准备

# 安装MCP Python SDKpipinstallmcp# 或者用uv（推荐）uvaddmcp

2.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