# https://github.com/modelcontextprotocol/python-sdk 项目说明书
生成时间：2026-05-13 12:56:39 UTC
## 目录
- [MCP 协议与 Python SDK 简介](#page-introduction)
- [安装与配置](#page-installation)
- [快速开始指南](#page-quickstart)
- [系统架构与核心组件](#page-architecture)
- [传输协议实现](#page-transports)
- [FastMCP 服务器框架](#page-fastmcp)
- [资源、资源模板、提示与工具](#page-resources-prompts-tools)
- [上下文与会话管理](#page-context-session)
- [客户端开发指南](#page-client-development)
- [客户端 OAuth 认证](#page-client-auth)
## MCP 协议与 Python SDK 简介
### 相关页面
相关主题：[安装与配置](#page-installation), [系统架构与核心组件](#page-architecture), [快速开始指南](#page-quickstart)
相关源码文件
以下源码文件用于生成本页说明：
- [AGENTS.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/AGENTS.md)
- [CONTRIBUTING.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/CONTRIBUTING.md)
- [examples/snippets/servers/completion.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/completion.py)
- [src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
- [src/mcp/server/mcpserver/resources/base.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/base.py)
- [src/mcp/server/mcpserver/resources/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
- [examples/servers/simple-prompt/mcp_simple_prompt/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-prompt/mcp_simple_prompt/server.py)
- [examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py)
- [RELEASE.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/RELEASE.md)
# MCP 协议与 Python SDK 简介
## 概述
MCP（Model Context Protocol，模型上下文协议）是一种开放协议，用于在大型语言模型（LLM）应用程序与外部数据源、工具和服务之间建立标准化的通信桥梁。Python SDK 是该协议的核心实现库，为开发者提供了构建 MCP 服务器和客户端的能力。
MCP 协议的核心价值在于**标准化**——它定义了 LLM 应用与外部世界交互的统一方式，使开发者能够创建可复用的组件，让 AI 系统能够安全、受控地访问本地文件、数据库、API 以及各类服务。
## 核心概念
### 服务器与客户端架构
MCP 采用典型的客户端-服务器架构，支持多种通信传输方式：
| 组件 | 说明 | 支持的传输方式 |
|------|------|----------------|
| MCP 服务器 | 托管资源、工具和提示的服务器 | stdio、streamable-http |
| MCP 客户端 | 连接服务器并调用功能的客户端 | stdio、streamable-http |
```mermaid
graph TD
A["MCP 客户端"] -->|"stdio/streamable-http"| B["MCP 服务器"]
B --> C["资源管理器"]
B --> D["工具管理器"]
B --> E["提示管理器"]
B --> F["完成处理器"]
C --> G["FunctionResource"]
C --> H["FileResource"]
G --> I["fn: Callable"]
H --> J["path: Path"]
```
### 资源系统
MCP 的资源系统允许服务器向客户端暴露可读取的数据。SDK 支持两种主要资源类型：
| 资源类型 | 说明 | 关键字段 |
|----------|------|----------|
| FunctionResource | 通过函数动态生成的资源内容 | `fn: Callable` |
| FileResource | 从文件系统读取的静态资源 | `path: Path`, `is_binary: bool` |
所有资源都继承自基类 `Resource`，包含以下通用字段：
- `uri`: 资源的唯一标识符
- `name`: 资源名称
- `title`: 人类可读的标题
- `description`: 资源描述
- `mime_type`: 资源的 MIME 类型，默认 `text/plain`
- `annotations`: 资源的注解信息
- `meta`: 额外的元数据
资料来源：[src/mcp/server/mcpserver/resources/base.py:1-42](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/base.py)
### 工具系统
工具是 MCP 服务器暴露的可执行函数，客户端可以通过 JSON-RPC 调用这些工具。工具系统支持：
- **结构化输出**: 使用 Pydantic 模型定义返回格式
- **元数据**: 支持标题、描述、图标和注解
- **异步执行**: 所有工具函数支持异步调用
```python
def add_tool(
self,
fn: Callable[..., Any],
name: str | None = None,
title: str | None = None,
description: str | None = None,
annotations: Annotations | None = None,
icons: list[Icon] | None = None,
meta: ToolMeta | None = None,
structured_output: type[BaseModel] | None = None,
) -> None:
```
资料来源：[src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
### 提示系统
提示（Prompts）允许服务器定义可复用的消息模板，这些模板可以接受参数并生成动态内容：
```python
class Prompt(BaseModel, abc.ABC):
name: str
description: str | None = None
arguments: list[PromptArgument] | None = None
icons: list[Icon] | None = None
```
使用示例：
```python
return types.GetPromptResult(
messages=create_messages(context=arguments.get("context"), topic=arguments.get("topic")),
description="A simple prompt with optional context and topic arguments",
)
```
资料来源：[examples/servers/simple-prompt/mcp_simple_prompt/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-prompt/mcp_simple_prompt/server.py)
### 完成处理器
完成处理器为资源模板和提示参数提供动态补全建议：
```python
@mcp.completion()
async def handle_completion(ref, argument, context):
if isinstance(ref, ResourceTemplateReference):
if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
if context and context.arguments and context.arguments.get("owner") == "modelcontextprotocol":
repos = ["python-sdk", "typescript-sdk", "specification"]
return Completion(values=repos, has_more=False)
return None
```
资料来源：[examples/snippets/servers/completion.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/completion.py)
## 交互式任务系统
MCP 支持长时间运行的任务（Tasks），包括两个关键功能：
### elicitation（征询）
用于在执行过程中请求用户确认或输入：
```mermaid
graph LR
A["工具调用"] --> B["task.elicit()"]
B --> C["input_required 状态"]
C --> D["用户响应"]
D --> E["任务完成"]
```
### Sampling（采样）
允许服务器请求 LLM 生成内容：
```mermaid
graph LR
A["工具调用"] --> B["task.create_message()"]
B --> C["input_required 状态"]
C --> D["LLM 响应"]
D --> E["任务完成"]
```
资料来源：[examples/servers/simple-task-interactive/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-task-interactive/README.md)
## 认证系统
SDK 提供了 OAuth 认证支持，包括完整的授权码流程：
```mermaid
graph TD
A["客户端"] --> B["授权请求"]
B --> C["用户登录"]
C --> D["授权码回调"]
D --> E["令牌交换"]
E --> F["认证完成"]
```
认证服务器可以返回 HTML 表单进行交互：
```python
async def handle_login_callback(self, request: Request) -> Response:
form = await request.form()
username = form.get("username")
password = form.get("password")
state = form.get("state")
```
资料来源：[examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py)
## 开发指南
### 环境配置
```bash
# 安装 uv（如果尚未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh
# 克隆仓库
git clone https://github.com/modelcontextprotocol/python-sdk.git
# 安装依赖
uv sync --frozen --all-extras --dev
```
资料来源：[CONTRIBUTING.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/CONTRIBUTING.md)
### 代码质量要求
| 检查项 | 命令 | 说明 |
|--------|------|------|
| 格式化 | `uv run --frozen ruff format .` | 使用 ruff 格式化代码 |
| 静态检查 | `uv run --frozen ruff check . --fix` | 自动修复 lint 问题 |
| 类型检查 | `uv run --frozen pyright` | 运行 pyright 类型检查 |
| 测试 | `uv run --frozen pytest` | 执行测试套件 |
### 核心规范
1. **类型提示**: 所有公共 API 必须包含类型提示
2. **文档字符串**: 公共 API 必须有 docstring，包含 `Raises:` 部分说明可能抛出的异常
3. **异常处理**:
- 使用 `logger.exception()` 而非 `logger.error()`
- 捕获具体异常类型而非通用 `Exception`
- 禁止在顶级处理器之外使用 `except Exception:`
4. **导入规范**: 所有导入必须位于文件顶部，禁止内联导入
```python
# 正确示例
try:
resource = await self._resource_manager.get_resource(uri, context)
except ValueError as exc:
raise ResourceError(f"Unknown resource: {uri}") from exc
```
资料来源：[AGENTS.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/AGENTS.md)
### 测试规范
- 使用 `anyio` 进行异步测试，而非 `asyncio`
- 测试应快速且确定性，使用内存异步执行
- 使用线程仅在必要时，使用子进程作为最后手段
- 不要使用 `Test` 前缀的类，使用顶层 `test_*` 函数
```python
# 推荐
def test_resource_read():
assert isinstance(resource, FunctionResource)
# 避免
class TestResource:
def test_read(self):
...
```
## 分支模型
| 分支 | 用途 | 变更类型 |
|------|------|----------|
| `main` | V2 开发分支 | 新功能、破坏性变更 |
| `v1.x` | 稳定版发布分支 | 安全修复、关键 bug 修复 |
资料来源：[AGENTS.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/AGENTS.md)
## 发布流程
1. 在 `pyproject.toml` 中更新依赖版本
2. 运行 `uv lock --resolution lowest-direct` 升级锁文件
3. 通过 GitHub UI 创建 Release，标签格式为 `vX.Y.Z`
4. 版本号将从标签自动设置
资料来源：[RELEASE.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/RELEASE.md)
## 快速开始示例
### 创建简单服务器
```python
from mcp.server.mcpserver import Server
from mcp.server.stdio import stdio_server
app = Server("my-mcp-server")
@app.list_tools()
async def handle_list_tools():
return [types.Tool(
name="greet",
description="Greet someone",
inputSchema={"type": "object", "properties": {"name": {"type": "string"}}}
)]
async def main():
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
```
### 启动 HTTP 服务器
```python
import uvicorn
if transport == "streamable-http":
uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
```
资料来源：[examples/servers/simple-prompt/mcp_simple_prompt/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-prompt/mcp_simple_prompt/server.py)
## 总结
MCP Python SDK 提供了一套完整、健壮的框架，用于构建与 LLM 应用交互的服务器和客户端。通过标准化的资源、工具、提示和完成处理器接口，开发者可以创建可复用、安全且易于维护的 MCP 组件。SDK 遵循严格的代码质量标准，包括类型提示要求、完整的文档字符串规范以及全面的测试覆盖，确保其稳定性和可维护性。
---
## 安装与配置
### 相关页面
相关主题：[MCP 协议与 Python SDK 简介](#page-introduction), [快速开始指南](#page-quickstart)
相关源码文件
以下源码文件用于生成本页说明：
- [CONTRIBUTING.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/CONTRIBUTING.md)
- [AGENTS.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/AGENTS.md)
- [examples/clients/simple-auth-client/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-auth-client/README.md)
- [examples/servers/simple-auth/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-auth/README.md)
- [examples/clients/simple-task-client/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-task-client/README.md)
- [examples/servers/simple-task-interactive/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-task-interactive/README.md)
# 安装与配置
本文档详细说明 Model Context Protocol (MCP) Python SDK 的安装、环境配置以及开发环境搭建流程。
## 环境要求
### 系统依赖
| 依赖项 | 版本要求 | 说明 |
|--------|----------|------|
| Python | 3.10+ | 支持 Python 3.10 至 3.14 |
| uv | 最新稳定版 | 包管理工具，详见 [uv 官方文档](https://docs.astral.sh/uv/getting-started/installation/) |
资料来源：[CONTRIBUTING.md:1-3]()
### 验证 Python 版本
```bash
python --version
```
确保输出为 Python 3.10 或更高版本。
## 项目克隆与依赖安装
### 1. Fork 并克隆仓库
```bash
# 1. 在 GitHub 上 Fork 仓库
# 2. 克隆你的 Fork
git clone https://github.com/YOUR-USERNAME/python-sdk.git
# 3. 进入项目目录
cd python-sdk
```
资料来源：[CONTRIBUTING.md:3-6]()
### 2. 安装依赖
MCP Python SDK **仅支持 uv 作为包管理工具**，严格禁止使用 pip 进行安装或依赖管理。
```bash
# 安装所有依赖（包括开发依赖和可选依赖）
uv sync --frozen --all-extras --dev
```
参数说明：
- `--frozen`：锁定版本，不更新 lock 文件
- `--all-extras`：安装所有可选依赖
- `--dev`：安装开发依赖
资料来源：[CONTRIBUTING.md:7]()
### 3. 设置预提交钩子
```bash
uv tool install pre-commit --with pre-commit-uv --force-reinstall
```
预提交钩子会自动执行代码格式化、类型检查等操作。
资料来源：[CONTRIBUTING.md:9-11]()
## 开发环境配置
### uv 工具链使用规范
| 操作类型 | 命令格式 | 说明 |
|----------|----------|------|
| 运行工具 | `uv run --frozen ` | 始终使用 `--frozen` 避免意外修改 lock 文件 |
| 添加依赖 | `uv add ` | 仅用于开发环境 |
| 升级依赖 | `uv lock --upgrade-package ` | 升级指定包版本 |
| 测试跨版本 | `uv run --frozen --python 3.10 pytest ...` | 指定 Python 版本运行测试 |
**禁止操作：**
- 禁止使用 `uv pip install`
- 禁止使用 `@latest` 语法指定版本
- 不要仅为修复 CVE 而提升依赖版本下限，约束条件 `>=` 已允许用户自行升级
资料来源：[AGENTS.md:1-20]()
### 代码质量检查
开发过程中应运行以下检查：
```bash
# 运行测试
uv run pytest
# 类型检查
uv run pyright
# 代码格式化
uv run ruff format .
# 代码检查
uv run ruff check . --fix
```
全部检查命令可通过预提交钩子一键执行：
```bash
pre-commit run --all-files
```
资料来源：[CONTRIBUTING.md:20-35]()
## 服务器配置
### 基本配置参数
客户端连接 MCP 服务器时可通过环境变量配置连接参数：
| 环境变量 | 说明 | 默认值 |
|----------|------|--------|
| `MCP_SERVER_PORT` | MCP 服务器端口号 | `8000` |
| `MCP_TRANSPORT_TYPE` | 传输类型：`streamable-http` 或 `sse` | `streamable-http` |
| `MCP_CLIENT_METADATA_URL` | 客户端元数据 (CIMD) 的可选 URL | `None` |
资料来源：[examples/clients/simple-auth-client/README.md:47-52]()
### 启动服务器示例
#### Streamable HTTP 传输
```bash
# 基本启动
uv run mcp-simple-auth --port=8000 --transport=streamable-http
# 带 OAuth 严格模式
uv run mcp-simple-auth --port=8000 --transport=streamable-http --oauth-strict
```
#### SSE 传输
```bash
uv run mcp-simple-auth --port=8000 --transport=sse
```
资料来源：[examples/servers/simple-auth/README.md:30-45]()
### 任务服务器配置
交互式任务服务器默认启动地址为 `http://localhost:8000/mcp`，可通过 `--port` 参数修改：
```bash
uv run mcp-simple-task-interactive --port=9000
```
资料来源：[examples/servers/simple-task-interactive/README.md:8-12]()
## 项目分支模型
MCP SDK 采用双分支维护策略：
| 分支 | 用途 | 变更类型 |
|------|------|----------|
| `main` | V2 开发分支 | 新功能、重构，允许破坏性变更 |
| `v1.x` | V1 稳定分支 | 仅接收安全修复和关键 bug 修复 |
开发新功能或破坏性变更应基于 `main` 分支创建新分支。安全修复和关键 bug 修复应基于 `v1.x` 分支。
资料来源：[AGENTS.md:1-15]()
## 开发工作流
```mermaid
graph TD
A[选择目标分支] --> B[基于目标分支创建新分支]
B --> C[编写代码]
C --> D[运行测试: uv run pytest]
D --> E{测试通过?}
E -->|否| F[修复问题]
F --> C
E -->|是| G[运行类型检查: uv run pyright]
G --> H[运行代码检查: uv run ruff check]
H --> I[提交代码]
I --> J[创建 Pull Request]
J --> K[等待 Code Review]
```
### 代码提交检查清单
1. ✅ 更新相关文档
2. ✅ 为新功能添加测试
3. ✅ 确保 CI 通过
4. ✅ 处理代码审查反馈
资料来源：[CONTRIBUTING.md:37-60]()
## 调试与日志
### 异常处理规范
- **必须使用** `logger.exception()` 而非 `logger.error()` 捕获异常
- **禁止** 在日志消息中包含异常信息：`logger.exception("Failed")` 而非 `logger.exception(f"Failed: {e}")`
- **优先捕获具体异常类型**：
- 文件操作：`except (OSError, PermissionError):`
- JSON 解析：`except json.JSONDecodeError:`
- 网络操作：`except (ConnectionError, TimeoutError):`
- **禁止使用** `except Exception:`（顶层处理器除外）
资料来源：[AGENTS.md:1-30]()
### 常见配置问题排查
| 问题 | 可能原因 | 解决方案 |
|------|----------|----------|
| 连接被拒绝 | 服务器未启动或端口错误 | 检查 `MCP_SERVER_PORT` 环境变量 |
| 传输类型错误 | 不支持的传输类型 | 确认使用 `streamable-http` 或 `sse` |
| 依赖冲突 | 使用了 pip 而非 uv | 使用 `uv sync` 重新安装依赖 |
## 验证安装
安装完成后，可通过以下命令验证环境：
```bash
# 验证 uv 可用
uv --version
# 验证 Python 版本
python --version
# 验证 MCP SDK 安装
uv run python -c "import mcp; print(mcp.__version__)"
# 运行项目测试
uv run pytest -v
```
## 下一步
- 阅读 [工具开发指南](./tools.md) 了解如何创建 MCP 工具
- 阅读 [资源管理](./resources.md) 了解如何暴露资源
- 阅读 [提示模板](./prompts.md) 了解如何定义提示
---
## 快速开始指南
### 相关页面
相关主题：[FastMCP 服务器框架](#page-fastmcp), [资源、资源模板、提示与工具](#page-resources-prompts-tools)
相关源码文件
以下源码文件用于生成本页说明：
- [examples/snippets/servers/fastmcp_quickstart.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/fastmcp_quickstart.py)
- [examples/snippets/servers/basic_tool.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_tool.py)
- [examples/snippets/servers/basic_resource.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_resource.py)
- [examples/snippets/servers/basic_prompt.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_prompt.py)
- [src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
- [src/mcp/server/mcpserver/resources/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
# 快速开始指南
本指南帮助开发者快速上手 Model Context Protocol (MCP) Python SDK，涵盖服务器创建、工具（Tool）、资源（Resource）和提示（Prompt）的定义与注册，以及如何启动服务。
## 环境准备
### 前置要求
| 要求 | 版本说明 |
|------|----------|
| Python | 3.10+ |
| 包管理器 | uv |
| 操作系统 | Windows、macOS、Linux |
### 安装 SDK
```bash
# 使用 uv 安装稳定版本
uv add mcp
# 安装开发版本（包含所有额外依赖和开发工具）
uv sync --frozen --all-extras --dev
```
资料来源：[CONTRIBUTING.md:5-8]()
## MCP 服务器架构
MCP 服务器是 MCP 协议的核心组件，负责向客户端提供服务能力。
```mermaid
graph TD
A[客户端] --> B[MCP Server]
B --> C[工具管理器
Tool Manager]
B --> D[资源管理器
Resource Manager]
B --> E[提示管理器
Prompt Manager]
C --> F[工具函数]
D --> G[资源数据]
E --> H[提示模板]
```
MCPServer 类是 SDK 的主入口，负责管理所有组件的生命周期并处理客户端请求。
资料来源：[src/mcp/server/mcpserver/server.py:1-50]()
## 创建基础服务器
### 最小示例
```python
from mcp.server import MCPServer
server = MCPServer(
name="my-server",
version="1.0.0",
)
```
### 完整配置示例
```python
from mcp.server import MCPServer, ServerSettings
settings = ServerSettings(
host="0.0.0.0",
port=8000,
streamable_http_transport=True,
)
server = MCPServer(
name="example-server",
version="1.0.0",
settings=settings,
)
```
| ServerSettings 参数 | 类型 | 默认值 | 说明 |
|---------------------|------|--------|------|
| host | str | "127.0.0.1" | 监听地址 |
| port | int | 8000 | 监听端口 |
| streamable_http_transport | bool | False | 启用流式 HTTP 传输 |
| oauth_strict | bool | False | 启用严格 OAuth 认证 |
资料来源：[examples/snippets/servers/fastmcp_quickstart.py:1-30]()
## 定义工具（Tool）
工具是服务器暴露给客户端的可调用函数。
### 基本工具定义
```python
from mcp.server import MCPServer
from mcp.types import Tool
def get_weather(location: str) -> str:
"""获取指定位置的天气信息"""
return f"{location} 天气晴朗，温度 25°C"
server = MCPServer(name="weather-server")
server.add_tool(
fn=get_weather,
name="get_weather",
description="获取天气信息",
)
```
### add_tool 参数说明
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| fn | Callable | 是 | 工具函数 |
| name | str | 否 | 工具名称，默认为函数名 |
| title | str | 否 | 工具标题 |
| description | str | 否 | 工具描述 |
| annotations | Annotations | 否 | 工具注解 |
函数必须包含类型提示，SDK 会自动验证参数类型。
资料来源：[examples/snippets/servers/basic_tool.py:1-25]()
## 定义资源（Resource）
资源用于向客户端提供数据访问能力。
### 资源类型
| 类型 | 说明 | 用途 |
|------|------|------|
| FunctionResource | 函数资源 | 动态生成数据 |
| FileResource | 文件资源 | 读取本地文件 |
### 函数资源
```python
from mcp.server import MCPServer
from mcp.server.resources import FunctionResource
def read_config() -> str:
"""读取配置数据"""
return '{"debug": true, "version": "1.0"}'
server = MCPServer(name="config-server")
server.add_resource(
FunctionResource.from_function(
fn=read_config,
uri="config://app",
name="app_config",
title="应用配置",
description="返回当前应用配置",
)
)
```
### 文件资源
```python
from mcp.server.resources import FileResource
from pathlib import Path
server.add_resource(
FileResource(
path=Path("/etc/app/config.json"),
name="config_file",
uri="file:///etc/app/config.json",
mime_type="application/json",
)
)
```
| FileResource 字段 | 类型 | 说明 |
|-------------------|------|------|
| path | Path | 文件路径（必须为绝对路径） |
| is_binary | bool | 是否以二进制模式读取 |
| mime_type | str | MIME 类型 |
资料来源：[examples/snippets/servers/basic_resource.py:1-45]()
## 定义提示（Prompt）
提示模板用于生成可复用的对话片段。
```python
from mcp.server import MCPServer
from mcp.server.prompts import Prompt
server = MCPServer(name="prompt-server")
server.add_prompt(
Prompt(
name="code_review",
description="代码审查提示",
arguments=[
PromptArgument(
name="language",
description="编程语言",
required=True,
),
PromptArgument(
name="code",
description="待审查代码",
required=True,
),
],
)
)
```
| PromptArgument 字段 | 类型 | 说明 |
|---------------------|------|------|
| name | str | 参数名称 |
| description | str | 参数描述 |
| required | bool | 是否必填 |
资料来源：[examples/snippets/servers/basic_prompt.py:1-30]()
## 读取资源
### read_resource 方法
```python
async def handle_resource_read(server: MCPServer):
# 使用 URI 读取资源
contents = await server.read_resource("config://app")
for content in contents:
print(content.content)
```
| 参数 | 类型 | 说明 |
|------|------|------|
| uri | AnyUrl \| str | 资源标识符 |
| context | Context | 请求上下文 |
异常处理：未知资源返回 `ResourceError`，读取失败时异常被捕获并记录日志。
资料来源：[src/mcp/server/mcpserver/server.py:45-70]()
## 启动服务器
### 运行服务器
```python
import uvicorn
if __name__ == "__main__":
uvicorn.run(
"my_server:app",
host="0.0.0.0",
port=8000,
reload=True,
)
```
### 命令行启动
```bash
# 方式一：直接运行模块
uv run python -m my_server
# 方式二：使用 uvicorn
uv run uvicorn my_server:app --host 0.0.0.0 --port 8000
```
### 配置传输方式
MCP 支持多种传输协议：
| 传输类型 | 说明 | 使用场景 |
|----------|------|----------|
| streamable-http | 流式 HTTP | 推荐默认选项 |
| sse | Server-Sent Events | 需要长连接的场景 |
```bash
# 启动支持 OAuth 的流式 HTTP 服务器
MCP_TRANSPORT=streamable-http MCP_OAUTH_STRICT=true uv run mcp-my-server
```
## 项目结构示例
```
mcp_project/
├── pyproject.toml
├── src/
│ └── my_server/
│ ├── __init__.py
│ ├── server.py # 服务器定义
│ ├── tools.py # 工具实现
│ ├── resources.py # 资源定义
│ └── prompts.py # 提示定义
└── uv.lock
```
## 常见工作流程
```mermaid
graph LR
A[定义工具] --> B[定义资源]
B --> C[定义提示]
C --> D[创建 MCPServer 实例]
D --> E[注册组件]
E --> F[启动服务]
F --> G[客户端连接]
G --> H[调用工具/访问资源]
```
## 下一步
- 阅读 [examples/servers/](examples/servers/) 目录下的完整示例
- 了解 [认证与授权](examples/servers/simple-auth/README.md) 功能
- 探索 [实验性功能](examples/servers/simple-task/README.md) 如任务支持
---
## 系统架构与核心组件
### 相关页面
相关主题：[传输协议实现](#page-transports), [上下文与会话管理](#page-context-session)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
- [src/mcp/server/session.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/session.py)
- [src/mcp/client/session.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/session.py)
- [src/mcp/shared/session.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/session.py)
- [src/mcp/shared/message.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/message.py)
- [src/mcp/types/_types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/types/_types.py)
- [src/mcp/types/jsonrpc.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/types/jsonrpc.py)
- [src/mcp/server/mcpserver/resources/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
- [src/mcp/server/auth/routes.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/auth/routes.py)
- [src/mcp/server/experimental/task_context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/experimental/task_context.py)
# 系统架构与核心组件
Model Context Protocol (MCP) Python SDK 是一个用于构建 Model Context Protocol 客户端和服务器的框架。MCP 是一种标准化协议，使 AI 模型能够与外部数据源和工具进行交互。本文详细介绍 MCP SDK 的系统架构和核心组件。
## 1. 架构概述
MCP 采用客户端-服务器架构，支持多种传输协议（stdio、streamable-http）。协议基于 JSON-RPC 2.0 规范实现双向通信。
```mermaid
graph TD
subgraph 客户端层
Client[Client 客户端]
ClientSession[ClientSession 客户端会话]
end
subgraph 传输层
Stdio[Stdio 传输]
Http[Streamable HTTP 传输]
end
subgraph 服务器层
Server[Server 服务器]
MCPServer[MCPServer MCP 服务器]
ServerSession[ServerSession 服务端会话]
end
subgraph 核心服务
ResourceManager[ResourceManager 资源管理]
PromptManager[PromptManager 提示管理]
ToolHandler[ToolHandler 工具处理]
Auth[AuthModule 认证模块]
end
Client --> ClientSession
ClientSession <-->|JSON-RPC| Stdio
ClientSession <-->|JSON-RPC| Http
Stdio <--> Server
Http <--> Server
Server --> ServerSession
ServerSession --> MCPServer
MCPServer --> ResourceManager
MCPServer --> PromptManager
MCPServer --> ToolHandler
MCPServer --> Auth
```
## 2. 核心组件关系
### 2.1 组件层次结构
```mermaid
graph TD
subgraph 高层API
MCPServer[MCP Server]
Client[Client]
end
subgraph 会话层
ClientSession[ClientSession]
ServerSession[ServerSession]
SharedSession[SharedSession]
end
subgraph 消息层
MessageHandler[MessageHandler]
JSONRPCHandler[JSONRPCHandler]
end
subgraph 传输层
StdioTransport[StdioTransport]
StreamableHttpTransport[StreamableHttpTransport]
end
MCPServer --> ServerSession
Client --> ClientSession
ClientSession --> SharedSession
ServerSession --> SharedSession
SharedSession --> MessageHandler
MessageHandler --> JSONRPCHandler
StdioTransport --> JSONRPCHandler
StreamableHttpTransport --> JSONRPCHandler
```
### 2.2 组件职责表
| 组件 | 文件路径 | 职责 |
|------|----------|------|
| `Client` | `src/mcp/client/client.py` | 管理客户端连接生命周期，处理服务端发现和传输协商 |
| `ClientSession` | `src/mcp/client/session.py` | 客户端会话管理，处理请求/响应、notifications 和 call cancellation |
| `ServerSession` | `src/mcp/server/session.py` | 服务端会话管理，处理请求路由和响应 |
| `SharedSession` | `src/mcp/shared/session.py` | 会话基类，封装 JSON-RPC 消息处理逻辑 |
| `MCPServer` | `src/mcp/server/mcpserver/server.py` | 高层服务器 API，整合资源、工具、提示管理 |
| `ResourceManager` | `src/mcp/server/mcpserver/resources/` | 资源注册、模板管理和资源读取 |
| `ToolHandler` | `内置于 MCPServer` | 工具注册、参数验证和执行 |
资料来源：[src/mcp/shared/session.py:1-50](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/session.py)
## 3. 会话层详解
### 3.1 SharedSession 基类
`SharedSession` 是客户端和服务端会话的共同基类，封装了 JSON-RPC 2.0 协议的通用逻辑。
**核心职责：**
- 消息队列管理
- 请求-响应对应关系维护
- Cancellation token 处理
- 消息发送和接收
```mermaid
graph LR
subgraph SharedSession
Queue[消息队列]
Requests[待处理请求]
Handlers[消息处理器]
end
subgraph 消息类型
Request[Request 请求]
Response[Response 响应]
Notification[Notification 通知]
end
Queue --> Handlers
Requests --> Handlers
Request -->|等待响应| Requests
Response -->|匹配请求| Requests
Notification -->|无需响应| Queue
```
资料来源：[src/mcp/shared/session.py:50-150](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/session.py)
### 3.2 客户端会话
`ClientSession` 扩展了 SharedSession，添加了客户端特定的功能：
```python
class ClientSession(SharedSession):
# 核心方法
async def initialize() -> InitializeResult
async def list_tools() -> list[Tool]
async def call_tool(name: str, arguments: dict) -> CallToolResult
async def list_resources() -> list[Resource]
async def list_resource_templates() -> list[ResourceTemplate]
async def read_resource(uri: str | AnyUrl) -> ReadResourceResult
async def list_prompts() -> list[Prompt]
async def get_prompt(name: str, arguments: dict) -> GetPromptResult
```
**客户端初始化流程：**
```mermaid
sequenceDiagram
participant Client
participant ClientSession
participant Server
Client->>ClientSession: 创建会话
ClientSession->>Server: 发送 initialize 请求
Server->>ServerSession: 处理初始化
ServerSession-->>ClientSession: 返回 InitializeResult
ClientSession->>Client: 通知初始化完成
Client->>ClientSession: 发送已接收通知
ClientSession->>Server: 发送 notifications/initialized
```
资料来源：[src/mcp/client/session.py:100-200](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/session.py)
### 3.3 服务端会话
`ServerSession` 负责处理来自客户端的请求和调用服务端处理器。
```python
class ServerSession(SharedSession):
# 核心回调
on_list_tools: Optional[Callback[...]]
on_call_tool: Optional[Callback[...]]
on_list_resources: Optional[Callback[...]]
on_read_resource: Optional[Callback[...]]
on_list_resource_templates: Optional[Callback[...]]
on_list_prompts: Optional[Callback[...]]
on_get_prompt: Optional[Callback[...]]
```
资料来源：[src/mcp/server/session.py:50-120](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/session.py)
## 4. 消息协议
### 4.1 JSON-RPC 2.0 实现
MCP 基于 JSON-RPC 2.0 规范，定义了三类消息：
| 消息类型 | 描述 | 必须包含字段 |
|---------|------|-------------|
| `JSONRPCRequest` | 请求消息，需要对方响应 | `jsonrpc`, `id`, `method` |
| `JSONRPCResponse` | 响应消息 | `jsonrpc`, `id`, `result` |
| `JSONRPCError` | 错误响应 | `jsonrpc`, `id`, `error` |
| `JSONRPCNotification` | 通知消息，无需响应 | `jsonrpc`, `method` |
```python
class JSONRPCMessage(BaseModel):
"""JSON-RPC 消息基类"""
jsonrpc: Literal["2.0"] = "2.0"
class JSONRPCRequest(JSONRPCMessage):
"""带方法调用的请求"""
id: RequestId
method: str
params: Optional[dict[str, Any]] = None
class JSONRPCResponse(JSONRPCMessage):
"""成功响应"""
id: Union[int, str]
result: Any
class JSONRPCError(JSONRPCMessage):
"""错误响应"""
id: Union[int, str]
error: ErrorData
```
资料来源：[src/mcp/types/jsonrpc.py:20-80](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/types/jsonrpc.py)
### 4.2 错误代码定义
| 错误码 | 名称 | 用途 |
|-------|------|------|
| `-32700` | Parse Error | JSON 解析失败 |
| `-32600` | Invalid Request | 无效的请求对象 |
| `-32601` | Method Not Found | 方法不存在 |
| `-32602` | Invalid Params | 无效的参数 |
| `-32603` | Internal Error | 内部错误 |
| `-32000` | Connection Closed | 连接已关闭 |
| `-32001` | Request Timeout | 请求超时 |
## 5. MCP Server 高级 API
`MCPServer` 是构建 MCP 服务器的高层入口，封装了资源管理、工具处理、提示管理和认证功能。
### 5.1 初始化参数
```python
class MCPServer:
def __init__(
self,
name: str = "mcp-server",
version: str = "1.0.0",
settings: ServerSettings | None = None,
resources: list[Resource] | None = None,
tools: list[Callable] | None = None,
prompts: list[Prompt] | None = None,
instructions: str | None = None,
lifespan: LifespanContextManager | None = None,
) -> None:
```
| 参数 | 类型 | 描述 |
|------|------|------|
| `name` | `str` | 服务器名称 |
| `version` | `str` | 服务器版本 |
| `settings` | `ServerSettings` | 服务器配置选项 |
| `resources` | `list[Resource]` | 初始资源列表 |
| `tools` | `list[Callable]` | 可调用工具列表 |
| `prompts` | `list[Prompt]` | 预定义提示列表 |
| `instructions` | `str` | 服务器使用说明 |
| `lifespan` | `LifespanContextManager` | 生命周期管理器 |
资料来源：[src/mcp/server/mcpserver/server.py:100-150](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
### 5.2 核心方法
```python
class MCPServer:
# 资源管理
async def list_resources() -> list[MCPResource]
async def list_resource_templates() -> list[MCPResourceTemplate]
async def read_resource(uri: AnyUrl | str) -> Iterable[ReadResourceContents]
# 工具管理
def add_tool(
fn: Callable[..., Any],
name: str | None = None,
title: str | None = None,
description: str | None = None,
annotations: Annotations | None = None,
) -> None:
# 提示管理
def add_prompt(prompt: Prompt) -> None
async def list_prompts() -> list[MCPPrompt]
async def get_prompt(name: str, arguments: dict | None) -> MCPPrompt
# 生命周期
async def run_async()
def streamable_http_app() -> ASGIApp
```
资料来源：[src/mcp/server/mcpserver/server.py:200-300](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
## 6. 资源系统
### 6.1 资源类型
MCP 支持两种主要资源类型：
| 类型 | 文件路径 | 描述 |
|------|----------|------|
| `FunctionResource` | `resources/types.py` | 从函数动态生成的资源 |
| `FileResource` | `resources/types.py` | 读取本地文件的资源 |
```python
class FunctionResource(Resource):
"""从函数动态生成的资源"""
uri: AnyUrl
name: str
title: Optional[str] = None
description: Optional[str] = None
mime_type: str = "text/plain"
fn: Callable[..., Any] # 实际读取函数
class FileResource(Resource):
"""读取本地文件的资源"""
path: Path # 必须是绝对路径
is_binary: bool = False # 是否二进制读取
mime_type: str = "text/plain"
```
资料来源：[src/mcp/server/mcpserver/resources/types.py:50-100](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
### 6.2 资源读取流程
```mermaid
graph TD
Client[Client] -->|read_resource| ServerSession
ServerSession -->|路由| MCPServer
MCPServer -->|获取资源| ResourceManager
ResourceManager -->|验证URI| URI解析
URI解析 -->|查找资源| ResourceRegistry
ResourceRegistry -->|返回| Resource
Resource -->|调用read方法| FunctionResource
FunctionResource -->|执行函数| ReadFunction
ReadFunction -->|返回内容| ReadResourceContents
```
## 7. 认证系统
### 7.1 OAuth 2.0 支持
MCP 支持 OAuth 2.0 认证流程，包括：
- **授权码流程** (Authorization Code Flow)
- **客户端凭证流程** (Client Credentials Flow)
- **令牌刷新** (Token Refresh)
- **令牌撤销** (Token Revocation)
```mermaid
graph LR
subgraph 认证流程
Client -->|1. 授权请求| AuthServer
AuthServer -->|2. 授权码| Client
Client -->|3. 令牌请求| AuthServer
AuthServer -->|4. 访问令牌| Client
Client -->|5. 受保护资源| ResourceServer
ResourceServer -->|6. 资源| Client
end
subgraph 组件
AuthServer[授权服务器]
ResourceServer[资源服务器]
end
```
资料来源：[src/mcp/server/auth/routes.py:50-150](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/auth/routes.py)
### 7.2 认证配置
```python
class AuthServerProvider(Protocol):
"""认证服务器提供者协议"""
async def get_auth_server_metadata(self, issuer_url: AnyHttpUrl) -> OAuthServerMetadata: ...
async def create_authorization_url(self, ...) -> tuple[str, str]: ... # url, state
async def handle_authorization_callback(self, ...) -> Tokens: ...
async def refresh_access_token(self, ...) -> Tokens: ...
class TokenVerifier(Protocol):
"""令牌验证器协议"""
async def verify(self, token: str) -> TokenMetadata: ...
```
### 7.3 受保护资源元数据
RFC 9728 兼容的受保护资源元数据端点：
```python
def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
"""
构建 RFC 9728 合规的受保护资源元数据 URL
在主机和资源路径之间插入 /.well-known/oauth-protected-resource
例如: https://example.com/mcp -> https://example.com/.well-known/oauth-protected-resource/mcp
"""
```
## 8. 交互式任务
### 8.1 Elicitation 与 Sampling
MCP 支持服务端向客户端发起交互请求：
| 交互类型 | 方向 | 用途 |
|---------|------|------|
| `Elicitation` | Server → Client | 请求用户确认或输入 |
| `Sampling` | Server → Client | 请求 LLM 补全 |
```mermaid
graph TD
subgraph Elicitation 流程
Server -->|task.elicit()| Task
Task -->|发送请求| Client
Client -->|用户输入| ElicitResult
ElicitResult -->|返回| Task
Task -->|完成| Result
end
subgraph Sampling 流程
Server2 -->|task.create_message()| Task2
Task2 -->|发送请求| Client2
Client2 -->|LLM响应| CreateMessageResult
CreateMessageResult -->|返回| Task2
Task2 -->|完成| Result2
end
```
### 8.2 任务上下文
```python
class TaskContext:
"""任务上下文，用于处理交互式任务"""
async def elicit(
self,
message: str,
requested_schema: dict[str, Any],
**kwargs,
) -> ElicitResult:
"""发送 elicitation 请求给客户端"""
async def create_message(
self,
messages: list[CreateMessageRequestParam],
**kwargs,
) -> CreateMessageResult:
"""发送 sampling 请求给客户端的 LLM"""
```
资料来源：[src/mcp/server/experimental/task_context.py:100-200](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/experimental/task_context.py)
## 9. 传输层
### 9.1 Stdio 传输
适用于本地进程间通信，通过标准输入/输出流传输 JSON-RPC 消息。
```python
from mcp.server.stdio import stdio_server
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
```
### 9.2 Streamable HTTP 传输
适用于网络通信，支持长轮询和 Server-Sent Events。
```python
from mcp.server import MCPServer
import uvicorn
app = MCPServer(...)
uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=8000)
```
| 传输方式 | 适用场景 | 特点 |
|---------|---------|------|
| Stdio | 本地进程、CLI 工具 | 低延迟，简单可靠 |
| Streamable HTTP | 网络服务、分布式部署 | 支持双向通信，可扩展 |
## 10. 消息类型定义
### 10.1 核心类型
| 类型 | 描述 |
|------|------|
| `Tool` | 可调用工具的定义 |
| `Resource` | 数据资源的定义 |
| `ResourceTemplate` | 参数化资源模板 |
| `Prompt` | 预定义提示模板 |
| `SamplingMessage` | 采样消息 |
| `Completion` | 自动补全建议 |
```python
class Tool(BaseModel):
name: str
description: Optional[str] = None
inputSchema: dict[str, Any]
annotations: Optional[Annotations] = None
class Resource(BaseModel):
uri: AnyUrl
name: str
description: Optional[str] = None
mimeType: Optional[str] = None
class Prompt(BaseModel):
name: str
description: Optional[str] = None
arguments: Optional[list[PromptArgument]] = None
```
资料来源：[src/mcp/types/_types.py:50-150](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/types/_types.py)
## 11. 服务器设置
```python
class ServerSettings(BaseModel):
"""服务器配置选项"""
host: str | None = None
port: int | None = None
streamable_http_path: str | None = None
streamable_http_session_timeout: float = 60.0
streamable_http_max_session_duration: float | None = None
streamable_http_session_keepalive: bool = True
# 功能开关
capabilities: ServerCapabilities = Field(default_factory=ServerCapabilities)
# 重复检测
warn_on_duplicate_tools: bool = True
warn_on_duplicate_resources: bool = True
warn_on_duplicate_prompts: bool = True
# 认证配置
auth: AuthSettings | None = None
# 生命周期
lifespan: LifespanContextManager | None = None
```
资料来源：[src/mcp/server/mcpserver/server.py:50-100](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
## 12. 总结
MCP Python SDK 的架构设计遵循以下原则：
1. **分层设计**：清晰的分层架构，从传输层到高层 API
2. **协议标准化**：基于 JSON-RPC 2.0，确保互操作性
3. **灵活性**：支持多种传输方式和认证机制
4. **可扩展性**：通过回调和中间件支持自定义行为
5. **类型安全**：全面的类型提示和 Pydantic 模型验证
这套架构使得开发者能够快速构建支持外部数据和工具交互的 AI 应用服务器和客户端。
---
## 传输协议实现
### 相关页面
相关主题：[系统架构与核心组件](#page-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/client/stdio.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/stdio.py)
- [src/mcp/client/sse.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/sse.py)
- [src/mcp/client/websocket.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/websocket.py)
- [src/mcp/client/streamable_http.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/streamable_http.py)
- [src/mcp/server/stdio.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/stdio.py)
- [src/mcp/server/sse.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/sse.py)
- [src/mcp/server/websocket.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/websocket.py)
- [src/mcp/server/streamable_http.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/streamable_http.py)
- [src/mcp/server/streamable_http_manager.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/streamable_http_manager.py)
- [examples/snippets/servers/streamable_http_basic_mounting.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_http_basic_mounting.py)
- [examples/snippets/servers/streamable_starlette_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_starlette_mount.py)
# 传输协议实现
## 概述
MCP (Model Context Protocol) Python SDK 支持多种传输协议实现，用于客户端与服务器之间的通信。这些传输层负责在进程间或网络间传递 MCP 协议消息，支持不同的部署场景和性能需求。
## 支持的传输协议类型
| 传输类型 | 客户端模块 | 服务器模块 | 典型用途 |
|---------|-----------|-----------|---------|
| stdio | `mcp.client.stdio` | `mcp.server.stdio` | 本地进程通信 |
| Streamable HTTP | `mcp.client.streamable_http` | `mcp.server.streamable_http` | HTTP-based 远程通信 |
| SSE | `mcp.client.sse` | `mcp.server.sse` | 服务端推送事件 |
| WebSocket | `mcp.client.websocket` | `mcp.server.websocket` | 双向实时通信 |
## 核心传输协议
### Stdio 传输
Stdio 传输是最基础的进程间通信方式，适用于本地运行的 MCP 服务器。客户端通过标准输入/输出流与服务器通信，协议消息以 JSON 格式编码。
```mermaid
graph LR
A[Client Process] -->|stdin| B[Server Process]
B -->|stdout| A
```
**客户端用法示例：**
```python
from mcp.client.stdio import stdio_client
from mcp.client import ClientSession
async def main():
async with stdio_client() as streams:
async with ClientSession(*streams) as session:
await session.initialize()
# 使用 session 进行 RPC 调用
```
**服务器用法示例：**
```python
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
```
资料来源：[src/mcp/client/stdio.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/stdio.py)，[src/mcp/server/stdio.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/stdio.py)
### Streamable HTTP 传输
Streamable HTTP 是为 Web 环境设计的传输协议，支持通过 HTTP 请求/响应流式传输 MCP 消息。该协议兼容 RESTful 架构，便于与现有 Web 基础设施集成。
**协议特点：**
- 支持分块传输编码 (Chunked Transfer Encoding)
- 支持 Server-Sent Events 用于服务端推送
- 与 Starlette、FastAPI 等 ASGI 框架兼容
- 支持 SSE 作为服务端推送机制
资料来源：[src/mcp/client/streamable_http.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/streamable_http.py)，[src/mcp/server/streamable_http.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/streamable_http.py)
**服务器配置示例：**
```python
from mcp.server import Server
from mcp.server.streamable_http import StreamableHTTPServerHandler
import uvicorn
app = Server("my-mcp-server")
if __name__ == "__main__":
uvicorn.run(
app.streamable_http_app(),
host="127.0.0.1",
port=8000
)
```
资料来源：[examples/snippets/servers/streamable_http_basic_mounting.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_http_basic_mounting.py)
### SSE 传输
Server-Sent Events (SSE) 传输允许服务器主动向客户端推送消息，适用于需要服务器端实时通知的场景。
```mermaid
graph LR
A[HTTP Client] -->|POST /mcp| B[MCP Server]
B -->|SSE Stream| A
A -->|GET /sse| B
```
资料来源：[src/mcp/client/sse.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/sse.py)，[src/mcp/server/sse.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/sse.py)
### WebSocket 传输
WebSocket 传输提供全双工通信通道，适用于需要低延迟、双向实时交互的应用场景。
**WebSocket 服务器配置：**
```python
from mcp.server.websocket import websocket_server
import uvicorn
async def main():
async with websocket_server(app) as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
if __name__ == "__main__":
uvicorn.run(main, host="127.0.0.1", port=8000)
```
资料来源：[src/mcp/server/websocket.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/websocket.py)，[src/mcp/client/websocket.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/websocket.py)
## Streamable HTTP 服务器管理
`streamable_http_manager` 模块提供了高级的 HTTP 传输管理功能，包括会话管理、请求路由和生命周期管理。
### 核心组件
| 组件 | 职责 |
|------|------|
| `StreamableHTTPServerHandler` | 处理 HTTP 请求，解析 MCP 协议消息 |
| `HTTPAuthorizationMiddleware` | 处理认证头和授权验证 |
| `HTTPStreamManager` | 管理流式响应和 SSE 连接 |
资料来源：[src/mcp/server/streamable_http_manager.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/streamable_http_manager.py)
### 挂载到 ASGI 应用
MCP HTTP 服务器可以挂载到现有的 ASGI 应用中：
```python
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from mcp.server.streamable_http_manager import StreamableHTTPManager
async def app_scope(request):
# MCP 服务器逻辑
pass
mcp_app = Starlette(
routes=[
Mount("/mcp", app=app_scope),
]
)
```
资料来源：[examples/snippets/servers/streamable_starlette_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_starlette_mount.py)
## 传输协议选择指南
```mermaid
graph TD
A[选择传输协议] --> B{部署场景}
B -->|本地进程| C[Stdio]
B -->|Web 服务| D[Streamable HTTP]
B -->|实时推送| E[SSE]
B -->|低延迟交互| F[WebSocket]
C --> G[最简单、最可靠]
D --> H[兼容性最佳]
E --> I[单向实时通信]
F --> J[全双工通信]
```
**选择建议：**
| 场景 | 推荐协议 | 原因 |
|------|---------|------|
| Claude Desktop 集成 | Stdio | 原生支持，最稳定 |
| Web 应用后端 | Streamable HTTP | 标准化 HTTP 接口 |
| 实时监控面板 | SSE | 服务端推送能力强 |
| 游戏/协作工具 | WebSocket | 最低延迟 |
## 协议消息格式
所有传输协议都使用相同的 JSON-RPC 2.0 消息格式：
```json
// 请求消息
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
// 响应消息
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [...]
}
}
// 错误消息
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32600,
"message": "Invalid Request"
}
}
```
## 服务器示例对比
### Stdio 模式
```python
from mcp.server import Server
from mcp.server.stdio import stdio_server
app = Server("example-server")
async def main():
async with stdio_server() as streams:
await app.run(
streams[0],
streams[1],
app.create_initialization_options()
)
```
资料来源：[src/mcp/server/stdio.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/stdio.py)
### HTTP 模式
```python
from mcp.server import Server
from mcp.server.streamable_http import streamable_http_server
import uvicorn
app = Server("example-server")
@click.command()
@click.option("--port", default=8000, help="Port to listen on")
@click.option(
"--transport",
type=click.Choice(["stdio", "streamable-http"]),
default="stdio",
)
def main(port: int, transport: str) -> int:
if transport == "streamable-http":
uvicorn.run(
app.streamable_http_app(),
host="127.0.0.1",
port=port
)
else:
# stdio 模式
...
```
资料来源：[examples/servers/simple-pagination/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-pagination/server.py)
## 传输层安全性
MCP SDK 的传输层本身不实现加密，安全性依赖于底层传输协议：
| 传输类型 | 安全建议 |
|---------|---------|
| Stdio | 仅用于本地通信，使用文件系统权限控制 |
| Streamable HTTP | 配合 HTTPS/TLS 使用，添加认证中间件 |
| WebSocket | 使用 WSS (WebSocket Secure) |
| SSE | 通过 HTTPS 提供服务 |
## 常见问题
### Q: Stdio 和 HTTP 传输可以同时使用吗？
可以。服务器可以同时监听多种传输协议，通过不同的启动参数选择：
```bash
# Stdio 模式
python server.py --transport stdio
# HTTP 模式
python server.py --transport streamable-http --port 8000
```
### Q: 如何调试传输层问题？
1. 启用调试日志：`export MCP_LOG_LEVEL=debug`
2. 使用 HTTP 抓包工具查看请求/响应
3. 对于 stdio，使用 `script` 命令记录会话
## 相关资源
- [MCP 协议规范](https://modelcontextprotocol.io/specification)
- [Python SDK 源码仓库](https://github.com/modelcontextprotocol/python-sdk)
- [示例服务器集合](../servers/)
---
## FastMCP 服务器框架
### 相关页面
相关主题：[资源、资源模板、提示与工具](#page-resources-prompts-tools), [上下文与会话管理](#page-context-session)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
- [src/mcp/server/mcpserver/resources/base.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/base.py)
- [src/mcp/server/mcpserver/resources/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
- [src/mcp/cli/cli.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/cli/cli.py)
- [examples/snippets/servers/completion.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/completion.py)
- [examples/servers/simple-pagination/mcp_simple_pagination/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-pagination/mcp_simple_pagination/server.py)
- [examples/servers/simple-task-interactive/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-task-interactive/README.md)
- [examples/servers/simple-task/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-task/README.md)
- [examples/clients/simple-task-interactive-client/README.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-task-interactive-client/README.md)
# FastMCP 服务器框架
FastMCP 是 MCP (Model Context Protocol) Python SDK 中提供的高级服务器框架，旨在简化 MCP 服务器的开发流程。与底层 `Server` 类相比，FastMCP 提供了更简洁的 API 设计，让开发者能够以更少的代码量快速构建功能完整的 MCP 服务器。
## 框架概述
### 设计目标
FastMCP 框架的核心设计理念是**简化**与**约定优于配置**。它将 MCP 协议中复杂的内部机制抽象化，为开发者提供直观的装饰器和上下文管理器，使服务器开发变得更加直观和高效。
### 架构分层
```
┌─────────────────────────────────────────────────────────┐
│ FastMCP API 层 │
│ (装饰器 @tool, @resource, @prompt, 生命周期管理) │
├─────────────────────────────────────────────────────────┤
│ Server 核心层 │
│ (工具管理、资源管理、提示管理、认证) │
├─────────────────────────────────────────────────────────┤
│ 传输层 │
│ (stdio, streamable-http, SSE) │
├─────────────────────────────────────────────────────────┤
│ MCP 协议层 │
│ (JSON-RPC 消息处理、协议序列化) │
└─────────────────────────────────────────────────────────┘
```
资料来源：[src/mcp/server/mcpserver/server.py:1-150]()
## 核心组件
### 1. 服务器实例
FastMCP 服务器通过 `FastMCP` 类进行初始化和配置。该类封装了底层的 `Server` 实例，并提供了便捷的方法来注册工具、资源和提示。
**初始化参数：**
| 参数 | 类型 | 说明 | 默认值 |
|------|------|------|--------|
| `name` | `str` | 服务器名称 | `"fastmcp"` |
| `title` | `str \| None` | 服务器显示标题 | `None` |
| `description` | `str \| None` | 服务器描述 | `None` |
| `version` | `str \| None` | 服务器版本 | `None` |
| `lifespan` | `Lifespan \| None` | 生命周期上下文管理器 | `None` |
资料来源：[src/mcp/server/mcpserver/server.py:80-120]()
### 2. 工具系统 (Tools)
工具是 MCP 服务器中最核心的交互组件。FastMCP 提供了 `@tool` 装饰器来简化工具注册过程。
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
@mcp.tool()
async def get_weather(city: str) -> str:
"""获取城市天气信息"""
return f"{city} 的天气是晴天"
```
**工具装饰器参数：**
| 参数 | 类型 | 说明 |
|------|------|------|
| `name` | `str \| None` | 工具名称，默认使用函数名 |
| `description` | `str \| None` | 工具描述，来自函数 docstring |
| `annotations` | `Annotations \| None` | 工具注解（只读、幂等等） |
资料来源：[src/mcp/server/fastmcp/tools/base.py](), [src/mcp/server/fastmcp/tools/tool_manager.py]()
### 3. 资源系统 (Resources)
资源用于向客户端暴露可读取的数据。FastMCP 支持两种类型的资源：
#### 3.1 静态资源
```python
@mcp.resource("config://app")
def get_config() -> str:
"""返回应用配置"""
return '{"version": "1.0.0"}'
```
#### 3.2 资源模板
```python
@mcp.resource("users://{user_id}/profile")
async def get_user_profile(user_id: str) -> str:
"""返回用户资料"""
return f'{{"id": "{user_id}", "name": "User {user_id}"}}'
```
资料来源：[src/mcp/server/mcpserver/resources/base.py:1-50](), [src/mcp/server/mcpserver/resources/types.py]()
### 4. 提示系统 (Prompts)
提示用于定义可复用的 LLM 交互模板。
```python
@mcp.prompt()
def code_review_prompt(repo: str, pr: str) -> str:
"""代码审查提示模板"""
return f"请审查仓库 {repo} 的 PR #{pr} 中的代码变更"
```
资料来源：[src/mcp/server/fastmcp/prompts/base.py](), [src/mcp/server/fastmcp/prompts/manager.py]()
## 上下文管理
### Context 对象
每个工具和资源函数都可以接收一个 `Context` 对象作为第一个参数，该对象提供了访问服务器功能的能力。
```python
@mcp.tool()
async def search_files(
context: Context,
pattern: str,
path: str = "."
) -> list[str]:
# 访问资源管理器
resources = await context.list_resources()
# 创建子任务
task = await context.create_task("another_tool", {"arg": "value"})
return results
```
资料来源：[src/mcp/server/fastmcp/context.py]()
### 生命周期管理
FastMCP 支持通过 `lifespan` 参数管理服务器启动和关闭时的资源。
```python
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP
@asynccontextmanager
async def lifespan(server: FastMCP):
# 启动时初始化资源
await db.connect()
await cache.initialize()
yield
# 关闭时清理资源
await cache.close()
await db.disconnect()
mcp = FastMCP("my-server", lifespan=lifespan)
```
资料来源：[examples/snippets/servers/lifespan_example.py](), [src/mcp/server/mcpserver/server.py:100-110]()
## 异常处理
FastMCP 定义了框架专用的异常类型，便于细粒度的错误处理。
| 异常类型 | 说明 |
|----------|------|
| `ResourceError` | 资源访问错误 |
| `ToolError` | 工具执行错误 |
| `PromptError` | 提示处理错误 |
| `ValidationError` | 参数验证错误 |
资料来源：[src/mcp/server/fastmcp/exceptions.py]()
## 运行方式
### 标准输入输出模式 (Stdio)
适用于命令行工具和 Claude Desktop 集成：
```python
# my_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
@mcp.tool()
async def hello(name: str) -> str:
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run()
```
运行命令：
```bash
mcp run my_server.py
```
资料来源：[examples/servers/simple-pagination/mcp_simple_pagination/server.py:40-60]()
### HTTP 传输模式
适用于 Web 环境，支持流式响应：
```python
import uvicorn
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
# ... 定义工具和资源 ...
if __name__ == "__main__":
uvicorn.run(mcp.streamable_http_app(), host="0.0.0.0", port=8000)
```
资料来源：[examples/servers/simple-pagination/mcp_simple_pagination/server.py:45-55]()
## 任务系统 (Experimental)
FastMCP 支持实验性的异步任务功能，允许长时间运行的工具返回任务引用而非直接结果。
```python
@mcp.tool()
async def long_running_task(context: Context) -> str:
# 创建任务
task = await context.create_task()
# 发送进度更新
await task.send_progress("开始处理...")
# 模拟长时间操作
await asyncio.sleep(2)
# 完成任务
return "任务完成"
```
使用 `call_tool_as_task` 方法调用：
```python
result = await session.experimental.call_tool_as_task("long_running_task", {})
task_id = result.task.task_id
# 轮询任务状态
final = await session.experimental.get_task_result(task_id, CallToolResult)
```
资料来源：[examples/servers/simple-task/README.md](), [examples/clients/simple-task-interactive-client/README.md]()
## 高级功能
### 资源订阅与通知
FastMCP 支持资源变更通知机制，服务器可以主动推送更新给客户端：
```python
@mcp.resource("data://stream")
async def streaming_data(context: Context):
async for update in data_generator():
# 发送资源更新通知
await context.notify_resources_changed()
yield update
```
### 交互式任务 (Elicitation)
服务器可以请求用户输入确认：
```python
@mcp.tool()
async def confirm_delete(context: Context, filename: str) -> str:
# 请求用户确认
result = await context.elicit(
message=f"确认删除 {filename}?",
actions=[{"label": "确认", "value": "yes"}, {"label": "取消", "value": "no"}]
)
if result.action == "accept":
return f"已删除 {filename}"
return "操作已取消"
```
资料来源：[examples/servers/simple-task-interactive/README.md]()
### 采样 (Sampling)
服务器可以请求 LLM 生成内容：
```python
@mcp.tool()
async def generate_text(context: Context, topic: str) -> str:
result = await context.sample(
prompt=f"写一首关于 {topic} 的诗",
system="你是一位诗人"
)
return result.content
```
资料来源：[examples/clients/simple-task-interactive-client/README.md]()
## 与底层 Server 类的关系
FastMCP 是对底层 `Server` 类的高级封装。两者的主要区别：
| 特性 | `Server` | `FastMCP` |
|------|----------|-----------|
| API 复杂度 | 高 | 低 |
| 代码量 | 较多 | 简洁 |
| 灵活性 | 高 | 中等 |
| 装饰器支持 | 无 | 有 |
| 适用场景 | 深度定制 | 快速开发 |
资料来源：[src/mcp/server/mcpserver/server.py:80-150]()
## 最佳实践
### 1. 类型注解
所有公共 API 必须包含类型注解：
```python
@mcp.tool()
async def process_data(context: Context, data: list[str], options: dict[str, Any] | None = None) -> dict[str, Any]:
"""处理数据并返回结果"""
...
```
### 2. 错误处理
使用框架定义的异常类型，并在文档中说明可能抛出的异常：
```python
@mcp.tool()
async def read_resource(context: Context, uri: str) -> str:
"""读取资源内容
Raises:
ResourceError: 资源不存在或无法访问
"""
try:
return await context.read_resource(uri)
except Exception:
logger.exception(f"Error reading resource {uri}")
raise ResourceError(f"Error reading resource {uri}")
```
### 3. 资源管理
在 `lifespan` 中正确管理外部资源连接：
```python
@asynccontextmanager
async def lifespan(server: FastMCP):
# 使用 try/finally 确保清理
connection = await create_connection()
try:
server._connection = connection
yield
finally:
await connection.close()
```
资料来源：[AGENTS.md - Exception Handling]()
---
## 资源、资源模板、提示与工具
### 相关页面
相关主题：[FastMCP 服务器框架](#page-fastmcp)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/server/mcpserver/resources/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/types.py)
- [src/mcp/server/mcpserver/resources/templates.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/templates.py)
- [src/mcp/server/mcpserver/utilities/func_metadata.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/fastmcp/utilities/func_metadata.py)
- [examples/snippets/servers/basic_tool.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_tool.py)
- [examples/snippets/servers/basic_resource.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_resource.py)
- [examples/snippets/servers/basic_prompt.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_prompt.py)
- [examples/mcpserver/weather_structured.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/mcpserver/weather_structured.py)
- [examples/servers/simple-prompt/mcp_simple_prompt/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-prompt/mcp_simple_prompt/server.py)
# 资源、资源模板、提示与工具
## 概述
MCP SDK 提供了四类核心功能用于构建 MCP 服务器：**资源（Resources）**、**资源模板（Resource Templates）**、**提示（Prompts）** 和 **工具（Tools）**。这些功能使服务器能够向 LLM 客户端暴露数据、模板和可执行函数，从而实现丰富的工具调用和数据检索能力。
---
## 资源（Resources）
### 定义与作用
资源是 MCP 协议中用于向 LLM 客户端暴露数据的机制。资源可以是静态内容（如文件）或动态生成的内容（如函数返回值）。客户端可以通过 URI 来读取特定的资源内容。
### 资源类型
MCP SDK 支持两种主要的资源类型：
| 类型 | 说明 | 适用场景 |
|------|------|----------|
| `FunctionResource` | 由函数动态生成内容 | API 响应、实时数据 |
| `FileResource` | 读取本地文件内容 | 配置文件、文档、静态资源 |
### FunctionResource
`FunctionResource` 允许将函数作为资源暴露给客户端。服务器调用该函数并返回其结果作为资源内容。
**创建方式：**
```python
from mcp.server.mcpserver.resources.types import FunctionResource
resource = FunctionResource.from_function(
fn=my_function,
uri="myapp://data",
name="my_data",
title="我的数据",
description="获取应用程序数据",
mime_type="application/json"
)
```
**关键特性：**
- 自动类型验证：`from_function` 方法内部使用 `validate_call` 验证函数参数 资料来源：[src/mcp/server/mcpserver/resources/types.py:40]()
### FileResource
`FileResource` 用于读取本地文件系统中的文件。路径必须为绝对路径。
```python
from mcp.server.mcpserver.resources.types import FileResource
from pathlib import Path
resource = FileResource(
path=Path("/etc/app/config.json"),
name="config",
mime_type="application/json"
)
```
**路径验证：**
FileResource 使用 Pydantic 验证器确保路径是绝对路径：
```python
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:
"""确保路径是绝对路径"""
if not path.is_absolute():
raise ValueError("Path must be absolute")
return path
```
资料来源：[src/mcp/server/mcpserver/resources/types.py:65-72]()
### 注册资源到服务器
资源通过装饰器或直接方法注册到服务器：
```python
@server.resource(
uri="myapp://config",
name="config",
title="应用程序配置",
description="获取当前配置"
)
def get_config():
return {"setting": "value"}
```
---
## 资源模板（Resource Templates）
### 概念
资源模板是带有 URI 参数的资源，允许动态生成资源内容。它们使用 URI 模板语法（如 `github://repos/{owner}/{repo}`）来定义参数化的资源端点。
### 工作原理
```mermaid
graph TD
A[客户端请求] --> B[解析 URI 模板]
B --> C{参数验证}
C -->|通过| D[调用处理函数]
C -->|失败| E[返回错误]
D --> F[返回资源内容]
```
### URI 模板与参数验证
资源模板通过比对 URI 参数和函数参数来确保正确性：
```python
uri_params = extract_uri_params(uri)
func_params = extract_func_params(fn)
if uri_params != func_params:
raise ValueError(
f"Mismatch between URI parameters {uri_params} and function parameters {func_params}"
)
```
资料来源：[src/mcp/server/mcpserver/server.py:95-100]()
### 注册资源模板
```python
@server.list_resource_templates
async def handle_list_resource_templates():
return [
types.ResourceTemplate(
uri_template="github://repos/{owner}/{repo}",
name="github_repo",
title="GitHub 仓库",
description="获取 GitHub 仓库信息",
arguments=[
types.PromptArgument(
name="owner",
description="仓库所有者",
required=True,
),
types.PromptArgument(
name="repo",
description="仓库名称",
required=True,
),
],
)
]
```
### 模板资源示例
在补全示例中，模板资源可以提供上下文感知的自动补全：
```python
# 基于 GitHub 资源模板的补全示例
if isinstance(ref, ResourceTemplateReference):
if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
if context and context.arguments.get("owner") == "modelcontextprotocol":
repos = ["python-sdk", "typescript-sdk", "specification"]
return Completion(values=repos, has_more=False)
```
资料来源：[examples/snippets/servers/completion.py:10-18]()
---
## 提示（Prompts）
### 定义
提示是预定义的模板，可包含消息和参数。它们允许服务器定义可重用的提示模式，客户端可以根据需要调用。
### 提示结构
```mermaid
graph LR
A[Prompt 定义] --> B[名称和描述]
A --> C[参数列表]
A --> D[消息模板]
C --> C1[必需参数]
C --> C2[可选参数]
D --> D1[角色]
D --> D2[内容]
```
### 提示注册
使用装饰器方式注册提示：
```python
@server.prompt(
name="simple",
title="简单提示",
description="一个基础提示示例"
)
def simple_prompt(context: str = None, topic: str = None):
return {
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": f"请帮我{context or ''}关于{topic or '通用'}的内容"
}
}
]
}
```
资料来源：[examples/snippets/servers/basic_prompt.py:8-30]()
### 提示处理器
对于更复杂的提示逻辑，可以使用处理器函数：
```python
async def handle_list_prompts() -> list[types.ListPromptsResult]:
return [
types.ListPromptsResult(
name="simple",
description="一个简单提示",
arguments=[
types.PromptArgument(
name="context",
description="额外上下文",
required=False,
),
types.PromptArgument(
name="topic",
description="关注的主题",
required=False,
),
],
)
]
async def handle_get_prompt(
ctx: ServerRequestContext,
params: types.GetPromptRequestParams
) -> types.GetPromptResult:
if params.name != "simple":
raise ValueError(f"Unknown prompt: {params.name}")
arguments = params.arguments or {}
return types.GetPromptResult(
messages=create_messages(
context=arguments.get("context"),
topic=arguments.get("topic")
),
description="一个带有可选参数的简单提示",
)
```
资料来源：[examples/servers/simple-prompt/mcp_simple_prompt/server.py:28-58]()
### 使用 Prompt 类
也可以直接实例化 Prompt 类：
```python
from mcp.types import Prompt
prompt = Prompt(
name="my_prompt",
description="我的提示",
arguments=[
PromptArgument(
name="param1",
description="参数说明",
required=True
)
]
)
server.add_prompt(prompt)
```
---
## 工具（Tools）
### 概述
工具是服务器暴露给 LLM 客户端的可执行函数。客户端可以调用工具来执行服务器上的操作。
### 工具注册
使用 `@server.tool()` 装饰器注册工具：
```python
@server.tool(
name="get_weather",
description="获取指定城市的天气信息",
)
async def get_weather(city: str, country: str | None = None) -> str:
"""获取天气信息"""
return f"城市 {city} 的天气是晴天"
```
资料来源：[examples/snippets/servers/basic_tool.py:1-12]()
### 工具方法
#### add_tool()
直接方法添加工具：
```python
server.add_tool(
fn=my_function,
name="tool_name",
title="工具标题",
description="工具描述",
annotations={"readOnlyHint": True}
)
```
#### tool() 装饰器
装饰器方式添加工具，支持更多配置：
```python
@server.tool(
name="custom_tool",
title="自定义工具",
description="带参数的工具",
annotations={
"readOnlyHint": True,
"openWorldHint": False
}
)
async def custom_tool(param1: str, param2: int = 10):
return {"result": f"处理了 {param1} 和 {param2}"}
```
### 工具注解（Annotations）
工具可以附带注解来提供额外元数据：
| 注解 | 类型 | 说明 |
|------|------|------|
| `readOnlyHint` | bool | 指示工具是否只读 |
| `openWorldHint` | bool | 指示工具是否访问外部世界 |
| `idempotentHint` | bool | 指示工具是否幂等 |
| `interactiveHint` | bool | 指示工具是否需要交互 |
### 工具返回类型
工具可以返回多种格式的内容：
```python
@server.tool()
async def structured_tool():
"""返回结构化内容"""
return {
"content": [
{
"type": "text",
"text": "文本内容"
},
{
"type": "resource",
"resource": {
"uri": "myapp://data",
"mimeType": "application/json",
"text": '{"key": "value"}'
}
}
],
"isError": False
}
```
资料来源：[examples/mcpserver/weather_structured.py:20-45]()
### 函数元数据提取
工具注册时会自动提取函数元数据：
```python
fn = validate_call(fn)
func_name = name or fn.__name__
```
关键元数据包括：
- **函数名称**：用于标识工具
- **参数信息**：名称、类型、默认值、文档字符串
- **返回类型注解**：用于文档和验证
---
## 完整示例
### 集成资源、提示和工具
```python
from mcp.server import Server
from mcp.types import Resource, Tool, Prompt
# 创建服务器实例
app = Server("my-mcp-server")
# 注册资源
@app.resource(uri="myapp://config", name="config")
def get_config():
return {"version": "1.0", "debug": True}
# 注册资源模板
@app.list_resource_templates
async def handle_list_templates():
return [
types.ResourceTemplate(
uri_template="myapp://users/{user_id}",
name="user_profile",
description="获取用户资料",
arguments=[
types.PromptArgument(name="user_id", description="用户ID", required=True)
],
)
]
# 注册提示
@app.prompt(name="greet", title="问候语")
def greet_prompt(name: str):
return {
"messages": [
{"role": "user", "content": {"type": "text", "text": f"你好，{name}！"}}
]
}
# 注册工具
@app.tool(name="calculate", description="执行计算")
def calculate(a: int, b: int, operation: str = "add"):
if operation == "add":
return a + b
return a - b
# 运行服务器
async def main():
from mcp.server.stdio import stdio_server
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
```
---
## 架构关系图
```mermaid
graph TD
subgraph MCP服务器
A[Server 实例]
subgraph 资源层
B[Resources 资源]
C[Resource Templates 资源模板]
end
subgraph 交互层
D[Prompts 提示]
E[Tools 工具]
end
end
F[LLM 客户端] --> |读取资源| B
F --> |读取模板| C
F --> |调用提示| D
F --> |执行工具| E
B --> |内容| F
C --> |动态内容| F
D --> |消息| F
E --> |结果| F
G[函数] --> |FunctionResource| B
H[文件] --> |FileResource| B
```
---
## 最佳实践
### 资源设计
1. **使用有意义的 URI**：资源 URI 应清晰表达资源的用途
2. **选择正确的 MIME 类型**：确保客户端正确解析资源内容
3. **提供描述**：为每个资源添加清晰的描述
### 提示设计
1. **保持简洁**：提示应清晰、聚焦
2. **提供默认值**：可选参数应有合理的默认值
3. **文档完整**：说明每个参数的用途
### 工具设计
1. **幂等性**：优先设计幂等工具
2. **错误处理**：使用适当的异常处理，返回有意义的错误信息
3. **类型注解**：为所有参数和返回值提供类型注解
4. **使用注解**：通过注解描述工具行为
---
## 错误处理
### 资源错误
```python
async def read_resource(self, uri: AnyUrl | str, context: Context | None = None):
try:
resource = await self._resource_manager.get_resource(uri, context)
except ValueError as exc:
raise ResourceError(f"Unknown resource: {uri}") from exc
try:
content = await resource.read()
return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
except Exception as exc:
logger.exception(f"Error getting resource {uri}")
raise ResourceError(f"Error reading resource {uri}") from exc
```
资料来源：[src/mcp/server/mcpserver/server.py:54-66]()
### 关键原则
- **捕获具体异常**：使用 `except (OSError, PermissionError)` 等具体异常类型
- **使用 logger.exception()**：记录异常时不包含异常对象在消息中
- **避免裸 except**：除非在顶级处理器中
- **转换异常**：将内部异常转换为客户端可理解的错误
---
## 上下文与会话管理
### 相关页面
相关主题：[FastMCP 服务器框架](#page-fastmcp), [系统架构与核心组件](#page-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/server/context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/context.py)
- [src/mcp/client/context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/context.py)
- [src/mcp/shared/_context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/_context.py)
- [src/mcp/server/fastmcp/utilities/context_injection.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/fastmcp/utilities/context_injection.py)
- [src/mcp/server/fastmcp/utilities/types.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/fastmcp/utilities/types.py)
- [examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/tool_progress.py)
# 上下文与会话管理
## 概述
在 Model Context Protocol (MCP) SDK 中，**上下文（Context）** 和 **会话管理** 是连接服务器端与客户端、传递请求元数据、管理生命周期状态的核心机制。上下文对象在请求处理链中携带关键信息，使开发者能够在工具执行、资源访问、提示渲染等场景中访问请求级别的元数据和服务器状态。
上下文管理系统的设计遵循以下原则：
- **请求级别隔离**：每个请求都有独立的上下文实例
- **类型安全**：通过泛型参数确保上下文类型的一致性
- **懒加载初始化**：在需要时才创建完整的上下文对象
- **与生命周期集成**：上下文与服务器的 lifespan 机制协同工作
资料来源：[src/mcp/shared/_context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/_context.py)
---
## 架构设计
### 上下文分层模型
MCP SDK 的上下文系统采用分层架构，不同层次的上下文服务于不同的功能范围：
```mermaid
graph TB
subgraph "服务端上下文"
S1[ServerContext]
S2[RequestContext]
S3[ToolContext]
end
subgraph "客户端上下文"
C1[ClientContext]
C2[SessionContext]
end
subgraph "共享上下文"
SH[SharedContext]
end
S1 --> S2
S2 --> S3
C1 --> C2
S1 -.->|共享基类| SH
C1 -.->|共享基类| SH
```
### 核心组件表
| 组件 | 文件位置 | 职责 |
|------|---------|------|
| `Context` | `src/mcp/shared/_context.py` | 共享上下文基类，提供通用上下文功能 |
| `ServerContext` | `src/mcp/server/context.py` | 服务端上下文，包含服务器引用和请求信息 |
| `ClientContext` | `src/mcp/client/context.py` | 客户端上下文，管理会话状态和请求追踪 |
| `ContextInjector` | `src/mcp/server/fastmcp/utilities/context_injection.py` | 上下文注入工具，用于 FastMCP 简化开发 |
| `ProgressToken` | `相关类型定义 | 支持进度回调和长时间运行任务 |
资料来源：[src/mcp/shared/_context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/_context.py)、[src/mcp/server/context.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/context.py)
---
## 服务端上下文
### ServerContext 结构
服务端上下文在 `src/mcp/server/context.py` 中定义，包含以下核心属性：
| 属性 | 类型 | 说明 |
|------|------|------|
| `mcp_server` | `MCPServer` | 对应的 MCP 服务器实例 |
| `request_id` | `str \| None` | 当前请求的唯一标识符 |
| `session_id` | `str \| None` | 客户端会话标识符 |
| `client_info` | `ClientInfo \| None` | 客户端信息（元数据） |
| `capabilities` | `ServerCapabilities \| None` | 客户端支持的能力列表 |
### 上下文创建与管理
在服务端请求处理中，上下文通过以下模式创建：
```python
async def read_resource(
self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
"""通过 URI 读取资源。"""
if context is None:
context = Context(mcp_server=self)
try:
resource = await self._resource_manager.get_resource(uri, context)
except ValueError as exc:
raise ResourceError(f"Unknown resource: {uri}") from exc
```
资料来源：[src/mcp/server/mcpserver/server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/server.py)
### 生命周期集成
上下文与服务器的 lifespan 机制深度集成，支持泛型类型参数 `LifespanResultT`：
```python
Context[LifespanResultT, Any]
```
这种设计允许：
- 在上下文初始化阶段注入特定的生命周期数据
- 在请求处理过程中访问生命周期状态
- 类型安全的上下文传递
---
## 客户端上下文
### ClientContext 结构
客户端上下文管理会话状态和请求追踪：
| 属性 | 类型 | 说明 |
|------|------|------|
| `session` | `ClientSession` | 活跃的客户端会话 |
| `request_id` | `str` | 当前请求的唯一标识符 |
| `timeout` | `timedelta \| None` | 请求超时配置 |
### 会话管理流程
```mermaid
sequenceDiagram
participant Client as 客户端
participant Session as ClientSession
participant Context as ClientContext
participant Server as MCP Server
Client->>Session: 建立连接
Session->>Context: 创建上下文
Context->>Server: 初始化请求
Server-->>Context: 初始化响应
Context-->>Session: 会话就绪
Session-->>Client: 连接就绪
loop 请求处理
Client->>Context: 发起请求
Context->>Server: 发送请求
Server-->>Context: 返回结果
Context-->>Client: 传递结果
end
```
---
## 上下文注入机制
### ContextInjector 用法
FastMCP 提供了 `ContextInjector` 工具类，用于在函数调用时自动注入上下文：
```python
from mcp.server.fastmcp.utilities.context_injection import ContextInjector
# 定义注入器
injector = ContextInjector()
# 在工具函数中使用
@injector.inject
async def my_tool(ctx: Context, arg1: str, arg2: int):
"""带上下文注入的工具函数。"""
# 直接使用 ctx，无需手动传递
await ctx.info(f"Processing {arg1} with {arg2}")
```
### 注入工作流程
```mermaid
graph LR
A[工具调用] --> B[ContextInjector]
B --> C{检查上下文参数}
C -->|已存在| D[使用现有上下文]
C -->|不存在| E[创建新上下文]
D --> F[执行业务逻辑]
E --> F
F --> G[返回结果]
```
资料来源：[src/mcp/server/fastmcp/utilities/context_injection.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/fastmcp/utilities/context_injection.py)
---
## 进度令牌与长时间运行任务
### ProgressToken 支持
MCP SDK 通过 `ProgressToken` 支持长时间运行任务的进度报告：
| 参数 | 类型 | 说明 |
|------|------|------|
| `token` | `str \| NumberToken` | 进度令牌标识符 |
| `progress` | `float \| None` | 当前进度值（0-100） |
| `total` | `float \| None` | 任务总量 |
### 工具进度示例
```python
async def handle_call_tool(
ctx: ServerRequestContext, params: types.CallToolRequestParams
) -> types.CallToolResult:
"""处理工具调用，支持进度报告。"""
progress_token = ctx.progress_token
if progress_token:
await progress_token(
progress=0,
total=100,
)
# 模拟长时间任务
for i in range(10):
await progress_token(progress=(i + 1) * 10)
await asyncio.sleep(0.1)
return types.CallToolResult(
content=[types.TextContent(type="text", text="Task completed")]
)
```
资料来源：[examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/tool_progress.py)
### 进度报告时序图
```mermaid
sequenceDiagram
participant Client as 客户端
participant Server as MCP 服务器
participant Tool as 工具函数
Client->>Server: call_tool with progressToken
Server->>Tool: 初始化任务
Tool->>Tool: 步骤 1
Tool->>Client: progress(25)
Tool->>Tool: 步骤 2
Tool->>Client: progress(50)
Tool->>Tool: 步骤 3
Tool->>Client: progress(75)
Tool->>Tool: 步骤 4
Tool->>Client: progress(100)
Tool->>Client: 返回结果
```
---
## 快速参考
### 上下文获取模式
| 场景 | 推荐方式 |
|------|---------|
| 服务端请求处理 | 通过函数参数接收 `Context` |
| 资源读取 | `context = Context(mcp_server=self)` |
| FastMCP 工具 | 使用 `@injector.inject` 装饰器 |
| 客户端调用 | 通过 `ClientSession` 的上下文方法 |
### 常见错误处理
| 错误类型 | 处理方式 |
|---------|---------|
| `ValueError: Unknown resource` | 检查 URI 格式是否正确 |
| `ResourceError` | 资源读取失败，查看日志详情 |
| `Context not available` | 确保在正确的请求上下文中调用 |
### 日志记录规范
根据项目规范，异常处理必须使用 `logger.exception()`：
```python
except Exception as exc:
logger.exception(f"Error getting resource {uri}")
raise ResourceError(f"Error reading resource {uri}") from exc
```
资料来源：[AGENTS.md](https://github.com/modelcontextprotocol/python-sdk/blob/main/AGENTS.md)
---
## 最佳实践
1. **始终传递上下文**：在工具函数签名中包含 `Context` 参数，以便访问服务器状态和元数据
2. **避免硬编码依赖**：使用上下文提供的接口而非直接引用全局状态
3. **正确处理异步**：上下文操作可能涉及异步 I/O，确保在 async 函数中使用
4. **利用进度报告**：对于耗时的工具调用，使用 `progress_token` 提供用户反馈
5. **类型安全**：使用泛型参数确保上下文类型的编译时检查
---
## 相关资源
- [MCP 协议规范](https://modelcontextprotocol.io/specification)
- [FastMCP 文档](./fastmcp.md)
- [服务器开发指南](./servers.md)
- [客户端开发指南](./clients.md)
---
## 客户端开发指南
### 相关页面
相关主题：[客户端 OAuth 认证](#page-client-auth), [传输协议实现](#page-transports)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/client/client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/client.py)
- [src/mcp/client/session.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/session.py)
- [src/mcp/client/session_group.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/session_group.py)
- [src/mcp/shared/response_router.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/response_router.py)
- [examples/snippets/clients/completion_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/completion_client.py)
- [examples/snippets/clients/stdio_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/stdio_client.py)
- [examples/snippets/clients/parsing_tool_results.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/parsing_tool_results.py)
# 客户端开发指南
本指南详细介绍如何使用 Model Context Protocol (MCP) Python SDK 开发客户端应用程序。MCP 客户端负责连接到 MCP 服务器，发现服务器提供的功能（工具、资源、提示），并通过标准化的协议与服务器进行交互。
## 核心概念
### 客户端架构概述
MCP 客户端采用分层架构设计，主要包含以下组件：
| 组件 | 职责 | 源码位置 |
|------|------|----------|
| `Client` | 连接管理和生命周期控制 | `src/mcp/client/client.py` |
| `ClientSession` | 会话状态管理和请求处理 | `src/mcp/client/session.py` |
| `SessionGroup` | 多会话协调管理 | `src/mcp/client/session_group.py` |
| `ResponseRouter` | 响应路由和请求-响应匹配 | `src/mcp/shared/response_router.py` |
### 传输层支持
MCP SDK 支持多种传输方式，客户端可根据场景选择合适的传输协议：
```mermaid
graph TD
A[MCP 客户端] --> B{传输方式选择}
B --> C[stdio 传输]
B --> D[streamable-http 传输]
B --> E[SSE 传输]
C --> F[标准输入输出通信]
D --> G[HTTP 流式传输]
E --> H[Server-Sent Events]
```
## 连接管理
### 建立 stdio 连接
标准输入输出传输适用于本地进程间通信，是最常用的连接方式之一。
```python
# 资料来源：examples/snippets/clients/stdio_client.py
import asyncio
from mcp.client.stdio import stdio_client
from mcp.client.client import Client
async def main():
async with stdio_client() as (read, write):
async with Client(
read,
write,
client_protocol,
asyncio.create_task
) as client:
await client.initialize()
# 使用客户端功能
```
### 建立 HTTP 连接
对于需要远程访问的场景，可使用 streamable-http 传输方式：
```python
# 连接到远程 MCP 服务器
from mcp.client.streamable_http import streamable_http_client
async def connect_to_remote():
async with streamable_http_client(
"http://localhost:8000/mcp",
timeout=30
) as (read, write, get_next_message):
# 处理连接
```
### 客户端初始化流程
```mermaid
sequenceDiagram
participant 客户端 as MCP 客户端
participant 服务器 as MCP 服务器
客户端->>服务器: 发送 Initialize 请求
客户端->>服务器: 包含协议版本和客户端能力
服务器-->>客户端: 返回服务器信息
服务器-->>客户端: 协议版本确认
服务器-->>客户端: 服务器能力集
客户端->>服务器: 发送 Initialized 通知
Note over 客户端: 连接建立完成
```
初始化时客户端需要指定支持的协议版本和自身能力集，服务器确认后将返回其能力信息。
## 会话管理
### ClientSession 的核心职责
`ClientSession` 类是客户端与服务器交互的核心抽象，负责管理会话状态和协调各类请求：
```python
# 资料来源：src/mcp/client/session.py
class ClientSession:
"""MCP 客户端会话核心类"""
async def initialize(self) -> InitializeResult:
"""初始化与服务器的连接"""
async def list_tools(self) -> list[Tool]:
"""列出服务器提供的所有工具"""
async def call_tool(
self,
name: str,
arguments: dict[str, Any] | None = None
) -> CallToolResult:
"""调用服务器上的工具"""
async def list_resources(self) -> list[Resource]:
"""列出服务器提供的资源"""
async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
"""读取指定资源的内容"""
async def list_prompts(self) -> list[Prompt]:
"""列出服务器提供的提示模板"""
async def get_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None
) -> GetPromptResult:
"""获取并渲染提示模板"""
```
### 调用工具
工具调用是 MCP 协议最核心的功能之一，允许客户端请求服务器执行特定操作：
```python
# 资料来源：examples/snippets/clients/stdio_client.py
async def call_server_tool():
result = await session.call_tool(
"get_time",
arguments={"timezone": "UTC"}
)
# 处理返回结果
for content in result.content:
if content.type == "text":
print(f"工具返回: {content.text}")
```
### 解析工具返回结果
工具调用可能返回多种类型的内容，客户端需要正确处理：
```python
# 资料来源：examples/snippets/clients/parsing_tool_results.py
from mcp.types import TextContent, ImageContent, EmbeddedResource
def parse_tool_result(result):
"""解析工具调用结果"""
for content in result.content:
# 处理文本内容
if isinstance(content, TextContent):
print(f"文本: {content.text}")
# 处理图片内容
elif isinstance(content, ImageContent):
print(f"图片 MIME 类型: {content.mimeType}")
# content.data 包含 base64 编码的图片数据
# 处理嵌入资源
elif isinstance(content, EmbeddedResource):
print(f"嵌入资源: {content.resource.uri}")
```
### 资源管理
#### 列出可用资源
```python
async def list_available_resources():
resources = await session.list_resources()
for resource in resources:
print(f"资源: {resource.name}")
print(f" URI: {resource.uri}")
print(f" 描述: {resource.description}")
```
#### 读取资源内容
```python
async def read_resource_content(uri: str):
contents = await session.read_resource(uri)
for content in contents:
print(f"内容类型: {content.mimeType}")
print(f"数据: {content.content}")
```
### 提示模板
服务器可以定义可复用的提示模板，客户端可以获取并使用这些模板：
```python
# 资料来源：examples/snippets/clients/completion_client.py
async def use_prompt_template():
# 列出所有可用提示
prompts = await session.list_prompts()
# 获取特定提示
result = await session.get_prompt(
"simple",
arguments={"context": "额外上下文", "topic": "技术主题"}
)
# result.messages 包含渲染后的消息列表
for message in result.messages:
print(f"角色: {message.role}")
print(f"内容: {message.content}")
```
## 自动补全支持
MCP 支持在客户端实现自动补全功能，提升用户体验：
```python
# 资料来源：examples/snippets/servers/completion.py
from mcp.types import ResourceTemplateReference, Completion
async def request_completion(ref: ResourceTemplateReference, argument: str):
"""请求服务器提供补全建议"""
# 对于 GitHub 资源模板的特定参数提供补全
if ref.uri == "github://repos/{owner}/{repo}":
if argument == "repo":
# 返回预定义的补全值
return Completion(
values=["python-sdk", "typescript-sdk", "specification"],
has_more=False
)
```
## 响应路由机制
`ResponseRouter` 负责管理异步请求与响应的匹配，确保请求能够正确路由到对应的处理程序：
```mermaid
graph LR
A[请求发送] --> B[RequestTracker]
B --> C{匹配响应}
C -->|找到匹配| D[触发回调]
C -->|未找到| E[等待响应]
D --> F[返回结果]
E --> C
```
```python
# 资料来源：src/mcp/shared/response_router.py
class ResponseRouter:
"""响应路由器，管理请求-响应映射"""
async def send_request(self, method: str, params: dict) -> Any:
"""发送请求并等待响应"""
def __aiter__(self):
"""异步迭代器，用于接收传入的消息"""
```
## 多会话管理
### SessionGroup 的使用
当客户端需要同时管理多个服务器连接时，可使用 `SessionGroup` 进行统一管理：
```python
# 资料来源：src/mcp/client/session_group.py
from mcp.client.session_group import SessionGroup
async def manage_multiple_sessions():
async with SessionGroup() as group:
# 添加多个会话
session1 = await group.add_server(stdio_server())
session2 = await group.add_server(http_server("http://server2/mcp"))
# 批量操作
all_tools = await group.request(
lambda s: s.list_tools(),
is_error_ok=True
)
```
### SessionGroup 主要方法
| 方法 | 功能 | 源码位置 |
|------|------|----------|
| `add_server()` | 添加新服务器会话 | `src/mcp/client/session_group.py` |
| `remove_session()` | 移除指定会话 | `src/mcp/client/session_group.py` |
| `request()` | 向所有会话发送请求 | `src/mcp/client/session_group.py` |
| `notify()` | 向所有会话发送通知 | `src/mcp/client/session_group.py` |
## 错误处理
### 异常处理规范
根据项目开发规范，客户端代码应遵循以下异常处理原则：
```python
# 资料来源：AGENTS.md
import logging
logger = logging.getLogger(__name__)
try:
result = await session.call_tool("tool_name", {})
except Exception as exc:
# 使用 logger.exception() 而非 logger.error()
logger.exception("工具调用失败")
# 不要在日志消息中包含异常详情
```
### 推荐实践
| 场景 | 异常类型 | 处理方式 |
|------|----------|----------|
| JSON 解析 | `json.JSONDecodeError` | 捕获并返回有意义的错误信息 |
| 网络错误 | `ConnectionError`, `TimeoutError` | 重试或降级处理 |
| 资源不存在 | `ValueError` | 捕获并提示用户 |
| 授权失败 | `PermissionError` | 清理会话并重新认证 |
**禁止使用裸的 `except Exception:`** — 应始终捕获具体异常类型。
## 认证集成
### OAuth 认证流程
对于需要认证的 MCP 服务器，客户端需要实现完整的 OAuth 流程：
```python
# 资料来源：examples/clients/simple-auth-client/mcp_simple_auth_client/main.py
class CallbackHandler(BaseHTTPRequestHandler):
"""处理 OAuth 回调"""
def do_GET(self):
query_params = parse_qs(urlparse(self.path).query)
if "code" in query_params and "state" in query_params:
# 获取授权码
self.callback_data["authorization_code"] = query_params["code"][0]
self.callback_data["state"] = query_params["state"][0]
self.send_response(200)
elif "error" in query_params:
# 处理错误
self.callback_data["error"] = query_params["error"][0]
self.send_response(400)
```
### 环境配置
| 环境变量 | 描述 | 默认值 |
|----------|------|--------|
| `MCP_SERVER_PORT` | MCP 服务器端口 | `8000` |
| `MCP_TRANSPORT_TYPE` | 传输类型 | `streamable-http` |
| `MCP_CLIENT_METADATA_URL` | 客户端元数据地址 | `None` |
## 完整示例
以下是一个功能完整的 MCP 客户端实现示例：
```python
# 资料来源：examples/snippets/clients/stdio_client.py
import asyncio
from mcp.client.stdio import stdio_client
from mcp.client.client import Client
from mcp.protocol import ClientProtocol
async def run():
async with stdio_client() as (read, write):
client_protocol = ClientProtocol(
read,
write,
asyncio.create_task
)
async with Client(
read,
write,
client_protocol,
asyncio.create_task
) as client:
session = client_session = await client.create_session()
await session.initialize()
# 列出并调用工具
tools = await session.list_tools()
print(f"可用工具: {[t.name for t in tools]}")
# 调用工具并处理结果
result = await session.call_tool("get_time", {"a": 1, "b": 3})
# 解析返回内容
result_unstructured = result.content[0]
if hasattr(result_unstructured, 'text'):
print(f"工具结果: {result_unstructured.text}")
def main():
asyncio.run(run())
if __name__ == "__main__":
main()
```
## 开发工作流
### 本地开发设置
```bash
# 克隆仓库
git clone https://github.com/modelcontextprotocol/python-sdk.git
# 安装依赖
uv sync --frozen --all-extras --dev
# 安装预提交钩子
uv tool install pre-commit --with pre-commit-uv --force-reinstall
# 运行测试
uv run pytest
# 类型检查
uv run pyright
# 代码格式化
uv run ruff format .
uv run ruff check . --fix
```
### 调试技巧
1. **启用详细日志**：在开发环境中启用 `logging.DEBUG` 级别可查看完整的协议交互
2. **使用 stdio 传输**：本地调试时优先使用 stdio 传输，便于观察输入输出
3. **单元测试**：为客户端代码编写单元测试，确保异常处理路径覆盖
## 最佳实践总结
| 类别 | 建议 |
|------|------|
| 连接管理 | 使用上下文管理器自动处理连接生命周期 |
| 错误处理 | 捕获具体异常类型，使用 `logger.exception()` |
| 类型提示 | 所有公共 API 必须包含类型注解 |
| 异步代码 | 使用 `anyio` 而非 `asyncio` 进行测试 |
| 资源清理 | 确保所有资源在异常情况下也能正确释放 |
通过遵循本指南中的模式和实践，开发者可以构建稳定、可靠的 MCP 客户端应用程序，充分利用 Model Context Protocol 提供的强大功能。
---
## 客户端 OAuth 认证
### 相关页面
相关主题：[客户端开发指南](#page-client-development)
相关源码文件
以下源码文件用于生成本页说明：
- [src/mcp/client/auth/oauth2.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/auth/oauth2.py)
- [src/mcp/client/auth/exceptions.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/auth/exceptions.py)
- [src/mcp/client/auth/extensions/client_credentials.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/client/auth/extensions/client_credentials.py)
- [src/mcp/shared/auth.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/auth.py)
- [src/mcp/shared/auth_utils.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/auth_utils.py)
- [examples/snippets/clients/oauth_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/oauth_client.py)
# 客户端 OAuth 认证
## 概述
MCP Python SDK 的客户端 OAuth 认证模块为 MCP 客户端提供了完整的 OAuth 2.0 认证支持，包括授权码流程（Authorization Code Flow）和客户端凭证流程（Client Credentials Flow）。该模块基于 [RFC 6749](https://datatracker.ietf.org/doc/html/rfc6749)、[RFC 7636](https://datatracker.ietf.org/doc/html/rfc7636)（PKCE）、[RFC 8414](https://www.rfc-editor.org/rfc/rfc8414)（OAuth 授权服务器元数据）以及 [RFC 9728](https://www.rfc-editor.org/rfc/rfc9728)（受保护资源元数据）等标准规范实现。
主要功能包括：
- **动态客户端注册（DCR）**：自动向授权服务器注册客户端
- **PKCE 支持**：增强授权码流程的安全性
- **Token 管理**：自动刷新过期 Token
- **元数据发现**：支持 RFC 8414 和 RFC 9728 规定的标准发现端点
- **多认证方式**：支持 `client_secret_basic` 和 `client_secret_post`
- **URL-based Client ID (CIMD)**：支持基于 URL 的客户端标识符
资料来源：[src/mcp/client/auth/oauth2.py:1-50]()
## 架构设计
### 模块结构
```
src/mcp/client/auth/
├── oauth2.py # 核心 OAuth2 实现
├── exceptions.py # OAuth 异常定义
├── utils.py # 工具函数
└── extensions/
└── client_credentials.py # 客户端凭证扩展
src/mcp/shared/
├── auth.py # 共享数据模型
└── auth_utils.py # 共享工具函数
```
### 核心组件关系图
```mermaid
graph TD
subgraph "客户端应用层"
A[MCP Client]
B[OAuthClientProvider]
end
subgraph "认证核心"
C[OAuthContext]
D[TokenStorage]
E[PKCEParameters]
end
subgraph "认证扩展"
F[ClientCredentialsOAuthProvider]
end
subgraph "共享模型"
G[OAuthClientMetadata]
H[OAuthToken]
I[OAuthMetadata]
J[ProtectedResourceMetadata]
end
subgraph "外部服务"
K[授权服务器]
L[资源服务器]
end
A --> B
B --> C
B --> D
B --> E
F --> C
G --> H
I --> J
B <--> K
B <--> L
```
### OAuth 上下文数据流
```mermaid
graph LR
subgraph "初始化阶段"
A1[server_url] --> C[OAuthContext]
A2[client_metadata] --> C
A3[storage] --> C
end
subgraph "发现阶段"
C --> D1[发现受保护资源元数据]
C --> D2[发现授权服务器元数据]
D1 --> E1[protected_resource_metadata]
D2 --> E2[oauth_metadata]
end
subgraph "注册阶段"
C --> R1[动态客户端注册]
R1 --> R2[client_info]
end
subgraph "认证阶段"
R2 --> A1[授权请求]
A1 --> T1[Token 交换]
T1 --> T2[current_tokens]
end
```
资料来源：[src/mcp/client/auth/oauth2.py:140-200]()
## 核心数据模型
### OAuthClientMetadata
客户端元数据模型，用于动态客户端注册：
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `client_id` | `str` | 是 | 客户端标识符 |
| `client_secret` | `str` | 否 | 客户端密钥 |
| `redirect_uris` | `list[AnyUrl]` | 是 | 重定向 URI 列表 |
| `grant_types` | `list[str]` | 否 | 支持的授权类型 |
| `token_endpoint_auth_method` | `str` | 否 | Token 端点认证方法 |
| `scope` | `str` | 否 | 请求的权限范围 |
资料来源：[src/mcp/shared/auth.py:200-260]()
### OAuthToken
Token 数据模型：
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `access_token` | `str` | 是 | 访问令牌 |
| `token_type` | `str` | 是 | Token 类型，通常为 "Bearer" |
| `expires_at` | `float` | 是 | Token 过期时间戳 |
| `refresh_token` | `str` | 否 | 刷新令牌 |
| `expires_in` | `int` | 否 | Token 有效期（秒） |
| `scope` | `str` | 否 | 授予的权限范围 |
资料来源：[src/mcp/shared/auth.py:300-340]()
### OAuthMetadata
授权服务器元数据模型：
| 字段 | 类型 | 说明 |
|------|------|------|
| `issuer` | `str` | 授权服务器签发者标识 |
| `authorization_endpoint` | `str` | 授权端点 URL |
| `token_endpoint` | `str` | Token 端点 URL |
| `registration_endpoint` | `str` | 客户端注册端点 |
| `scopes_supported` | `list[str]` | 支持的权限范围 |
| `response_types_supported` | `list[str]` | 支持的响应类型 |
| `grant_types_supported` | `list[str]` | 支持的授权类型 |
| `code_challenge_methods_supported` | `list[str]` | 支持的 PKCE 方法 |
资料来源：[src/mcp/shared/auth.py:80-150]()
### ProtectedResourceMetadata
受保护资源元数据（RFC 9728）：
| 字段 | 类型 | 说明 |
|------|------|------|
| `resource` | `str` | 资源服务器标识符 |
| `authorization_servers` | `list[str]` | 关联的授权服务器列表 |
| `scopes_supported` | `list[str]` | 资源支持的范围 |
| `bearer_methods_supported` | `list[str]` | 支持的 Bearer Token 方法 |
资料来源：[src/mcp/shared/auth.py:400-450]()
### OAuthContext
OAuth 流程上下文数据类：
```python
@dataclass
class OAuthContext:
server_url: str
client_metadata: OAuthClientMetadata
storage: TokenStorage
redirect_handler: Callable[[str], Awaitable[None]] | None
callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None
timeout: float = 300.0
client_metadata_url: str | None = None
# 元数据发现结果
protected_resource_metadata: ProtectedResourceMetadata | None = None
oauth_metadata: OAuthMetadata | None = None
auth_server_url: str | None = None
protocol_version: str | None = None
# 客户端信息
client_info: OAuthClientInformationFull | None = None
# Token 管理
current_tokens: OAuthToken | None = None
token_expiry_time: float | None = None
```
资料来源：[src/mcp/client/auth/oauth2.py:140-185]()
## Token 存储接口
### TokenStorage Protocol
开发者需要实现 `TokenStorage` 接口来持久化 Token：
```python
class TokenStorage(Protocol):
"""Token 存储接口"""
async def get_tokens(self) -> OAuthToken | None:
"""获取存储的 Token"""
...
async def set_tokens(self, tokens: OAuthToken) -> None:
"""存储 Token"""
...
async def get_client_info(self) -> OAuthClientInformationFull | None:
"""获取存储的客户端信息"""
...
async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
"""存储客户端信息"""
...
```
资料来源：[src/mcp/client/auth/oauth2.py:100-130]()
## 认证流程详解
### 授权码流程（Authorization Code Flow）
```mermaid
sequenceDiagram
participant Client as MCP 客户端
participant AS as 授权服务器
participant RS as 资源服务器
Note over Client: 1. 元数据发现
Client->>RS: GET /.well-known/oauth-protected-resource
RS-->>Client: ProtectedResourceMetadata
Client->>AS: GET /.well-known/oauth-authorization-server
AS-->>Client: OAuthMetadata
Note over Client: 2. 客户端注册
Client->>AS: POST /register (DCR)
AS-->>Client: client_id, client_secret
Note over Client: 3. PKCE 准备
Client->>Client: 生成 code_verifier
Client->>Client: 生成 code_challenge = BASE64URL(SHA256(code_verifier))
Note over Client: 4. 授权请求
Client->>Client: 构建授权 URL
Client->>Browser: 重定向到授权页面
User->>Browser: 用户登录
Note over Client: 5. 回调处理
Browser->>Client: /callback?code=xxx&state=xxx
Client->>AS: POST /token (code + code_verifier)
AS-->>Client: access_token, refresh_token
Note over Client: 6. 资源访问
Client->>RS: MCP 请求 (Bearer token)
RS->>AS: Token 内省 (可选)
RS-->>Client: MCP 响应
```
#### 第一阶段：元数据发现
```python
# 发现受保护资源元数据
prm_discovery_urls = build_protected_resource_metadata_discovery_urls(context.server_url)
for url in prm_discovery_urls:
response = yield create_protected_resource_request(url)
ok, prm = await handle_protected_resource_response(response)
if ok and prm:
context.protected_resource_metadata = prm
break
# 发现授权服务器元数据
asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
context.auth_server_url, context.server_url
)
for url in asm_discovery_urls:
response = yield create_oauth_metadata_request(url)
ok, asm = await handle_auth_metadata_response(response)
if ok and asm:
context.oauth_metadata = asm
break
```
资料来源：[src/mcp/client/auth/oauth2.py:250-310]()
#### 第二阶段：客户端注册
支持两种客户端标识方式：
**方式一：动态客户端注册（DCR）**
```python
registration_request = create_client_registration_request(
context.oauth_metadata,
context.client_metadata,
context.get_authorization_base_url(context.server_url),
)
# ... 发送注册请求
client_info = await handle_registration_response(response)
```
资料来源：[src/mcp/client/auth/oauth2.py:340-360]()
**方式二：URL-based Client ID (CIMD)**
当服务器支持 `client_id_metadata_document_supported` 时使用：
```python
if should_use_client_metadata_url(context.oauth_metadata, context.client_metadata_url):
client_information = create_client_info_from_metadata_url(
context.client_metadata_url,
redirect_uris=context.client_metadata.redirect_uris,
)
```
资料来源：[src/mcp/client/auth/utils.py:80-100]()
#### 第三阶段：授权请求与 Token 交换
```python
# 生成 PKCE 参数
pkce = PKCEParameters()
# 构建授权 URL
authorization_url = build_authorization_url(
context.oauth_metadata.authorization_endpoint,
client_id=context.client_info.client_id,
redirect_uri=redirect_uri,
scope=selected_scope,
state=state,
code_challenge=pkce.code_challenge,
code_challenge_method="S256",
)
# 打开浏览器进行授权
await context.redirect_handler(authorization_url)
# 处理回调
code, state = await context.callback_handler()
# 交换 Token
token_request = create_token_request(
context.oauth_metadata.token_endpoint,
grant_type="authorization_code",
code=code,
redirect_uri=redirect_uri,
client_id=context.client_info.client_id,
client_secret=context.client_info.client_secret,
code_verifier=pkce.code_verifier,
)
```
资料来源：[src/mcp/client/auth/oauth2.py:400-480]()
### Token 自动刷新
```mermaid
graph TD
A[MCP 请求] --> B{Token 有效?}
B -->|是| C[发送请求]
B -->|否| D{有 Refresh Token?}
D -->|有| E[刷新 Token]
D -->|无| F[重新授权]
E --> G{刷新成功?}
G -->|是| H[存储新 Token]
G -->|否| I[重新授权]
H --> C
F --> A
I --> A
```
```python
async def _handle_token_refresh(self, context: OAuthContext) -> bool:
"""处理 Token 刷新"""
if not context.oauth_metadata or not context.current_tokens:
return False
refresh_token = context.current_tokens.refresh_token
if not refresh_token:
return False
# 创建刷新请求
token_request = create_token_request(
context.oauth_metadata.token_endpoint,
grant_type="refresh_token",
refresh_token=refresh_token,
client_id=context.client_info.client_id,
client_secret=context.client_info.client_secret,
)
# 发送请求并处理响应
response = yield token_request
token_response = await handle_token_response(response, context.current_tokens.scope)
if token_response:
context.current_tokens = token_response
await context.storage.set_tokens(token_response)
return True
return False
```
资料来源：[src/mcp/client/auth/oauth2.py:550-600]()
## PKCE 支持
### PKCEParameters 数据类
```python
class PKCEParameters(BaseModel):
"""PKCE 参数"""
code_verifier: str = Field(..., min_length=43, max_length=128)
code_challenge: str
code_challenge_method: Literal["S256"] = "S256"
```
资料来源：[src/mcp/client/auth/oauth2.py:60-75]()
### Code Verifier 与 Code Challenge 生成
```mermaid
graph LR
A[随机字符串
43-128 字符] --> B[Base64URL 编码]
B --> C[code_verifier]
C --> D[SHA256 哈希]
D --> E[Base64URL 编码]
E --> F[code_challenge]
```
## 客户端凭证流程
对于服务端到服务端的认证场景，可以使用简化的客户端凭证流程：
```mermaid
graph LR
A[服务端] -->|client_id + client_secret| B[授权服务器]
B -->|access_token| A
A -->|Bearer Token| C[资源服务器]
```
### ClientCredentialsOAuthProvider
```python
from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider
# 初始化
provider = ClientCredentialsOAuthProvider(
server_url="http://localhost:8000/mcp",
storage=InMemoryTokenStorage(),
client_id="my-service",
client_secret="secret",
scopes="tools read resources",
)
# 使用 httpx.AsyncClient
async with httpx.AsyncClient(auth=provider) as client:
response = await client.post("http://localhost:8000/mcp", json=request_data)
```
资料来源：[src/mcp/client/auth/extensions/client_credentials.py:30-80]()
## 异常处理
### OAuthFlowError
授权流程异常：
```python
class OAuthFlowError(Exception):
"""OAuth 流程错误"""
error: OAuthErrorCode
error_description: str | None = None
```
### OAuthTokenError
Token 相关错误：
```python
class OAuthTokenError(Exception):
"""Token 处理错误"""
error: OAuthErrorCode
error_description: str | None = None
```
资料来源：[src/mcp/client/auth/exceptions.py:1-30]()
## 使用示例
### 完整授权码流程示例
```python
import asyncio
from httpx import AsyncClient
from mcp.client.auth.oauth2 import OAuthClientProvider, OAuthClientMetadata
from mcp.client.auth.in_memory_storage import InMemoryTokenStorage
async def main():
# 1. 配置客户端元数据
client_metadata = OAuthClientMetadata(
redirect_uris=["http://localhost:8000/callback"],
grant_types=["authorization_code", "refresh_token"],
token_endpoint_auth_method="client_secret_post",
scope="tools read",
)
# 2. 创建 Token 存储
storage = InMemoryTokenStorage()
# 3. 创建 OAuth 提供者
oauth_provider = OAuthClientProvider(
server_url="http://localhost:8000/mcp",
client_metadata=client_metadata,
storage=storage,
redirect_handler=lambda url: print(f"打开浏览器: {url}"),
callback_handler=lambda: input("输入授权码: ").split(","),
)
# 4. 使用 OAuth 认证
async with AsyncClient(auth=oauth_provider) as client:
response = await client.post(
"http://localhost:8000/mcp",
json={"jsonrpc": "2.0", "method": "tools/list", "id": 1}
)
print(response.json())
asyncio.run(main())
```
### 带资源验证的示例
```python
from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url
# 验证请求的资源与配置的资源是否匹配
allowed = check_resource_allowed(
requested_resource="http://localhost:8001",
configured_resource="http://localhost:8001"
)
# 从服务器 URL 提取资源 URL
resource_url = resource_url_from_server_url("http://localhost:8001/mcp")
# 结果: "http://localhost:8001"
```
资料来源：[src/mcp/shared/auth_utils.py:50-100]()
## 配置选项
### OAuthClientProvider 初始化参数
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| `server_url` | `str` | 是 | - | MCP 服务器 URL |
| `client_metadata` | `OAuthClientMetadata` | 是 | - | OAuth 客户端元数据 |
| `storage` | `TokenStorage` | 是 | - | Token 存储实现 |
| `redirect_handler` | `Callable` | 否 | `None` | 授权重定向处理函数 |
| `callback_handler` | `Callable` | 否 | `None` | 回调处理函数 |
| `timeout` | `float` | 否 | `300.0` | OAuth 流程超时时间 |
| `client_metadata_url` | `str` | 否 | `None` | URL-based Client ID |
| `validate_resource_url` | `Callable` | 否 | `None` | 自定义资源 URL 验证 |
### AuthSettings 配置
| 字段 | 类型 | 说明 |
|------|------|------|
| `issuer_url` | `AnyHttpUrl` | 授权服务器 URL |
| `resource_server_url` | `AnyHttpUrl` | 资源服务器 URL |
| `service_documentation_url` | `AnyHttpUrl` | 服务文档 URL |
| `client_registration_options` | `ClientRegistrationOptions` | 客户端注册选项 |
| `revocation_options` | `RevocationOptions` | Token 撤销选项 |
| `required_scopes` | `list[str]` | 必需的权限范围 |
资料来源：[src/mcp/server/auth/settings.py:1-40]()
## 安全考虑
### PKCE 必要性
对于公开客户端（如移动应用或单页应用），强烈建议使用 PKCE。SDK 默认使用 S256 方法进行 code challenge：
```python
code_challenge_method: Literal["S256"] = "S256"
```
### Token 安全存储
- 生产环境应使用加密存储（如 `EncryptedTokenStorage`）
- Refresh Token 应安全保存，防止泄露
- Token 过期时间应合理设置
### HTTPS 要求
SDK 对客户端元数据 URL 有严格的验证要求：
```python
def is_valid_client_metadata_url(url: str | None) -> bool:
"""验证客户端元数据 URL 是否有效"""
if not url:
return False
try:
parsed = urlparse(url)
return parsed.scheme == "https" and parsed.path not in ("", "/")
except Exception:
return False
```
资料来源：[src/mcp/client/auth/utils.py:30-50]()
## 相关文档
- [授权服务器实现](../servers/auth/README.md)
- [Streamable HTTP 传输](../transports/streamable-http.md)
- [RFC 6749 - OAuth 2.0 授权框架](https://datatracker.ietf.org/doc/html/rfc6749)
- [RFC 7636 - PKCE](https://datatracker.ietf.org/doc/html/rfc7636)
- [RFC 8414 - OAuth 授权服务器元数据](https://www.rfc-editor.org/rfc/rfc8414)
- [RFC 9728 - 受保护资源元数据](https://www.rfc-editor.org/rfc/rfc9728)
---
---
## Doramagic 踩坑日志
项目：modelcontextprotocol/python-sdk
摘要：发现 22 个潜在踩坑项，其中 3 个为 high/blocking；最高优先级：运行坑 - 来源证据：[packit] Pull from upstream failed for release 1.27.1。
## 1. 运行坑 · 来源证据：[packit] Pull from upstream failed for release 1.27.1
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：[packit] Pull from upstream failed for release 1.27.1
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_7cd3c6cd2e994e6cbd60b3cbb19c96e6 | https://github.com/modelcontextprotocol/python-sdk/issues/2563 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 2. 安全/权限坑 · 来源证据：MCP client does not retry authenticated request after successful OAuth token exchange
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：MCP client does not retry authenticated request after successful OAuth token exchange
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_437e528a826e41e6ae0d654e0e858cbc | https://github.com/modelcontextprotocol/python-sdk/issues/2577 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 3. 安全/权限坑 · 来源证据：OAuth token refresh sends RFC 8707 resource parameter that Entra ID v2.0 rejects (AADSTS9010010)
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：OAuth token refresh sends RFC 8707 resource parameter that Entra ID v2.0 rejects (AADSTS9010010)
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_f4669be4619541d3bc0001e432fa5416 | https://github.com/modelcontextprotocol/python-sdk/issues/2578 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 4. 安装坑 · 来源证据：Race condition between SSE session init and tools/call rejection (-32602)
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Race condition between SSE session init and tools/call rejection (-32602)
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_f707b6204fd541d3aea0e5c1ef97e432 | https://github.com/modelcontextprotocol/python-sdk/issues/2583 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 5. 安装坑 · 来源证据：Types-only install option
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Types-only install option
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_1582e38e22384d509ffb75f91a25c9d0 | https://github.com/modelcontextprotocol/python-sdk/issues/2581 | 来源类型 github_issue 暴露的待验证使用条件。
## 6. 配置坑 · 来源证据：Expose `schema_generator` on `FastMCP` for tool params schema generation
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：Expose `schema_generator` on `FastMCP` for tool params schema generation
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_2a614d750d3f4ec59807cdd800d8ff75 | https://github.com/modelcontextprotocol/python-sdk/issues/2582 | 来源类型 github_issue 暴露的待验证使用条件。
## 7. 配置坑 · 来源证据：v1.24.0
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：v1.24.0
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_51df2611be1e47b586edd80b1adb1cfb | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.24.0 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 8. 能力坑 · 能力判断依赖假设
- 严重度：medium
- 证据强度：source_linked
- 发现：README/documentation is current enough for a first validation pass.
- 对用户的影响：假设不成立时，用户拿不到承诺的能力。
- 建议检查：将假设转成下游验证清单。
- 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
- 证据：capability.assumptions | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | README/documentation is current enough for a first validation pass.
## 9. 运行坑 · 来源证据：fix: 4 additional raise sites missing exception chaining with 'from'
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：fix: 4 additional raise sites missing exception chaining with 'from'
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_71303d3f16c942efb8521432eba345a1 | https://github.com/modelcontextprotocol/python-sdk/issues/2575 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 10. 维护坑 · 来源证据：v1.23.2
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：v1.23.2
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_3c7b4addde6a4da3ae29d5f3f01e7455 | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.23.2 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 11. 维护坑 · 维护活跃度未知
- 严重度：medium
- 证据强度：source_linked
- 发现：未记录 last_activity_observed。
- 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
- 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
- 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
- 证据：evidence.maintainer_signals | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | last_activity_observed missing
## 12. 安全/权限坑 · 下游验证发现风险项
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：下游已经要求复核，不能在页面中弱化。
- 建议检查：进入安全/权限治理复核队列。
- 防护动作：下游风险存在时必须保持 review/recommendation 降级。
- 证据：downstream_validation.risk_items | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | no_demo; severity=medium
## 13. 安全/权限坑 · 存在安全注意事项
- 严重度：medium
- 证据强度：source_linked
- 发现：No sandbox install has been executed yet; downstream must verify before user use.
- 对用户的影响：用户安装前需要知道权限边界和敏感操作。
- 建议检查：转成明确权限清单和安全审查提示。
- 防护动作：安全注意事项必须面向用户前置展示。
- 证据：risks.safety_notes | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | No sandbox install has been executed yet; downstream must verify before user use.
## 14. 安全/权限坑 · 存在评分风险
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：风险会影响是否适合普通用户安装。
- 建议检查：把风险写入边界卡，并确认是否需要人工复核。
- 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
- 证据：risks.scoring_risks | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | no_demo; severity=medium
## 15. 安全/权限坑 · 来源证据：Claude Code client sends tools/call without initialize handshake — SSE transport rejects with -32602
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Claude Code client sends tools/call without initialize handshake — SSE transport rejects with -32602
- 对用户的影响：可能阻塞安装或首次运行。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_48d1c0f7a5d040568e55a2fa6ead27d8 | https://github.com/modelcontextprotocol/python-sdk/issues/2579 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 16. 安全/权限坑 · 来源证据：v1.23.0
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：v1.23.0
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_ca7e48aa359f4c0b86a3ca42172b40d9 | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.23.0 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 17. 安全/权限坑 · 来源证据：v1.23.3
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：v1.23.3
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_202f1fef816145aa912dae7104ecf977 | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.23.3 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 18. 安全/权限坑 · 来源证据：v1.25.0
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：v1.25.0
- 对用户的影响：可能影响升级、迁移或版本选择。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_9cdf542dc72c4e31998478a1e41b9c8e | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.25.0 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 19. 安全/权限坑 · 来源证据：v1.27.0
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：v1.27.0
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_eb0db07c9d9940e8af311c6e3165dada | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.27.0 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 20. 安全/权限坑 · 来源证据：v1.27.1
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：v1.27.1
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_bcdd360ffa1641fbb2ed4c5be2be5ebe | https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.27.1 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 21. 维护坑 · issue/PR 响应质量未知
- 严重度：low
- 证据强度：source_linked
- 发现：issue_or_pr_quality=unknown。
- 对用户的影响：用户无法判断遇到问题后是否有人维护。
- 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
- 防护动作：issue/PR 响应未知时，必须提示维护风险。
- 证据：evidence.maintainer_signals | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | issue_or_pr_quality=unknown
## 22. 维护坑 · 发布节奏不明确
- 严重度：low
- 证据强度：source_linked
- 发现：release_recency=unknown。
- 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
- 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
- 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
- 证据：evidence.maintainer_signals | github_repo:862584018 | https://github.com/modelcontextprotocol/python-sdk | release_recency=unknown