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/plain
  • annotations: 资源的注解信息
  • 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执行测试套件

核心规范

  1. 类型提示: 所有公共 API 必须包含类型提示
  2. 文档字符串: 公共 API 必须有 docstring,包含 Raises: 部分说明可能抛出的异常
  3. 异常处理:
  • 使用 logger.exception() 而非 logger.error()
  • 捕获具体异常类型而非通用 Exception
  • 禁止在顶级处理器之外使用 except Exception:
  1. 导入规范: 所有导入必须位于文件顶部,禁止内联导入
# 正确示例
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):
        ...

分支模型

分支用途变更类型
mainV2 开发分支新功能、破坏性变更
v1.x稳定版发布分支安全修复、关键 bug 修复

资料来源:AGENTS.md

发布流程

  1. pyproject.toml 中更新依赖版本
  2. 运行 uv lock --resolution lowest-direct 升级锁文件
  3. 通过 GitHub UI 创建 Release,标签格式为 vX.Y.Z
  4. 版本号将从标签自动设置

资料来源: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 版本

继续阅读本节完整说明和来源证据。

章节 1. Fork 并克隆仓库

继续阅读本节完整说明和来源证据。

环境要求

系统依赖

依赖项版本要求说明
Python3.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

资料来源:CONTRIBUTING.md:20-35

服务器配置

基本配置参数

客户端连接 MCP 服务器时可通过环境变量配置连接参数:

环境变量说明默认值
MCP_SERVER_PORTMCP 服务器端口号8000
MCP_TRANSPORT_TYPE传输类型:streamable-httpssestreamable-http
MCP_CLIENT_METADATA_URL客户端元数据 (CIMD) 的可选 URLNone

资料来源: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 采用双分支维护策略:

分支用途变更类型
mainV2 开发分支新功能、重构,允许破坏性变更
v1.xV1 稳定分支仅接收安全修复和关键 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]

代码提交检查清单

  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-httpsse
依赖冲突使用了 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)的定义与注册,以及如何启动服务。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 前置要求

继续阅读本节完整说明和来源证据。

章节 安装 SDK

继续阅读本节完整说明和来源证据。

章节 最小示例

继续阅读本节完整说明和来源证据。

环境准备

前置要求

要求版本说明
Python3.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 参数类型默认值说明
hoststr"127.0.0.1"监听地址
portint8000监听端口
streamable_http_transportboolFalse启用流式 HTTP 传输
oauth_strictboolFalse启用严格 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 参数说明

参数类型必填说明
fnCallable工具函数
namestr工具名称,默认为函数名
titlestr工具标题
descriptionstr工具描述
annotationsAnnotations工具注解

函数必须包含类型提示,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 字段类型说明
pathPath文件路径(必须为绝对路径)
is_binarybool是否以二进制模式读取
mime_typestrMIME 类型

资料来源: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 字段类型说明
namestr参数名称
descriptionstr参数描述
requiredbool是否必填

资料来源: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)
参数类型说明
uriAnyUrl \str资源标识符
contextContext请求上下文

异常处理:未知资源返回 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推荐默认选项
sseServer-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[调用工具/访问资源]

下一步

资料来源:[CONTRIBUTING.md:5-8]()

系统架构与核心组件

Model Context Protocol (MCP) Python SDK 是一个用于构建 Model Context Protocol 客户端和服务器的框架。MCP 是一种标准化协议,使 AI 模型能够与外部数据源和工具进行交互。本文详细介绍 MCP SDK 的系统架构和核心组件。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 2.1 组件层次结构

继续阅读本节完整说明和来源证据。

章节 2.2 组件职责表

继续阅读本节完整说明和来源证据。

章节 3.1 SharedSession 基类

继续阅读本节完整说明和来源证据。

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 --> Auth

2. 核心组件关系

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 --> JSONRPCHandler

2.2 组件职责表

组件文件路径职责
Clientsrc/mcp/client/client.py管理客户端连接生命周期,处理服务端发现和传输协商
ClientSessionsrc/mcp/client/session.py客户端会话管理,处理请求/响应、notifications 和 call cancellation
ServerSessionsrc/mcp/server/session.py服务端会话管理,处理请求路由和响应
SharedSessionsrc/mcp/shared/session.py会话基类,封装 JSON-RPC 消息处理逻辑
MCPServersrc/mcp/server/mcpserver/server.py高层服务器 API,整合资源、工具、提示管理
ResourceManagersrc/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 错误代码定义

错误码名称用途
-32700Parse ErrorJSON 解析失败
-32600Invalid Request无效的请求对象
-32601Method Not Found方法不存在
-32602Invalid Params无效的参数
-32603Internal Error内部错误
-32000Connection Closed连接已关闭
-32001Request 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:
参数类型描述
namestr服务器名称
versionstr服务器版本
settingsServerSettings服务器配置选项
resourceslist[Resource]初始资源列表
toolslist[Callable]可调用工具列表
promptslist[Prompt]预定义提示列表
instructionsstr服务器使用说明
lifespanLifespanContextManager生命周期管理器

资料来源: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 支持两种主要资源类型:

类型文件路径描述
FunctionResourceresources/types.py从函数动态生成的资源
FileResourceresources/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 -->|返回内容| ReadResourceContents

7. 认证系统

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 支持服务端向客户端发起交互请求:

交互类型方向用途
ElicitationServer → Client请求用户确认或输入
SamplingServer → 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
    end

8.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 的架构设计遵循以下原则:

  1. 分层设计:清晰的分层架构,从传输层到高层 API
  2. 协议标准化:基于 JSON-RPC 2.0,确保互操作性
  3. 灵活性:支持多种传输方式和认证机制
  4. 可扩展性:通过回调和中间件支持自定义行为
  5. 类型安全:全面的类型提示和 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 协议消息,支持不同的部署场景和性能需求。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 Stdio 传输

继续阅读本节完整说明和来源证据。

章节 Streamable HTTP 传输

继续阅读本节完整说明和来源证据。

章节 SSE 传输

继续阅读本节完整说明和来源证据。

概述

MCP (Model Context Protocol) Python SDK 支持多种传输协议实现,用于客户端与服务器之间的通信。这些传输层负责在进程间或网络间传递 MCP 协议消息,支持不同的部署场景和性能需求。

支持的传输协议类型

传输类型客户端模块服务器模块典型用途
stdiomcp.client.stdiomcp.server.stdio本地进程通信
Streamable HTTPmcp.client.streamable_httpmcp.server.streamable_httpHTTP-based 远程通信
SSEmcp.client.ssemcp.server.sse服务端推送事件
WebSocketmcp.client.websocketmcp.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.pysrc/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.pysrc/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.pysrc/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.pysrc/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()
        )

资料来源:src/mcp/server/stdio.py

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: 如何调试传输层问题?

  1. 启用调试日志:export MCP_LOG_LEVEL=debug
  2. 使用 HTTP 抓包工具查看请求/响应
  3. 对于 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 服务器。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 设计目标

继续阅读本节完整说明和来源证据。

章节 架构分层

继续阅读本节完整说明和来源证据。

章节 1. 服务器实例

继续阅读本节完整说明和来源证据。

框架概述

设计目标

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 实例,并提供了便捷的方法来注册工具、资源和提示。

初始化参数:

参数类型说明默认值
namestr服务器名称"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 类的高级封装。两者的主要区别:

特性ServerFastMCP
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

资料来源:[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 服务器,发现服务器提供的功能(工具、资源、提示),并通过标准化的协议与服务器进行交互。

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 客户端架构概述

继续阅读本节完整说明和来源证据。

章节 传输层支持

继续阅读本节完整说明和来源证据。

章节 建立 stdio 连接

继续阅读本节完整说明和来源证据。

核心概念

客户端架构概述

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_PORTMCP 服务器端口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

调试技巧

  1. 启用详细日志:在开发环境中启用 logging.DEBUG 级别可查看完整的协议交互
  2. 使用 stdio 传输:本地调试时优先使用 stdio 传输,便于观察输入输出
  3. 单元测试:为客户端代码编写单元测试,确保异常处理路径覆盖

最佳实践总结

类别建议
连接管理使用上下文管理器自动处理连接生命周期
错误处理捕获具体异常类型,使用 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 ...

章节 相关页面

继续阅读本节完整说明和来源证据。

章节 模块结构

继续阅读本节完整说明和来源证据。

章节 核心组件关系图

继续阅读本节完整说明和来源证据。

章节 OAuth 上下文数据流

继续阅读本节完整说明和来源证据。

概述

MCP Python SDK 的客户端 OAuth 认证模块为 MCP 客户端提供了完整的 OAuth 2.0 认证支持,包括授权码流程(Authorization Code Flow)和客户端凭证流程(Client Credentials Flow)。该模块基于 RFC 6749RFC 7636(PKCE)、RFC 8414(OAuth 授权服务器元数据)以及 RFC 9728(受保护资源元数据)等标准规范实现。

主要功能包括:

  • 动态客户端注册(DCR):自动向授权服务器注册客户端
  • PKCE 支持:增强授权码流程的安全性
  • Token 管理:自动刷新过期 Token
  • 元数据发现:支持 RFC 8414 和 RFC 9728 规定的标准发现端点
  • 多认证方式:支持 client_secret_basicclient_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 <--> L

OAuth 上下文数据流

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_idstr客户端标识符
client_secretstr客户端密钥
redirect_urislist[AnyUrl]重定向 URI 列表
grant_typeslist[str]支持的授权类型
token_endpoint_auth_methodstrToken 端点认证方法
scopestr请求的权限范围

资料来源:src/mcp/shared/auth.py:200-260

OAuthToken

Token 数据模型:

字段类型必填说明
access_tokenstr访问令牌
token_typestrToken 类型,通常为 "Bearer"
expires_atfloatToken 过期时间戳
refresh_tokenstr刷新令牌
expires_inintToken 有效期(秒)
scopestr授予的权限范围

资料来源:src/mcp/shared/auth.py:300-340

OAuthMetadata

授权服务器元数据模型:

字段类型说明
issuerstr授权服务器签发者标识
authorization_endpointstr授权端点 URL
token_endpointstrToken 端点 URL
registration_endpointstr客户端注册端点
scopes_supportedlist[str]支持的权限范围
response_types_supportedlist[str]支持的响应类型
grant_types_supportedlist[str]支持的授权类型
code_challenge_methods_supportedlist[str]支持的 PKCE 方法

资料来源:src/mcp/shared/auth.py:80-150

ProtectedResourceMetadata

受保护资源元数据(RFC 9728):

字段类型说明
resourcestr资源服务器标识符
authorization_serverslist[str]关联的授权服务器列表
scopes_supportedlist[str]资源支持的范围
bearer_methods_supportedlist[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 --> A
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 数据类

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_urlstr-MCP 服务器 URL
client_metadataOAuthClientMetadata-OAuth 客户端元数据
storageTokenStorage-Token 存储实现
redirect_handlerCallableNone授权重定向处理函数
callback_handlerCallableNone回调处理函数
timeoutfloat300.0OAuth 流程超时时间
client_metadata_urlstrNoneURL-based Client ID
validate_resource_urlCallableNone自定义资源 URL 验证

AuthSettings 配置

字段类型说明
issuer_urlAnyHttpUrl授权服务器 URL
resource_server_urlAnyHttpUrl资源服务器 URL
service_documentation_urlAnyHttpUrl服务文档 URL
client_registration_optionsClientRegistrationOptions客户端注册选项
revocation_optionsRevocationOptionsToken 撤销选项
required_scopeslist[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 在发现、验证和编译中沉淀的项目专属风险,不把社区讨论只当作装饰信息。

high 来源证据:[packit] Pull from upstream failed for release 1.27.1

可能增加新用户试用和生产接入成本。

high 来源证据:MCP client does not retry authenticated request after successful OAuth token exchange

可能影响授权、密钥配置或安全边界。

high 来源证据:OAuth token refresh sends RFC 8707 resource parameter that Entra ID v2.0 rejects (AADSTS9010010)

可能影响授权、密钥配置或安全边界。

medium 来源证据:Race condition between SSE session init and tools/call rejection (-32602)

可能增加新用户试用和生产接入成本。

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_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

来源:Doramagic 发现、验证与编译记录