[!NOTE]
本文档适用于 MCP Python SDK v1.x（当前稳定版本）。v1.x 的代码和文档，请查看
v1.x分支。
即将发布的 v2 文档（预 alpha 版，正在main分支开发），请查看README.v2.md。
Model Context Protocol（模型上下文协议）允许应用程序以标准化的方式为 LLM 提供上下文，将提供上下文与实际 LLM 交互的关注点分离开来。这个 Python SDK 实现了完整的 MCP 规范，可以轻松地：
我们推荐使用 uv 来管理你的 Python 项目。
如果你还没有创建 uv 管理的项目，请先创建一个：
bash
uv init mcp-server-demo
cd mcp-server-demo
然后将 MCP 添加到你的项目依赖中：
bash
uv add "mcp[cli]"
或者，对于使用 pip 管理依赖的项目：
pip install "mcp[cli]"
使用 uv 运行 mcp 命令：
uv run mcp
让我们创建一个简单的 MCP 服务器，暴露一个计算器工具和一些数据：
"""
FastMCP 快速开始示例。
从仓库根目录运行：
uv run examples/snippets/servers/fastmcp_quickstart.py
"""
from mcp.server.fastmcp import FastMCP
# 创建一个 MCP 服务器
mcp = FastMCP("Demo", json_response=True)
# 添加一个加法工具
@mcp.tool()
def add(a: int, b: int) -> int:
"""将两个数字相加"""
return a + b
# 添加一个动态问候资源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""获取个性化问候语"""
return f"Hello, {name}!"
# 添加一个提示词
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
"""生成问候提示词"""
styles = {
"friendly": "请写一个温暖、友好的问候语",
"formal": "请写一个正式、专业的问候语",
"casual": "请写一个随意、轻松的问候语",
}
return f"{styles.get(style, styles['friendly'])} 给一个名叫 {name} 的人。"
# 使用可流式 HTTP 传输运行
if __name__ == "__main__":
mcp.run(transport="streamable-http")
完整示例：examples/snippets/servers/fastmcp_quickstart.py
你可以将此服务器安装到 Claude Code 并立即与之交互。首先，运行服务器：
uv run --with mcp examples/snippets/servers/fastmcp_quickstart.py
然后将其添加到 Claude Code：
claude mcp add --transport http my-server http://localhost:8000/mcp
或者，你也可以使用 MCP Inspector 进行测试。如上所述启动服务器，然后在另一个终端中：
npx -y @modelcontextprotocol/inspector
在 inspector UI 中，连接到 http://localhost:8000/mcp。
Model Context Protocol (MCP) 让你可以构建服务器，以安全、标准化的方式向 LLM 应用程序暴露数据和功能。可以把它想象成一个 Web API，但专门为 LLM 交互设计。MCP 服务器可以：
FastMCP 服务器是你与 MCP 协议交互的核心接口。它处理连接管理、协议合规性和消息路由：
"""展示具有强类型支持的启动/关闭生命周期示例。"""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
# 示例用的模拟数据库类
class Database:
"""示例用的模拟数据库类。"""
@classmethod
async def connect(cls) -> "Database":
"""连接到数据库。"""
return cls()
async def disconnect(self) -> None:
"""断开数据库连接。"""
pass
def query(self) -> str:
"""执行查询。"""
return "查询结果"
@dataclass
class AppContext:
"""具有类型化依赖项的应用程序上下文。"""
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""使用类型安全的上下文管理应用程序生命周期。"""
# 启动时初始化
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
# 关闭时清理
await db.disconnect()
# 将生命周期传递给服务器
mcp = FastMCP("My App", lifespan=app_lifespan)
# 在工具中访问类型安全的生命周期上下文
@mcp.tool()
def query_db(ctx: Context[ServerSession, AppContext]) -> str:
"""使用已初始化资源的工具。"""
db = ctx.request_context.lifespan_context.db
return db.query()
完整示例：examples/snippets/servers/lifespan_example.py
资源是你向 LLM 暴露数据的方式。它们类似于 REST API 中的 GET 端点——它们提供数据，但不应该执行大量计算或产生副作用：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="Resource Example")
@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
"""按名称读取文档。"""
# 通常这里会从磁盘读取
return f"Content of {name}"
@mcp.resource("config://settings")
def get_settings() -> str:
"""获取应用程序设置。"""
return """{
"theme": "dark",
"language": "en",
"debug": false
}"""
完整示例：examples/snippets/servers/basic_resource.py
工具让 LLM 可以通过你的服务器执行操作。与资源不同，工具被期望执行计算并产生副作用：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="Tool Example")
@mcp.tool()
def sum(a: int, b: int) -> int:
"""将两个数字相加。"""
return a + b
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
"""获取城市天气。"""
# 通常这里会调用天气 API
return f"Weather in {city}: 22degrees{unit[0].upper()}"
完整示例：examples/snippets/servers/basic_tool.py
工具可以选择性地接收一个 Context 对象，只需包含一个带有 Context 类型注解的参数即可。这个上下文由 FastMCP 框架自动注入，并提供对 MCP 功能的访问：
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
mcp = FastMCP(name="Progress Example")
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
"""执行带有进度更新的任务。"""
await ctx.info(f"Starting: {task_name}")
for i in range(steps):
progress = (i + 1) / steps
await ctx.report_progress(
progress=progress,
total=1.0,
message=f"Step {i + 1}/{steps}",
)
await ctx.debug(f"Completed step {i + 1}")
return f"Task '{task_name}' completed"
完整示例：examples/snippets/servers/tool_progress.py
如果工具的返回类型注解兼容，默认将返回结构化结果。否则，将返回非结构化结果。
结构化输出支持以下返回类型：
dict[str, T]（其中 T 是任何 JSON 可序列化类型）{"result": value} 中{"result": value} 中没有类型提示的类不能用于结构化输出。只有具有正确注解属性的类才会被转换为 Pydantic 模型以进行模式生成和验证。
结构化结果会根据从注解生成的输出模式自动验证。这确保了工具返回类型良好、经过验证的数据，客户端可以轻松处理。
注意： 为了向后兼容，也返回非结构化结果。非结构化结果是为了与 MCP 规范的先前版本向后兼容而提供的，并且在当前版本的 SDK 中与 FastMCP 的先前版本具有兼容性。
注意： 如果工具函数的返回类型注解导致工具被分类为结构化输出且这是不希望的，可以通过向 @tool 装饰器传递 structured_output=False 来抑制此分类。
为了完全控制工具响应，包括 _meta 字段（用于向客户端应用程序传递数据而不暴露给模型），你可以直接返回 CallToolResult：
"""展示直接返回 CallToolResult 以进行高级控制的示例。"""
from typing import Annotated
from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP
from mcp.types import CallToolResult, TextContent
mcp = FastMCP("CallToolResult Example")
class ValidationModel(BaseModel):
"""用于验证结构化输出的模型。"""
status: str
data: dict[str, int]
@mcp.tool()
def advanced_tool() -> CallToolResult:
"""直接返回 CallToolResult 以完全控制，包括 _meta 字段。"""
return CallToolResult(
content=[TextContent(type="text", text="模型可见的响应")],
_meta={"hidden": "仅限客户端应用程序的数据"},
)
@mcp.tool()
def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
"""返回经过结构化输出验证的 CallToolResult。"""
return CallToolResult(
content=[TextContent(type="text", text="已验证的响应")],
structuredContent={"status": "success", "data": {"result": 42}},
_meta={"internal": "元数据"},
)
@mcp.tool()
def empty_result_tool() -> CallToolResult:
"""对于空结果，返回内容为空的 CallToolResult。"""
return CallToolResult(content=[])
完整示例：examples/snippets/servers/direct_call_tool_result.py
重要： 必须始终返回 CallToolResult（不能是 Optional 或 Union）。对于空结果，使用 CallToolResult(content=[])。对于可选简单类型，使用 str | None 而不使用 CallToolResult。
```python
"""展示工具的结构化输出示例。"""
from typing import TypedDict
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Structured Output Example")
class WeatherData(BaseModel):
"""天气信息结构。"""
temperature: float = Field(description="摄氏温度")
humidity: float = Field(description="湿度百分比")
condition: str
wind_speed: float
@mcp.tool()
def get_weather