Mcp Builder

MCP 服务器开发指南

概述

创建 MCP（模型上下文协议）服务器，使 LLM 能够通过精心设计的工具与外部服务交互。MCP 服务器的质量衡量标准是它能在多大程度上帮助 LLM 完成现实世界的任务。

---

流程

🚀 高级工作流程

创建高质量的 MCP 服务器涉及四个主要阶段：

第一阶段：深入研究和规划

1.1 理解现代 MCP 设计

API 覆盖率 vs 工作流工具： 平衡全面的 API 端点覆盖率和专业化的工作流工具。工作流工具对于特定任务可能更方便，而全面的覆盖率给予代理组合操作的灵活性。性能因客户端而异——有些客户端受益于组合基础工具的代码执行，而其他客户端在处理更高级别的工作流时表现更好。不确定时，优先考虑全面的 API 覆盖率。

工具命名和可发现性： 清晰、描述性的工具名称帮助代理快速找到正确的工具。使用一致的前缀（例如 github_create_issue、github_list_repos）和面向操作的命名。

上下文管理： 代理受益于简洁的工具描述和过滤/分页结果的能力。设计返回专注、相关数据的工具。一些客户端支持代码执行，这可以帮助代理高效地过滤和处理数据。

可操作的错误消息： 错误消息应该通过具体建议和后续步骤指导代理找到解决方案。

1.2 研究 MCP 协议文档

导航 MCP 规范：

从站点地图开始查找相关页面：https://modelcontextprotocol.io/sitemap.xml

然后使用 .md 后缀获取特定页面以获得 markdown 格式（例如 https://modelcontextprotocol.io/specification/draft.md）。

需要查看的关键页面：

• 规范概述和架构
• 传输机制（可流式 HTTP、stdio）
• 工具、资源和提示定义

1.3 研究框架文档

推荐技术栈：

• 语言：TypeScript（高质量的 SDK 支持和在许多执行环境（如 MCPB）中的良好兼容性。另外 AI 模型擅长生成 TypeScript 代码，受益于其广泛使用、静态类型和良好的 linting 工具）
• 传输：远程服务器使用可流式 HTTP，使用无状态 JSON（比有状态会话和流式响应更容易扩展和维护）。本地服务器使用 stdio。

加载框架文档：

• MCP 最佳实践：📋 查看最佳实践 - 核心指导原则

TypeScript（推荐）：

• TypeScript SDK：使用 WebFetch 加载 https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md
• ⚡ TypeScript 指南 - TypeScript 模式和示例

Python：

• Python SDK：使用 WebFetch 加载 https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md
• 🐍 Python 指南 - Python 模式和示例

1.4 规划你的实现

理解 API： 审查服务的 API 文档以识别关键端点、认证要求和数据模型。根据需要使用网络搜索和 WebFetch。

工具选择： 优先考虑全面的 API 覆盖率。列出要实现的端点，从最常见的操作开始。

---

第二阶段：实现

2.1 设置项目结构

查看特定语言的指南以进行项目设置：

• ⚡ TypeScript 指南 - 项目结构、package.json、tsconfig.json
• 🐍 Python 指南 - 模块组织、依赖项

2.2 实现核心基础设施

创建共享实用程序：

• 带认证的 API 客户端
• 错误处理助手
• 响应格式化（JSON/Markdown）
• 分页支持

2.3 实现工具

对于每个工具：

输入模式：

• 使用 Zod（TypeScript）或 Pydantic（Python）
• 包含约束和清晰的描述
• 在字段描述中添加示例

输出模式：

• 在可能的情况下为结构化数据定义 outputSchema
• 在工具响应中使用 structuredContent（TypeScript SDK 功能）
• 帮助客户端理解并处理工具输出

工具描述：

• 功能的简明摘要
• 参数描述
• 返回类型模式

实现：

• I/O 操作使用 async/await
• 具有可操作消息的适当错误处理
• 在适用的地方支持分页
• 使用现代 SDK 时返回文本内容和结构化数据

注释：

• readOnlyHint：true/false
• destructiveHint：true/false
• idempotentHint：true/false
• openWorldHint：true/false

---

第三阶段：审查和测试

3.1 代码质量

审查以下方面：

• 没有重复代码（DRY 原则）
• 一致的错误处理
• 完整的类型覆盖
• 清晰的工具描述

3.2 构建和测试

TypeScript：

• 运行 npm run build 验证编译
• 使用 MCP Inspector 测试：npx @modelcontextprotocol/inspector

Python：

• 验证语法：python -m py_compile your_server.py
• 使用 MCP Inspector 测试

查看特定语言的指南以了解详细的测试方法和质量检查清单。

---

第四阶段：创建评估

实现你的 MCP 服务器后，创建全面的评估以测试其有效性。

加载 [✅ 评估指南](./reference/evaluation.md) 获取完整的评估指导。

4.1 理解评估目的

使用评估来测试 LLM 是否能有效使用你的 MCP 服务器来回答现实、复杂的问题。

4.2 创建 10 个评估问题

要创建有效的评估，请遵循评估指南中概述的流程：

1. 工具检查：列出可用工具并了解它们的功能 2. 内容探索：使用只读操作探索可用数据 3. 问题生成：创建 10 个复杂的、现实的问题 4. 答案验证：自己解决每个问题以验证答案

4.3 评估要求

确保每个问题都满足：

• 独立性：不依赖于其他问题
• 只读：只需要非破坏性操作
• 复杂性：需要多个工具调用和深度探索
• 现实性：基于人类关心的真实用例
• 可验证性：可通过字符串比较验证的单一、清晰答案
• 稳定性：答案不会随时间变化

4.4 输出格式

创建具有以下结构的 XML 文件：

<evaluation>
  <qa_pair>
    <question>查找关于以动物代号命名的 AI 模型发布的讨论。一个模型需要使用格式 ASL-X 的特定安全指定。哪个数字 X 正在为以斑点野猫命名的模型确定？</question>
    <answer>3</answer>
  </qa_pair>
<!-- 更多 qa_pairs... -->
</evaluation>

---

参考文件

📚 文档库

在开发过程中根据需要加载这些资源：

核心 MCP 文档（首先加载）

• MCP 协议：从 https://modelcontextprotocol.io/sitemap.xml 的站点地图开始，然后使用 .md 后缀获取特定页面
• 📋 MCP 最佳实践 - 通用 MCP 指导原则，包括：
• 服务器和工具命名约定
• 响应格式指导原则（JSON vs Markdown）
• 分页最佳实践
• 传输选择（可流式 HTTP vs stdio）
• 安全和错误处理标准

SDK 文档（在第一/二阶段加载）

• Python SDK：从 https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md 获取
• TypeScript SDK：从 https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md 获取

特定语言实现指南（在第二阶段加载）

• 🐍 Python 实现指南 - 完整的 Python/FastMCP 指南，包含：
• 服务器初始化模式
• Pydantic 模型示例
• 使用 @mcp.tool 的工具注册
• 完整的工作示例
• 质量检查清单

• ⚡ TypeScript 实现指南 - 完整的 TypeScript 指南，包含：
• 项目结构
• Zod 模式模式
• 使用 server.registerTool 的工具注册
• 完整的工作示例
• 质量检查清单

评估指南（在第四阶段加载）

• ✅ 评估指南 - 完整的评估创建指南，包含：
• 问题创建指导原则
• 答案验证策略
• XML 格式规范
• 示例问题和答案
• 使用提供的脚本运行评估

MCP 服务器评估指南

概述

本文档提供了为 MCP 服务器创建全面评估的指导。评估测试 LLM 是否能仅使用提供的工具有效使用你的 MCP 服务器来回答现实、复杂的问题。

---

快速参考

评估要求

• 创建 10 个人类可读的问题
• 问题必须是只读的、独立的、非破坏性的
• 每个问题需要多个工具调用（可能数十个）
• 答案必须是单一、可验证的值
• 答案必须是稳定的（不会随时间变化）

输出格式

<evaluation>
   <qa_pair>
      <question>你的问题在这里</question>
      <answer>单一可验证答案</answer>
   </qa_pair>
</evaluation>

---

评估的目的

MCP 服务器的质量衡量标准不是服务器实现工具的好坏或全面程度，而是这些实现（输入/输出模式、文档字符串/描述、功能）如何在没有其他上下文且只能访问 MCP 服务器的情况下，使 LLM 能够回答现实且困难的问题。

评估概述

创建 10 个人类可读的问题，仅需要只读、独立、非破坏性和幂等操作来回答。每个问题应该：

• 现实的
• 清晰简洁的
• 无歧义的
• 复杂的，可能需要几十个工具调用或步骤
• 可以用你提前识别的单一、可验证值回答

问题指导原则

核心要求

1. 问题必须是独立的

• 每个问题不应依赖于任何其他问题的答案
• 不应假设处理另一个问题时的先写操作

2. 问题必须只需要非破坏性和幂等工具使用

• 不应指示或要求修改状态以获得正确答案

3. 问题必须是现实的、清晰的、简洁的和复杂的

• 必须要求另一个 LLM 使用多个（可能几十个）工具或步骤来回答

复杂性和深度

4. 问题必须需要深度探索

• 考虑需要多个子问题和顺序工具调用的多跳问题
• 每一步都应该受益于在之前问题中发现的信息

5. 问题可能需要大量分页

• 可能需要分页浏览多个结果页面
• 可能需要查询旧数据（1-2 年前的）以找到小众信息
• 问题必须是困难的

6. 问题必须需要深度理解

• 而不是表面知识
• 可能将复杂思想提出为需要证据的真/假问题
• 可能使用多项选择格式，其中 LLM 必须搜索不同的假设

7. 问题不能通过直接关键词搜索解决

• 不要包含目标内容中的特定关键词
• 使用同义词、相关概念或转述
• 需要多个搜索、分析多个相关项目、提取上下文，然后推导答案

工具测试

8. 问题应该对工具返回值进行压力测试

• 可能引发工具返回大型 JSON 对象或列表，压垮 LLM
• 应该需要理解多种数据模式：
• ID 和名称
• 时间戳和日期时间（月、日、年、秒）
• 文件 ID、名称、扩展名和 mime 类型
• URL、GID 等
• 应该探查工具返回所有有用形式数据的能力

9. 问题应该主要反映真实的人类用例

• LLM 辅助的人类会关心的信息检索任务类型

10. 问题可能需要几十个工具调用

• 这对具有有限上下文的 LLM 构成挑战
• 鼓励 MCP 服务器工具减少返回的信息

11. 包含模糊问题

• 可能是模糊的或需要关于调用哪些工具的困难决策
• 强迫 LLM 可能犯错或误解
• 确保尽管存在模糊性，仍然有单一可验证答案

稳定性

12. 问题必须设计成答案不会改变

• 不要问依赖于"当前状态"的问题，这是动态的
• 例如，不要计算：
• 帖子的反应数
• 线程的回复数
• 频道中的成员数

13. 不要让 MCP 服务器限制你创建的问题类型

• 创建具有挑战性和复杂性的问题
• 有些可能无法用可用的 MCP 服务器工具解决
• 问题可能需要特定的输出格式（datetime vs epoch time，JSON vs MARKDOWN）
• 问题可能需要几十个工具调用才能完成

答案指导原则

验证

1. 答案必须可通过直接字符串比较验证

• 如果答案可以用多种格式重写，在问题中明确指定输出格式
• 示例："使用 YYYY/MM/DD。"，"回答 True 或 False。"，"回答 A、B、C 或 D，仅此而已。"
• 答案应该是单一可验证值，如：
• 用户 ID、用户名、显示名、名字、姓氏
• 频道 ID、频道名称
• 消息 ID、字符串
• URL、标题
• 数值
• 时间戳、日期时间
• 布尔值（对于真/假问题）
• 电子邮件地址、电话号码
• 文件 ID、文件名、文件扩展名
• 多项选择答案
• 答案不需要特殊格式或复杂的结构化输出
• 答案将使用直接字符串比较验证

可读性

2. 答案通常应该偏好人类可读的格式

• 示例：名称、名字、姓氏、日期时间、文件名、消息字符串、URL、是/否、真/假、a/b/c/d
• 而不是不透明的 ID（尽管 ID 是可接受的）
• 绝大多数答案应该是人类可读的

稳定性

3. 答案必须是稳定/静止的

• 查看旧内容（例如，已结束的对话、已启动的项目、已回答的问题）
• 基于"关闭"概念创建问题，这些概念将始终返回相同答案
• 问题可能要求考虑固定时间窗口以避免非静止答案
• 依赖于不太可能改变的上下文
• 示例：如果查找论文名称，要足够具体，以免与后来发表的论文混淆

4. 答案必须是清晰和无歧义的

• 问题必须设计成有单一、清晰的答案
• 答案可以从使用 MCP 服务器工具推导出来

多样性

5. 答案必须是多样的

• 答案应该是单一可验证值，具有多种模式和格式
• 用户概念：用户 ID、用户名、显示名、名字、姓氏、电子邮件地址、电话号码
• 频道概念：频道 ID、频道名称、频道主题
• 消息概念：消息 ID、消息字符串、时间戳、月、日、年

6. 答案不能是复杂结构

• 不是值列表
• 不是复杂对象
• 不是 ID 或字符串列表
• 不是自然语言文本
• 除非答案可以通过直接字符串比较直接验证
• 并且可以现实地重现
• LLM 不太可能以任何其他顺序或格式返回相同列表的可能性应该很小

评估流程

第 1 步：文档检查

阅读目标 API 的文档以了解：

• 可用的端点和功能
• 如果存在模糊性，从网络获取额外信息
• 尽可能并行化此步骤
• 确保每个子代理只检查文件系统或网络上的文档

第 2 步：工具检查

列出 MCP 服务器中可用的工具：

• 直接检查 MCP 服务器
• 理解输入/输出模式、文档字符串和描述
• 在此阶段不要调用工具本身

第 3 步：发展理解

重复步骤 1 和 2，直到你有良好的理解：

• 多次迭代
• 考虑你想要创建的任务类型
• 完善你的理解
• 在任何阶段都不应阅读 MCP 服务器实现本身的代码
• 使用你的直觉和理解来创建合理、现实但非常具有挑战性的任务

第 4 步：只读内容检查

理解 API 和工具后，使用 MCP 服务器工具：

• 仅使用只读和非破坏性操作检查内容
• 目标：识别特定内容（例如，用户、频道、消息、项目、任务）以创建现实问题
• 不应调用任何修改状态的工具
• 不会阅读 MCP 服务器实现本身的代码
• 与追求独立探索的各个子代理并行化此步骤
• 确保每个子代理只执行只读、非破坏性和幂等操作
• 注意：一些工具可能返回大量数据，这会使你超出上下文限制
• 进行增量、小而有针对性的工具调用进行探索
• 在所有工具调用请求中，使用 limit 参数限制结果（<10）
• 使用分页

第 5 步：任务生成

检查内容后，创建 10 个人类可读的问题：

• LLM 应该能够使用 MCP 服务器回答这些问题
• 遵循上述所有问题和答案指导原则

输出格式

每个 QA 对由一个问题和一个答案组成。输出应该是具有以下结构的 XML 文件：

<evaluation>
   <qa_pair>
      <question>查找在 2024 年第二季度创建的具有最高已完成任务数的项目的项目名称。</question>
      <answer>网站重新设计</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索标记为"bug"且在 2024 年 3 月关闭的问题。哪个用户关闭了最多问题？提供他们的用户名。</question>
      <answer>sarah_dev</answer>
   </qa_pair>
   <qa_pair>
      <question>查找修改了 /api 目录中的文件并在 2024 年 1 月 1 日至 2024 年 1 月 31 日之间合并的拉取请求。有多少不同的贡献者处理了这些 PR？</question>
      <answer>7</answer>
   </qa_pair>
   <qa_pair>
      <question>查找在 2023 年之前创建的星数最多的存储库。存储库名称是什么？</question>
      <answer>data-pipeline</answer>
   </qa_pair>
</evaluation>

评估示例

好的问题

示例 1：需要深度探索的多跳问题（GitHub MCP）

<qa_pair>
   <question>查找在 2023 年第三季度归档且之前是组织中最多被分叉项目的存储库。该存储库中使用的编程语言是什么？</question>
   <answer>Python</answer>
</qa_pair>

这个问题是好的，因为：

• 需要多次搜索来找到归档的存储库
• 需要识别在归档前哪个具有最多分叉
• 需要检查存储库详细信息以了解语言
• 答案是简单、可验证的值
• 基于历史（已关闭）数据，不会改变

示例 2：需要在没有关键词匹配的情况下理解上下文（项目管理 MCP）

<qa_pair>
   <question>定位在 2023 年底完成的专注于改善客户入职的倡议。项目主管在完成后创建了一个回顾文档。主管当时的角色头衔是什么？</question>
   <answer>产品经理</answer>
</qa_pair>

这个问题是好的，因为：

• 不使用特定项目名称（"专注于改善客户入职的倡议"）
• 需要从特定时间范围查找已完成项目
• 需要识别项目主管和他们的角色
• 需要从回顾文档理解上下文
• 答案是人类可读且稳定的
• 基于已完成的工作（不会改变）

示例 3：需要多个步骤的复杂聚合（问题跟踪器 MCP）

<qa_pair>
   <question>在 2024 年 1 月报告并标记为关键优先级的所有错误中，哪个受让人在 48 小时内解决了其分配错误的最高百分比？提供受让人的用户名。</question>
   <answer>alex_eng</answer>
</qa_pair>

这个问题是好的，因为：

• 需要按日期、优先级和状态过滤错误
• 需要按受让人分组并计算解决率
• 需要理解时间戳以确定 48 小时窗口
• 测试分页（可能需要处理许多错误）
• 答案是单一用户名
• 基于特定时间段的历史数据

示例 4：需要跨多种数据类型综合（CRM MCP）

<qa_pair>
   <question>查找在 2023 年第四季度从 Starter 升级到 Enterprise 计划且具有最高年度合同价值的账户。该账户在哪个行业运营？</question>
   <answer>医疗保健</answer>
</qa_pair>

这个问题是好的，因为：

• 需要理解订阅级别变化
• 需要识别特定时间范围内的升级事件
• 需要比较合同价值
• 必须访问账户行业信息
• 答案简单且可验证
• 基于已完成的历史交易

差的问题

示例 1：答案随时间变化

<qa_pair>
   <question>当前分配给工程团队有多少个开放问题？</question>
   <answer>47</answer>
</qa_pair>

这个问题是差的，因为：

• 答案会随着问题被创建、关闭或重新分配而改变
• 不基于稳定/静止数据
• 依赖于动态的"当前状态"

示例 2：关键词搜索太容易

<qa_pair>
   <question>查找标题为"Add authentication feature"的拉取请求，告诉我谁创建了它。</question>
   <answer>developer123</answer>
</qa_pair>

这个问题是差的，因为：

• 可以通过对确切标题的简单关键词搜索解决
• 不需要深度探索或理解
• 不需要综合或分析

示例 3：答案格式模糊

<qa_pair>
   <question>列出所有以 Python 为主要语言的存储库。</question>
   <answer>repo1, repo2, repo3, data-pipeline, ml-tools</answer>
</qa_pair>

这个问题是差的，因为：

• 答案是可能以任何顺序返回的列表
• 难以通过直接字符串比较验证
• LLM 可能格式不同（JSON 数组、逗号分隔、换行符分隔）
• 更好的方式是询问特定聚合（计数）或最高级（最多星数）

验证流程

创建评估后：

1. 检查 XML 文件以了解模式 2. 加载每个任务指令并使用 MCP 服务器和工具，通过尝试自己解决任务来识别正确答案 3. 标记任何需要写入或破坏性操作的操作 4. 累积所有正确答案并替换文档中任何不正确的答案 5. 删除任何 `<qa_pair>`需要写入或破坏性操作

记住要并行化解决任务以避免超出上下文，然后累积所有答案并在最后对文件进行更改。

创建质量评估的技巧

1. 努力思考并提前规划然后再生成任务 2. 在机会出现时并行化以加速流程和管理上下文 3. 专注于现实用例，这些是人类实际想要完成的 4. 创建具有挑战性的问题，测试 MCP 服务器能力的极限 5. 通过使用历史数据和关闭概念确保稳定性 6. 通过自己使用 MCP 服务器工具解决问题来验证答案 7. 根据你在过程中学到的东西迭代和完善

---

运行评估

创建评估文件后，你可以使用提供的评估工具来测试你的 MCP 服务器。

设置

1. 安装依赖项

   pip install -r scripts/requirements.txt

或手动安装：

   pip install anthropic mcp

2. 设置 API 密钥

   export ANTHROPIC_API_KEY=your_api_key_here

评估文件格式

评估文件使用带有 <qa_pair> 元素的 XML 格式：

<evaluation>
   <qa_pair>
      <question>查找在 2024 年第二季度创建的具有最高已完成任务数的项目的项目名称。</question>
      <answer>网站重新设计</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索标记为"bug"且在 2024 年 3 月关闭的问题。哪个用户关闭了最多问题？提供他们的用户名。</question>
      <answer>sarah_dev</answer>
   </qa_pair>
</evaluation>

运行评估

评估脚本（scripts/evaluation.py）支持三种传输类型：

重要说明：

• stdio 传输：评估脚本自动为你启动和管理 MCP 服务器进程。不要手动运行服务器。
• sse/http 传输：在运行评估前必须单独启动 MCP 服务器。脚本连接到指定 URL 上已运行的服务器。

1. 本地 STDIO 服务器

对于本地运行的 MCP 服务器（脚本自动启动服务器）：

python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a my_mcp_server.py \
  evaluation.xml

使用环境变量：

python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a my_mcp_server.py \
  -e API_KEY=abc123 \
  -e DEBUG=true \
  evaluation.xml

2. 服务器发送事件 (SSE)

对于基于 SSE 的 MCP 服务器（必须先启动服务器）：

python scripts/evaluation.py \
  -t sse \
  -u https://example.com/mcp \
  -H "Authorization: Bearer token123" \
  -H "X-Custom-Header: value" \
  evaluation.xml

3. HTTP（可流式 HTTP）

对于基于 HTTP 的 MCP 服务器（必须先启动服务器）：

python scripts/evaluation.py \
  -t http \
  -u https://example.com/mcp \
  -H "Authorization: Bearer token123" \
  evaluation.xml

命令行选项

usage: evaluation.py [-h] [-t {stdio,sse,http}] [-m MODEL] [-c COMMAND]
                     [-a ARGS [ARGS ...]] [-e ENV [ENV ...]] [-u URL]
                     [-H HEADERS [HEADERS ...]] [-o OUTPUT]
                     eval_file

positional arguments:
  eval_file             评估 XML 文件的路径

optional arguments:
  -h, --help            显示帮助消息
  -t, --transport       传输类型：stdio、sse 或 http（默认：stdio）
  -m, --model           要使用的 Claude 模型（默认：claude-3-7-sonnet-20250219）
  -o, --output          报告的输出文件（默认：打印到 stdout）

stdio 选项:
  -c, --command         运行 MCP 服务器的命令（例如，python、node）
  -a, --args            命令的参数（例如，server.py）
  -e, --env             格式为 KEY=VALUE 的环境变量

sse/http 选项:
  -u, --url             MCP 服务器 URL
  -H, --header          格式为 'Key: Value' 的 HTTP 头

输出

评估脚本生成详细的报告，包括：

• 摘要统计：
• 准确性（正确/总数）
• 平均任务持续时间
• 每个任务的平均工具调用
• 总工具调用

• 每个任务结果：
• 提示和预期响应
• 来自代理的实际响应
• 答案是否正确（✅/❌）
• 持续时间和工具调用详细信息
• 代理对其方法的总结
• 代理对工具的反馈

将报告保存到文件

python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a my_server.py \
  -o evaluation_report.md \
  evaluation.xml

完整示例工作流程

以下是创建和运行评估的完整示例：

1. 创建你的评估文件（my_evaluation.xml）：

<evaluation>
   <qa_pair>
      <question>查找在 2024 年 1 月创建最多问题的用户。他们的用户名是什么？</question>
      <answer>alice_developer</answer>
   </qa_pair>
   <qa_pair>
      <question>在 2024 年第一季度合并的所有拉取请求中，哪个存储库的数量最多？提供存储库名称。</question>
      <answer>backend-api</answer>
   </qa_pair>
   <qa_pair>
      <question>查找在 2023 年 12 月完成且从开始到结束持续时间最长的项目。它花费了多少天？</question>
      <answer>127</answer>
   </qa_pair>
</evaluation>

2. 安装依赖项：

pip install -r scripts/requirements.txt
export ANTHROPIC_API_KEY=your_api_key

3. 运行评估：

python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a github_mcp_server.py \
  -e GITHUB_TOKEN=ghp_xxx \
  -o github_eval_report.md \
  my_evaluation.xml

4. 查看报告在 github_eval_report.md 中：

• 查看哪些问题通过/失败
• 阅读代理对你的工具的反馈
• 识别改进领域
• 迭代你的 MCP 服务器设计

故障排除

连接错误

如果遇到连接错误：

• STDIO：验证命令和参数是否正确
• SSE/HTTP：检查 URL 是否可访问且头是否正确
• 确保在环境变量或头中设置任何所需的 API 密钥

准确性低

如果许多评估失败：

• 查看每个任务的代理反馈
• 检查工具描述是否清晰全面
• 验证输入参数是否记录良好
• 考虑工具是否返回过多或过少的数据
• 确保错误消息是可操作的

超时问题

如果任务超时：

• 使用更有能力的模型（例如，claude-3-7-sonnet-20250219）
• 检查工具是否返回过多数据
• 验证分页是否正常工作
• 考虑简化复杂问题

MCP 服务器最佳实践

快速参考

服务器命名

• Python：{service}_mcp（例如 slack_mcp）
• Node/TypeScript：{service}-mcp-server（例如 slack-mcp-server）

工具命名

• 使用 snake_case 加服务前缀
• 格式：{service}_{action}_{resource}
• 示例：slack_send_message、github_create_issue

响应格式

• 同时支持 JSON 和 Markdown 格式
• JSON 用于程序化处理
• Markdown 用于人类可读性

分页

• 始终尊重 limit 参数
• 返回 has_more、next_offset、total_count
• 默认为 20-50 项

传输

• 可流式 HTTP：用于远程服务器、多客户端场景
• stdio：用于本地集成、命令行工具
• 避免 SSE（已弃用，改用可流式 HTTP）

---

服务器命名约定

遵循这些标准化的命名模式：

Python：使用格式 {service}_mcp（小写加下划线）

• 示例：slack_mcp、github_mcp、jira_mcp

Node/TypeScript：使用格式 {service}-mcp-server（小写加连字符）

• 示例：slack-mcp-server、github-mcp-server、jira-mcp-server

名称应该是：

• 通用的，描述被集成的服务
• 容易从任务描述推断出来
• 不包含版本号

---

工具命名和设计

工具命名

1. 使用 snake_case：search_users、create_project、get_channel_info 2. 包含服务前缀：预期你的 MCP 服务器可能与其他 MCP 服务器一起使用

• 使用 slack_send_message 而不仅仅是 send_message
• 使用 github_create_issue 而不仅仅是 create_issue

3. 以行动为导向：以动词开头（get、list、search、create 等） 4. 具体化：避免可能与其它服务器冲突的通用名称

工具设计

• 工具描述必须精确且无歧义地描述功能
• 描述必须与实际功能精确匹配
• 提供工具注释（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
• 保持工具操作专注和原子化

---

响应格式

所有返回数据的工具都应该支持多种格式：

JSON 格式（response_format="json"）

• 机器可读的结构化数据
• 包含所有可用字段和元数据
• 一致的字段名称和类型
• 用于程序化处理

Markdown 格式（response_format="markdown"，通常为默认）

• 人类可读的格式化文本
• 使用标题、列表和格式化以清晰显示
• 将时间戳转换为人类可读格式
• 在括号中显示 ID 的显示名称
• 省略详细的元数据

---

分页

对于列出资源的工具：

• 始终尊重 `limit` 参数
• 实现分页：使用 offset 或基于游标的分页
• 返回分页元数据：包含 has_more、next_offset/next_cursor、total_count
• 永远不要将所有结果加载到内存中：对于大数据集尤其重要
• 默认为合理的限制：通常 20-50 项

示例分页响应：

{
  "total": 150,
  "count": 20,
  "offset": 0,
  "items": [...],
  "has_more": true,
  "next_offset": 20
}

---

传输选项

可流式 HTTP

最适合：远程服务器、Web 服务、多客户端场景

特性：

• 通过 HTTP 的双向通信
• 支持多个同时客户端
• 可部署为 Web 服务
• 启用服务器到客户端通知

在以下情况使用：

• 同时为多个客户端提供服务
• 部署为云服务
• 与 Web 应用程序集成

stdio

最适合：本地集成、命令行工具

特性：

• 标准输入/输出流通信
• 设置简单，不需要网络配置
• 作为客户端的子进程运行

在以下情况使用：

• 为本地开发环境构建工具
• 与桌面应用程序集成
• 单用户、单会话场景

注意：stdio 服务器不应记录到 stdout（使用 stderr 进行记录）

传输选择

---

安全最佳实践

认证和授权

OAuth 2.1：

• 使用来自受认可机构的证书进行安全 OAuth 2.1
• 在处理请求之前验证访问令牌
• 只接受专门为你的服务器指定的令牌

API 密钥：

• 将 API 密钥存储在环境变量中，绝不在代码中存储
• 在服务器启动时验证密钥
• 当认证失败时提供清晰的错误消息

输入验证

• 清理文件路径以防止目录遍历
• 验证 URL 和外部标识符
• 检查参数大小和范围
• 在系统调用中防止命令注入
• 对所有输入使用模式验证（Pydantic/Zod）

错误处理

• 不要向客户端暴露内部错误
• 在服务器端记录安全相关错误
• 提供有帮助但不揭示信息的错误消息
• 在错误后清理资源

DNS 重新绑定保护

对于本地运行的可流式 HTTP 服务器：

• 启用 DNS 重新绑定保护
• 验证所有传入连接上的 Origin 头
• 绑定到 127.0.0.1 而不是 0.0.0.0

---

工具注释

提供注释以帮助客户端理解工具行为：

重要：注释是提示，不是安全保证。客户端不应仅基于注释做出安全关键决策。

---

错误处理

• 使用标准 JSON-RPC 错误代码
• 在结果对象中报告工具错误（而不是协议级错误）
• 提供有用的、具体的错误消息并建议后续步骤
• 不要暴露内部实现细节
• 在错误时正确清理资源

示例错误处理：

try {
  const result = performOperation();
  return { content: [{ type: "text", text: result }] };
} catch (error) {
  return {
    isError: true,
    content: [{
      type: "text",
      text: `错误：${error.message}。尝试使用 filter='active_only' 来减少结果。`
    }]
  };
}

---

测试要求

全面的测试应该覆盖：

• 功能测试：使用有效/无效输入验证正确执行
• 集成测试：与外部系统的交互测试
• 安全测试：验证认证、输入清理、速率限制
• 性能测试：负载下的行为、超时检查
• 错误处理：确保正确的错误报告和清理

---

文档要求

• 提供所有工具和功能的清晰文档
• 包含工作示例（每个主要功能至少 3 个）
• 记录安全注意事项
• 指定所需权限和访问级别
• 记录速率限制和性能特征

Node/TypeScript MCP 服务器实现指南

概述

本文档提供了使用 MCP TypeScript SDK 实现 MCP 服务器的 Node/TypeScript 特定最佳实践和示例。它涵盖了项目结构、服务器设置、工具注册模式、使用 Zod 进行输入验证、错误处理和完整的工作示例。

---

快速参考

关键导入

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import express from "express";
import { z } from "zod";

服务器初始化

const server = new McpServer({
  name: "service-mcp-server",
  version: "1.0.0"
});

工具注册模式

server.registerTool(
  "tool_name",
  {
    title: "工具显示名称",
    description: "工具的作用",
    inputSchema: { param: z.string() },
    outputSchema: { result: z.string() }
  },
  async ({ param }) => {
    const output = { result: `已处理：${param}` };
    return {
      content: [{ type: "text", text: JSON.stringify(output) }],
      structuredContent: output // 结构化数据的现代模式
    };
  }
);

---

MCP TypeScript SDK

官方 MCP TypeScript SDK 提供：

• 用于服务器初始化的 McpServer 类
• 用于工具注册的 registerTool 方法
• 用于运行时输入验证的 Zod 模式集成
• 类型安全的工具处理器实现

重要 - 仅使用现代 API：

• 应该使用：server.registerTool()、server.registerResource()、server.registerPrompt()
• 不应使用：旧弃用 API 如 server.tool()、server.setRequestHandler(ListToolsRequestSchema, ...) 或手动处理器注册
• register* 方法提供更好的类型安全性、自动模式处理，是推荐的方法

有关完整详细信息，请参阅参考文档中的 MCP SDK 文档。

服务器命名约定

Node/TypeScript MCP 服务器必须遵循此命名模式：

• 格式：{service}-mcp-server（小写加连字符）
• 示例：github-mcp-server、jira-mcp-server、stripe-mcp-server

名称应该是：

• 通用的（不绑定到特定功能）
• 描述被集成的服务/API
• 容易从任务描述推断
• 不包含版本号或日期

项目结构

为 Node/TypeScript MCP 服务器创建以下结构：

{service}-mcp-server/
├── package.json
├── tsconfig.json
├── README.md
├── src/
│   ├── index.ts          # 带有 McpServer 初始化的主入口点
│   ├── types.ts          # TypeScript 类型定义和接口
│   ├── tools/            # 工具实现（每个域一个文件）
│   ├── services/         # API 客户端和共享实用程序
│   ├── schemas/          # Zod 验证模式
│   └── constants.ts      # 共享常量（API_URL、CHARACTER_LIMIT 等）
└── dist/                 # 构建的 JavaScript 文件（入口点：dist/index.js）

工具实现

工具命名

使用 snake_case 作为工具名称（例如，"search_users"、"create_project"、"get_channel_info"），具有清晰、面向操作的名称。

避免命名冲突：包含服务上下文以防止重叠：

• 使用 "slack_send_message" 而不仅仅是 "send_message"
• 使用 "github_create_issue" 而不仅仅是 "create_issue"
• 使用 "asana_list_tasks" 而不仅仅是 "list_tasks"

工具结构

使用 registerTool 方法注册工具，具有以下要求：

• 使用 Zod 模式进行运行时输入验证和类型安全
• description 字段必须明确提供 - JSDoc 注释不会自动提取
• 明确提供 title、description、inputSchema 和 annotations
• inputSchema 必须是 Zod 模式对象（不是 JSON 模式）
• 明确键入所有参数和返回值

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

const server = new McpServer({
  name: "example-mcp",
  version: "1.0.0"
});

// 输入验证的 Zod 模式
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "查询必须至少 2 个字符")
    .max(200, "查询不能超过 200 个字符")
    .describe("匹配姓名/电子邮件的搜索字符串"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("要返回的最大结果数"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("分页要跳过的结果数"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("输出格式：'markdown' 用于人类可读或 'json' 用于机器可读")
}).strict();

// 从 Zod 模式的类型定义
type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

server.registerTool(
  "example_search_users",
  {
    title: "搜索示例用户",
    description: `在示例系统中按姓名、电子邮件或团队搜索用户。

此工具搜索示例平台中的所有用户配置文件，支持部分匹配和各种搜索过滤器。它不创建或修改用户，仅搜索现有用户。

参数：
  - query (string)：匹配姓名/电子邮件的搜索字符串
  - limit (number)：要返回的最大结果数，介于 1-100 之间（默认：20）
  - offset (number)：分页要跳过的结果数（默认：0）
  - response_format ('markdown' | 'json')：输出格式（默认：'markdown'）

返回：
  JSON 格式：具有以下模式的结构化数据：
  {
    "total": number,           // 找到的匹配总数
    "count": number,           // 此响应中的结果数
    "offset": number,          // 当前分页偏移量
    "users": [
      {
        "id": string,          // 用户 ID（例如，"U123456789"）
        "name": string,        // 全名（例如，"John Doe"）
        "email": string,       // 电子邮件地址
        "team": string,        // 团队名称（可选）
        "active": boolean      // 用户是否活跃
      }
    ],
    "has_more": boolean,       // 是否有更多结果可用
    "next_offset": number      // 下一页的偏移量（如果 has_more 为 true）
  }

示例：
  - 使用时机："查找所有营销团队成员" -> 参数 query="team:marketing"
  - 使用时机："搜索 John 的账户" -> 参数 query="john"
  - 不使用时机：你需要创建用户（使用 example_create_user 代替）

错误处理：
  - 如果请求过多返回 "错误：速率限制"（429 状态）
  - 如果搜索返回空值返回 "未找到匹配 '<query>' 的用户"`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    try {
      // 输入验证由 Zod 模式处理
      // 使用验证参数发出 API 请求
      const data = await makeApiRequest<any>(
        "users/search",
        "GET",
        undefined,
        {
          q: params.query,
          limit: params.limit,
          offset: params.offset
        }
      );

      const users = data.users || [];
      const total = data.total || 0;

      if (!users.length) {
        return {
          content: [{
            type: "text",
            text: `未找到匹配 '${params.query}' 的用户`
          }]
        };
      }

      // 准备结构化输出
      const output = {
        total,
        count: users.length,
        offset: params.offset,
        users: users.map((user: any) => ({
          id: user.id,
          name: user.name,
          email: user.email,
          ...(user.team ? { team: user.team } : {}),
          active: user.active ?? true
        })),
        has_more: total > params.offset + users.length,
        ...(total > params.offset + users.length ? {
          next_offset: params.offset + users.length
        } : {})
      };

      // 根据请求格式格式化文本表示
      let textContent: string;
      if (params.response_format === ResponseFormat.MARKDOWN) {
        const lines = [`# 用户搜索结果：'${params.query}'`, "",
          `找到 ${total} 个用户（显示 ${users.length} 个）`, ""];
        for (const user of users) {
          lines.push(`## ${user.name} (${user.id})`);
          lines.push(`- **电子邮件**：${user.email}`);
          if (user.team) lines.push(`- **团队**：${user.team}`);
          lines.push("");
        }
        textContent = lines.join("\n");
      } else {
        textContent = JSON.stringify(output, null, 2);
      }

      return {
        content: [{ type: "text", text: textContent }],
        structuredContent: output // 结构化数据的现代模式
      };
    } catch (error) {
      return {
        content: [{
          type: "text",
          text: handleApiError(error)
        }]
      };
    }
  }
);

用于输入验证的 Zod 模式

Zod 提供运行时类型验证：

import { z } from "zod";

// 带验证的基本模式
const CreateUserSchema = z.object({
  name: z.string()
    .min(1, "姓名是必需的")
    .max(100, "姓名不能超过 100 个字符"),
  email: z.string()
    .email("无效的电子邮件格式"),
  age: z.number()
    .int("年龄必须是整数")
    .min(0, "年龄不能为负")
    .max(150, "年龄不能超过 150")
}).strict();  // 使用 .strict() 禁止额外字段

// 枚举
enum ResponseFormat {
  MARKDOWN = "markdown",
  JSON = "json"
}

const SearchSchema = z.object({
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("输出格式")
});

// 带默认值的可选字段
const PaginationSchema = z.object({
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("要返回的最大结果数"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("要跳过的结果数")
});

响应格式选项

为灵活性支持多种输出格式：

enum ResponseFormat {
  MARKDOWN = "markdown",
  JSON = "json"
}

const inputSchema = z.object({
  query: z.string(),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("输出格式：'markdown' 用于人类可读或 'json' 用于机器可读")
});

Markdown 格式：

• 使用标题、列表和格式化以清晰显示
• 将时间戳转换为人类可读格式
• 在括号中显示 ID 的显示名称
• 省略详细的元数据
• 逻辑地分组相关信息

JSON 格式：

• 返回适合程序化处理的完整结构化数据
• 包含所有可用字段和元数据
• 使用一致的字段名称和类型

分页实现

对于列出资源的工具：

const ListSchema = z.object({
  limit: z.number().int().min(1).max(100).default(20),
  offset: z.number().int().min(0).default(0)
});

async function listItems(params: z.infer<typeof ListSchema>) {
  const data = await apiRequest(params.limit, params.offset);

  const response = {
    total: data.total,
    count: data.items.length,
    offset: params.offset,
    items: data.items,
    has_more: data.total > params.offset + data.items.length,
    next_offset: data.total > params.offset + data.items.length
      ? params.offset + data.items.length
      : undefined
  };

  return JSON.stringify(response, null, 2);
}

字符限制和截断

添加 CHARACTER_LIMIT 常量以防止响应过大：

// 在 constants.ts 中的模块级别
export const CHARACTER_LIMIT = 25000;  // 以字符为单位的最大响应大小

async function searchTool(params: SearchInput) {
  let result = generateResponse(data);

  // 检查字符限制并在需要时截断
  if (result.length > CHARACTER_LIMIT) {
    const truncatedData = data.slice(0, Math.max(1, data.length / 2));
    response.data = truncatedData;
    response.truncated = true;
    response.truncation_message =
      `响应从 ${data.length} 截断到 ${truncatedData.length} 项。` +
      `使用 'offset' 参数或添加过滤器以查看更多结果。`;
    result = JSON.stringify(response, null, 2);
  }

  return result;
}

错误处理

提供清晰的、可操作的错误消息：

import axios, { AxiosError } from "axios";

function handleApiError(error: unknown): string {
  if (error instanceof AxiosError) {
    if (error.response) {
      switch (error.response.status) {
        case 404:
          return "错误：未找到资源。请检查 ID 是否正确。";
        case 403:
          return "错误：权限被拒绝。你无权访问此资源。";
        case 429:
          return "错误：速率限制。请等待后再发出更多请求。";
        default:
          return `错误：API 请求失败，状态为 ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "错误：请求超时。请重试。";
    }
  }
  return `错误：发生意外错误：${error instanceof Error ? error.message : String(error)}`;
}

共享实用程序

将常用功能提取到可重用函数中：

// 共享 API 请求函数
async function makeApiRequest<T>(
  endpoint: string,
  method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
  data?: any,
  params?: any
): Promise<T> {
  try {
    const response = await axios({
      method,
      url: `${API_BASE_URL}/${endpoint}`,
      data,
      params,
      timeout: 30000,
      headers: {
        "Content-Type": "application/json",
        "Accept": "application/json"
      }
    });
    return response.data;
  } catch (error) {
    throw error;
  }
}

异步/等待最佳实践

始终对网络请求和 I/O 操作使用 async/await：

// 好：异步网络请求
async function fetchData(resourceId: string): Promise<ResourceData> {
  const response = await axios.get(`${API_URL}/resource/${resourceId}`);
  return response.data;
}

// 坏：Promise 链
function fetchData(resourceId: string): Promise<ResourceData> {
  return axios.get(`${API_URL}/resource/${resourceId}`)
    .then(response => response.data);  // 更难阅读和维护
}

TypeScript 最佳实践

1. 使用严格 TypeScript：在 tsconfig.json 中启用严格模式 2. 定义接口：为所有数据结构创建清晰的接口定义 3. 避免 `any`：使用适当的类型或 unknown 代替 any 4. 运行时验证使用 Zod：使用 Zod 模式验证外部数据 5. 类型保护：为复杂类型检查创建类型保护函数 6. 错误处理：始终使用适当的错误类型检查的 try-catch 7. 空安全：使用可选链（?.）和空值合并（??）

// 好：使用 Zod 和接口进行类型安全
interface UserResponse {
  id: string;
  name: string;
  email: string;
  team?: string;
  active: boolean;
}

const UserSchema = z.object({
  id: z.string(),
  name: z.string(),
  email: z.string().email(),
  team: z.string().optional(),
  active: z.boolean()
});

type User = z.infer<typeof UserSchema>;

async function getUser(id: string): Promise<User> {
  const data = await apiCall(`/users/${id}`);
  return UserSchema.parse(data);  // 运行时验证
}

// 坏：使用 any
async function getUser(id: string): Promise<any> {
  return await apiCall(`/users/${id}`);  // 没有类型安全
}

包配置

package.json

{
  "name": "{service}-mcp-server",
  "version": "1.0.0",
  "description": "用于 {Service} API 集成的 MCP 服务器",
  "type": "module",
  "main": "dist/index.js",
  "scripts": {
    "start": "node dist/index.js",
    "dev": "tsx watch src/index.ts",
    "build": "tsc",
    "clean": "rm -rf dist"
  },
  "engines": {
    "node": ">=18"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.6.1",
    "axios": "^1.7.9",
    "zod": "^3.23.8"
  },
  "devDependencies": {
    "@types/node": "^22.10.0",
    "tsx": "^4.19.2",
    "typescript": "^5.7.2"
  }
}

tsconfig.json

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "lib": ["ES2022"],
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true,
    "declaration": true,
    "declarationMap": true,
    "sourceMap": true,
    "allowSyntheticDefaultImports": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist"]
}

完整示例

#!/usr/bin/env node
/**
 * 示例服务的 MCP 服务器。
 *
 * 此服务器提供与示例 API 交互的工具，包括用户搜索、
 * 项目管理和数据导出功能。
 */

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import axios, { AxiosError } from "axios";

// 常量
const API_BASE_URL = "https://api.example.com/v1";
const CHARACTER_LIMIT = 25000;

// 枚举
enum ResponseFormat {
  MARKDOWN = "markdown",
  JSON = "json"
}

// Zod 模式
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "查询必须至少 2 个字符")
    .max(200, "查询不能超过 200 个字符")
    .describe("匹配姓名/电子邮件的搜索字符串"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("要返回的最大结果数"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("分页要跳过的结果数"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("输出格式：'markdown' 用于人类可读或 'json' 用于机器可读")
}).strict();

type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

// 共享实用程序函数
async function makeApiRequest<T>(
  endpoint: string,
  method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
  data?: any,
  params?: any
): Promise<T> {
  try {
    const response = await axios({
      method,
      url: `${API_BASE_URL}/${endpoint}`,
      data,
      params,
      timeout: 30000,
      headers: {
        "Content-Type": "application/json",
        "Accept": "application/json"
      }
    });
    return response.data;
  } catch (error) {
    throw error;
  }
}

function handleApiError(error: unknown): string {
  if (error instanceof AxiosError) {
    if (error.response) {
      switch (error.response.status) {
        case 404:
          return "错误：未找到资源。请检查 ID 是否正确。";
        case 403:
          return "错误：权限被拒绝。你无权访问此资源。";
        case 429:
          return "错误：速率限制。请等待后再发出更多请求。";
        default:
          return `错误：API 请求失败，状态为 ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "错误：请求超时。请重试。";
    }
  }
  return `错误：发生意外错误：${error instanceof Error ? error.message : String(error)}`;
}

// 创建 MCP 服务器实例
const server = new McpServer({
  name: "example-mcp",
  version: "1.0.0"
});

// 注册工具
server.registerTool(
  "example_search_users",
  {
    title: "搜索示例用户",
    description: `[如上所示的完整描述]`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    // 如上所示的实现
  }
);

// 主函数
// 对于 stdio（本地）：
async function runStdio() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("错误：需要 EXAMPLE_API_KEY 环境变量");
    process.exit(1);
  }

  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP 服务器通过 stdio 运行");
}

// 对于可流式 HTTP（远程）：
async function runHTTP() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("错误：需要 EXAMPLE_API_KEY 环境变量");
    process.exit(1);
  }

  const app = express();
  app.use(express.json());

  app.post('/mcp', async (req, res) => {
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: undefined,
      enableJsonResponse: true
    });
    res.on('close', () => transport.close());
    await server.connect(transport);
    await transport.handleRequest(req, res, req.body);
  });

  const port = parseInt(process.env.PORT || '3000');
  app.listen(port, () => {
    console.error(`MCP 服务器在 http://localhost:${port}/mcp 上运行`);
  });
}

// 根据环境选择传输
const transport = process.env.TRANSPORT || 'stdio';
if (transport === 'http') {
  runHTTP().catch(error => {
    console.error("服务器错误：", error);
    process.exit(1);
  });
} else {
  runStdio().catch(error => {
    console.error("服务器错误：", error);
    process.exit(1);
  });
}

---

高级 MCP 功能

资源注册

将数据暴露为资源以进行高效的基于 URI 的访问：

import { ResourceTemplate } from "@modelcontextprotocol/sdk/types.js";

// 使用 URI 模板注册资源
server.registerResource(
  {
    uri: "file://documents/{name}",
    name: "文档资源",
    description: "按名称访问文档",
    mimeType: "text/plain"
  },
  async (uri: string) => {
    // 从 URI 提取参数
    const match = uri.match(/^file:\/\/documents\/(.+)$/);
    if (!match) {
      throw new Error("无效的 URI 格式");
    }

    const documentName = match[1];
    const content = await loadDocument(documentName);

    return {
      contents: [{
        uri,
        mimeType: "text/plain",
        text: content
      }]
    };
  }
);

// 动态列出可用资源
server.registerResourceList(async () => {
  const documents = await getAvailableDocuments();
  return {
    resources: documents.map(doc => ({
      uri: `file://documents/${doc.name}`,
      name: doc.name,
      mimeType: "text/plain",
      description: doc.description
    }))
  };
});

何时使用资源 vs 工具：

• 资源：对于具有简单基于 URI 参数的数据访问
• 工具：对于需要验证和业务逻辑的复杂操作
• 资源：当数据相对静态或基于模板时
• 工具：当操作有副作用或复杂工作流时

传输选项

TypeScript SDK 支持两种主要传输机制：

可流式 HTTP（推荐用于远程服务器）

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";

const app = express();
app.use(express.json());

app.post('/mcp', async (req, res) => {
  // 为每个请求创建新的传输（无状态，防止请求 ID 冲突）
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: undefined,
    enableJsonResponse: true
  });

  res.on('close', () => transport.close());

  await server.connect(transport);
  await transport.handleRequest(req, res, req.body);
});

app.listen(3000);

stdio（用于本地集成）

import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const transport = new StdioServerTransport();
await server.connect(transport);

传输选择：

• 可流式 HTTP：Web 服务、远程访问、多个客户端
• stdio：命令行工具、本地开发、子进程集成

通知支持

当服务器状态改变时通知客户端：

// 当工具列表改变时通知
server.notification({
  method: "notifications/tools/list_changed"
});

// 当资源改变时通知
server.notification({
  method: "notifications/resources/list_changed"
});

谨慎使用通知 - 仅在服务器功能真正改变时使用。

---

代码最佳实践

代码组合性和可重用性

你的实现必须优先考虑组合性和代码重用：

1. 提取常用功能：

• 为在多个工具中使用的操作创建可重用的助手函数
• 为 HTTP 请求构建共享的 API 客户端，而不是复制代码
• 在实用程序函数中集中错误处理逻辑
• 将业务逻辑提取到可以组合的专用函数中
• 提取共享的 markdown 或 JSON 字段选择和格式化功能

2. 避免重复：

• 绝不在工具之间复制粘贴类似代码
• 如果你发现自己写了两次类似逻辑，将其提取到函数中
• 分页、过滤、字段选择和格式化等常见操作应该是共享的
• 认证/授权逻辑应该集中化

构建和运行

运行之前始终构建你的 TypeScript 代码：

# 构建项目
npm run build

# 运行服务器
npm start

# 带自动重新加载的开发
npm run dev

在认为实现完成之前，始终确保 npm run build 成功完成。

质量检查清单

在完成 Node/TypeScript MCP 服务器实现之前，确保：

战略设计

• [ ] 工具启用完整工作流，而不仅仅是 API 端点包装器
• [ ] 工具名称反映自然任务细分
• [ ] 响应格式优化代理上下文效率
• [ ] 在适当的地方使用人类可读标识符
• [ ] 错误消息指导代理正确使用

实现质量

• [ ] 专注实现：实现了最重要和最有价值的工具
• [ ] 所有工具都使用带有完整配置的 registerTool 注册
• [ ] 所有工具都包含 title、description、inputSchema 和 annotations
• [ ] 注释正确设置（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
• [ ] 所有工具都使用 Zod 模式进行运行时输入验证，使用 .strict() 强制执行
• [ ] 所有 Zod 模式都有适当的约束和描述性错误消息
• [ ] 所有工具都有带有显式输入/输出类型的全面描述
• [ ] 描述包括返回值示例和完整的模式文档
• [ ] 错误消息清晰、可操作且具有教育性

TypeScript 质量

• [ ] 为所有数据结构定义了 TypeScript 接口
• [ ] 在 tsconfig.json 中启用了严格 TypeScript
• [ ] 没有使用 any 类型 - 使用 unknown 或适当的类型代替
• [ ] 所有异步函数都有显式的 Promise<T> 返回类型
• [ ] 错误处理使用适当的类型保护（例如，axios.isAxiosError、z.ZodError）

高级功能（如适用）

• [ ] 为适当的数据端点注册了资源
• [ ] 配置了适当的传输（stdio 或可流式 HTTP）
• [ ] 为动态服务器功能实现了通知
• [ ] 与 SDK 接口类型安全

项目配置

• [ ] Package.json 包含所有必需的依赖项
• [ ] 构建脚本在 dist/ 目录中生成工作的 JavaScript
• [ ] 主入口点正确配置为 dist/index.js
• [ ] 服务器名称遵循格式：{service}-mcp-server
• [ ] tsconfig.json 正确配置了严格模式

代码质量

• [ ] 在适用的地方正确实现了分页
• [ ] 大型响应检查 CHARACTER_LIMIT 常量并用清晰消息截断
• [ ] 为可能的大型结果集提供过滤选项
• [ ] 所有网络操作都优雅地处理超时和连接错误
• [ ] 常用功能提取到可重用函数中
• [ ] 类似操作的返回类型一致

测试和构建

• [ ] npm run build 成功完成，没有错误
• [ ] 创建了 dist/index.js 并且可执行
• [ ] 服务器运行：node dist/index.js --help
• [ ] 所有导入都正确解析
• [ ] 示例工具调用按预期工作

Python MCP 服务器实现指南

概述

本文档提供了使用 MCP Python SDK 实现 MCP 服务器的 Python 特定最佳实践和示例。它涵盖了服务器设置、工具注册模式、使用 Pydantic 进行输入验证、错误处理和完整的工作示例。

---

快速参考

关键导入

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx

服务器初始化

mcp = FastMCP("service_mcp")

工具注册模式

@mcp.tool(name="tool_name", annotations={...})
async def tool_function(params: InputModel) -> str:
    # 实现
    pass

---

MCP Python SDK 和 FastMCP

官方 MCP Python SDK 提供 FastMCP，一个用于构建 MCP 服务器的高级框架。它提供：

• 从函数签名和文档字符串自动生成描述和 inputSchema
• 用于输入验证的 Pydantic 模型集成
• 使用 @mcp.tool 的基于装饰器的工具注册

有关完整的 SDK 文档，使用 WebFetch 加载： https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md

服务器命名约定

Python MCP 服务器必须遵循此命名模式：

• 格式：{service}_mcp（小写加下划线）
• 示例：github_mcp、jira_mcp、stripe_mcp

名称应该是：

• 通用的（不绑定到特定功能）
• 描述被集成的服务/API
• 容易从任务描述推断
• 不包含版本号或日期

工具实现

工具命名

使用 snake_case 作为工具名称（例如，"search_users"、"create_project"、"get_channel_info"），具有清晰、面向操作的名称。

避免命名冲突：包含服务上下文以防止重叠：

• 使用 "slack_send_message" 而不仅仅是 "send_message"
• 使用 "github_create_issue" 而不仅仅是 "create_issue"
• 使用 "asana_list_tasks" 而不仅仅是 "list_tasks"

使用 FastMCP 的工具结构

工具使用 @mcp.tool 装饰器和 Pydantic 模型进行输入验证来定义：

from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("example_mcp")

# 定义输入验证的 Pydantic 模型
class ServiceToolInput(BaseModel):
    '''服务工具操作的输入模型。'''
    model_config = ConfigDict(
        str_strip_whitespace=True,  # 自动去除字符串空白字符
        validate_assignment=True,    # 赋值时验证
        extra='forbid'              # 禁止额外字段
    )

    param1: str = Field(..., description="第一个参数描述（例如，'user123'、'project-abc'）", min_length=1, max_length=100)
    param2: Optional[int] = Field(default=None, description="带约束的可选整数参数", ge=0, le=1000)
    tags: Optional[List[str]] = Field(default_factory=list, description="要应用的标签列表", max_items=10)

@mcp.tool(
    name="service_tool_name",
    annotations={
        "title": "人类可读的工具标题",
        "readOnlyHint": True,     # 工具不修改环境
        "destructiveHint": False,  # 工具不执行破坏性操作
        "idempotentHint": True,    # 重复调用没有额外效果
        "openWorldHint": False     # 工具不与外部实体交互
    }
)
async def service_tool_name(params: ServiceToolInput) -> str:
    '''工具描述自动成为 'description' 字段。

    此工具对服务执行特定操作。它在处理之前使用 ServiceToolInput Pydantic 模型
    验证所有输入。

    参数：
        params (ServiceToolInput)：包含以下内容的验证输入参数：
            - param1 (str)：第一个参数描述
            - param2 (Optional[int])：带默认值的可选参数
            - tags (Optional[List[str]])：标签列表

    返回：
        str：包含操作结果的 JSON 格式化响应
    '''
    # 这里的实现
    pass

Pydantic v2 关键功能

• 使用 model_config 而不是嵌套的 Config 类
• 使用 field_validator 而不是弃用的 validator
• 使用 model_dump() 而不是弃用的 dict()
• 验证器需要 @classmethod 装饰器
• 验证器方法需要类型提示

from pydantic import BaseModel, Field, field_validator, ConfigDict

class CreateUserInput(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    name: str = Field(..., description="用户全名", min_length=1, max_length=100)
    email: str = Field(..., description="用户电子邮件地址", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., description="用户年龄", ge=0, le=150)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("电子邮件不能为空")
        return v.lower()

响应格式选项

为灵活性支持多种输出格式：

from enum import Enum

class ResponseFormat(str, Enum):
    '''工具响应的输出格式。'''
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    query: str = Field(..., description="搜索查询")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="输出格式：'markdown' 用于人类可读或 'json' 用于机器可读"
    )

Markdown 格式：

• 使用标题、列表和格式化以清晰显示
• 将时间戳转换为人类可读格式（例如，"2024-01-15 10:30:00 UTC" 而不是 epoch）
• 在括号中显示 ID 的显示名称（例如，"@john.doe (U123456)"）
• 省略详细的元数据（例如，只显示一个个人资料图像 URL，而不是所有大小）
• 逻辑地分组相关信息

JSON 格式：

• 返回适合程序化处理的完整结构化数据
• 包含所有可用字段和元数据
• 使用一致的字段名称和类型

分页实现

对于列出资源的工具：

class ListInput(BaseModel):
    limit: Optional[int] = Field(default=20, description="要返回的最大结果数", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="分页要跳过的结果数", ge=0)

async def list_items(params: ListInput) -> str:
    # 使用分页发出 API 请求
    data = await api_request(limit=params.limit, offset=params.offset)

    # 返回分页信息
    response = {
        "total": data["total"],
        "count": len(data["items"]),
        "offset": params.offset,
        "items": data["items"],
        "has_more": data["total"] > params.offset + len(data["items"]),
        "next_offset": params.offset + len(data["items"]) if data["total"] > params.offset + len(data["items"]) else None
    }
    return json.dumps(response, indent=2)

错误处理

提供清晰的、可操作的错误消息：

def _handle_api_error(e: Exception) -> str:
    '''所有工具的一致错误格式化。'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "错误：未找到资源。请检查 ID 是否正确。"
        elif e.response.status_code == 403:
            return "错误：权限被拒绝。你无权访问此资源。"
        elif e.response.status_code == 429:
            return "错误：速率限制。请等待后再发出更多请求。"
        return f"错误：API 请求失败，状态为 {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "错误：请求超时。请重试。"
    return f"错误：发生意外错误：{type(e).__name__}"

共享实用程序

将常用功能提取到可重用函数中：

# 共享 API 请求函数
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''所有 API 调用的可重用函数。'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

异步/等待最佳实践

始终对网络请求和 I/O 操作使用 async/await：

# 好：异步网络请求
async def fetch_data(resource_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{API_URL}/resource/{resource_id}")
        response.raise_for_status()
        return response.json()

# 坏：同步请求
def fetch_data(resource_id: str) -> dict:
    response = requests.get(f"{API_URL}/resource/{resource_id}")  # 阻塞
    return response.json()

类型提示

在整个过程中使用类型提示：

from typing import Optional, List, Dict, Any

async def get_user(user_id: str) -> Dict[str, Any]:
    data = await fetch_user(user_id)
    return {"id": data["id"], "name": data["name"]}

工具文档字符串

每个工具都必须具有带有显式类型信息的全面文档字符串：

async def search_users(params: UserSearchInput) -> str:
    '''
    在示例系统中按姓名、电子邮件或团队搜索用户。

    此工具搜索示例平台中的所有用户配置文件，
    支持部分匹配和各种搜索过滤器。它不
    创建或修改用户，仅搜索现有用户。

    参数：
        params (UserSearchInput)：包含以下内容的验证输入参数：
            - query (str)：匹配姓名/电子邮件的搜索字符串（例如，"john"、"@example.com"、"team:marketing"）
            - limit (Optional[int])：要返回的最大结果数，介于 1-100 之间（默认：20）
            - offset (Optional[int])：分页要跳过的结果数（默认：0）

    返回：
        str：包含搜索结果的 JSON 格式化字符串，具有以下模式：

        成功响应：
        {
            "total": int,           # 找到的匹配总数
            "count": int,           # 此响应中的结果数
            "offset": int,          # 当前分页偏移量
            "users": [
                {
                    "id": str,      # 用户 ID（例如，"U123456789"）
                    "name": str,    # 全名（例如，"John Doe"）
                    "email": str,   # 电子邮件地址（例如，"john@example.com"）
                    "team": str     # 团队名称（例如，"Marketing"）- 可选
                }
            ]
        }

        错误响应：
        "错误：<错误消息>" 或 "未找到匹配 '<query>' 的用户"

    示例：
        - 使用时机："查找所有营销团队成员" -> 参数 query="team:marketing"
        - 使用时机："搜索 John 的账户" -> 参数 query="john"
        - 不使用时机：你需要创建用户（使用 example_create_user 代替）
        - 不使用时机：你有用户 ID 并需要完整详细信息（使用 example_get_user 代替）

    错误处理：
        - 输入验证错误由 Pydantic 模型处理
        - 如果请求过多返回 "错误：速率限制"（429 状态）
        - 如果 API 密钥无效返回 "错误：无效的 API 认证"（401 状态）
        - 返回格式化的结果列表或 "未找到匹配 'query' 的用户"
    '''

完整示例

下面查看一个完整的 Python MCP 服务器示例：

#!/usr/bin/env python3
'''
示例服务的 MCP 服务器。

此服务器提供与示例 API 交互的工具，包括用户搜索、
项目管理和数据导出功能。
'''

from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("example_mcp")

# 常量
API_BASE_URL = "https://api.example.com/v1"

# 枚举
class ResponseFormat(str, Enum):
    '''工具响应的输出格式。'''
    MARKDOWN = "markdown"
    JSON = "json"

# 输入验证的 Pydantic 模型
class UserSearchInput(BaseModel):
    '''用户搜索操作的输入模型。'''
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    query: str = Field(..., description="匹配姓名/电子邮件的搜索字符串", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, description="要返回的最大结果数", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="分页要跳过的结果数", ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="输出格式")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("查询不能为空或仅空白字符")
        return v.strip()

# 共享实用程序函数
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''所有 API 调用的可重用函数。'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

def _handle_api_error(e: Exception) -> str:
    '''所有工具的一致错误格式化。'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "错误：未找到资源。请检查 ID 是否正确。"
        elif e.response.status_code == 403:
            return "错误：权限被拒绝。你无权访问此资源。"
        elif e.response.status_code == 429:
            return "错误：速率限制。请等待后再发出更多请求。"
        return f"错误：API 请求失败，状态为 {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "错误：请求超时。请重试。"
    return f"错误：发生意外错误：{type(e).__name__}"

# 工具定义
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "搜索示例用户",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''在示例系统中按姓名、电子邮件或团队搜索用户。

    [如上所示的完整文档字符串]
    '''
    try:
        # 使用验证参数发出 API 请求
        data = await _make_api_request(
            "users/search",
            params={
                "q": params.query,
                "limit": params.limit,
                "offset": params.offset
            }
        )

        users = data.get("users", [])
        total = data.get("total", 0)

        if not users:
            return f"未找到匹配 '{params.query}' 的用户"

        # 根据请求格式格式化响应
        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [f"# 用户搜索结果：'{params.query}'", ""]
            lines.append(f"找到 {total} 个用户（显示 {len(users)} 个）")
            lines.append("")

            for user in users:
                lines.append(f"## {user['name']} ({user['id']})")
                lines.append(f"- **电子邮件**：{user['email']}")
                if user.get('team'):
                    lines.append(f"- **团队**：{user['team']}")
                lines.append("")

            return "\n".join(lines)

        else:
            # 机器可读的 JSON 格式
            import json
            response = {
                "total": total,
                "count": len(users),
                "offset": params.offset,
                "users": users
            }
            return json.dumps(response, indent=2)

    except Exception as e:
        return _handle_api_error(e)

if __name__ == "__main__":
    mcp.run()

---

高级 FastMCP 功能

上下文参数注入

FastMCP 可以自动将 Context 参数注入到工具中，用于日志记录、进度报告、资源读取和用户交互等高级功能：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("example_mcp")

@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
    '''具有上下文访问权限的高级工具，用于日志记录和进度。'''

    # 为长时间操作报告进度
    await ctx.report_progress(0.25, "开始搜索...")

    # 记录信息以进行调试
    await ctx.log_info("处理查询", {"query": query, "timestamp": datetime.now()})

    # 执行搜索
    results = await search_api(query)
    await ctx.report_progress(0.75, "格式化结果...")

    # 访问服务器配置
    server_name = ctx.fastmcp.name

    return format_results(results)

@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
    '''可以从用户请求额外输入的工具。'''

    # 在需要时请求敏感信息
    api_key = await ctx.elicit(
        prompt="请提供你的 API 密钥：",
        input_type="password"
    )

    # 使用提供的密钥
    return await api_call(resource_id, api_key)

上下文功能：

• ctx.report_progress(progress, message) - 为长时间操作报告进度
• ctx.log_info(message, data) / ctx.log_error() / ctx.log_debug() - 日志记录
• ctx.elicit(prompt, input_type) - 从用户请求输入
• ctx.fastmcp.name - 访问服务器配置
• ctx.read_resource(uri) - 读取 MCP 资源

资源注册

将数据暴露为资源以进行高效的基于模板的访问：

@mcp.resource("file://documents/{name}")
async def get_document(name: str) -> str:
    '''将文档暴露为 MCP 资源。

    资源对于不需要复杂参数的静态或半静态数据很有用。
    它们使用 URI 模板进行灵活访问。
    '''
    document_path = f"./docs/{name}"
    with open(document_path, "r") as f:
        return f.read()

@mcp.resource("config://settings/{key}")
async def get_setting(key: str, ctx: Context) -> str:
    '''将配置暴露为带上下文的资源。'''
    settings = await load_settings()
    return json.dumps(settings.get(key, {}))

何时使用资源 vs 工具：

• 资源：对于具有简单参数（URI 模板）的数据访问
• 工具：对于具有验证和业务逻辑的复杂操作

结构化输出类型

FastMCP 支持字符串以外的多种返回类型：

from typing import TypedDict
from dataclasses import dataclass
from pydantic import BaseModel

# 结构化返回的 TypedDict
class UserData(TypedDict):
    id: str
    name: str
    email: str

@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
    '''返回结构化数据 - FastMCP 处理序列化。'''
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# 复杂验证的 Pydantic 模型
class DetailedUser(BaseModel):
    id: str
    name: str
    email: str
    created_at: datetime
    metadata: Dict[str, Any]

@mcp.tool()
async def get_user_detailed(user_id: str) -> DetailedUser:
    '''返回 Pydantic 模型 - 自动生成模式。'''
    user = await fetch_user(user_id)
    return DetailedUser(**user)

生命周期管理

初始化跨请求持久化的资源：

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan():
    '''管理服务器生命周期内存在的资源。'''
    # 初始化连接、加载配置等
    db = await connect_to_database()
    config = load_configuration()

    # 使所有工具可用
    yield {"db": db, "config": config}

    # 关闭时清理
    await db.close()

mcp = FastMCP("example_mcp", lifespan=app_lifespan)

@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
    '''通过上下文访问生命周期资源。'''
    db = ctx.request_context.lifespan_state["db"]
    results = await db.query(query)
    return format_results(results)

传输选项

FastMCP 支持两种主要传输机制：

# stdio 传输（用于本地工具）- 默认
if __name__ == "__main__":
    mcp.run()

# 可流式 HTTP 传输（用于远程服务器）
if __name__ == "__main__":
    mcp.run(transport="streamable_http", port=8000)

传输选择：

• stdio：命令行工具、本地集成、子进程执行
• 可流式 HTTP：Web 服务、远程访问、多个客户端

---

代码最佳实践

代码组合性和可重用性

你的实现必须优先考虑组合性和代码重用：

1. 提取常用功能：

• 为在多个工具中使用的操作创建可重用的助手函数
• 为 HTTP 请求构建共享的 API 客户端，而不是复制代码
• 在实用程序函数中集中错误处理逻辑
• 将业务逻辑提取到可以组合的专用函数中
• 提取共享的 markdown 或 JSON 字段选择和格式化功能

2. 避免重复：

• 绝不在工具之间复制粘贴类似代码
• 如果你发现自己写了两次类似逻辑，将其提取到函数中
• 分页、过滤、字段选择和格式化等常见操作应该是共享的
• 认证/授权逻辑应该集中化

Python 特定最佳实践

1. 使用类型提示：始终包含函数参数和返回值的类型注解 2. Pydantic 模型：为所有输入验证定义清晰的 Pydantic 模型 3. 避免手动验证：让 Pydantic 处理带约束的输入验证 4. 适当导入：分组导入（标准库、第三方、本地） 5. 错误处理：使用特定异常类型（httpx.HTTPStatusError，而不是通用 Exception） 6. 异步上下文管理器：对需要清理的资源使用 async with 7. 常量：在 UPPER_CASE 中定义模块级常量

质量检查清单

在完成 Python MCP 服务器实现之前，确保：

战略设计

• [ ] 工具启用完整工作流，而不仅仅是 API 端点包装器
• [ ] 工具名称反映自然任务细分
• [ ] 响应格式优化代理上下文效率
• [ ] 在适当的地方使用人类可读标识符
• [ ] 错误消息指导代理正确使用

实现质量

• [ ] 专注实现：实现了最重要和最有价值的工具
• [ ] 所有工具都有描述性名称和文档
• [ ] 类似操作的返回类型一致
• [ ] 所有外部调用都实现了错误处理
• [ ] 服务器名称遵循格式：{service}_mcp
• [ ] 所有网络操作都使用 async/await
• [ ] 常用功能提取到可重用函数中
• [ ] 错误消息清晰、可操作且具有教育性
• [ ] 输出经过正确验证和格式化

工具配置

• [ ] 所有工具在装饰器中实现 'name' 和 'annotations'
• [ ] 注释正确设置（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
• [ ] 所有工具都使用 Pydantic BaseModel 进行输入验证，带有 Field() 定义
• [ ] 所有 Pydantic 字段都有显式类型和带约束的描述
• [ ] 所有工具都有带有显式输入/输出类型的全面文档字符串
• [ ] 文档字符串包括字典/JSON 返回的完整模式结构
• [ ] Pydantic 模型处理输入验证（不需要手动验证）

高级功能（如适用）

• [ ] 上下文注入用于日志记录、进度或请求
• [ ] 为适当的数据端点注册了资源
• [ ] 为持久连接实现了生命周期管理
• [ ] 使用了结构化输出类型（TypedDict、Pydantic 模型）
• [ ] 配置了适当的传输（stdio 或可流式 HTTP）

代码质量

• [ ] 文件包括适当的导入，包括 Pydantic 导入
• [ ] 在适用的地方正确实现了分页
• [ ] 为可能的大型结果集提供过滤选项
• [ ] 所有异步函数都用 async def 正确定义
• [ ] HTTP 客户端使用带有适当上下文管理器的异步模式
• [ ] 在整个代码中使用类型提示
• [ ] 在模块级别以 UPPER_CASE 定义常量

测试

• [ ] 服务器成功运行：python your_server.py --help
• [ ] 所有导入都正确解析
• [ ] 示例工具调用按预期工作
• [ ] 错误场景优雅处理