Doramagic 项目包 · 项目说明书
python-sdk 项目
生成时间:2026-05-13 12:56:39 UTC
MCP 协议与 Python SDK 简介
MCP(Model Context Protocol,模型上下文协议)是一种开放协议,用于在大型语言模型(LLM)应用程序与外部数据源、工具和服务之间建立标准化的通信桥梁。Python SDK 是该协议的核心实现库,为开发者提供了构建 MCP 服务器和客户端的能力。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
MCP(Model Context Protocol,模型上下文协议)是一种开放协议,用于在大型语言模型(LLM)应用程序与外部数据源、工具和服务之间建立标准化的通信桥梁。Python SDK 是该协议的核心实现库,为开发者提供了构建 MCP 服务器和客户端的能力。
MCP 协议的核心价值在于标准化——它定义了 LLM 应用与外部世界交互的统一方式,使开发者能够创建可复用的组件,让 AI 系统能够安全、受控地访问本地文件、数据库、API 以及各类服务。
核心概念
服务器与客户端架构
MCP 采用典型的客户端-服务器架构,支持多种通信传输方式:
| 组件 | 说明 | 支持的传输方式 |
|---|---|---|
| MCP 服务器 | 托管资源、工具和提示的服务器 | stdio、streamable-http |
| MCP 客户端 | 连接服务器并调用功能的客户端 | stdio、streamable-http |
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/plainannotations: 资源的注解信息meta: 额外的元数据
资料来源:src/mcp/server/mcpserver/resources/base.py:1-42
工具系统
工具是 MCP 服务器暴露的可执行函数,客户端可以通过 JSON-RPC 调用这些工具。工具系统支持:
- 结构化输出: 使用 Pydantic 模型定义返回格式
- 元数据: 支持标题、描述、图标和注解
- 异步执行: 所有工具函数支持异步调用
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
提示系统
提示(Prompts)允许服务器定义可复用的消息模板,这些模板可以接受参数并生成动态内容:
class Prompt(BaseModel, abc.ABC):
name: str
description: str | None = None
arguments: list[PromptArgument] | None = None
icons: list[Icon] | None = None
使用示例:
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
完成处理器
完成处理器为资源模板和提示参数提供动态补全建议:
@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
交互式任务系统
MCP 支持长时间运行的任务(Tasks),包括两个关键功能:
elicitation(征询)
用于在执行过程中请求用户确认或输入:
graph LR
A["工具调用"] --> B["task.elicit()"]
B --> C["input_required 状态"]
C --> D["用户响应"]
D --> E["任务完成"]Sampling(采样)
允许服务器请求 LLM 生成内容:
graph LR
A["工具调用"] --> B["task.create_message()"]
B --> C["input_required 状态"]
C --> D["LLM 响应"]
D --> E["任务完成"]资料来源:examples/servers/simple-task-interactive/README.md
认证系统
SDK 提供了 OAuth 认证支持,包括完整的授权码流程:
graph TD
A["客户端"] --> B["授权请求"]
B --> C["用户登录"]
C --> D["授权码回调"]
D --> E["令牌交换"]
E --> F["认证完成"]认证服务器可以返回 HTML 表单进行交互:
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
开发指南
环境配置
# 安装 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
代码质量要求
| 检查项 | 命令 | 说明 |
|---|---|---|
| 格式化 | uv run --frozen ruff format . | 使用 ruff 格式化代码 |
| 静态检查 | uv run --frozen ruff check . --fix | 自动修复 lint 问题 |
| 类型检查 | uv run --frozen pyright | 运行 pyright 类型检查 |
| 测试 | uv run --frozen pytest | 执行测试套件 |
核心规范
- 类型提示: 所有公共 API 必须包含类型提示
- 文档字符串: 公共 API 必须有 docstring,包含
Raises:部分说明可能抛出的异常 - 异常处理:
- 使用
logger.exception()而非logger.error() - 捕获具体异常类型而非通用
Exception - 禁止在顶级处理器之外使用
except Exception:
- 导入规范: 所有导入必须位于文件顶部,禁止内联导入
# 正确示例
try:
resource = await self._resource_manager.get_resource(uri, context)
except ValueError as exc:
raise ResourceError(f"Unknown resource: {uri}") from exc
资料来源:AGENTS.md
测试规范
- 使用
anyio进行异步测试,而非asyncio - 测试应快速且确定性,使用内存异步执行
- 使用线程仅在必要时,使用子进程作为最后手段
- 不要使用
Test前缀的类,使用顶层test_*函数
# 推荐
def test_resource_read():
assert isinstance(resource, FunctionResource)
# 避免
class TestResource:
def test_read(self):
...
分支模型
| 分支 | 用途 | 变更类型 |
|---|---|---|
main | V2 开发分支 | 新功能、破坏性变更 |
v1.x | 稳定版发布分支 | 安全修复、关键 bug 修复 |
资料来源:AGENTS.md
发布流程
- 在
pyproject.toml中更新依赖版本 - 运行
uv lock --resolution lowest-direct升级锁文件 - 通过 GitHub UI 创建 Release,标签格式为
vX.Y.Z - 版本号将从标签自动设置
资料来源:RELEASE.md
快速开始示例
创建简单服务器
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 服务器
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
总结
MCP Python SDK 提供了一套完整、健壮的框架,用于构建与 LLM 应用交互的服务器和客户端。通过标准化的资源、工具、提示和完成处理器接口,开发者可以创建可复用、安全且易于维护的 MCP 组件。SDK 遵循严格的代码质量标准,包括类型提示要求、完整的文档字符串规范以及全面的测试覆盖,确保其稳定性和可维护性。
资料来源:[src/mcp/server/mcpserver/resources/base.py:1-42](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/mcpserver/resources/base.py)
安装与配置
本文档详细说明 Model Context Protocol (MCP) Python SDK 的安装、环境配置以及开发环境搭建流程。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
环境要求
系统依赖
| 依赖项 | 版本要求 | 说明 |
|---|---|---|
| Python | 3.10+ | 支持 Python 3.10 至 3.14 |
| uv | 最新稳定版 | 包管理工具,详见 uv 官方文档 |
资料来源:CONTRIBUTING.md:1-3
验证 Python 版本
python --version
确保输出为 Python 3.10 或更高版本。
项目克隆与依赖安装
1. Fork 并克隆仓库
# 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 进行安装或依赖管理。
# 安装所有依赖(包括开发依赖和可选依赖)
uv sync --frozen --all-extras --dev
参数说明:
--frozen:锁定版本,不更新 lock 文件--all-extras:安装所有可选依赖--dev:安装开发依赖
资料来源:CONTRIBUTING.md:7
3. 设置预提交钩子
uv tool install pre-commit --with pre-commit-uv --force-reinstall
预提交钩子会自动执行代码格式化、类型检查等操作。
资料来源:CONTRIBUTING.md:9-11
开发环境配置
uv 工具链使用规范
| 操作类型 | 命令格式 | 说明 |
|---|---|---|
| 运行工具 | uv run --frozen <tool> | 始终使用 --frozen 避免意外修改 lock 文件 |
| 添加依赖 | uv add <package> | 仅用于开发环境 |
| 升级依赖 | uv lock --upgrade-package <package> | 升级指定包版本 |
| 测试跨版本 | uv run --frozen --python 3.10 pytest ... | 指定 Python 版本运行测试 |
禁止操作:
- 禁止使用
uv pip install - 禁止使用
@latest语法指定版本 - 不要仅为修复 CVE 而提升依赖版本下限,约束条件
>=已允许用户自行升级
资料来源:AGENTS.md:1-20
代码质量检查
开发过程中应运行以下检查:
# 运行测试
uv run pytest
# 类型检查
uv run pyright
# 代码格式化
uv run ruff format .
# 代码检查
uv run ruff check . --fix
全部检查命令可通过预提交钩子一键执行:
pre-commit run --all-files
服务器配置
基本配置参数
客户端连接 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 传输
# 基本启动
uv run mcp-simple-auth --port=8000 --transport=streamable-http
# 带 OAuth 严格模式
uv run mcp-simple-auth --port=8000 --transport=streamable-http --oauth-strict
#### SSE 传输
uv run mcp-simple-auth --port=8000 --transport=sse
资料来源:examples/servers/simple-auth/README.md:30-45
任务服务器配置
交互式任务服务器默认启动地址为 http://localhost:8000/mcp,可通过 --port 参数修改:
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
开发工作流
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]代码提交检查清单
- ✅ 更新相关文档
- ✅ 为新功能添加测试
- ✅ 确保 CI 通过
- ✅ 处理代码审查反馈
调试与日志
异常处理规范
- 必须使用
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 重新安装依赖 |
验证安装
安装完成后,可通过以下命令验证环境:
# 验证 uv 可用
uv --version
# 验证 Python 版本
python --version
# 验证 MCP SDK 安装
uv run python -c "import mcp; print(mcp.__version__)"
# 运行项目测试
uv run pytest -v
下一步
资料来源:[CONTRIBUTING.md:1-3]()
快速开始指南
本指南帮助开发者快速上手 Model Context Protocol (MCP) Python SDK,涵盖服务器创建、工具(Tool)、资源(Resource)和提示(Prompt)的定义与注册,以及如何启动服务。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
环境准备
前置要求
| 要求 | 版本说明 |
|---|---|
| Python | 3.10+ |
| 包管理器 | uv |
| 操作系统 | Windows、macOS、Linux |
安装 SDK
# 使用 uv 安装稳定版本
uv add mcp
# 安装开发版本(包含所有额外依赖和开发工具)
uv sync --frozen --all-extras --dev
资料来源:CONTRIBUTING.md:5-8
MCP 服务器架构
MCP 服务器是 MCP 协议的核心组件,负责向客户端提供服务能力。
graph TD
A[客户端] --> B[MCP Server]
B --> C[工具管理器<br/>Tool Manager]
B --> D[资源管理器<br/>Resource Manager]
B --> E[提示管理器<br/>Prompt Manager]
C --> F[工具函数]
D --> G[资源数据]
E --> H[提示模板]MCPServer 类是 SDK 的主入口,负责管理所有组件的生命周期并处理客户端请求。
资料来源:src/mcp/server/mcpserver/server.py:1-50
创建基础服务器
最小示例
from mcp.server import MCPServer
server = MCPServer(
name="my-server",
version="1.0.0",
)
完整配置示例
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)
工具是服务器暴露给客户端的可调用函数。
基本工具定义
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 | 文件资源 | 读取本地文件 |
函数资源
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="返回当前应用配置",
)
)
文件资源
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)
提示模板用于生成可复用的对话片段。
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 方法
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
启动服务器
运行服务器
import uvicorn
if __name__ == "__main__":
uvicorn.run(
"my_server:app",
host="0.0.0.0",
port=8000,
reload=True,
)
命令行启动
# 方式一:直接运行模块
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 | 需要长连接的场景 |
# 启动支持 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
常见工作流程
graph LR
A[定义工具] --> B[定义资源]
B --> C[定义提示]
C --> D[创建 MCPServer 实例]
D --> E[注册组件]
E --> F[启动服务]
F --> G[客户端连接]
G --> H[调用工具/访问资源]下一步
- 阅读 examples/servers/ 目录下的完整示例
- 了解 认证与授权 功能
- 探索 实验性功能 如任务支持
资料来源:[CONTRIBUTING.md:5-8]()
系统架构与核心组件
Model Context Protocol (MCP) Python SDK 是一个用于构建 Model Context Protocol 客户端和服务器的框架。MCP 是一种标准化协议,使 AI 模型能够与外部数据源和工具进行交互。本文详细介绍 MCP SDK 的系统架构和核心组件。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
1. 架构概述
MCP 采用客户端-服务器架构,支持多种传输协议(stdio、streamable-http)。协议基于 JSON-RPC 2.0 规范实现双向通信。
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 --> Auth2. 核心组件关系
2.1 组件层次结构
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 --> JSONRPCHandler2.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
3. 会话层详解
3.1 SharedSession 基类
SharedSession 是客户端和服务端会话的共同基类,封装了 JSON-RPC 2.0 协议的通用逻辑。
核心职责:
- 消息队列管理
- 请求-响应对应关系维护
- Cancellation token 处理
- 消息发送和接收
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
3.2 客户端会话
ClientSession 扩展了 SharedSession,添加了客户端特定的功能:
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
客户端初始化流程:
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
3.3 服务端会话
ServerSession 负责处理来自客户端的请求和调用服务端处理器。
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
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 |
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
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 初始化参数
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
5.2 核心方法
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
6. 资源系统
6.1 资源类型
MCP 支持两种主要资源类型:
| 类型 | 文件路径 | 描述 |
|---|---|---|
FunctionResource | resources/types.py | 从函数动态生成的资源 |
FileResource | resources/types.py | 读取本地文件的资源 |
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
6.2 资源读取流程
graph TD
Client[Client] -->|read_resource| ServerSession
ServerSession -->|路由| MCPServer
MCPServer -->|获取资源| ResourceManager
ResourceManager -->|验证URI| URI解析
URI解析 -->|查找资源| ResourceRegistry
ResourceRegistry -->|返回| Resource
Resource -->|调用read方法| FunctionResource
FunctionResource -->|执行函数| ReadFunction
ReadFunction -->|返回内容| ReadResourceContents7. 认证系统
7.1 OAuth 2.0 支持
MCP 支持 OAuth 2.0 认证流程,包括:
- 授权码流程 (Authorization Code Flow)
- 客户端凭证流程 (Client Credentials Flow)
- 令牌刷新 (Token Refresh)
- 令牌撤销 (Token Revocation)
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
7.2 认证配置
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 兼容的受保护资源元数据端点:
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 补全 |
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
end8.2 任务上下文
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
9. 传输层
9.1 Stdio 传输
适用于本地进程间通信,通过标准输入/输出流传输 JSON-RPC 消息。
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。
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 | 自动补全建议 |
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
11. 服务器设置
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
12. 总结
MCP Python SDK 的架构设计遵循以下原则:
- 分层设计:清晰的分层架构,从传输层到高层 API
- 协议标准化:基于 JSON-RPC 2.0,确保互操作性
- 灵活性:支持多种传输方式和认证机制
- 可扩展性:通过回调和中间件支持自定义行为
- 类型安全:全面的类型提示和 Pydantic 模型验证
这套架构使得开发者能够快速构建支持外部数据和工具交互的 AI 应用服务器和客户端。
资料来源:[src/mcp/shared/session.py:1-50](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/session.py)
传输协议实现
MCP (Model Context Protocol) Python SDK 支持多种传输协议实现,用于客户端与服务器之间的通信。这些传输层负责在进程间或网络间传递 MCP 协议消息,支持不同的部署场景和性能需求。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
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 格式编码。
graph LR
A[Client Process] -->|stdin| B[Server Process]
B -->|stdout| A客户端用法示例:
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 调用
服务器用法示例:
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,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,src/mcp/server/streamable_http.py
服务器配置示例:
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
SSE 传输
Server-Sent Events (SSE) 传输允许服务器主动向客户端推送消息,适用于需要服务器端实时通知的场景。
graph LR
A[HTTP Client] -->|POST /mcp| B[MCP Server]
B -->|SSE Stream| A
A -->|GET /sse| B资料来源:src/mcp/client/sse.py,src/mcp/server/sse.py
WebSocket 传输
WebSocket 传输提供全双工通信通道,适用于需要低延迟、双向实时交互的应用场景。
WebSocket 服务器配置:
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,src/mcp/client/websocket.py
Streamable HTTP 服务器管理
streamable_http_manager 模块提供了高级的 HTTP 传输管理功能,包括会话管理、请求路由和生命周期管理。
核心组件
| 组件 | 职责 |
|---|---|
StreamableHTTPServerHandler | 处理 HTTP 请求,解析 MCP 协议消息 |
HTTPAuthorizationMiddleware | 处理认证头和授权验证 |
HTTPStreamManager | 管理流式响应和 SSE 连接 |
资料来源:src/mcp/server/streamable_http_manager.py
挂载到 ASGI 应用
MCP HTTP 服务器可以挂载到现有的 ASGI 应用中:
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
传输协议选择指南
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 消息格式:
// 请求消息
{
"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 模式
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()
)
HTTP 模式
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
传输层安全性
MCP SDK 的传输层本身不实现加密,安全性依赖于底层传输协议:
| 传输类型 | 安全建议 |
|---|---|
| Stdio | 仅用于本地通信,使用文件系统权限控制 |
| Streamable HTTP | 配合 HTTPS/TLS 使用,添加认证中间件 |
| WebSocket | 使用 WSS (WebSocket Secure) |
| SSE | 通过 HTTPS 提供服务 |
常见问题
Q: Stdio 和 HTTP 传输可以同时使用吗?
可以。服务器可以同时监听多种传输协议,通过不同的启动参数选择:
# Stdio 模式
python server.py --transport stdio
# HTTP 模式
python server.py --transport streamable-http --port 8000
Q: 如何调试传输层问题?
- 启用调试日志:
export MCP_LOG_LEVEL=debug - 使用 HTTP 抓包工具查看请求/响应
- 对于 stdio,使用
script命令记录会话
相关资源
资料来源:[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)
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 装饰器来简化工具注册过程。
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 静态资源
@mcp.resource("config://app")
def get_config() -> str:
"""返回应用配置"""
return '{"version": "1.0.0"}'
#### 3.2 资源模板
@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 交互模板。
@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 对象作为第一个参数,该对象提供了访问服务器功能的能力。
@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 参数管理服务器启动和关闭时的资源。
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 集成:
# 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()
运行命令:
mcp run my_server.py
资料来源:examples/servers/simple-pagination/mcp_simple_pagination/server.py:40-60
HTTP 传输模式
适用于 Web 环境,支持流式响应:
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 支持实验性的异步任务功能,允许长时间运行的工具返回任务引用而非直接结果。
@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 方法调用:
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 支持资源变更通知机制,服务器可以主动推送更新给客户端:
@mcp.resource("data://stream")
async def streaming_data(context: Context):
async for update in data_generator():
# 发送资源更新通知
await context.notify_resources_changed()
yield update
交互式任务 (Elicitation)
服务器可以请求用户输入确认:
@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 生成内容:
@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 必须包含类型注解:
@mcp.tool()
async def process_data(context: Context, data: list[str], options: dict[str, Any] | None = None) -> dict[str, Any]:
"""处理数据并返回结果"""
...
2. 错误处理
使用框架定义的异常类型,并在文档中说明可能抛出的异常:
@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 中正确管理外部资源连接:
@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
资料来源:[src/mcp/server/mcpserver/server.py:1-150]()
资源、资源模板、提示与工具
MCP SDK 提供了四类核心功能用于构建 MCP 服务器:资源(Resources)、资源模板(Resource Templates)、提示(Prompts) 和 工具(Tools)。这些功能使服务器能够向 LLM 客户端暴露数据、模板和可执行函数,从而实现丰富的工具调用和数据检索能力。
继续阅读本节完整说明和来源证据。
概述
MCP SDK 提供了四类核心功能用于构建 MCP 服务器:资源(Resources)、资源模板(Resource Templates)、提示(Prompts) 和 工具(Tools)。这些功能使服务器能够向 LLM 客户端暴露数据、模板和可执行函数,从而实现丰富的工具调用和数据检索能力。
来源:https://github.com/modelcontextprotocol/python-sdk / 项目说明书
上下文与会话管理
在 Model Context Protocol (MCP) SDK 中,上下文(Context) 和 会话管理 是连接服务器端与客户端、传递请求元数据、管理生命周期状态的核心机制。上下文对象在请求处理链中携带关键信息,使开发者能够在工具执行、资源访问、提示渲染等场景中访问请求级别的元数据和服务器状态。
继续阅读本节完整说明和来源证据。
概述
在 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)
客户端开发指南
本指南详细介绍如何使用 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 支持多种传输方式,客户端可根据场景选择合适的传输协议:
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 连接
标准输入输出传输适用于本地进程间通信,是最常用的连接方式之一。
# 资料来源: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 传输方式:
# 连接到远程 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):
# 处理连接
客户端初始化流程
sequenceDiagram
participant 客户端 as MCP 客户端
participant 服务器 as MCP 服务器
客户端->>服务器: 发送 Initialize 请求
客户端->>服务器: 包含协议版本和客户端能力
服务器-->>客户端: 返回服务器信息
服务器-->>客户端: 协议版本确认
服务器-->>客户端: 服务器能力集
客户端->>服务器: 发送 Initialized 通知
Note over 客户端: 连接建立完成初始化时客户端需要指定支持的协议版本和自身能力集,服务器确认后将返回其能力信息。
会话管理
ClientSession 的核心职责
ClientSession 类是客户端与服务器交互的核心抽象,负责管理会话状态和协调各类请求:
# 资料来源: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 协议最核心的功能之一,允许客户端请求服务器执行特定操作:
# 资料来源: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}")
解析工具返回结果
工具调用可能返回多种类型的内容,客户端需要正确处理:
# 资料来源: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}")
资源管理
#### 列出可用资源
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}")
#### 读取资源内容
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}")
提示模板
服务器可以定义可复用的提示模板,客户端可以获取并使用这些模板:
# 资料来源: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 支持在客户端实现自动补全功能,提升用户体验:
# 资料来源: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 负责管理异步请求与响应的匹配,确保请求能够正确路由到对应的处理程序:
graph LR
A[请求发送] --> B[RequestTracker]
B --> C{匹配响应}
C -->|找到匹配| D[触发回调]
C -->|未找到| E[等待响应]
D --> F[返回结果]
E --> C# 资料来源:src/mcp/shared/response_router.py
class ResponseRouter:
"""响应路由器,管理请求-响应映射"""
async def send_request(self, method: str, params: dict) -> Any:
"""发送请求并等待响应"""
def __aiter__(self):
"""异步迭代器,用于接收传入的消息"""
多会话管理
SessionGroup 的使用
当客户端需要同时管理多个服务器连接时,可使用 SessionGroup 进行统一管理:
# 资料来源: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 |
错误处理
异常处理规范
根据项目开发规范,客户端代码应遵循以下异常处理原则:
# 资料来源: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 流程:
# 资料来源: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 客户端实现示例:
# 资料来源: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()
开发工作流
本地开发设置
# 克隆仓库
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
调试技巧
- 启用详细日志:在开发环境中启用
logging.DEBUG级别可查看完整的协议交互 - 使用 stdio 传输:本地调试时优先使用 stdio 传输,便于观察输入输出
- 单元测试:为客户端代码编写单元测试,确保异常处理路径覆盖
最佳实践总结
| 类别 | 建议 |
|---|---|
| 连接管理 | 使用上下文管理器自动处理连接生命周期 |
| 错误处理 | 捕获具体异常类型,使用 logger.exception() |
| 类型提示 | 所有公共 API 必须包含类型注解 |
| 异步代码 | 使用 anyio 而非 asyncio 进行测试 |
| 资源清理 | 确保所有资源在异常情况下也能正确释放 |
通过遵循本指南中的模式和实践,开发者可以构建稳定、可靠的 MCP 客户端应用程序,充分利用 Model Context Protocol 提供的强大功能。
来源:https://github.com/modelcontextprotocol/python-sdk / 项目说明书
客户端 OAuth 认证
MCP Python SDK 的客户端 OAuth 认证模块为 MCP 客户端提供了完整的 OAuth 2.0 认证支持,包括授权码流程(Authorization Code Flow)和客户端凭证流程(Client Credentials Flow)。该模块基于 RFC 6749、RFC 7636(PKCE)、RFC 8414(OAuth 授权服务器元数据)以及 RFC ...
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
MCP Python SDK 的客户端 OAuth 认证模块为 MCP 客户端提供了完整的 OAuth 2.0 认证支持,包括授权码流程(Authorization Code Flow)和客户端凭证流程(Client Credentials Flow)。该模块基于 RFC 6749、RFC 7636(PKCE)、RFC 8414(OAuth 授权服务器元数据)以及 RFC 9728(受保护资源元数据)等标准规范实现。
主要功能包括:
- 动态客户端注册(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 # 共享工具函数
核心组件关系图
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 <--> LOAuth 上下文数据流
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 流程上下文数据类:
@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:
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)
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 响应#### 第一阶段:元数据发现
# 发现受保护资源元数据
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)
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 时使用:
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 交换
# 生成 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 自动刷新
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 --> Aasync 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 数据类
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 生成
graph LR
A[随机字符串<br/>43-128 字符] --> B[Base64URL 编码]
B --> C[code_verifier]
C --> D[SHA256 哈希]
D --> E[Base64URL 编码]
E --> F[code_challenge]客户端凭证流程
对于服务端到服务端的认证场景,可以使用简化的客户端凭证流程:
graph LR
A[服务端] -->|client_id + client_secret| B[授权服务器]
B -->|access_token| A
A -->|Bearer Token| C[资源服务器]ClientCredentialsOAuthProvider
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
授权流程异常:
class OAuthFlowError(Exception):
"""OAuth 流程错误"""
error: OAuthErrorCode
error_description: str | None = None
OAuthTokenError
Token 相关错误:
class OAuthTokenError(Exception):
"""Token 处理错误"""
error: OAuthErrorCode
error_description: str | None = None
资料来源:src/mcp/client/auth/exceptions.py:1-30
使用示例
完整授权码流程示例
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())
带资源验证的示例
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:
code_challenge_method: Literal["S256"] = "S256"
Token 安全存储
- 生产环境应使用加密存储(如
EncryptedTokenStorage) - Refresh Token 应安全保存,防止泄露
- Token 过期时间应合理设置
HTTPS 要求
SDK 对客户端元数据 URL 有严格的验证要求:
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
相关文档
资料来源:[src/mcp/client/auth/oauth2.py:1-50]()
失败模式与踩坑日记
保留 Doramagic 在发现、验证和编译中沉淀的项目专属风险,不把社区讨论只当作装饰信息。
可能增加新用户试用和生产接入成本。
可能影响授权、密钥配置或安全边界。
可能影响授权、密钥配置或安全边界。
可能增加新用户试用和生产接入成本。
Pitfall Log / 踩坑日志
项目: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_generatoronFastMCPfor 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
来源:Doramagic 发现、验证与编译记录