PythonAI
MCP Python SDK 核心概念与实战指南
Model Context Protocol (MCP) 是标准化 AI 模型与外部数据源交互的开放标准。介绍 MCP Python SDK 的核心概念、环境搭建及开发实践。涵盖 FastMCP 快速构建工具、资源与提示词定义，以及底层 Server 异步处理机制。包含客户端连接示例与 SQLite 数据库管理实战案例。强调安全性校验、错误处理与性能优化建议，帮助开发者高效集成 LLM 能力。
Model Context Protocol (MCP) 是标准化 AI 模型与外部数据源交互的开放标准。介绍 MCP Python SDK 的核心概念、环境搭建及开发实践。涵盖 FastMCP 快速构建工具、资源与提示词定义，以及底层 Server 异步处理机制。包含客户端连接示例与 SQLite 数据库管理实战案例。强调安全性校验、错误处理与性能优化建议，帮助开发者高效集成 LLM 能力。
Model Context Protocol (MCP) 是一个开放标准，旨在标准化 AI 模型（如 Claude, GPT）与外部数据源（IDE, 数据库，生产工具）之间的交互。
MCP Python SDK 是该标准的官方 Python 实现，它屏蔽了底层的 JSON-RPC 通信细节，让开发者能够专注于业务逻辑。
Host (Claude Desktop/IDE) ↔ MCP Client ↔ Transport (Stdio/SSE) ↔ MCP Server ↔ Data Source
前置要求：Python 3.10+
# 安装核心 SDK
pip install mcp
# 建议安装 uv (现代 Python 包管理器) 以获得更好的体验
pip install uv
对于 90% 的场景，FastMCP 是最高效的选择。它类似于 FastAPI，通过装饰器模式快速构建服务。
创建一个名为 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()
MCP 服务通常通过标准输入输出（Stdio）运行。要在 Claude Desktop 中使用此服务：
~/Library/Application Support/Claude/claude_desktop_config.json 或 Windows 对应路径)。添加配置：
{
"mcpServers": {
"my-math": {
"command": "python",
"args": ["/绝对路径/到/math_server.py"]
}
}
}
除了工具 (Tools)，MCP 还定义了 资源 (Resources) 和 提示词 (Prompts)。
资源类似于 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()
提示词是预定义的上下文模板，用于引导用户与 AI 的对话。
@mcp.prompt()
def review_code(code: str) -> str:
"""创建一个代码审查的 Prompt 模板"""
return f"""
请作为一名资深架构师审查以下代码。
重点关注：安全性、性能 (O(n) 复杂度) 和可读性。
代码内容：{code}
"""
对于需要精细控制（如自定义生命周期、鉴权、复杂的异步流）的场景，不能只用 FastMCP，需要直接操作底层的 Server 类。
这里展示如何不使用 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())
MCP SDK 完全基于 Python 的 asyncio。这意味着你可以：
Context 对象向客户端发送进度通知或日志。并发执行时总延迟约等于各任务最大耗时：Total Latency ≈ max(T_api1, T_api2)。
@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}"
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())
我们将结合所有知识，构建一个具备资源读取和工具执行能力的 SQLite MCP Server。
sqlite://schema 查看所有表结构。query_db 执行只读 SQL 查询。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()
Tool 的输入进行严格校验（如上面的 SQL 检查）。Tool 中硬编码敏感密钥，使用环境变量。mcp-inspector（需要 npm install -g @modelcontextprotocol/inspector）来可视化调试你的 Python Server。async def 并避免在主线程中执行阻塞调用。假设任务耗时为 t，若阻塞主线程，吞吐量将降为 1/t。调试 (Debugging):
npx @modelcontextprotocol/inspector python your_server.py
通过本文档，你应该已经掌握了从最简单的函数包装到构建完整的、基于资源的 MCP 服务的全部技能。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online