Mcp Builder

MCP 服务器开发指南

概述

创建 MCP（模型上下文协议）服务器，使大语言模型能够通过精心设计的工具与外部服务进行交互。MCP 服务器的质量取决于它如何有效地帮助大语言模型完成实际任务。

---

流程

🚀 高级工作流

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

阶段 1：深入研究与规划

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、标准输入输出）
• 工具、资源和提示词定义

1.3 学习框架文档

推荐技术栈：

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

加载框架文档：

• 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：实现

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 操作使用异步/等待
• 使用可操作的错误消息进行适当的错误处理
• 在适用情况下支持分页
• 使用现代 SDK 时同时返回文本内容和结构化数据

注解：

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

对于拥有2个或多个工具的复杂任务开发，use patterns from tools patterns，尤其是当工具涉及以下讨论时：

---

阶段 3：审查和测试

3.1 代码质量

审查：

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

3.2 构建和测试

TypeScript：

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

Python：

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

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

---

阶段 4：创建评估

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

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

4.1 理解评估目的

使用评估来测试大语言模型是否能够有效地使用你的 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 与 Markdown）
• 分页最佳实践
• 传输层选择（可流式 HTTP 与标准输入输出）
• 安全和错误处理标准

SDK 文档（在阶段 1/2 加载）

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

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

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

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

评估指南（在阶段 4 加载）

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

MCP 服务器评估指南

概述

本文档提供创建 MCP 服务器全面评估的指导。评估测试大语言模型是否能有效使用您的 MCP 服务器来回答现实的、复杂的问题，仅使用提供的工具。

---

快速参考

评估要求

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

输出格式

<evaluation>
   <qa_pair>
      <question>您的问题</question>
      <answer>单一可验证的答案</answer>
   </qa_pair>
</evaluation>

---

评估目的

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

评估概述

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

• 现实的
• 清晰简洁的
• 明确的
• 复杂的，可能需要数十次工具调用或步骤
• 可以用您预先确定的单一、可验证的值来回答

问题指南

核心要求

1. 问题必须是独立的

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

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

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

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

• 必须需要另一个大语言模型使用多个（可能数十个）工具或步骤来回答

复杂度和深度

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

• 考虑需要多个子问题和顺序工具调用的多跳问题
• 每一步都应受益于先前步骤中找到的信息

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

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

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

• 而不是表面知识
• 可以将复杂的想法作为需要证据的真/假问题提出
• 可以使用多选格式，大语言模型必须搜索不同的假设

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

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

工具测试

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

• 可能引发返回大型 JSON 对象或列表的工具，使大语言模型不堪重负
• 应该需要理解多种数据形式：
• ID 和名称
• 时间戳和日期时间（月、日、年、秒）
• 文件 ID、名称、扩展名和 MIME 类型
• URL、GID 等
• 应该测试工具返回所有有用数据形式的能力

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

• 人类在大语言模型辅助下关心的信息检索任务类型

10. 问题可能需要数十次工具调用

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

11. 包含模糊的问题

• 可能是模糊的或需要难以决定调用哪些工具
• 迫使大语言模型可能犯错或误解
• 确保尽管存在模糊性，仍然有一个单一的可验证答案

稳定性

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

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

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

• 创建具有挑战性和复杂的问题
• 有些可能无法用可用的 MCP 服务器工具解决
• 问题可能需要特定的输出格式（日期时间 vs. 纪元时间，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 或字符串列表
• 不是自然语言文本
• 除非答案可以使用直接字符串比较直接验证
• 并且可以实际重现
• 大语言模型不太可能以任何其他顺序或格式返回相同的列表

评估流程

步骤 1：文档检查

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

• 可用的端点和功能
• 如果存在歧义，从网络上获取更多信息
• 尽可能并行化此步骤
• 确保每个子代理仅检查来自文件系统或网络上的文档

步骤 2：工具检查

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

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

步骤 3：形成理解

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

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

步骤 4：只读内容检查

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

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

步骤 5：任务生成

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

• 大语言模型应该能够使用 MCP 服务器回答这些问题
• 遵循上述所有问题和答案指南

输出格式

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

<evaluation>
   <qa_pair>
      <question>查找在 2024 年第二季度创建的具有最多已完成任务的项目。项目名称是什么？</question>
      <answer>网站重新设计</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索在 2024 年 3 月关闭的标记为"bug"的问题。哪个用户关闭的问题最多？提供他们的用户名。</question>
      <answer>sarah_dev</answer>
   </qa_pair>
   <qa_pair>
      <question>查找在 2024 年 1 月 1 日至 1 月 31 日之间合并的修改了 /api 目录中文件的拉取请求。有多少不同的贡献者在这些 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 月报告的所有被标记为关键优先级的 bug 中，哪个被分配者在 48 小时内解决了他们分配的 bug 的最高百分比？提供被分配者的用户名。</question>
   <answer>alex_eng</answer>
</qa_pair>

这个问题很好，因为：

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

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

<qa_pair>
   <question>查找在 2023 年第四季度从入门版升级到企业版且年度合同价值最高的账户。该账户经营什么行业？</question>
   <answer>医疗保健</answer>
</qa_pair>

这个问题很好，因为：

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

差的问题

示例 1：答案随时间变化

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

这个问题很差，因为：

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

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

<qa_pair>
   <question>查找标题为"添加身份验证功能"的拉取请求，告诉我谁创建了它。</question>
   <answer>developer123</answer>
</qa_pair>

这个问题很差，因为：

• 可以通过直接搜索确切标题的关键词来解决
• 不需要深入探索或理解
• 不需要综合或分析

示例 3：答案格式不明确

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

这个问题很差，因为：

• 答案是可以以任何顺序返回的列表
• 难以使用直接字符串比较验证
• 大语言模型可能以不同方式格式化（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

评估文件格式

评估文件使用 XML 格式，包含 <qa_pair> 元素：

<evaluation>
   <qa_pair>
      <question>查找在 2024 年第二季度创建的具有最多已完成任务的项目。项目名称是什么？</question>
      <answer>网站重新设计</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索在 2024 年 3 月关闭的标记为"bug"的问题。哪个用户关闭的问题最多？提供他们的用户名。</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          报告输出文件（默认：输出到标准输出）

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

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

输出

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

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

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

将报告保存到文件

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: ${error.message}. Try using filter='active_only' to reduce results.`
    }]
  };
}

---

测试要求

全面的测试应涵盖:

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

---

文档要求

• 提供所有工具和功能的清晰文档
• 包含可工作的示例 (每个主要功能至少 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）

错误处理：
  - 如果请求过多返回 "Error: Rate limit exceeded"（429 状态）
  - 如果搜索返回空结果返回 "No users found matching '<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
    }))
  };
});

何时使用资源与工具：

• 资源：用于具有简单基于 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. 提取通用功能：

• 为跨多个工具使用的操作创建可重用的辅助函数
• 构建共享的 API 客户端进行 HTTP 请求，而不是复制代码
• 将错误处理逻辑集中到工具函数中
• 将业务逻辑提取到可以组合的专用函数中
• 提取共享的 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 服务器的高级框架。它提供：

• 从函数签名和文档字符串自动生成 description 和 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 模型验证所有输入。

    Args:
        params (ServiceToolInput): 经过验证的输入参数，包含：
            - param1 (str): 第一个参数描述
            - param2 (Optional[int]): 带有默认值的可选参数
            - tags (Optional[List[str]]): 标签列表

    Returns:
        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" 而不是时间戳）
• 显示带 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:
    '''
    按名称、邮箱或团队搜索 Example 系统中的用户。

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

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

    Returns:
        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>' 的用户"

    Examples:
        - 使用场景："查找所有营销团队成员" -> 使用 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
'''
Example Service 的 MCP 服务器。

此服务器提供与 Example 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": "搜索 Example 用户",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''按名称、邮箱或团队搜索 Example 系统中的用户。

    [完整文档字符串如上所示]
    '''
    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. 提取通用功能：

• 创建可重用的辅助函数，用于多个工具中使用的操作
• 构建共享的 API 客户端进行 HTTP 请求，而不是复制代码
• 将错误处理逻辑集中在工具函数中
• 将业务逻辑提取到可以组合的专用函数中
• 提取共享的 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 Fields 具有显式类型和带约束的描述
• [ ] 所有工具都有包含显式输入/输出类型的综合文档字符串
• [ ] 文档字符串包含 dict/JSON 返回的完整模式结构
• [ ] Pydantic 模型处理输入验证（无需手动验证）

高级特性（如适用）

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

代码质量

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

测试

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

PATs - Patterns for Agentic Tools

Design patterns for building quality tools for AI agents

https://arcade.dev/patterns

GUIDING PRINCIPLE

"Your agents are only as good as your tools."

Tools are how agents interact with the world. When tools are well-designed, orchestration stays simple and agents behave predictably. When tools are sloppy, the orchestration layer has to compensate - and it never does it well. These patterns capture what works, so you don't have to figure it out the hard way.

---

THREE AXES OF TOOL CLASSIFICATION

Every tool exists at coordinates across these dimensions.

AXIS 1: MATURITY

How sophisticated is the tool's implementation?

1. ATOMIC - Single-operation tools that do one thing. Direct API wrappers. 2. ENHANCED - Built-in validation, defaults, error handling. Production-ready. 3. COMPOSITE - Bundle multiple operations into coherent workflows. Task-oriented. 4. ORCHESTRATED - Coordinate other tools, manage state across calls, handle complex flows.

AXIS 2: INTEGRATION TYPE

What kind of system does the tool connect to?

1. API - REST, GraphQL, gRPC services. 2. DATABASE - SQL, NoSQL, vector stores. Direct data access. 3. FILE_SYSTEM - Local files, cloud storage, document processing. 4. SYSTEM - Shell commands, OS operations, infrastructure control.

AXIS 3: ACCESS PATTERN

How does the tool execute and return results?

1. SYNCHRONOUS - Request-response. Agent waits for completion. 2. ASYNCHRONOUS - Fire and poll. Long-running operations with job IDs. 3. STREAMING - Incremental delivery. Results arrive as they're ready. 4. EVENT_DRIVEN - Push-based. Tool notifies when something happens.

---

CROSS-CUTTING CONCERNS

These principles apply regardless of which axis or pattern you're working with.

MACHINE EXPERIENCE

"Design for the LLM, not the human." Tool descriptions, parameter names, and error messages should be optimized for machine comprehension. Clear, unambiguous, structured.

TOOL DAGs

"Hint at what comes next." Tools should suggest related operations and dependencies. Help agents build workflows without trial and error.

ERROR-GUIDED RECOVERY

"Errors should teach, not just fail." When things go wrong, provide actionable guidance. Suggest retries, alternatives, or parameter adjustments.

SECURITY BOUNDARIES

"Prompts express intent, code enforces rules." Never trust the agent to enforce security. Authorization, validation, and audit happen in the tool layer.

---

PATTERN CATEGORIES

1. TOOL (4 patterns) - "What kind of tool is this?"

The atomic unit. What IS a tool? The fundamental building blocks of agent tooling.

1. TOOL The atomic callable unit that an agent can invoke to perform work.

• Named function with typed parameters and return type
• Clear description for LLM comprehension
• Documented side effects

2. QUERY_TOOL A read-only tool that retrieves data without side effects.

• Safe to retry
• Results can be cached
• Parallelizable

3. COMMAND_TOOL A tool that performs actions with side effects.

• Modifies state or triggers external actions
• May require confirmation for destructive operations
• Document irreversibility clearly

4. DISCOVERY_TOOL A tool that reveals available operations, schema, or capabilities.

• list_tables(), describe_schema(), get_capabilities()
• Essential for schema-on-read scenarios

---

2. TOOL INTERFACE (7 patterns) - "How do agents call this tool?"

The contract between agent and tool. How agents see, understand, and call tools.

5. TOOL_DESCRIPTION Write descriptions optimized for LLM comprehension, not human reading.

• Comprehensive: Everything the LLM needs to know
• Include examples of valid inputs
• Link to prerequisites and follow-ups

6. CONSTRAINED_INPUT Use enums, ranges, and validation to limit inputs to valid values.

• Enums instead of free-form strings
• Ranges for numbers (min/max)
• Patterns for format validation

7. SMART_DEFAULTS Reduce required parameters by providing sensible defaults.

• Context-aware defaults (current user, time)
• Default to most common case
• Document defaults explicitly

8. NATURAL_IDENTIFIER Accept human-friendly identifiers and resolve them internally.

• Accept email, username, display name
• Resolve to system ID internally
• Handle ambiguity with suggestions

9. MUTUAL_EXCLUSIVITY Enforce "exactly one of X or Y" parameter constraints.

• Document valid combinations
• Validate early with clear errors
• Provide examples

10. PERFORMANCE_HINT Guide agents toward efficient usage patterns.

• "Prefer conversation_id (faster)"
• "For batch operations, use batch_get"
• Warn about expensive operations

11. PARAMETER_COERCION Accept flexible input formats and normalize internally.

• Dates: ISO, relative, natural language
• Numbers: string or numeric
• Lists: single item or array

---

3. TOOL DISCOVERY (5 patterns) - "How do agents find this tool?"

Navigation and selection in a tool ecosystem. How agents find the right tool.

12. TOOL_REGISTRY Provide a catalog of available tools with their capabilities.

• List all tools with descriptions
• Categorize by domain or function
• Include authentication requirements

13. SCHEMA_EXPLORER Progressively reveal structure through layered discovery.

• Layer 1: list_tables() - What exists
• Layer 2: describe_table() - Field details
• Layer 3: sample_data() - See examples

14. DEPENDENCY_HINT Embed "call X before Y" guidance in tool descriptions.

• Prerequisites: "If you don't have X, call Y first"
• Follow-ups: "After this, you might want Z"
• Include in error messages too

15. CAPABILITY_MATCHING Find tools by intent or capability, not just name.

• Semantic search across descriptions
• Capability tags
• Action mapping

16. HEALTH_CHECK Verify tool availability before relying on it.

• Fast availability probe
• Backend status
• Degraded mode indication

---

4. TOOL COMPOSITION (6 patterns) - "How do tools combine?"

Building complex operations from simple tools. How tools combine and chain.

17. ABSTRACTION_LADDER Provide tools at multiple levels of granularity.

• Low: create_file(path, bytes)
• Mid: create_document(title, content)
• High: draft_report(topic)

18. TASK_BUNDLE Combine multiple operations into a single tool.

• dm_user() = search + resolve + send
• Handle intermediate errors
• Name after task, not steps

19. BATCH_OPERATION Process multiple items in a single tool call.

• Accept arrays
• Return per-item results
• Handle partial failures

20. OPERATION_MODE Provide different modes for different access patterns.

• Explore: Read-only, safe
• Preview: Show what would happen
• Execute: Actually perform

21. TOOL_CHAIN Explicitly define sequences of tool calls.

• Ordered step definitions
• Data passing between steps
• Checkpoints for resume

22. SCATTER_GATHER_TOOL Fan out to multiple sources, then combine results.

• Query in parallel
• Gather and merge results
• Handle partial failures

---

5. TOOL EXECUTION (6 patterns) - "How does this tool execute?"

The internal processing patterns. How tools do their actual work.

23. SYNCHRONOUS_EXECUTION Immediate request-response execution.

• Returns result in same call
• Bounded time (seconds)
• Most tools should be synchronous

24. ASYNC_JOB Handle long-running operations with job IDs and polling.

• Start: Return job_id immediately
• Poll: check_job_status()
• Result: get_job_result()

25. IDEMPOTENT_OPERATION Make operations safe to retry with identical results.

• Accept idempotency key
• Cache and return same result
• Document guarantee

26. TRANSACTIONAL_BOUNDARY Ensure all-or-nothing execution for multi-step operations.

• All steps succeed or all roll back
• Data integrity maintained

27. COMPENSATION_HANDLER Undo completed steps when a multi-step operation fails.

• Track completed steps
• Define compensating actions
• Execute in reverse order

28. TIMEOUT_BOUNDARY Define maximum execution time with graceful termination.

• Set appropriate limits
• Return partial results if possible
• Clear timeout errors

---

6. TOOL OUTPUT (7 patterns) - "How does this tool return results?"

Communicating results back to agents. How tools return useful information.

29. RESPONSE_SHAPER Transform raw API responses into agent-friendly formats.

• Flatten nested structures
• Select relevant fields only
• Rename for clarity

30. TOKEN_EFFICIENT_RESPONSE Minimize response size while preserving essential information.

• Essential fields only
• Truncate long text
• Count, don't list

31. PAGINATED_RESULT Handle large result sets with cursor-based pagination.

• Cursor-based (not page numbers)
• has_more flag
• Include next-page hint

32. PROGRESSIVE_DETAIL Return summary by default, full detail on request.

• Summary mode (default)
• Full mode on request
• Selective expansion flags

33. NEXT_ACTION_HINT Suggest what the agent should do next.

• Suggested tools with parameters
• Required data for next step
• Alternative paths

34. GUI_URL Include links to view or edit results in a web interface.

• View URL for the resource
• Edit URL if applicable
• Dashboard links

35. PARTIAL_SUCCESS Report mixed success/failure results for batch operations.

• Per-item status
• Summary statistics
• Retry guidance for failures

---

7. TOOL CONTEXT (5 patterns) - "How is state managed?"

The MCP-native category. Stateful connections and context management.

36. IDENTITY_ANCHOR Establish user identity and context at the start of a session.

• who_am_i() returns user context
• User ID, roles, permissions
• Team and org context

37. SESSION_CONTEXT Maintain state across multiple tool calls in a conversation.

• Persist context across calls
• Set working project/scope
• Expire stale sessions

38. RESOURCE_REFERENCE Point to external data by URI instead of embedding it.

• URI-based references
• Resolvable by tools
• Include content type

39. CONTEXT_INJECTION Automatically inject relevant context the agent didn't request.

• User context (team, permissions)
• Temporal context (time, timezone)
• Allow explicit overrides

40. CONTEXT_BOUNDARY Define the scope or boundaries of tool operations.

• Root paths for file access
• Tenant scope for data
• Permission scope limits

---

8. TOOL RESILIENCE (6 patterns) - "How does this tool handle failures?"

Recovery, retry, and degradation patterns. How tools handle failures gracefully.

41. RECOVERY_GUIDE Provide actionable error messages that tell agents how to fix the problem.

• What went wrong
• Why it failed
• How to fix (specific steps)

42. ERROR_CLASSIFICATION Distinguish between retryable and permanent failures.

• Retryable: transient, will succeed later
• Permanent: won't work without changes
• Auth required: need to re-authenticate

43. CONFIRMATION_REQUEST Request clarification when input is ambiguous.

• Return matches with details
• Ask for selection
• Provide exact call for each option

44. FUZZY_MATCH_THRESHOLD Auto-accept high-confidence matches, confirm uncertain ones.

• >90%: auto-accept
• 50-90%: return options
• <50%: ask for different input

45. GRACEFUL_DEGRADATION Return partial results when full operation isn't possible.

• Return what worked
• Note what failed
• Suggest remediation

46. FALLBACK_TOOL Provide alternative tools when the primary is unavailable.

• Define fallback order
• Switch transparently or suggest
• Indicate when fallback used

---

9. TOOL SECURITY (4 patterns) - "How is access controlled?"

Trust, authorization, and data protection. How access is controlled.

47. SECRET_INJECTION Inject credentials at runtime, never passing them through the LLM.

• Context-based injection
• Never accept secrets as parameters
• Secure storage

48. PERMISSION_GATE Enforce access control in code, not prompts.

• Check permissions before executing
• Log all denials
• "Prompts express intent; code enforces rules"

49. SCOPE_DECLARATION Declare required OAuth scopes per tool.

• Required scopes documented
• Pre-check before execution
• Clear missing scope errors

50. AUDIT_TRAIL Log all tool invocations for security and debugging.

• What: tool, parameters (redacted)
• Who: user, session
• When: timestamp
• Result: success/failure, duration

---

10. COMPOSITIONAL (4 patterns) - "What patterns apply across the system?"

Cross-cutting patterns that span multiple categories. System-wide concerns.

51. TOOL_GATEWAY Provide a unified interface to multiple tool backends.

• Single entry point
• Route to correct backend
• Aggregate discovery

52. TOOL_ADAPTER Wrap legacy APIs as agent-friendly tools.

• Hide legacy complexity
• Add LLM-friendly descriptions
• Shape responses appropriately

53. CANONICAL_TOOL_MODEL Use standard data models across the tool ecosystem.

• Define canonical schemas (User, Task, Event)
• Map to/from canonical form
• Consistent naming

54. TOOL_VERSIONING Support multiple tool versions coexisting.

• Version in name or metadata
• Parallel support during migration
• Deprecation with migration guide

---

QUICK REFERENCE

When building a tool, consider:

1. WHAT kind of tool? → Tool, Query Tool, Command Tool, Discovery Tool 2. HOW called? → Description, Constraints, Defaults, Natural IDs 3. HOW found? → Registry, Schema, Dependencies, Matching 4. HOW composed? → Abstraction levels, Bundles, Batches, Chains 5. HOW executed? → Sync, Async, Idempotent, Transactional 6. HOW returned? → Shape, Efficiency, Pagination, Hints 7. HOW contextual? → Identity, Session, Resources, Boundaries 8. HOW resilient? → Recovery, Classification, Degradation, Fallback 9. HOW secure? → Secrets, Permissions, Scopes, Audit 10. HOW integrated? → Gateway, Adapter, Canonical, Versioning

---

THE PARADIGM SHIFT

---

Generated from arcade.dev/patterns