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()
def add(a: int, b: int) –> int:
"""计算两个整数的和"""
return a + b

@mcp.tool()
def calculate_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 中使用此服务：

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")
def get_system_logs(system_name: str) –> str:
"""获取指定系统的最新日志"""
# 在实际应用中，这里会读取文件或数据库
# 假设日志数据生成函数 $f(x)$
return f"Log entry for {system_name}: System status implies stability > 99%."

if __name__ == "__main__":
mcp.run()

4.2 提示词 (Prompts)

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

@mcp.prompt()
def review_code(code: str) –> str:
"""创建一个代码审查的 Prompt 模板"""
return f"""
请作为一名资深架构师审查以下代码。
重点关注：安全性、性能 ($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()
async def list_tools():
return [
Tool(
name="echo",
description="Echo back input",
inputSchema={
"type": "object",
"properties": {"message": {"type": "string"}},
"required": ["message"]
}
)
]

@app.call_tool()
async def call_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. 显式的主循环控制
async def main():
# 使用标准输入输出传输层
async with StdioServerTransport() as transport:
# 将 Server 连接到 Transport
await 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()
async def fetch_concurrent_data(ctx: Context):
# 向客户端发送日志，不阻塞主流程
await ctx.info("Starting concurrent fetch…")

# 假设这里有两个耗时的 IO 操作
res1, res2 = await asyncio.gather(task1(), task2())
return f"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

async def run_client():
# 定义要连接的 Server 参数
server_params = StdioServerParameters(
command="python",
args=["my_server.py"], # 指向你的 Server 脚本
)

# 建立连接
async with stdio_client(server_params) as (read, write):
async with 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。

功能设计

代码实现

import sqlite3
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

# 初始化数据库（演示用）
DB_PATH = "demo.db"
def init_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 —
@mcp.resource("sqlite://schema")
def get_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]])
return f"Database Schema:\\n{schema_str}"

# — Tool: 执行 SQL 查询 —
# 使用 Pydantic 定义更加严谨的输入结构
class QueryParams(BaseModel):
query: str = Field(description="SELECT SQL query to execute")

@mcp.tool()
def query_database(params: QueryParams) –> str:
"""执行只读 SQL 查询 (SELECT only)"""
query = params.query.strip()

# 简单的安全检查 (实际生产需更严格的校验)
if not 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()
# 将结果转换为易读的字符串格式
return str(results)
except Exception as e:
return f"Error executing query: {str(e)}"
finally:
conn.close()

if __name__ == "__main__":
mcp.run()

8. 最佳实践与注意事项

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