MCP Python SDK
模型上下文协议 (MCP) 的 Python 实现
目录
概述
模型上下文协议 (MCP) 允许应用程序以标准化的方式为 LLM 提供上下文，从而将提供上下文的关注点与实际的 LLM 交互分离开来。此 Python SDK 实现了完整的 MCP 规范，从而可以轻松实现以下功能：
- 构建可连接到任何 MCP 服务器的 MCP 客户端
- 创建公开资源、提示和工具的 MCP 服务器
- 使用标准传输，例如 stdio 和 SSE
- 处理所有 MCP 协议消息和生命周期事件
安装
我们建议使用uv来管理您的 Python 项目：
或者：
快速入门
让我们创建一个简单的 MCP 服务器，它公开一个计算器工具和一些数据：
# server.py
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
您可以在Claude Desktop中安装此服务器并通过运行以下命令立即与其交互：
或者，您可以使用 MCP 检查器进行测试：
什么是 MCP？
模型上下文协议 (MCP)允许您构建服务器，以安全、标准化的方式向 LLM 应用程序公开数据和功能。您可以将其视为一个 Web API，但专为 LLM 交互而设计。MCP 服务器可以：
- 通过资源公开数据（可以将其视为 GET 端点；它们用于将信息加载到 LLM 的上下文中）
- 通过工具提供功能（有点像 POST 端点；它们用于执行代码或产生副作用）
- 通过提示（LLM 交互的可重复使用模板）定义交互模式
- 还有更多！
核心概念
服务器
FastMCP 服务器是 MCP 协议的核心接口。它负责连接管理、协议合规性和消息路由：
from mcp.server.fastmcp import FastMCP
# Create a named server
mcp = FastMCP("My App")
# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
资源
资源是你向 LLM 公开数据的方式。它们类似于 REST API 中的 GET 端点——它们提供数据，但不应执行大量计算或产生副作用：
@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
"""Dynamic user data"""
return f"Profile data for user {user_id}"
工具
工具允许 LLM 通过你的服务器执行操作。与资源不同，工具需要执行计算并产生副作用：
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
"""Calculate BMI given weight in kg and height in meters"""
return weight_kg / (height_m ** 2)
@mcp.tool()
async def fetch_weather(city: str) -> str:
"""Fetch current weather for a city"""
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/{city}")
return response.text
提示
提示是可重复使用的模板，可帮助 LLM 有效地与您的服务器交互：
@mcp.prompt()
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
@mcp.prompt()
def debug_error(error: str) -> list[Message]:
return [
UserMessage("I'm seeing this error:"),
UserMessage(error),
AssistantMessage("I'll help debug that. What have you tried so far?")
]
图片
FastMCP 提供了一个Image类来自动处理图像数据：
from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage
@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
"""Create a thumbnail from an image"""
img = PILImage.open(image_path)
img.thumbnail((100, 100))
return Image(data=img.tobytes(), format="png")
语境
Context 对象使您的工具和资源能够访问 MCP 功能：
from mcp.server.fastmcp import FastMCP, Context
@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
"""Process multiple files with progress tracking"""
for i, file in enumerate(files):
ctx.info(f"Processing {file}")
await ctx.report_progress(i, len(files))
data = await ctx.read_resource(f"file://{file}")
return "Processing complete"
运行您的服务器
开发模式
测试和调试服务器的最快方法是使用 MCP Inspector：
mcp dev server.py
# Add dependencies
mcp dev server.py --with pandas --with numpy
# Mount local code
mcp dev server.py --with-editable .
Claude 桌面集成
服务器准备就绪后，请在 Claude Desktop 中安装它：
mcp install server.py
# Custom name
mcp install server.py --name "My Analytics Server"
# Environment variables
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env
直接执行
对于自定义部署等高级场景：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
if __name__ == "__main__":
mcp.run()
使用以下命令运行它：
python server.py
# or
mcp run server.py
示例
回显服务器
一个简单的服务器，演示资源、工具和提示：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Echo")
@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""Echo a message as a resource"""
return f"Resource echo: {message}"
@mcp.tool()
def echo_tool(message: str) -> str:
"""Echo a message as a tool"""
return f"Tool echo: {message}"
@mcp.prompt()
def echo_prompt(message: str) -> str:
"""Create an echo prompt"""
return f"Please process this message: {message}"
SQLite 资源管理器
显示数据库集成的更复杂示例：
from mcp.server.fastmcp import FastMCP
import sqlite3
mcp = FastMCP("SQLite Explorer")
@mcp.resource("schema://main")
def get_schema() -> str:
"""Provide the database schema as a resource"""
conn = sqlite3.connect("database.db")
schema = conn.execute(
"SELECT sql FROM sqlite_master WHERE type='table'"
).fetchall()
return "\n".join(sql[0] for sql in schema if sql[0])
@mcp.tool()
def query_data(sql: str) -> str:
"""Execute SQL queries safely"""
conn = sqlite3.connect("database.db")
try:
result = conn.execute(sql).fetchall()
return "\n".join(str(row) for row in result)
except Exception as e:
return f"Error: {str(e)}"
高级用法
低级服务器
为了获得更多控制权，您可以直接使用底层服务器实现。这赋予您完全的协议访问权限，并允许您自定义服务器的各个方面：
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
# Create a server instance
server = Server("example-server")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
return [
types.Prompt(
name="example-prompt",
description="An example prompt template",
arguments=[
types.PromptArgument(
name="arg1",
description="Example argument",
required=True
)
]
)
]
@server.get_prompt()
async def handle_get_prompt(
name: str,
arguments: dict[str, Any] | None
) -> types.GetPromptResult:
if name != "example-prompt":
raise ValueError(f"Unknown prompt: {name}")
return types.GetPromptResult(
description="Example prompt",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text="Example prompt text"
)
)
]
)
async def run():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="example",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
)
)
)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
编写 MCP 客户端
SDK 提供了用于连接 MCP 服务器的高级客户端接口：
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python", # Executable
args=["example_server.py"], # Optional command line arguments
env=None # Optional environment variables
)
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
# Get a prompt
prompt = await session.get_prompt("example-prompt", arguments={"arg1": "value"})
# List available resources
resources = await session.list_resources()
# List available tools
tools = await session.list_tools()
# Read a resource
resource = await session.read_resource("file://some/path")
# Call a tool
result = await session.call_tool("tool-name", arguments={"arg1": "value"})
if __name__ == "__main__":
import asyncio
asyncio.run(run())
MCP 原语
MCP 协议定义了服务器可以实现的三个核心原语：
| 原始 | 控制 | 描述 | 使用示例 |
|---|
| 提示 | 用户控制 | 由用户选择调用的交互式模板 | 斜线命令、菜单选项 |
| 资源 | 应用程序控制 | 客户端应用程序管理的上下文数据 | 文件内容、API 响应 |
| 工具 | 模型控制 | 向 LLM 公开的函数用于执行操作 | API 调用、数据更新 |
服务器功能
MCP 服务器在初始化期间声明功能：
| 能力 | 功能标志 | 描述 |
|---|
prompts | listChanged | 提示模板管理 |
resources | subscribe listChanged | 资源公开和更新 |
tools | listChanged | 工具发现和执行 |
logging | - | 服务器日志配置 |
completion | - | 论证完成建议 |
文档
贡献
我们热衷于支持各种经验水平的贡献者，并期待您参与到项目中来。请参阅贡献指南，开始您的贡献之旅。
执照
该项目根据 MIT 许可证获得许可 - 有关详细信息，请参阅 LICENSE 文件。