MCP Python SDK — 用 Python 构建 MCP 服务端与客户端

GitHub: modelcontextprotocol/python-sdk
Stars: 23,300+ | Language: Python | License: MIT
官网: modelcontextprotocol.io

目录

1. 项目速览
2. 功能概述
3. 适用场景
4. 快速上手
5. 源码架构
6. 实操 Demo
7. 同类对比
8. 参考资源

项目速览

MCP Python SDK（mcp 包）是 Model Context Protocol 协议的官方 Python 实现，由 Anthropic 维护。它提供了一套完整的工具链，让 Python 开发者可以轻松构建 MCP 服务端（Server）和客户端（Client），通过标准化的方式为 LLM 提供上下文（context）、工具（tools）、资源（resources）和提示模板（prompts）。

截至 2026 年 6 月，项目在 GitHub 上已获得超过 23,300 颗 Star，累计 883 次提交。最新版本支持 Python 3.10+，可通过 PyPI 直接安装。SDK 的核心设计理念是「低门槛、高上限」：通过 FastMCP 高级 API，初学者用不到 10 行代码就能跑起一个 MCP Server；而通过 MCPServer 低级 API 和丰富的类型系统，高级用户可以精细控制协议层面的每一个细节。

SDK 支持三种传输协议：stdio（标准输入输出，用于本地进程间通信）、SSE（Server-Sent Events，用于 HTTP 流式通信）和 Streamable HTTP（2025 年新增的推荐传输方式）。此外还内置了 OAuth 2.1 资源服务器认证、结构化输出验证、分页、Elicitation（表单/URL 交互式信息收集）、Sampling（服务端调用 LLM）、进度报告等高级特性。

功能概述

FastMCP 高级 API

FastMCP 是 SDK 的推荐入口，通过装饰器模式极大简化了 MCP 服务端的构建过程：

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Demo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    styles = {
        "friendly": "Please write a warm, friendly greeting",
        "formal": "Please write a formal, professional greeting",
        "casual": "Please write a casual, relaxed greeting",
    }
    return f"{styles.get(style, styles['friendly'])} for someone named {name}."

if __name__ == "__main__":
    mcp.run(transport="streamable-http", json_response=True)

FastMCP 提供三类核心装饰器：

• @mcp.tool() — 注册工具函数。支持 icons 参数为工具指定图标，structured_output=True（默认）自动为返回值生成 JSON Schema。
• @mcp.resource(uri_template) — 注册资源。URI 模板支持路径参数（如 "greeting://{name}"），LLM 可像读取文件一样访问资源。
• @mcp.prompt() — 注册提示模板。将参数化的提示逻辑封装为可复用的服务端功能。

FastMCP 运行时可配置项：

• json_response=True — 工具返回值包裹在 {"result": ...} 中
• stateless_http=False — 无状态 HTTP 模式，不维护会话
• lifespan — 通过 async context manager 管理应用生命周期
• token_verifier — OAuth 2.1 Token 验证器
• auth — AuthSettings 配置（issuer_url, resource_server_url, required_scopes）

结构化输出 (Structured Output)

SDK 支持四种方式定义工具的返回类型，LLM 客户端会收到自动生成的 JSON Schema：

1. Pydantic BaseModel：

from pydantic import BaseModel, Field

class WeatherData(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    humidity: float = Field(description="Humidity percentage")
    condition: str
    wind_speed: float

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    return WeatherData(temperature=22.5, humidity=45.0, condition="sunny", wind_speed=5.2)

2. TypedDict：

from typing import TypedDict

class LocationInfo(TypedDict):
    latitude: float
    longitude: float
    name: str

@mcp.tool()
def get_location(address: str) -> LocationInfo:
    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK")

3. 普通类型提示类：

class UserProfile:
    name: str
    age: int
    email: str | None = None

@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    return UserProfile(name="Alice", age=30, email="alice@example.com")

4. 简单类型与容器：

@mcp.tool()
def list_cities() -> list[str]:
    return ["London", "Paris", "Tokyo"]

@mcp.tool()
def get_temperature(city: str) -> float:
    return 22.5

应用生命周期管理 (Lifespan)

通过 async context manager 管理数据库连接、模型加载等资源的启动和清理：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.mcpserver import Context, MCPServer

class Database:
    @classmethod
    async def connect(cls) -> "Database":
        return cls()
    async def disconnect(self) -> None: ...
    def query(self) -> str:
        return "Query result"

@dataclass
class AppContext:
    db: Database

@asynccontextmanager
async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()

mcp = MCPServer("My App", lifespan=app_lifespan)

@mcp.tool()
def query_db(ctx: Context[AppContext]) -> str:
    db = ctx.request_context.lifespan_context.db
    return db.query()

Context 对象

每个工具/资源的处理函数可以通过 ctx: Context 参数访问丰富的上下文信息：

属性/方法	说明
ctx.request_id	当前请求的唯一 ID
ctx.client_id	客户端标识符
ctx.fastmcp	FastMCP 实例引用
ctx.session	ServerSession 对象（可发送通知、调用 LLM）
ctx.request_context	原始 RequestContext（含 lifespan_context）
await ctx.debug(msg)	发送 debug 级别日志
await ctx.info(msg)	发送 info 级别日志
await ctx.warning(msg)	发送 warning 级别日志
await ctx.error(msg)	发送 error 级别日志
await ctx.report_progress(progress, total, message)	报告任务进度
await ctx.read_resource(uri)	读取指定 URI 的资源
await ctx.elicit(message, schema)	弹出表单让用户输入信息

Sampling（服务端调用 LLM）

MCP 协议允许服务端通过 ctx.session.create_message() 反向调用 LLM 完成子任务：

@mcp.tool()
async def summarize_text(text: str, ctx: Context) -> str:
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(type="text", text=f"Summarize: {text}")
            )
        ],
        max_tokens=500
    )
    return result.content.text

Elicitation（交互式信息收集）

SDK 支持两种 elicitation 模式：表单模式（ctx.elicit）和 URL 模式（ctx.elicit_url），用于在工具执行过程中需要用户交互的场景（如授权确认、参数补充）。

OAuth 2.1 认证

通过 TokenVerifier 协议和 AuthSettings 配置实现基于 RFC 9728 的资源服务器认证：

from mcp.server.auth import TokenVerifier, AuthSettings

class MyTokenVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        # 验证 JWT token 并返回 AccessToken
        ...

mcp = MCPServer("Secure App",
    token_verifier=MyTokenVerifier(),
    auth=AuthSettings(
        issuer_url="https://auth.example.com",
        resource_server_url="https://api.example.com",
        required_scopes=["mcp:read", "mcp:write"]
    )
)

传输协议

SDK 支持三种传输协议，每种都有对应的启动方式：

传输方式	启动方法	适用场景
stdio	mcp.run(transport="stdio")	本地进程间通信，Claude Desktop 等客户端
SSE	mcp.sse_app() 返回 ASGI app	HTTP 流式传输，兼容性好
Streamable HTTP	mcp.run(transport="streamable-http")	推荐方式，支持多路由挂载

Streamable HTTP 模式支持挂载到现有 Starlette/FastAPI 应用、CORS 中间件配置、多服务共享端口、基于 Host 的路由分发等高级部署场景。

适用场景

构建 IDE/AI 编辑器插件

MCP Python SDK 是构建 AI 编码助手的首选方案之一。通过 stdio 传输协议，可以将 Python 工具（如代码分析、测试运行、数据库查询）暴露给 Claude Code、Zed、Continue 等 MCP 兼容客户端。

企业内部工具集成

通过 Streamable HTTP 传输，将企业内部系统（CRM、ERP、工单系统）的 API 封装为 MCP 工具，供组织内所有 AI 助手统一调用。结合 OAuth 2.1 认证，可实现细粒度的权限管控。

数据管道与自动化工作流

利用 Structured Output 特性，将数据处理函数注册为 MCP 工具，让 LLM 可以在推理过程中调用精确的数据处理逻辑，避免 LLM 自身进行不可靠的数值计算。

AI Agent 工具生态

通过 ctx.session.create_message() 的 Sampling 能力，构建可递归调用 LLM 的复合 Agent 工具。配合 Elicitation 能力，实现需要人工审批的关键操作。

教育与原型验证

FastMCP 不到 10 行代码即可创建完整 MCP 服务的特性，使其成为学习 MCP 协议和快速原型验证的理想工具。

快速上手

安装

uv add "mcp[cli]"
# 或
pip install "mcp[cli]"

创建第一个 MCP Server

mkdir my-mcp-server && cd my-mcp-server
cat > server.py << 'EOF'
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("My First MCP Server")

@mcp.tool()
def hello(name: str) -> str:
    """Say hello to someone"""
    return f"Hello, {name}! Welcome to MCP."

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
EOF

本地开发与调试

SDK 提供了 CLI 工具 mcp 用于开发工作流：

# 以开发模式启动（支持热重载）
uv run mcp dev server.py

# 安装到 Claude Desktop 配置
uv run mcp install server.py

# 直接运行
uv run mcp run server.py

编写 MCP Client

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env=None
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出所有工具
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")

            # 调用工具
            result = await session.call_tool("hello", {"name": "World"})
            print(result.content[0].text)

import asyncio
asyncio.run(main())

源码架构

包结构

src/mcp/
├── server/
│   ├── mcpserver.py      # MCPServer 低级 API + FastMCP 高级 API
│   ├── fastmcp/          # FastMCP 子模块（utilities, prompts, resources, tools）
│   ├── auth/             # OAuth 2.1 认证（TokenVerifier, AuthSettings）
│   ├── models.py         # 数据模型
│   └── session.py        # ServerSession 管理
├── client/
│   ├── session.py        # ClientSession
│   ├── stdio/            # stdio 传输客户端
│   └── streamable_http.py # Streamable HTTP 传输客户端
├── types.py              # 协议类型定义（Tool, Resource, Prompt 等）
├── shared/               # 共享工具
└── cli/                  # mcp CLI 工具

核心设计模式

高级/低级 API 分层：MCPServer 提供协议级别的精确控制（@server.list_tools()、@server.call_tool() 等装饰器），FastMCP 在此基础上封装了便捷的装饰器和高层抽象。

类型系统：types.py 定义了所有 MCP 协议消息类型，包括 Tool、Resource、Prompt、CallToolResult、TextContent、SamplingMessage、CreateMessageResult 等 20+ 个类型。

传输抽象：所有传输协议通过统一的 read/write stream 接口与 Server/Client 解耦，新增传输方式无需修改核心逻辑。

Lifespan 模式：借鉴 FastAPI 的 lifespan 设计，通过 async context manager 管理应用级资源的生命周期，并通过 dataclass + 泛型 Context[AppContext] 实现类型安全的上下文传递。

实操 Demo

Demo 1: 天气查询 MCP Server

完整版天气工具，集成结构化输出、模拟 API 调用和错误处理：

from mcp.server.mcpserver import MCPServer
from pydantic import BaseModel, Field
import random

mcp = MCPServer("Weather Service")

class WeatherReport(BaseModel):
    city: str
    temperature: float = Field(description="Temperature in Celsius")
    humidity: float
    condition: str
    forecast: str

@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> WeatherReport:
    """Get current weather for a city. Unit can be 'celsius' or 'fahrenheit'."""
    temp = round(random.uniform(-5, 38), 1)
    if unit == "fahrenheit":
        temp = round(temp * 9/5 + 32, 1)

    conditions = ["sunny", "cloudy", "rainy", "partly cloudy"]
    return WeatherReport(
        city=city,
        temperature=temp,
        humidity=round(random.uniform(30, 90), 1),
        condition=random.choice(conditions),
        forecast=f"Expect {random.choice(conditions)} tomorrow."
    )

if __name__ == "__main__":
    mcp.run(transport="streamable-http", json_response=True)

Demo 2: 带 Lifespan 的数据库查询 Server

模拟 PostgreSQL 连接管理，展示生命周期管理：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.mcpserver import Context, MCPServer

class Database:
    @classmethod
    async def connect(cls) -> "Database":
        print("[Lifespan] Connecting to database...")
        return cls()
    async def disconnect(self) -> None:
        print("[Lifespan] Disconnecting from database...")
    def query(self, sql: str) -> list[dict]:
        return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

@dataclass
class AppContext:
    db: Database

@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()

mcp = MCPServer("DB Query Service", lifespan=lifespan)

@mcp.tool()
def query_users(ctx: Context[AppContext], limit: int = 10) -> list[dict]:
    db = ctx.request_context.lifespan_context.db
    return db.query(f"SELECT * FROM users LIMIT {limit}")

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Demo 3: 部署到生产环境

使用 Starlette 挂载 MCP Server 到现有 ASGI 应用：

from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware.cors import CORSMiddleware
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Production API")

@mcp.tool()
def health_check() -> str:
    return "OK"

# 挂载到 Starlette 应用
app = Starlette(routes=[
    Mount("/mcp", app=mcp.streamable_http_app()),
])

# 添加 CORS 支持
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    expose_headers=["Mcp-Session-Id"],
)

启动：

uvicorn app:app --host 0.0.0.0 --port 8000

同类对比

维度	MCP Python SDK	MCP TypeScript SDK	FastAPI + 自定义协议
协议标准化	MCP 标准协议，跨客户端兼容	MCP 标准协议	自定义协议，不兼容
学习曲线	低（FastMCP 装饰器）	低（McpServer + Zod）	中（需自行定义协议）
结构化输出	Pydantic/TypedDict/dataclass 原生支持	Zod v4 / Valibot 等 Standard Schema	需手动序列化
传输协议	stdio + SSE + Streamable HTTP	stdio + Streamable HTTP	仅 HTTP
OAuth 认证	内置 RFC 9728 支持	内置 OAuth helpers	需自行集成
ASGI 挂载	支持 Mount 到 Starlette/FastAPI	支持 Express/Hono 中间件	天然 ASGI
生态成熟度	Beta 阶段，快速迭代	v1.x 稳定版（Q3 2026）	成熟稳定
适合场景	Python 生态工具集成	Node.js/前端生态工具集成	通用 HTTP API

参考资源

• GitHub 仓库: modelcontextprotocol/python-sdk
• 官方文档: py.sdk.modelcontextprotocol.io
• MCP 协议规范: modelcontextprotocol.io
• PyPI 包: pypi.org/project/mcp
• 示例代码: github.com/modelcontextprotocol/python-sdk/tree/main/examples — 包含 23 个 snippets 示例，涵盖基础工具、结构化输出、Lifespan、OAuth、分页、Elicitation、图片处理、Sampling 等
• 参考 Server 实现: modelcontextprotocol/servers — 7 个官方参考实现（Everything, Fetch, Filesystem, Git, Memory, Sequential Thinking, Time）