Model Context Protocol (MCP) Python SDK 权威指南

1. 简介与核心概念

Model Context Protocol (MCP) 是一个开放标准，旨在标准化 AI 模型（如 Claude, GPT）与外部数据源（IDE, 数据库, 生产工具）之间的交互。

MCP Python SDK 是该标准的官方 Python 实现，它屏蔽了底层的 JSON-RPC 通信细节，让开发者能够专注于业务逻辑。

github: https://github.com/modelcontextprotocol/python-sdk

核心架构图解

Host (Claude Desktop/IDE) ↔ MCP Client ↔ Transport (Stdio/SSE) ↔ MCP Server ↔ Data Source \text{Host (Claude Desktop/IDE)} \leftrightarrow \text{MCP Client} \leftrightarrow \text{Transport (Stdio/SSE)} \leftrightarrow \text{MCP Server} \leftrightarrow \text{Data Source} Host (Claude Desktop/IDE)↔MCP Client↔Transport (Stdio/SSE)↔MCP Server↔Data Source

• Server (服务端): 提供工具（Tools）、资源（Resources）和提示词（Prompts）。
• Client (客户端): 也就是 Host，负责连接 Server 并消费这些能力。
• Protocol: 基于 JSON-RPC 2.0。

2. 环境准备

前置要求：Python 3.10+

# 安装核心 SDK pip install mcp # 建议安装 uv (现代 Python 包管理器) 以获得更好的体验 pip install uv 

3. 入门篇：FastMCP 极速开发

对于 90% 的场景，FastMCP 是最高效的选择。它类似于 FastAPI，通过装饰器模式快速构建服务。

3.1 Hello World：构建一个数学工具服务器

创建一个名为 math_server.py 的文件：

from mcp.server.fastmcp import FastMCP # 1. 初始化服务，指定服务名称 mcp = FastMCP("MyMathServer")# 2. 定义工具 (Tools)# 工具是模型可以调用的函数，具有副作用（执行操作）或计算能力@mcp.tool()defadd(a:int, b:int)->int:"""计算两个整数的和"""return a + b @mcp.tool()defcalculate_hypotenuse(a:float, b:float)->float:"""计算直角三角形的斜边长"""# 公式：$c = \sqrt{a^2 + b^2}$return(a**2+ b**2)**0.5# 3. 运行服务if __name__ =="__main__": mcp.run()

3.2 运行与测试

MCP 服务通常通过标准输入输出（Stdio）运行。要在 Claude Desktop 中使用此服务：

1. 打开 Claude Desktop 配置文件 (~/Library/Application Support/Claude/claude_desktop_config.json 或 Windows 对应路径)。
2. 重启 Claude Desktop，你现在可以直接在这个 AI 聊天框中让它计算斜边长了。

添加配置：

{"mcpServers":{"my-math":{"command":"python","args":["/绝对路径/到/math_server.py"]}}}

4. 进阶篇：核心原语详解

除了工具 (Tools)，MCP 还定义了 资源 (Resources) 和 提示词 (Prompts)。

4.1 资源 (Resources)

资源类似于 API 中的 GET 请求，用于向 LLM 暴露只读数据。它们通过 URI 寻址。

假设我们需要让 AI 读取系统的日志文件：

from mcp.server.fastmcp import FastMCP, Context mcp = FastMCP("LogServer")# 定义动态资源# 这里的 pattern "logs://{system_name}/current" 就像 API 路由@mcp.resource("logs://{system_name}/current")defget_system_logs(system_name:str)->str:"""获取指定系统的最新日志"""# 在实际应用中，这里会读取文件或数据库# 假设日志数据生成函数 $f(x)$returnf"Log entry for {system_name}: System status implies stability > 99%."if __name__ =="__main__": mcp.run()

4.2 提示词 (Prompts)

提示词是预定义的上下文模板，用于引导用户与 AI 的对话。

@mcp.prompt()defreview_code(code:str)->str:"""创建一个代码审查的 Prompt 模板"""returnf""" 请作为一名资深架构师审查以下代码。 重点关注：安全性、性能 ($O(n)$ 复杂度) 和可读性。 代码内容： {code} """

5. 高级篇：底层架构与生命周期

对于需要精细控制（如自定义生命周期、鉴权、复杂的异步流）的场景，不能只用 FastMCP，需要直接操作底层的 Server 类。

5.1 手动构建 Server 与 Session

这里展示如何不使用 FastMCP，而是使用底层的 Server 和 StdioServerTransport。

import asyncio from mcp.server import Server, NotificationOptions from mcp.server.stdio import StdioServerTransport from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource # 1. 实例化低级 Server 对象 app = Server("LowLevelApp")# 2. 注册工具处理器@app.list_tools()asyncdeflist_tools():return[ Tool( name="echo", description="Echo back input", inputSchema={"type":"object","properties":{"message":{"type":"string"}},"required":["message"]})]@app.call_tool()asyncdefcall_tool(name:str, arguments:dict):if name =="echo": msg = arguments["message"]return[TextContent(type="text", text=f"Echo: {msg}")]raise ValueError(f"Unknown tool: {name}")# 3. 显式的主循环控制asyncdefmain():# 使用标准输入输出传输层asyncwith StdioServerTransport()as transport:# 将 Server 连接到 Transportawait app.run( transport.read_messages, transport.write_message,# 初始化时的 capabilities 设置 initialization_options=None)if __name__ =="__main__": asyncio.run(main())

5.2 异步与并发

MCP SDK 完全基于 Python 的 asyncio。这意味着你可以：

• 在 Tool 内部并发调用外部 API。
• 使用 Context 对象向客户端发送进度通知或日志。

Total Latency ≈ max ⁡ ( T a p i 1 , T a p i 2 ) ( if run concurrently ) \text{Total Latency} \approx \max(T_{api1}, T_{api2}) \quad (\text{if run concurrently}) Total Latency≈max(Tapi1​,Tapi2​)(if run concurrently)

@mcp.tool()asyncdeffetch_concurrent_data(ctx: Context):# 向客户端发送日志，不阻塞主流程await ctx.info("Starting concurrent fetch...")# 假设这里有两个耗时的 IO 操作 res1, res2 =await asyncio.gather(task1(), task2())returnf"Result: {res1}, {res2}"

6. 客户端篇：构建 MCP Client

MCP 是双向的。除了构建 Server 给 Claude 用，你也可以构建 Client 来调用别人的 Server（例如连接到本地的 Postgres MCP Server）。

import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client asyncdefrun_client():# 定义要连接的 Server 参数 server_params = StdioServerParameters( command="python", args=["my_server.py"],# 指向你的 Server 脚本)# 建立连接asyncwith stdio_client(server_params)as(read, write):asyncwith ClientSession(read, write)as session:# 1. 初始化await session.initialize()# 2. 列出可用工具 tools =await session.list_tools()print(f"Available tools: {[t.name for t in tools.tools]}")# 3. 调用工具 result =await session.call_tool("add", arguments={"a":10,"b":20})print(f"Calculation Result: {result.content[0].text}")if __name__ =="__main__": asyncio.run(run_client())

7. 实战案例：SQLite 数据库管理器

我们将结合所有知识，构建一个具备资源读取和工具执行能力的 SQLite MCP Server。

功能设计

1. Resource: sqlite://schema 查看所有表结构。
2. Tool: query_db 执行只读 SQL 查询。

代码实现

import sqlite3 from mcp.server.fastmcp import FastMCP from pydantic import BaseModel, Field # 初始化数据库（演示用） DB_PATH ="demo.db"definit_db(): conn = sqlite3.connect(DB_PATH) conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, role TEXT)") conn.execute("INSERT OR IGNORE INTO users (id, name, role) VALUES (1, 'Alice', 'Admin')") conn.commit() conn.close() init_db() mcp = FastMCP("SQLiteManager")# --- Resource: 暴露数据库 Schema [email protected]("sqlite://schema")defget_schema()->str:"""获取数据库当前的 Schema 信息""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT sql FROM sqlite_master WHERE type='table'") tables = cursor.fetchall() conn.close() schema_str ="\n".join([t[0]for t in tables if t[0]])returnf"Database Schema:\n{schema_str}"# --- Tool: 执行 SQL 查询 ---# 使用 Pydantic 定义更加严谨的输入结构classQueryParams(BaseModel): query:str= Field(description="SELECT SQL query to execute")@mcp.tool()defquery_database(params: QueryParams)->str:"""执行只读 SQL 查询 (SELECT only)""" query = params.query.strip()# 简单的安全检查 (实际生产需更严格的校验)ifnot query.upper().startswith("SELECT"):raise ValueError("Only SELECT queries are allowed for safety.") conn = sqlite3.connect(DB_PATH)try: cursor = conn.cursor() cursor.execute(query) results = cursor.fetchall()# 将结果转换为易读的字符串格式returnstr(results)except Exception as e:returnf"Error executing query: {str(e)}"finally: conn.close()if __name__ =="__main__": mcp.run()

8. 最佳实践与注意事项

1. 安全性 (Security):
   • MCP Server 本质上是在允许 LLM 在你的机器上执行代码。务必对 Tool 的输入进行严格校验（如上面的 SQL 检查）。
   • 不要在 Tool 中硬编码敏感密钥，使用环境变量。
2. 错误处理 (Error Handling):
   • 不要让 Server 崩溃。在 Tool 内部捕获异常并返回人类可读的错误信息字符串，这样 LLM 可以尝试自我修正参数并重试。
   • 可以使用 mcp-inspector（需要 npm install -g @modelcontextprotocol/inspector）来可视化调试你的 Python Server。
3. 性能 (Performance):
   • 对于重型计算任务，请务必使用 async def 并避免在主线程中执行阻塞调用。假设任务耗时为 t t t，若阻塞主线程，吞吐量将降为 1 / t 1/t 1/t。

调试 (Debugging):

npx @modelcontextprotocol/inspector python your_server.py 

通过本文档，你应该已经掌握了从最简单的函数包装到构建完整的、基于资源的 MCP 服务的全部技能。