From 54a9c167be9d6d6942fdd8a141304853e8d18b21 Mon Sep 17 00:00:00 2001 From: wehub-skill-sync Date: Mon, 13 Jul 2026 21:35:36 +0800 Subject: [PATCH] chore: import zh skill mcp-builder --- LICENSE.txt | 73 +++ README.wehub.md | 9 + SKILL.md | 236 ++++++++ reference/evaluation.md | 602 ++++++++++++++++++++ reference/mcp_best_practices.md | 249 ++++++++ reference/node_mcp_server.md | 970 ++++++++++++++++++++++++++++++++ reference/python_mcp_server.md | 718 +++++++++++++++++++++++ scripts/connections.py | 151 +++++ scripts/evaluation.py | 373 ++++++++++++ scripts/example_evaluation.xml | 22 + scripts/requirements.txt | 30 + 11 files changed, 3433 insertions(+) create mode 100644 LICENSE.txt create mode 100644 README.wehub.md create mode 100644 SKILL.md create mode 100644 reference/evaluation.md create mode 100644 reference/mcp_best_practices.md create mode 100644 reference/node_mcp_server.md create mode 100644 reference/python_mcp_server.md create mode 100644 scripts/connections.py create mode 100644 scripts/evaluation.py create mode 100644 scripts/example_evaluation.xml create mode 100644 scripts/requirements.txt diff --git a/LICENSE.txt b/LICENSE.txt new file mode 100644 index 0000000..90a1735 --- /dev/null +++ b/LICENSE.txt @@ -0,0 +1,73 @@ +Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + 使用、复制与分发的条款和条件 + + 1. 定义 + + "许可" 指本文件第 1 至 9 条所定义的有关使用、复制与分发的条款和条件。 + + "许可方" 指版权所有人或其授权授予许可的实体。 + + "法律实体" 指行为实体以及控制该实体、受该实体控制或与该实体共同受控制的所有其他实体的联合。就本定义而言，"控制" 指 (i) 通过合同或其他方式，直接或间接拥有指导该实体管理方向的权力；(ii) 持有该实体百分之五十（50%）或以上的已发行股份；或 (iii) 对该实体的受益所有权。 + + "您"（或"您的"）指行使本许可所授予权限的个人或法律实体。 + + "源代码" 形式指进行修改时首选的形式，包括但不限于软件源代码、文档源代码和配置文件。 + + "目标代码" 形式指对源代码形式进行机械转换或翻译后产生的任何形式，包括但不限于编译后的目标代码、生成的文档以及转换为其他媒体类型的内容。 + + "作品" 指根据本许可提供的、在作品所附或包含的版权声明（附录中提供示例）中指明的作者作品，无论是源代码还是目标代码形式。 + + "衍生作品" 指基于（或源自）作品的任何作品，无论是源代码还是目标代码形式，且其中编辑性修订、注释、阐述或其他修改整体上构成了原创性的作者作品。就本许可而言，衍生作品不包括与作品及其衍生作品的接口保持独立、或仅链接（或按名称绑定）到该接口的作品。 + + "贡献" 指由版权所有人或经版权所有人授权的个人或法律实体，有意提交给许可方以纳入作品的任何作者作品，包括作品的原始版本以及对作品或其衍生作品的任何修改或增补。就本定义而言，"提交" 指以电子、口头或书面形式发送给许可方或其代表的任何通信，包括但不限于在许可方或其代表为讨论和改进作品而管理的电子邮件列表、源代码控制系统和问题跟踪系统上进行的通信，但不包括版权所有人以书面形式明确标记或以其他方式指定为"非贡献"的通信。 + + "贡献者" 指许可方以及其贡献已被许可方接收并随后纳入作品的任何个人或法律实体。 + + 2. 版权许可的授予。 在遵守本许可条款和条件的前提下，各贡献者特此授予您一项永久的、全球性的、非排他性的、免费的、免版税的、不可撤销的版权许可，以复制、准备衍生作品、公开展示、公开表演、再许可以及分发作品及其衍生作品（源代码或目标代码形式）。 + + 3. 专利许可的授予。 在遵守本许可条款和条件的前提下，各贡献者特此授予您一项永久的、全球性的、非排他性的、免费的、免版税的、不可撤销的（本节另有规定除外）专利许可，以制造、委托制造、使用、许诺销售、销售、进口以及以其他方式转让作品，但该许可仅适用于该贡献者可以许可的、必然会因其贡献本身或将其贡献与提交该贡献时所针对的作品相结合而侵犯的专利权利要求。如果您对任何实体提起专利诉讼（包括诉讼中的交叉请求或反请求），指控作品或纳入作品中的某项贡献构成直接或间接专利侵权，则根据本许可授予您的与该作品相关的任何专利许可应在该诉讼提起之日起终止。 + + 4. 分发。 您可以在任何介质中，以源代码或目标代码形式，复制和分发作品或其衍生作品的副本，无论是否经过修改，但须满足以下条件： + + (a) 您必须向作品或衍生作品的任何其他接收者提供本许可的副本；且 + + (b) 您必须确保任何修改过的文件带有显著声明，说明您已修改了这些文件；且 + + (c) 对于您分发的任何衍生作品的源代码形式，您必须保留作品源代码形式中的所有版权、专利、商标和归属声明，但那些与衍生作品任何部分无关的声明除外；且 + + (d) 如果作品在其分发中附带"NOTICE"文本文件，则您分发的任何衍生作品必须包含该 NOTICE 文件中所含归属声明的可读副本，但不包括与衍生作品任何部分无关的声明，且至少应放置在以下位置之一：作为衍生作品一部分分发的 NOTICE 文本文件中；随衍生作品提供的源代码形式或文档中（如有）；或在衍生作品生成的显示界面中，且以第三方声明通常出现的方式和位置呈现。NOTICE 文件的内容仅供参考，不修改本许可。您可以在您分发的衍生作品中添加您自己的归属声明，作为作品 NOTICE 文本的补充或附录，前提是此类额外的归属声明不得被解释为对本许可的修改。 + + 您可以在您的修改中添加您自己的版权声明，并为您修改的使用、复制或分发，或为任何此类衍生作品整体，提供附加或不同的许可条款和条件，前提是您对作品的使用、复制和分发在其他方面遵守本许可规定的条件。 + + 5. 贡献的提交。 除非您另有明确声明，否则您有意向许可方提交以供纳入作品的任何贡献，均应按本许可的条款和条件进行，不附加任何额外条款或条件。尽管有上述规定，本协议中的任何内容均不得取代或修改您可能与许可方就此类贡献已签订的任何单独许可协议的条款。 + + 6. 商标。 本许可不授予使用许可方商号、商标、服务标识或产品名称的权限，除非在合理且惯常地描述作品来源以及复制 NOTICE 文件内容时所必需的范围内。 + + 7. 免责声明。 除非适用法律要求或书面同意，许可方按"现状"提供作品（各贡献者按"现状"提供其贡献），不附带任何明示或暗示的保证或条件，包括但不限于对所有权、不侵权、适销性或特定用途适用性的任何保证或条件。您自行确定使用或再分发作品的适当性，并承担根据本许可行使权限所附带的任何风险。 + + 8. 责任限制。 在任何情况下且在任何法律理论下（包括侵权（含过失）、合同或其他），除非适用法律要求（如蓄意和重大过失行为）或书面同意，任何贡献者均不对您因本许可或因使用或无法使用作品（包括但不限于商誉损失、停工、计算机故障或失灵，或任何及所有其他商业损害或损失）而产生的任何性质的直接、间接、特殊、附带或后果性损害承担责任，即使该贡献者已被告知可能发生此类损害。 + + 9. 接受保证或附加责任。 在再分发作品或其衍生作品时，您可以选择提供并收取费用以接受与本许可一致的支持、保证、赔偿或其他责任义务和/或权利。但是，在接受此类义务时，您只能代表您自己并自行承担责任，而不能代表任何其他贡献者，并且仅在您同意赔偿、辩护并使各贡献者免受因您接受任何此类保证或附加责任而导致的任何责任或索赔。 + + 条款和条件结束 + + 附录：如何将 Apache License 应用于您的作品。 + + 要将 Apache License 应用于您的作品，请附加以下样板声明，将方括号 "[]" 中的字段替换为您自己的标识信息（不要包含方括号！）。文本应使用文件格式相应的注释语法。我们还建议在版权声明的同一"印刷页面"上包含文件或类名及用途说明，以便在第三方存档中更容易识别。 + + Copyright 2026 Anthropic, PBC. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.wehub.md b/README.wehub.md new file mode 100644 index 0000000..f0fe542 --- /dev/null +++ b/README.wehub.md @@ -0,0 +1,9 @@ +# WeHub 来源说明 + +- Skill 名称：`mcp-builder` +- 中文类目：MCP 服务器设计 + 评估 +- 上游仓库：`anthropics__skills` +- 上游路径：`skills/mcp-builder/SKILL.md` +- 上游链接：https://github.com/anthropics/skills/blob/HEAD/skills/mcp-builder/SKILL.md +- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理 +- 原作者、版权和许可证信息以上游仓库为准 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..a493e7f --- /dev/null +++ b/SKILL.md @@ -0,0 +1,236 @@ +--- +name: mcp-builder +description: 创建高质量 MCP（模型上下文协议）服务器的指南，使 LLM 能够通过精心设计的工具与外部服务进行交互。在构建 MCP 服务器以集成外部 API 或服务时使用，无论是用 Python（FastMCP）还是 Node/TypeScript（MCP SDK）。 +license: Complete terms in LICENSE.txt +--- + +# MCP 服务器开发指南 + +## 概述 + +创建 MCP（模型上下文协议）服务器，使 LLM 能够通过精心设计的工具与外部服务进行交互。MCP 服务器的质量取决于它在多大程度上帮助 LLM 完成真实世界任务。 + +--- + +# 流程 + +## 🚀 高级工作流 + +创建一个高质量的 MCP 服务器包含四个主要阶段： + +### 阶段一：深入调研与规划 + +#### 1.1 理解现代 MCP 设计 + +**API 覆盖与工作流工具：** +在全面的 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 代码，得益于其广泛使用、静态类型和良好的 lint 工具） +- **传输方式**：远程服务器使用可流式 HTTP，采用无状态 JSON（相比于有状态会话和流式响应，更易于扩展和维护）。本地服务器使用 stdio。 + +**加载框架文档：** + +- **MCP 最佳实践**：[📋 查看最佳实践](./reference/mcp_best_practices.md) — 核心指南 + +**对于 TypeScript（推荐）：** +- **TypeScript SDK**：使用 WebFetch 加载 `https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md` +- [⚡ TypeScript 指南](./reference/node_mcp_server.md) — TypeScript 模式与示例 + +**对于 Python：** +- **Python SDK**：使用 WebFetch 加载 `https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md` +- [🐍 Python 指南](./reference/python_mcp_server.md) — Python 模式与示例 + +#### 1.4 规划你的实现 + +**理解 API：** +查阅服务的 API 文档，识别关键端点、认证要求和数据模型。根据需要配合使用 Web 搜索和 WebFetch。 + +**工具选择：** +优先考虑全面的 API 覆盖。列出需要实现的端点，从最常见的操作开始。 + +--- + +### 阶段二：实现 + +#### 2.1 设置项目结构 + +项目设置请参见特定语言的指南： +- [⚡ TypeScript 指南](./reference/node_mcp_server.md) — 项目结构、package.json、tsconfig.json +- [🐍 Python 指南](./reference/python_mcp_server.md) — 模块组织、依赖项 + +#### 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 文件： + +```xml + + + 寻找关于使用动物代号命名 AI 模型的讨论。某个模型需要一种格式为 ASL-X 的特殊安全标识。以斑点野生猫科动物命名的模型，其 X 数字是多少？ + 3 + + + +``` + +--- + +# 参考文件 + +## 📚 文档库 + +在开发过程中根据需要加载以下资源： + +### 核心 MCP 文档（优先加载） +- **MCP 协议**：从 `https://modelcontextprotocol.io/sitemap.xml` 的站点地图开始，然后使用 `.md` 后缀获取具体页面 +- [📋 MCP 最佳实践](./reference/mcp_best_practices.md) — 通用 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 实现指南](./reference/python_mcp_server.md) — 完整的 Python/FastMCP 指南，包括： + - 服务器初始化模式 + - Pydantic 模型示例 + - 使用 `@mcp.tool` 注册工具 + - 完整的可运行示例 + - 质量检查清单 + +- [⚡ TypeScript 实现指南](./reference/node_mcp_server.md) — 完整的 TypeScript 指南，包括： + - 项目结构 + - Zod 模式模式 + - 使用 `server.registerTool` 注册工具 + - 完整的可运行示例 + - 质量检查清单 + +### 评估指南（在阶段四加载） +- [✅ 评估指南](./reference/evaluation.md) — 完整的评估创建指南，包括： + - 问题创建指南 + - 答案验证策略 + - XML 格式规范 + - 示例问题与答案 + - 使用提供的脚本运行评估 diff --git a/reference/evaluation.md b/reference/evaluation.md new file mode 100644 index 0000000..ee8053e --- /dev/null +++ b/reference/evaluation.md @@ -0,0 +1,602 @@ +# MCP 服务器评估指南 + +## 概述 + +本文档提供了为 MCP 服务器创建全面评估的指导。评估用于测试大语言模型能否仅通过所提供的工具，有效使用你的 MCP 服务器来回答真实、复杂的问题。 + +--- + +## 快速参考 + +### 评估要求 +- 创建 10 个人类可读的问题 +- 问题必须是只读、独立、非破坏性的 +- 每个问题需要多次工具调用（可能多达数十次） +- 答案必须是单个可验证的值 +- 答案必须稳定（不会随时间变化） + +### 输出格式 +```xml + + + 你的问题在此 + 单个可验证的答案 + + +``` + +--- + +## 评估的目的 + +衡量 MCP 服务器质量的关键不在于服务器实现工具的好坏或全面程度，而在于这些实现（输入/输出模式、文档字符串/描述、功能）能否让大语言模型——在没有其他上下文、仅能访问 MCP 服务器的情况下——回答真实且困难的问题。 + +## 评估概述 + +创建 10 个人类可读的问题，仅需通过只读、独立、非破坏性且幂等的操作即可回答。每个问题应具备以下特点： +- 真实 +- 清晰简洁 +- 无歧义 +- 复杂，可能需要数十次工具调用或步骤 +- 可用一个你事先确定的、可验证的单一值来回答 + +## 问题编写指南 + +### 核心要求 + +1. **问题必须是独立的** + - 每个问题不应依赖于任何其他问题的答案 + - 不应假设在处理另一个问题之前已执行过写入操作 + +2. **问题必须仅涉及非破坏性和幂等的工具使用** + - 不应指示或要求通过修改状态来得出正确答案 + +3. **问题必须真实、清晰、简洁且复杂** + - 必须要求另一个大语言模型使用多个（可能数十个）工具或步骤才能回答 + +### 复杂度与深度 + +4. **问题必须要求深入探索** + - 考虑多跳问题，需要多个子问题和连续的工具调用 + - 每一步应能从之前问题中发现的信息中获益 + +5. **问题可能需要大量翻页** + - 可能需要翻阅多页结果 + - 可能需要查询旧数据（已过时 1-2 年）以找到小众信息 + - 问题必须困难 + +6. **问题必须要求深入理解** + - 而非表层知识 + - 可将复杂观点设为是非题，要求提供证据 + - 可采用选择题形式，让大语言模型搜索不同的假设 + +7. **问题不能通过简单的关键词搜索解决** + - 不要包含目标内容中的特定关键词 + - 使用同义词、相关概念或意译 + - 需要多次搜索、分析多个相关项目、提取上下文，然后推导出答案 + +### 工具测试 + +8. **问题应对工具返回值进行压力测试** + - 可能使工具返回大型 JSON 对象或列表，压倒大语言模型 + - 应要求理解多种数据形式： + - ID 和名称 + - 时间戳和日期时间（月、日、年、秒） + - 文件 ID、名称、扩展名和 MIME 类型 + - URL、GID 等 + - 应探查工具返回所有有用数据形式的能力 + +9. **问题应主要反映真实的人类使用场景** + - 人类在 LLM 辅助下会关心的信息检索任务类型 + +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 文件： + +```xml + + + 找到在 2024 年第二季度创建、完成任务数量最多的项目。项目名称是什么？ + 网站改版 + + + 搜索在 2024 年 3 月关闭的、标记为"bug"的问题。哪个用户关闭的问题最多？请提供其用户名。 + sarah_dev + + + 查找修改了 /api 目录下文件、并在 2024 年 1 月 1 日至 1 月 31 日期间合并的拉取请求。有多少个不同的贡献者参与了这些 PR？ + 7 + + + 找到在 2023 年之前创建、获得星标数最多的仓库。仓库名称是什么？ + data-pipeline + + +``` + +## 评估示例 + +### 好的问题 + +**示例 1：需要深入探索的多跳问题（GitHub MCP）** +```xml + + 找到在 2023 年第三季度归档、且此前是该组织中被 fork 最多的项目仓库。该仓库使用的主要编程语言是什么？ + Python + +``` + +这个问题好的原因在于： +- 需要多次搜索才能找到已归档的仓库 +- 需要确定在归档前哪个仓库的 fork 数最多 +- 需要查看仓库详情以获取语言信息 +- 答案是一个简单、可验证的值 +- 基于历史（已闭合）数据，不会变化 + +**示例 2：需要理解上下文而无关键词匹配（项目管理 MCP）** +```xml + + 找到在 2023 年底完成、专注于改善客户引导的倡议项目。项目负责人在完成后创建了一份回顾文档。当时该负责人的职位是什么？ + 产品经理 + +``` + +这个问题好的原因在于： +- 没有使用具体项目名称（"专注于改善客户引导的倡议项目"） +- 需要从特定时间范围内找到已完成的项目 +- 需要确定项目负责人及其职位 +- 需要从回顾文档中理解上下文 +- 答案是人类可读且稳定的 +- 基于已完成的工作（不会变化） + +**示例 3：需要多步聚合的复杂问题（问题追踪器 MCP）** +```xml + + 在 2024 年 1 月报告的所有标记为"严重"优先级的 Bug 中，哪个受理人在 48 小时内解决了其被分配 Bug 的最高百分比？请提供该受理人的用户名。 + alex_eng + +``` + +这个问题好的原因在于： +- 需要按日期、优先级和状态过滤 Bug +- 需要按受理人分组并计算解决率 +- 需要理解时间戳以确定 48 小时窗口 +- 测试分页功能（可能需要处理大量 Bug） +- 答案是单一用户名 +- 基于特定时间段的历史数据 + +**示例 4：需要跨多种数据类型综合（CRM MCP）** +```xml + + 找到在 2023 年第四季度从 Starter 方案升级到 Enterprise 方案、且年合同价值最高的客户账户。该账户属于哪个行业？ + 医疗健康 + +``` + +这个问题好的原因在于： +- 需要理解订阅套餐变更 +- 需要在特定时间范围内识别升级事件 +- 需要比较合同价值 +- 必须获取账户行业信息 +- 答案简单且可验证 +- 基于已完成的过往交易 + +### 不好的问题 + +**示例 1：答案随时间变化** +```xml + + 当前分配给工程团队的未解决问题有多少？ + 47 + +``` + +这个问题不好的原因在于： +- 答案会随着问题的创建、关闭或重新分配而变化 +- 不是基于稳定/不变的数据 +- 依赖动态的"当前状态" + +**示例 2：关键词搜索过于简单** +```xml + + 找到标题为"Add authentication feature"的拉取请求，并告诉我谁创建的。 + developer123 + +``` + +这个问题不好的原因在于： +- 可以通过精确标题的关键词搜索直接解决 +- 不需要深入探索或理解 +- 无需综合或分析 + +**示例 3：答案格式有歧义** +```xml + + 列出所有以 Python 为主要编程语言的仓库。 + repo1, repo2, repo3, data-pipeline, ml-tools + +``` + +这个问题不好的原因在于： +- 答案是一个列表，可能以任意顺序返回 +- 难以通过直接字符串比较来验证 +- 大语言模型可能以不同格式输出（JSON 数组、逗号分隔、换行分隔） +- 更好的做法是询问具体的聚合值（计数）或最高值（最多星标） + +## 验证流程 + +创建评估后： + +1. **检查 XML 文件** 以了解模式 +2. **加载每个任务指令**，并使用 MCP 服务器和工具并行找出正确答案——亲自尝试解决问题 +3. **标记所有需要写入或破坏性操作的操作** +4. **汇总所有正确答案**，替换文档中的任何错误答案 +5. **移除所有需要写入或破坏性操作的 ``** + +请记住并行求解任务以避免上下文溢出，然后在最后汇总所有答案并对文件进行修改。 + +## 创建高质量评估的提示 + +1. **在生成任务之前深入思考并提前规划** +2. **在有机会时进行并行处理** 以加快速度并管理上下文 +3. **聚焦于真实的使用场景**——人类真正想要完成的任务 +4. **创建有挑战性的问题** 以测试 MCP 服务器能力的极限 +5. **确保稳定性**——使用历史数据和已闭合的概念 +6. **验证答案**——亲自使用 MCP 服务器工具解决问题 +7. **根据过程中的发现迭代并优化** + +--- + +# 运行评估 + +创建评估文件后，你可以使用提供的评估框架来测试你的 MCP 服务器。 + +## 环境设置 + +1. **安装依赖** + + ```bash + pip install -r scripts/requirements.txt + ``` + + 或手动安装： + ```bash + pip install anthropic mcp + ``` + +2. **设置 API 密钥** + + ```bash + export ANTHROPIC_API_KEY=your_api_key_here + ``` + +## 评估文件格式 + +评估文件使用 XML 格式，包含 `` 元素： + +```xml + + + 找到在 2024 年第二季度创建、完成任务数量最多的项目。项目名称是什么？ + 网站改版 + + + 搜索在 2024 年 3 月关闭的、标记为"bug"的问题。哪个用户关闭的问题最多？请提供其用户名。 + sarah_dev + + +``` + +## 运行评估 + +评估脚本（`scripts/evaluation.py`）支持三种传输类型： + +**重要说明：** +- **stdio 传输**：评估脚本会自动为你启动并管理 MCP 服务器进程。请勿手动运行服务器。 +- **SSE/HTTP 传输**：你必须在运行评估之前单独启动 MCP 服务器。脚本会连接到指定 URL 上已运行的服务器。 + +### 1. 本地 STDIO 服务器 + +适用于本地运行的 MCP 服务器（脚本自动启动服务器）： + +```bash +python scripts/evaluation.py \ + -t stdio \ + -c python \ + -a my_mcp_server.py \ + evaluation.xml +``` + +使用环境变量： + +```bash +python scripts/evaluation.py \ + -t stdio \ + -c python \ + -a my_mcp_server.py \ + -e API_KEY=abc123 \ + -e DEBUG=true \ + evaluation.xml +``` + +### 2. Server-Sent Events（SSE） + +适用于基于 SSE 的 MCP 服务器（需先启动服务器）： + +```bash +python scripts/evaluation.py \ + -t sse \ + -u https://example.com/mcp \ + -H "Authorization: Bearer token123" \ + -H "X-Custom-Header: value" \ + evaluation.xml +``` + +### 3. HTTP（Streamable HTTP） + +适用于基于 HTTP 的 MCP 服务器（需先启动服务器）： + +```bash +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 options: + -c, --command 运行 MCP 服务器的命令（例如：python、node） + -a, --args 命令的参数（例如：server.py） + -e, --env 环境变量，格式为 KEY=VALUE + +sse/http options: + -u, --url MCP 服务器 URL + -H, --header HTTP 头，格式为 'Key: Value' + +## 输出 + +评估脚本会生成一份详细报告，包括： + +- **汇总统计**： + - 准确率（正确数/总数） + - 平均任务耗时 + - 每任务平均工具调用次数 + - 工具调用总数 + +- **每项任务的结果**： + - 提示语及预期响应 + - 智能体的实际响应 + - 答案是否正确（✅/❌） + - 耗时及工具调用详情 + - 智能体对其方法的总结 + - 智能体对工具的反馈 + +### 将报告保存到文件 + +```bash +python scripts/evaluation.py \ + -t stdio \ + -c python \ + -a my_server.py \ + -o evaluation_report.md \ + evaluation.xml +``` + +## 完整工作流程示例 + +以下是创建和运行评估的完整示例： + +1. **创建评估文件**（`my_evaluation.xml`）： + +```xml + + + 找到在 2024 年 1 月创建问题最多的用户。该用户的用户名是什么？ + alice_developer + + + 在 2024 年第一季度合并的所有拉取请求中，哪个仓库的数量最多？请提供仓库名称。 + backend-api + + + 找到在 2023 年 12 月完成、且从开始到结束持续时间最长的项目。它花了多少天？ + 127 + + +``` + +2. **安装依赖**： + +```bash +pip install -r scripts/requirements.txt +export ANTHROPIC_API_KEY=your_api_key +``` + +3. **运行评估**： + +```bash +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`） +- 检查工具是否返回了过多数据 +- 确认分页功能是否正常工作 +- 考虑简化复杂问题 diff --git a/reference/mcp_best_practices.md b/reference/mcp_best_practices.md new file mode 100644 index 0000000..f308e64 --- /dev/null +++ b/reference/mcp_best_practices.md @@ -0,0 +1,249 @@ +# MCP 服务器最佳实践 + +## 快速参考 + +### 服务器命名 +- **Python**：`{service}_mcp`（例如 `slack_mcp`） +- **Node/TypeScript**：`{service}-mcp-server`（例如 `slack-mcp-server`） + +### 工具命名 +- 使用蛇形命名法，并带服务前缀 +- 格式：`{service}_{action}_{resource}` +- 示例：`slack_send_message`、`github_create_issue` + +### 响应格式 +- 同时支持 JSON 和 Markdown 两种格式 +- JSON 用于程序化处理 +- Markdown 用于人类可读 + +### 分页 +- 始终尊重 `limit` 参数 +- 返回 `has_more`、`next_offset`、`total_count` +- 默认每页 20–50 条 + +### 传输方式 +- **Streamable HTTP**：适用于远程服务器、多客户端场景 +- **stdio**：适用于本地集成、命令行工具 +- 避免使用 SSE（已弃用，推荐使用 streamable 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. **使用蛇形命名法**：`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 条 + +分页响应示例： +```json +{ + "total": 150, + "count": 20, + "offset": 0, + "items": [...], + "has_more": true, + "next_offset": 20 +} +``` + +--- + +## 传输方式选项 + +### Streamable HTTP + +**最适合**：远程服务器、Web 服务、多客户端场景 + +**特点**： +- 基于 HTTP 的双向通信 +- 支持多个同时连接的客户端 +- 可部署为 Web 服务 +- 支持服务器到客户端的通知 + +**使用场景**： +- 同时服务多个客户端 +- 部署为云服务 +- 与 Web 应用集成 + +### stdio + +**最适合**：本地集成、命令行工具 + +**特点**： +- 标准输入/输出流通信 +- 设置简单，无需网络配置 +- 作为客户端的子进程运行 + +**使用场景**： +- 为本地开发环境构建工具 +- 与桌面应用集成 +- 单用户、单会话场景 + +**注意**：stdio 服务器不应将日志输出到 stdout（请使用 stderr 记录日志） + +### 传输方式选择 + +| 标准 | stdio | Streamable HTTP | +|-----------|-------|-----------------| +| **部署** | 本地 | 远程 | +| **客户端** | 单个 | 多个 | +| **复杂度** | 低 | 中 | +| **实时性** | 否 | 是 | + +--- + +## 安全最佳实践 + +### 认证与授权 + +**OAuth 2.1**： +- 使用带有受认可机构证书的安全 OAuth 2.1 +- 在处理请求前验证访问令牌 +- 仅接受专门针对你的服务器的令牌 + +**API 密钥**： +- 将 API 密钥存储在环境变量中，切勿写在代码里 +- 在服务器启动时验证密钥 +- 认证失败时提供清晰的错误消息 + +### 输入验证 + +- 对文件路径进行清理以防止目录遍历攻击 +- 验证 URL 和外部标识符 +- 检查参数的大小和范围 +- 防止系统调用中的命令注入 +- 对所有输入使用模式验证（Pydantic/Zod） + +### 错误处理 + +- 不要向客户端暴露内部错误 +- 在服务端记录安全相关的错误 +- 提供有帮助但不泄露内部信息的错误消息 +- 错误发生后正确清理资源 + +### DNS 重绑定保护 + +对于本地运行的 streamable HTTP 服务器： +- 启用 DNS 重绑定保护 +- 验证所有传入连接的 `Origin` 头 +- 绑定到 `127.0.0.1` 而非 `0.0.0.0` + +--- + +## 工具注解 + +提供注解以帮助客户端理解工具行为： + +| 注解 | 类型 | 默认值 | 描述 | +|-----------|------|---------|-------------| +| `readOnlyHint` | boolean | false | 工具不会修改其运行环境 | +| `destructiveHint` | boolean | true | 工具可能执行破坏性更新 | +| `idempotentHint` | boolean | false | 使用相同参数的重复调用不会产生额外影响 | +| `openWorldHint` | boolean | true | 工具与外部实体交互 | + +**重要**：注解是提示，而非安全保证。客户端不应仅基于注解做出安全关键决策。 + +--- + +## 错误处理 + +- 使用标准 JSON-RPC 错误码 +- 在结果对象内报告工具错误（而非协议级别的错误） +- 提供有帮助、具体的错误消息，并附带建议的下一步操作 +- 不要暴露内部实现细节 +- 在发生错误时正确清理资源 + +错误处理示例： +```typescript +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 个） +- 记录安全注意事项 +- 说明所需的权限和访问级别 +- 记录速率限制和性能特征 diff --git a/reference/node_mcp_server.md b/reference/node_mcp_server.md new file mode 100644 index 0000000..ff34a40 --- /dev/null +++ b/reference/node_mcp_server.md @@ -0,0 +1,970 @@ +# Node/TypeScript MCP 服务端实现指南 + +## 概述 + +本文档提供了使用 MCP TypeScript SDK 实现 MCP 服务端时的 Node/TypeScript 特定最佳实践与示例。内容涵盖项目结构、服务端初始化、工具注册模式、基于 Zod 的输入验证、错误处理以及完整的可运行示例。 + +--- + +## 快速参考 + +### 关键导入 +```typescript +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"; +``` + +### 服务端初始化 +```typescript +const server = new McpServer({ + name: "service-mcp-server", + version: "1.0.0" +}); +``` + +### 工具注册模式 +```typescript +server.registerTool( + "tool_name", + { + title: "Tool Display Name", + description: "What the tool does", + inputSchema: { param: z.string() }, + outputSchema: { result: z.string() } + }, + async ({ param }) => { + const output = { result: `Processed: ${param}` }; + return { + content: [{ type: "text", text: JSON.stringify(output) }], + structuredContent: output // 结构化数据的现代模式 + }; + } +); +``` + +--- + +## MCP TypeScript SDK + +官方 MCP TypeScript SDK 提供： +- `McpServer` 类用于服务端初始化 +- `registerTool` 方法用于工具注册 +- Zod schema 集成实现运行时输入验证 +- 类型安全的工具处理函数实现 + +**重要——仅使用现代 API：** +- **应该使用**：`server.registerTool()`、`server.registerResource()`、`server.registerPrompt()` +- **不要使用**：旧的已弃用 API，如 `server.tool()`、`server.setRequestHandler(ListToolsRequestSchema, ...)` 或手动注册处理函数 +- `register*` 方法提供更好的类型安全性、自动 schema 处理，是推荐做法 + +详情请参阅参考资料中的 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 验证 schema +│ └── 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 schema 进行运行时输入验证和类型安全 +- `description` 字段必须显式提供——JSDoc 注释不会被自动提取 +- 显式提供 `title`、`description`、`inputSchema` 和 `annotations` +- `inputSchema` 必须是 Zod schema 对象（而非 JSON schema） +- 所有参数和返回值类型需显式声明 + +```typescript +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { z } from "zod"; + +const server = new McpServer({ + name: "example-mcp", + version: "1.0.0" +}); + +// 用于输入验证的 Zod schema +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 schema 推导的类型定义 +type UserSearchInput = z.infer; + +server.registerTool( + "example_search_users", + { + title: "Search Example Users", + description: `在 Example 系统中按姓名、邮箱或团队搜索用户。 + +该工具搜索 Example 平台中的所有用户资料，支持部分匹配和多种搜索筛选条件。不会创建或修改用户，仅搜索已有用户。 + +参数： + - query (string)：用于匹配姓名/邮箱的搜索字符串 + - limit (number)：返回的最大结果数，范围 1-100（默认值：20） + - offset (number)：分页时跳过的结果数（默认值：0） + - response_format ('markdown' | 'json')：输出格式（默认值：'markdown'） + +返回： + JSON 格式：结构化数据，schema 如下： + { + "total": number, // 找到的匹配总数 + "count": number, // 本次响应中的结果数 + "offset": number, // 当前分页偏移量 + "users": [ + { + "id": string, // 用户 ID（例如 "U123456789"） + "name": string, // 全名（例如 "张三"） + "email": string, // 邮箱地址 + "team": string, // 团队名称（可选） + "active": boolean // 用户是否活跃 + } + ], + "has_more": boolean, // 是否还有更多结果 + "next_offset": number // 下一页的偏移量（若 has_more 为 true） + } + +示例： + - 何时使用："查找所有营销团队成员" -> 参数为 query="team:marketing" + - 何时使用："搜索张三的账号" -> 参数为 query="zhang" + - 不要使用：需要创建用户时（应使用 example_create_user） + +错误处理： + - 请求过多时返回 "Error: Rate limit exceeded"（429 状态码） + - 搜索结果为空时返回 "未找到与 '' 匹配的用户"`, + inputSchema: UserSearchInputSchema, + annotations: { + readOnlyHint: true, + destructiveHint: false, + idempotentHint: true, + openWorldHint: true + } + }, + async (params: UserSearchInput) => { + try { + // 输入验证由 Zod schema 处理 + // 使用验证后的参数发起 API 请求 + const data = await makeApiRequest( + "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 Schema 用于输入验证 + +Zod 提供运行时类型验证： + +```typescript +import { z } from "zod"; + +// 带验证的基本 schema +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("跳过的结果数") +}); +``` + +## 响应格式选项 + +支持多种输出格式以提高灵活性： + +```typescript +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 格式**： +- 返回适合程序化处理的完整结构化数据 +- 包含所有可用字段和元数据 +- 使用一致的字段名和类型 + +## 分页实现 + +适用于列出资源的工具： + +```typescript +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) { + 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 常量以防止响应过长： + +```typescript +// 在 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; +} +``` + +## 错误处理 + +提供清晰、可操作的错误信息： + +```typescript +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)}`; +} +``` + +## 共享工具函数 + +将通用功能提取为可复用的函数： + +```typescript +// 共享 API 请求函数 +async function makeApiRequest( + endpoint: string, + method: "GET" | "POST" | "PUT" | "DELETE" = "GET", + data?: any, + params?: any +): Promise { + 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; + } +} +``` + +## Async/Await 最佳实践 + +网络请求和 I/O 操作始终使用 async/await： + +```typescript +// 好的做法：异步网络请求 +async function fetchData(resourceId: string): Promise { + const response = await axios.get(`${API_URL}/resource/${resourceId}`); + return response.data; +} + +// 不好的做法：Promise 链 +function fetchData(resourceId: string): Promise { + return axios.get(`${API_URL}/resource/${resourceId}`) + .then(response => response.data); // 可读性和可维护性较差 +} +``` + +## TypeScript 最佳实践 + +1. **使用严格 TypeScript**：在 tsconfig.json 中启用严格模式 +2. **定义接口**：为所有数据结构创建清晰的接口定义 +3. **避免使用 `any`**：使用正确类型或 `unknown` 替代 `any` +4. **使用 Zod 进行运行时验证**：使用 Zod schema 验证外部数据 +5. **类型守卫**：为复杂类型检查创建类型守卫函数 +6. **错误处理**：始终使用 try-catch 并配合正确的错误类型检查 +7. **空值安全**：使用可选链（`?.`）和空值合并（`??`） + +```typescript +// 好的做法：通过 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; + +async function getUser(id: string): Promise { + const data = await apiCall(`/users/${id}`); + return UserSchema.parse(data); // 运行时验证 +} + +// 不好的做法：使用 any +async function getUser(id: string): Promise { + return await apiCall(`/users/${id}`); // 没有类型安全 +} +``` + +## 包配置 + +### package.json + +```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 + +```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"] +} +``` + +## 完整示例 + +```typescript +#!/usr/bin/env node +/** + * Example 服务的 MCP 服务端。 + * + * 本服务端提供与 Example 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 schema +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; + +// 共享工具函数 +async function makeApiRequest( + endpoint: string, + method: "GET" | "POST" | "PUT" | "DELETE" = "GET", + data?: any, + params?: any +): Promise { + 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: "Search Example Users", + 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 的高效访问： + +```typescript +import { ResourceTemplate } from "@modelcontextprotocol/sdk/types.js"; + +// 使用 URI 模板注册资源 +server.registerResource( + { + uri: "file://documents/{name}", + name: "Document Resource", + 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（推荐用于远程服务端） + +```typescript +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（用于本地集成） + +```typescript +import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; + +const transport = new StdioServerTransport(); +await server.connect(transport); +``` + +**传输方式选择：** +- **流式 HTTP**：Web 服务、远程访问、多客户端 +- **stdio**：命令行工具、本地开发、子进程集成 + +### 通知支持 + +在服务端状态变化时通知客户端： + +```typescript +// 工具列表变更时通知 +server.notification({ + method: "notifications/tools/list_changed" +}); + +// 资源变更时通知 +server.notification({ + method: "notifications/resources/list_changed" +}); +``` + +适度使用通知——仅当服务端能力确实发生变化时使用。 + +--- + +## 代码最佳实践 + +### 代码组合性与可复用性 + +你的实现必须优先考虑组合性和代码复用： + +1. **提取通用功能**： + - 为多个工具中使用的操作创建可复用的辅助函数 + - 构建共享的 API 客户端来处理 HTTP 请求，而非重复代码 + - 将错误处理逻辑集中到工具函数中 + - 将业务逻辑提取到可组合的专用函数中 + - 提取共享的 Markdown 或 JSON 字段选择与格式化功能 + +2. **避免重复**： + - 切勿在不同工具之间复制粘贴相似代码 + - 如果发现自己两次编写相似逻辑，将其提取为函数 + - 分页、过滤、字段选择和格式化等常见操作应共享 + - 认证/授权逻辑应集中管理 + +## 构建与运行 + +始终先构建 TypeScript 代码再运行： + +```bash +# 构建项目 +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 schema 进行运行时输入验证，并启用 `.strict()` 约束 +- [ ] 所有 Zod schema 具有适当的约束条件和描述性错误信息 +- [ ] 所有工具具有全面的描述，包含显式的输入/输出类型 +- [ ] 描述包含返回值示例和完整的 schema 文档 +- [ ] 错误信息清晰、可操作且具有教育意义 + +### TypeScript 质量 +- [ ] 为所有数据结构定义 TypeScript 接口 +- [ ] 在 tsconfig.json 中启用严格 TypeScript 模式 +- [ ] 不使用 `any` 类型——使用 `unknown` 或正确的类型替代 +- [ ] 所有异步函数具有显式的 `Promise` 返回类型 +- [ ] 错误处理使用正确的类型守卫（例如 `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` +- [ ] 所有导入正确解析 +- [ ] 示例工具调用按预期工作 diff --git a/reference/python_mcp_server.md b/reference/python_mcp_server.md new file mode 100644 index 0000000..8d19b7b --- /dev/null +++ b/reference/python_mcp_server.md @@ -0,0 +1,718 @@ +# Python MCP 服务器实现指南 + +## 概述 + +本文档提供了使用 MCP Python SDK 实现 MCP 服务器的 Python 特定最佳实践和示例。内容涵盖服务器设置、工具注册模式、基于 Pydantic 的输入验证、错误处理以及完整的工作示例。 + +--- + +## 快速参考 + +### 关键导入 +```python +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 +``` + +### 服务器初始化 +```python +mcp = FastMCP("service_mcp") +``` + +### 工具注册模式 +```python +@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 模型进行输入验证： + +```python +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` 装饰器 +- 验证器方法必须包含类型提示 + +```python +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() +``` + +## 响应格式选项 + +支持多种输出格式以提高灵活性： + +```python +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 格式**： +- 返回适合程序化处理的完整、结构化数据 +- 包含所有可用字段和元数据 +- 使用一致的字段名称和类型 + +## 分页实现 + +对于列出资源的工具： + +```python +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) +``` + +## 错误处理 + +提供清晰、可操作的错误消息： + +```python +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__}" +``` + +## 共享工具函数 + +将通用功能提取为可复用的函数： + +```python +# 共享的 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() +``` + +## Async/Await 最佳实践 + +始终对网络请求和 I/O 操作使用 async/await： + +```python +# 推荐：异步网络请求 +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() +``` + +## 类型提示 + +全程使用类型提示： + +```python +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"]} +``` + +## 工具文档字符串 + +每个工具必须包含带有显式类型信息的全面文档字符串： + +```python +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 格式字符串，schema 如下： + + 成功响应： + { + "total": int, # 找到的匹配总数 + "count": int, # 当前响应中的结果数 + "offset": int, # 当前分页偏移量 + "users": [ + { + "id": str, # 用户 ID（例如 "U123456789"） + "name": str, # 全名（例如 "John Doe"） + "email": str, # 电子邮件地址（例如 "john@example.com"） + "team": str # 团队名称（例如 "Marketing"）- 可选 + } + ] + } + + 错误响应： + "错误：<错误消息>" 或 "未找到匹配 '' 的用户" + + Examples: + - 何时使用："查找所有市场团队成员" -> 参数 query="team:marketing" + - 何时使用："搜索 John 的账号" -> 参数 query="john" + - 何时不使用：需要创建用户时（应使用 example_create_user） + - 何时不使用：已有用户 ID 并需要完整详情时（应使用 example_get_user） + + 错误处理： + - 输入验证错误由 Pydantic 模型处理 + - 如果请求过多（429 状态），返回"错误：超出速率限制" + - 如果 API 密钥无效（401 状态），返回"错误：API 身份验证无效" + - 返回格式化的结果列表，或"未找到匹配 'query' 的用户" + ''' +``` + +## 完整示例 + +参见以下完整的 Python MCP 服务器示例： + +```python +#!/usr/bin/env python3 +''' +Example 服务的 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` 参数注入工具，以实现日志记录、进度报告、资源读取和用户交互等高级功能： + +```python +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) +``` + +**Context 功能：** +- `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 资源 + +### 资源注册 + +将数据暴露为资源，实现高效的基于模板的访问： + +```python +@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, {})) +``` + +**何时使用资源与工具：** +- **资源**：适用于带简单参数的数据访问（URI 模板） +- **工具**：适用于需要验证和业务逻辑的复杂操作 + +### 结构化输出类型 + +FastMCP 支持多种返回类型，不仅限于字符串： + +```python +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 模型 - 自动生成 schema。''' + user = await fetch_user(user_id) + return DetailedUser(**user) +``` + +### 生命周期管理 + +初始化跨请求持久化的资源： + +```python +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 支持两种主要的传输机制： + +```python +# stdio 传输（适用于本地工具）- 默认 +if __name__ == "__main__": + mcp.run() + +# Streamable HTTP 传输（适用于远程服务器） +if __name__ == "__main__": + mcp.run(transport="streamable_http", port=8000) +``` + +**传输选择：** +- **stdio**：命令行工具、本地集成、子进程执行 +- **Streamable HTTP**：Web 服务、远程访问、多客户端 + +--- + +## 代码最佳实践 + +### 代码可组合性与可复用性 + +您的实现必须优先考虑可组合性和代码复用： + +1. **提取通用功能**： + - 为跨多个工具使用的操作创建可复用的辅助函数 + - 构建共享的 API 客户端用于 HTTP 请求，而不是重复代码 + - 将错误处理逻辑集中到工具函数中 + - 将业务逻辑提取到可组合的专用函数中 + - 提取共享的 markdown 或 JSON 字段选择与格式化功能 + +2. **避免重复**： + - 切勿在工具之间复制粘贴相似的代码 + - 如果发现自己编写了两次相似的逻辑，将其提取为函数 + - 分页、过滤、字段选择和格式化等通用操作应共享 + - 身份验证/授权逻辑应集中处理 + +### Python 特定最佳实践 + +1. **使用类型提示**：始终包含函数参数和返回值的类型注解 +2. **Pydantic 模型**：为所有输入验证定义清晰的 Pydantic 模型 +3. **避免手动验证**：让 Pydantic 使用约束来处理输入验证 +4. **正确的导入**：对导入进行分组（标准库、第三方、本地） +5. **错误处理**：使用特定的异常类型（httpx.HTTPStatusError，而非通用的 Exception） +6. **异步上下文管理器**：对需要清理的资源使用 `async with` +7. **常量**：使用大写字母定义模块级常量 + +## 质量检查清单 + +在最终确定您的 Python MCP 服务器实现之前，请确保： + +### 策略设计 +- [ ] 工具支持完整的工作流程，而不仅仅是 API 端点封装 +- [ ] 工具名称反映自然的任务划分 +- [ ] 响应格式针对 agent 上下文效率进行优化 +- [ ] 在适当之处使用人类可读的标识符 +- [ ] 错误消息引导 agent 正确使用 + +### 实现质量 +- [ ] 聚焦实现：实现最重要和最具有价值的工具 +- [ ] 所有工具都有描述性名称和文档 +- [ ] 相似操作的返回类型保持一致 +- [ ] 所有外部调用都实现了错误处理 +- [ ] 服务器名称遵循格式：`{service}_mcp` +- [ ] 所有网络操作使用 async/await +- [ ] 通用功能被提取为可复用的函数 +- [ ] 错误消息清晰、可操作且具有教育意义 +- [ ] 输出经过正确验证和格式化 + +### 工具配置 +- [ ] 所有工具在装饰器中实现了 'name' 和 'annotations' +- [ ] 注解设置正确（readOnlyHint、destructiveHint、idempotentHint、openWorldHint） +- [ ] 所有工具使用 Pydantic BaseModel 进行输入验证，并附带 Field() 定义 +- [ ] 所有 Pydantic Field 具有显式类型、描述和约束 +- [ ] 所有工具具有包含显式输入/输出类型的全面文档字符串 +- [ ] 文档字符串包含 dict/JSON 返回的完整 schema 结构 +- [ ] Pydantic 模型处理输入验证（无需手动验证） + +### 高级特性（如适用） +- [ ] 使用上下文注入进行日志记录、进度报告或信息征求 +- [ ] 为适当的数据端点注册了资源 +- [ ] 为持久连接实现了生命周期管理 +- [ ] 使用了结构化输出类型（TypedDict、Pydantic 模型） +- [ ] 配置了合适的传输方式（stdio 或 Streamable HTTP） + +### 代码质量 +- [ ] 文件包含正确的导入，包括 Pydantic 导入 +- [ ] 在适用之处正确实现了分页 +- [ ] 为可能较大的结果集提供了过滤选项 +- [ ] 所有异步函数使用 `async def` 正确定义 +- [ ] HTTP 客户端使用遵循异步模式并配有正确的上下文管理器 +- [ ] 整个代码中使用了类型提示 +- [ ] 常量使用大写字母在模块级别定义 + +### 测试 +- [ ] 服务器成功运行：`python your_server.py --help` +- [ ] 所有导入正确解析 +- [ ] 示例工具调用按预期工作 +- [ ] 错误场景得到优雅处理 diff --git a/scripts/connections.py b/scripts/connections.py new file mode 100644 index 0000000..ffcd0da --- /dev/null +++ b/scripts/connections.py @@ -0,0 +1,151 @@ +"""Lightweight connection handling for MCP servers.""" + +from abc import ABC, abstractmethod +from contextlib import AsyncExitStack +from typing import Any + +from mcp import ClientSession, StdioServerParameters +from mcp.client.sse import sse_client +from mcp.client.stdio import stdio_client +from mcp.client.streamable_http import streamablehttp_client + + +class MCPConnection(ABC): + """Base class for MCP server connections.""" + + def __init__(self): + self.session = None + self._stack = None + + @abstractmethod + def _create_context(self): + """Create the connection context based on connection type.""" + + async def __aenter__(self): + """Initialize MCP server connection.""" + self._stack = AsyncExitStack() + await self._stack.__aenter__() + + try: + ctx = self._create_context() + result = await self._stack.enter_async_context(ctx) + + if len(result) == 2: + read, write = result + elif len(result) == 3: + read, write, _ = result + else: + raise ValueError(f"Unexpected context result: {result}") + + session_ctx = ClientSession(read, write) + self.session = await self._stack.enter_async_context(session_ctx) + await self.session.initialize() + return self + except BaseException: + await self._stack.__aexit__(None, None, None) + raise + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Clean up MCP server connection resources.""" + if self._stack: + await self._stack.__aexit__(exc_type, exc_val, exc_tb) + self.session = None + self._stack = None + + async def list_tools(self) -> list[dict[str, Any]]: + """Retrieve available tools from the MCP server.""" + response = await self.session.list_tools() + return [ + { + "name": tool.name, + "description": tool.description, + "input_schema": tool.inputSchema, + } + for tool in response.tools + ] + + async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + """Call a tool on the MCP server with provided arguments.""" + result = await self.session.call_tool(tool_name, arguments=arguments) + return result.content + + +class MCPConnectionStdio(MCPConnection): + """MCP connection using standard input/output.""" + + def __init__(self, command: str, args: list[str] = None, env: dict[str, str] = None): + super().__init__() + self.command = command + self.args = args or [] + self.env = env + + def _create_context(self): + return stdio_client( + StdioServerParameters(command=self.command, args=self.args, env=self.env) + ) + + +class MCPConnectionSSE(MCPConnection): + """MCP connection using Server-Sent Events.""" + + def __init__(self, url: str, headers: dict[str, str] = None): + super().__init__() + self.url = url + self.headers = headers or {} + + def _create_context(self): + return sse_client(url=self.url, headers=self.headers) + + +class MCPConnectionHTTP(MCPConnection): + """MCP connection using Streamable HTTP.""" + + def __init__(self, url: str, headers: dict[str, str] = None): + super().__init__() + self.url = url + self.headers = headers or {} + + def _create_context(self): + return streamablehttp_client(url=self.url, headers=self.headers) + + +def create_connection( + transport: str, + command: str = None, + args: list[str] = None, + env: dict[str, str] = None, + url: str = None, + headers: dict[str, str] = None, +) -> MCPConnection: + """Factory function to create the appropriate MCP connection. + + Args: + transport: Connection type ("stdio", "sse", or "http") + command: Command to run (stdio only) + args: Command arguments (stdio only) + env: Environment variables (stdio only) + url: Server URL (sse and http only) + headers: HTTP headers (sse and http only) + + Returns: + MCPConnection instance + """ + transport = transport.lower() + + if transport == "stdio": + if not command: + raise ValueError("Command is required for stdio transport") + return MCPConnectionStdio(command=command, args=args, env=env) + + elif transport == "sse": + if not url: + raise ValueError("URL is required for sse transport") + return MCPConnectionSSE(url=url, headers=headers) + + elif transport in ["http", "streamable_http", "streamable-http"]: + if not url: + raise ValueError("URL is required for http transport") + return MCPConnectionHTTP(url=url, headers=headers) + + else: + raise ValueError(f"Unsupported transport type: {transport}. Use 'stdio', 'sse', or 'http'") diff --git a/scripts/evaluation.py b/scripts/evaluation.py new file mode 100644 index 0000000..4177856 --- /dev/null +++ b/scripts/evaluation.py @@ -0,0 +1,373 @@ +"""MCP Server Evaluation Harness + +This script evaluates MCP servers by running test questions against them using Claude. +""" + +import argparse +import asyncio +import json +import re +import sys +import time +import traceback +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Any + +from anthropic import Anthropic + +from connections import create_connection + +EVALUATION_PROMPT = """You are an AI assistant with access to tools. + +When given a task, you MUST: +1. Use the available tools to complete the task +2. Provide summary of each step in your approach, wrapped in tags +3. Provide feedback on the tools provided, wrapped in tags +4. Provide your final response, wrapped in tags + +Summary Requirements: +- In your tags, you must explain: + - The steps you took to complete the task + - Which tools you used, in what order, and why + - The inputs you provided to each tool + - The outputs you received from each tool + - A summary for how you arrived at the response + +Feedback Requirements: +- In your tags, provide constructive feedback on the tools: + - Comment on tool names: Are they clear and descriptive? + - Comment on input parameters: Are they well-documented? Are required vs optional parameters clear? + - Comment on descriptions: Do they accurately describe what the tool does? + - Comment on any errors encountered during tool usage: Did the tool fail to execute? Did the tool return too many tokens? + - Identify specific areas for improvement and explain WHY they would help + - Be specific and actionable in your suggestions + +Response Requirements: +- Your response should be concise and directly address what was asked +- Always wrap your final response in tags +- If you cannot solve the task return NOT_FOUND +- For numeric responses, provide just the number +- For IDs, provide just the ID +- For names or text, provide the exact text requested +- Your response should go last""" + + +def parse_evaluation_file(file_path: Path) -> list[dict[str, Any]]: + """Parse XML evaluation file with qa_pair elements.""" + try: + tree = ET.parse(file_path) + root = tree.getroot() + evaluations = [] + + for qa_pair in root.findall(".//qa_pair"): + question_elem = qa_pair.find("question") + answer_elem = qa_pair.find("answer") + + if question_elem is not None and answer_elem is not None: + evaluations.append({ + "question": (question_elem.text or "").strip(), + "answer": (answer_elem.text or "").strip(), + }) + + return evaluations + except Exception as e: + print(f"Error parsing evaluation file {file_path}: {e}") + return [] + + +def extract_xml_content(text: str, tag: str) -> str | None: + """Extract content from XML tags.""" + pattern = rf"<{tag}>(.*?)" + matches = re.findall(pattern, text, re.DOTALL) + return matches[-1].strip() if matches else None + + +async def agent_loop( + client: Anthropic, + model: str, + question: str, + tools: list[dict[str, Any]], + connection: Any, +) -> tuple[str, dict[str, Any]]: + """Run the agent loop with MCP tools.""" + messages = [{"role": "user", "content": question}] + + response = await asyncio.to_thread( + client.messages.create, + model=model, + max_tokens=4096, + system=EVALUATION_PROMPT, + messages=messages, + tools=tools, + ) + + messages.append({"role": "assistant", "content": response.content}) + + tool_metrics = {} + + while response.stop_reason == "tool_use": + tool_use = next(block for block in response.content if block.type == "tool_use") + tool_name = tool_use.name + tool_input = tool_use.input + + tool_start_ts = time.time() + try: + tool_result = await connection.call_tool(tool_name, tool_input) + tool_response = json.dumps(tool_result) if isinstance(tool_result, (dict, list)) else str(tool_result) + except Exception as e: + tool_response = f"Error executing tool {tool_name}: {str(e)}\n" + tool_response += traceback.format_exc() + tool_duration = time.time() - tool_start_ts + + if tool_name not in tool_metrics: + tool_metrics[tool_name] = {"count": 0, "durations": []} + tool_metrics[tool_name]["count"] += 1 + tool_metrics[tool_name]["durations"].append(tool_duration) + + messages.append({ + "role": "user", + "content": [{ + "type": "tool_result", + "tool_use_id": tool_use.id, + "content": tool_response, + }] + }) + + response = await asyncio.to_thread( + client.messages.create, + model=model, + max_tokens=4096, + system=EVALUATION_PROMPT, + messages=messages, + tools=tools, + ) + messages.append({"role": "assistant", "content": response.content}) + + response_text = next( + (block.text for block in response.content if hasattr(block, "text")), + None, + ) + return response_text, tool_metrics + + +async def evaluate_single_task( + client: Anthropic, + model: str, + qa_pair: dict[str, Any], + tools: list[dict[str, Any]], + connection: Any, + task_index: int, +) -> dict[str, Any]: + """Evaluate a single QA pair with the given tools.""" + start_time = time.time() + + print(f"Task {task_index + 1}: Running task with question: {qa_pair['question']}") + response, tool_metrics = await agent_loop(client, model, qa_pair["question"], tools, connection) + + response_value = extract_xml_content(response, "response") + summary = extract_xml_content(response, "summary") + feedback = extract_xml_content(response, "feedback") + + duration_seconds = time.time() - start_time + + return { + "question": qa_pair["question"], + "expected": qa_pair["answer"], + "actual": response_value, + "score": int(response_value == qa_pair["answer"]) if response_value else 0, + "total_duration": duration_seconds, + "tool_calls": tool_metrics, + "num_tool_calls": sum(len(metrics["durations"]) for metrics in tool_metrics.values()), + "summary": summary, + "feedback": feedback, + } + + +REPORT_HEADER = """ +# Evaluation Report + +## Summary + +- **Accuracy**: {correct}/{total} ({accuracy:.1f}%) +- **Average Task Duration**: {average_duration_s:.2f}s +- **Average Tool Calls per Task**: {average_tool_calls:.2f} +- **Total Tool Calls**: {total_tool_calls} + +--- +""" + +TASK_TEMPLATE = """ +### Task {task_num} + +**Question**: {question} +**Ground Truth Answer**: `{expected_answer}` +**Actual Answer**: `{actual_answer}` +**Correct**: {correct_indicator} +**Duration**: {total_duration:.2f}s +**Tool Calls**: {tool_calls} + +**Summary** +{summary} + +**Feedback** +{feedback} + +--- +""" + + +async def run_evaluation( + eval_path: Path, + connection: Any, + model: str = "claude-3-7-sonnet-20250219", +) -> str: + """Run evaluation with MCP server tools.""" + print("🚀 Starting Evaluation") + + client = Anthropic() + + tools = await connection.list_tools() + print(f"📋 Loaded {len(tools)} tools from MCP server") + + qa_pairs = parse_evaluation_file(eval_path) + print(f"📋 Loaded {len(qa_pairs)} evaluation tasks") + + results = [] + for i, qa_pair in enumerate(qa_pairs): + print(f"Processing task {i + 1}/{len(qa_pairs)}") + result = await evaluate_single_task(client, model, qa_pair, tools, connection, i) + results.append(result) + + correct = sum(r["score"] for r in results) + accuracy = (correct / len(results)) * 100 if results else 0 + average_duration_s = sum(r["total_duration"] for r in results) / len(results) if results else 0 + average_tool_calls = sum(r["num_tool_calls"] for r in results) / len(results) if results else 0 + total_tool_calls = sum(r["num_tool_calls"] for r in results) + + report = REPORT_HEADER.format( + correct=correct, + total=len(results), + accuracy=accuracy, + average_duration_s=average_duration_s, + average_tool_calls=average_tool_calls, + total_tool_calls=total_tool_calls, + ) + + report += "".join([ + TASK_TEMPLATE.format( + task_num=i + 1, + question=qa_pair["question"], + expected_answer=qa_pair["answer"], + actual_answer=result["actual"] or "N/A", + correct_indicator="✅" if result["score"] else "❌", + total_duration=result["total_duration"], + tool_calls=json.dumps(result["tool_calls"], indent=2), + summary=result["summary"] or "N/A", + feedback=result["feedback"] or "N/A", + ) + for i, (qa_pair, result) in enumerate(zip(qa_pairs, results)) + ]) + + return report + + +def parse_headers(header_list: list[str]) -> dict[str, str]: + """Parse header strings in format 'Key: Value' into a dictionary.""" + headers = {} + if not header_list: + return headers + + for header in header_list: + if ":" in header: + key, value = header.split(":", 1) + headers[key.strip()] = value.strip() + else: + print(f"Warning: Ignoring malformed header: {header}") + return headers + + +def parse_env_vars(env_list: list[str]) -> dict[str, str]: + """Parse environment variable strings in format 'KEY=VALUE' into a dictionary.""" + env = {} + if not env_list: + return env + + for env_var in env_list: + if "=" in env_var: + key, value = env_var.split("=", 1) + env[key.strip()] = value.strip() + else: + print(f"Warning: Ignoring malformed environment variable: {env_var}") + return env + + +async def main(): + parser = argparse.ArgumentParser( + description="Evaluate MCP servers using test questions", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Evaluate a local stdio MCP server + python evaluation.py -t stdio -c python -a my_server.py eval.xml + + # Evaluate an SSE MCP server + python evaluation.py -t sse -u https://example.com/mcp -H "Authorization: Bearer token" eval.xml + + # Evaluate an HTTP MCP server with custom model + python evaluation.py -t http -u https://example.com/mcp -m claude-3-5-sonnet-20241022 eval.xml + """, + ) + + parser.add_argument("eval_file", type=Path, help="Path to evaluation XML file") + parser.add_argument("-t", "--transport", choices=["stdio", "sse", "http"], default="stdio", help="Transport type (default: stdio)") + parser.add_argument("-m", "--model", default="claude-3-7-sonnet-20250219", help="Claude model to use (default: claude-3-7-sonnet-20250219)") + + stdio_group = parser.add_argument_group("stdio options") + stdio_group.add_argument("-c", "--command", help="Command to run MCP server (stdio only)") + stdio_group.add_argument("-a", "--args", nargs="+", help="Arguments for the command (stdio only)") + stdio_group.add_argument("-e", "--env", nargs="+", help="Environment variables in KEY=VALUE format (stdio only)") + + remote_group = parser.add_argument_group("sse/http options") + remote_group.add_argument("-u", "--url", help="MCP server URL (sse/http only)") + remote_group.add_argument("-H", "--header", nargs="+", dest="headers", help="HTTP headers in 'Key: Value' format (sse/http only)") + + parser.add_argument("-o", "--output", type=Path, help="Output file for evaluation report (default: stdout)") + + args = parser.parse_args() + + if not args.eval_file.exists(): + print(f"Error: Evaluation file not found: {args.eval_file}") + sys.exit(1) + + headers = parse_headers(args.headers) if args.headers else None + env_vars = parse_env_vars(args.env) if args.env else None + + try: + connection = create_connection( + transport=args.transport, + command=args.command, + args=args.args, + env=env_vars, + url=args.url, + headers=headers, + ) + except ValueError as e: + print(f"Error: {e}") + sys.exit(1) + + print(f"🔗 Connecting to MCP server via {args.transport}...") + + async with connection: + print("✅ Connected successfully") + report = await run_evaluation(args.eval_file, connection, args.model) + + if args.output: + args.output.write_text(report) + print(f"\n✅ Report saved to {args.output}") + else: + print("\n" + report) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/example_evaluation.xml b/scripts/example_evaluation.xml new file mode 100644 index 0000000..41e4459 --- /dev/null +++ b/scripts/example_evaluation.xml @@ -0,0 +1,22 @@ + + + Calculate the compound interest on $10,000 invested at 5% annual interest rate, compounded monthly for 3 years. What is the final amount in dollars (rounded to 2 decimal places)? + 11614.72 + + + A projectile is launched at a 45-degree angle with an initial velocity of 50 m/s. Calculate the total distance (in meters) it has traveled from the launch point after 2 seconds, assuming g=9.8 m/s². Round to 2 decimal places. + 87.25 + + + A sphere has a volume of 500 cubic meters. Calculate its surface area in square meters. Round to 2 decimal places. + 304.65 + + + Calculate the population standard deviation of this dataset: [12, 15, 18, 22, 25, 30, 35]. Round to 2 decimal places. + 7.61 + + + Calculate the pH of a solution with a hydrogen ion concentration of 3.5 × 10^-5 M. Round to 2 decimal places. + 4.46 + + diff --git a/scripts/requirements.txt b/scripts/requirements.txt new file mode 100644 index 0000000..5ef9379 --- /dev/null +++ b/scripts/requirements.txt @@ -0,0 +1,30 @@ +- claude：通用代理，适用于任何不适合更专业代理的任务。未指定代理名称时 FleetView 的默认选项。（工具：*） +- Explore：只读搜索代理，用于广泛的发散搜索——当回答需要扫描大量文件、目录或命名约定，而你只需要结论而非文件内容时使用。它读取片段而非完整文件，因此只能定位代码，无法审查或审计代码。指定搜索广度："medium" 表示适度探索，"very thorough" 表示多个位置和命名约定。（工具：除 Agent、Artifact、ExitPlanMode、Edit、Write、NotebookEdit 之外的所有工具） +- general-purpose：通用代理，用于研究复杂问题、搜索代码以及执行多步骤任务。当你搜索关键词或文件但不确定能否在头几次尝试中正确匹配时，使用此代理为你执行搜索。（工具：*） +- Plan：软件架构代理，用于设计实现方案。当你需要规划任务的实现策略时使用。返回逐步方案，识别关键文件，并考虑架构权衡。（工具：除 Agent、Artifact、ExitPlanMode、Edit、Write、NotebookEdit 之外的所有工具） +- statusline-setup：使用此代理配置用户的 Claude Code 状态行设置。（工具：Read、Edit） + +当你同时启动多个代理执行独立工作时，在一条消息中使用多个工具调用发送，以便它们并发运行。 + +以下技能可供 Skill 工具使用： + +- deep-research：深度研究工具——展开网络搜索、获取来源、对抗性验证声明、综合生成带引用的报告。当用户需要关于任何主题的深度、多来源、经过事实核查的研究报告时使用。调用前，检查问题是否足够具体可以直接研究——如果不够明确（例如"买什么车"没有预算/用途/地区），先提出 2-3 个澄清问题以缩小范围。然后将完善后的问题作为参数传递，并将答案编织进去。 +- dataviz：每当你准备创建任何图表、图形、绘图、仪表盘或数据可视化时使用此技能——无论是 HTML 或 React 产物、内联 SVG、任何库（matplotlib、plotly、d3、Recharts 等）中的绘图代码、将要渲染和上传的图像/PNG，还是分享到 Slack 的图表。在编写第一行图表代码、选择图表颜色、构建统计图块/仪表/KPI 行或布局仪表盘之前**先阅读它**。生成风格统一、优雅、易访问、在浅色和深色模式下保持一致的可视化效果，使用可替换为你自己的品牌色的品牌中立占位调色板。教授一种设计系统无关的方法：表单启发式、带可运行验证器的颜色公式、标记规范以及交互规则。经过验证的默认调色板记录在 `references/palette.md` 中——将该文件的值替换为你的品牌色。触发词："chart"、"graph"、"plot"、"data viz"、"visualization"、"dashboard"、"analytics"、"visualize data"、"categorical colors"、"sequential / diverging palette"、"stat tile"、"sparkline"、"heatmap"、"legend"、"axis"、"tooltip"、"chart colors"、"color by series"。 +- update-config：使用此技能通过 settings.json 配置 Claude Code 运行环境。自动化行为（"从现在起当 X 时"、"每次 X 时"、"每当 X 时"、"在 X 之前/之后"）需要在 settings.json 中配置钩子——这些由运行环境执行，而非 Claude，因此记忆/偏好设置无法满足此类需求。也用于：权限（"允许 X"、"添加权限"、"将权限移至"）、环境变量（"设置 X=Y"）、钩子故障排查，或对 settings.json/settings.local.json 文件的任何更改。示例："允许 npm 命令"、"将 bq 权限添加到全局设置"、"将权限移至用户设置"、"设置 DEBUG=true"、"当 claude 停止时显示 X"。对于简单的设置如主题/模型，建议使用 /config 命令。 +- keybindings-help：当用户想要自定义键盘快捷键、重新绑定按键、添加和弦绑定或修改 ~/.claude/keybindings.json 时使用。示例："重新绑定 ctrl+s"、"添加快捷键和弦"、"更改提交键"、"自定义键绑定"。 +- verify：通过端到端执行代码并观察行为来验证代码更改是否确实按预期工作——驱动受影响的流程，而不仅仅是运行测试或类型检查。在提交重要更改之前运行。不要在只涉及测试、文档或其他没有运行时表面的代码的 diff 上调用它（对产品源代码的更改始终具有运行时表面）——因为没有什么可观察的。 +- code-review：审查当前 diff，查找正确性错误以及重用/简化/效率方面的清理机会，按指定的努力级别（low/medium：较少但高置信度的发现；high→max：更广泛的覆盖范围，可能包含不确定的发现）。传递 --comment 以将发现作为内联 PR 评论发布，或传递 --fix 以在审查后将修复应用于工作树。 +- simplify：审查已更改的代码，查找重用、简化、效率和层次方面的清理机会，然后应用修复。仅关注质量——它不寻找错误；请使用 /code-review 进行错误检查。 +- fewer-permission-prompts：扫描你的对话记录以查找常见的只读 Bash 和 MCP 工具调用，然后向项目 .claude/settings.json 添加优先允许列表以减少权限提示。 +- loop：按固定间隔运行提示或斜杠命令（例如 /loop 5m /foo，默认为 10m）。当用户想要设置重复任务、轮询状态或以固定间隔重复运行某些操作时使用（例如"每 5 分钟检查一次部署"、"持续运行 /babysit-prs"）。不要为一过性任务调用。 +- claude-api：Claude API / Anthropic SDK 的参考资料——模型 ID、定价、参数、流式传输、工具使用、MCP、代理、缓存、令牌计数、模型迁移。 +触发——在打开目标文件之前**先阅读**；不要因为"看起来像一行内容"就跳过——以下情况触发：提示中提到了任何形式的 Claude/Anthropic（Claude、Anthropic、Fable、Opus、Sonnet、Haiku、`anthropic`、`@anthropic-ai`、`claude-*`、`us.anthropic.*`、`[1m]`）；用户询问有关 LLM 的问题（定价/模型选择/限制/缓存）——绝不凭记忆回答；或者任务是 LLM 相关的但未指明提供商（agent/MCP/工具定义/多代理/RAG/LLM 评审/计算机使用；对自然语言进行生成/摘要/提取/分类/重写/对话；调试拒绝/截断/流式传输/工具调用/令牌）。 +仅在以下情况**跳过**（覆盖所有触发条件）：正在处理其他提供商（查询中提到了 OpenAI/GPT/Gemini/Llama/Mistral/Cohere/Ollama）；或者项目中 grep 到了相关关键字（如果未指明提供商，先运行此 grep——不要直接读取文件）。 +- run：启动并驱动此项目的应用程序以查看更改的效果。当要求运行、启动或截取应用程序屏幕截图，或确认更改在实际应用程序（而不仅仅是测试）中生效时使用。首先查找已涵盖启动应用程序的项目技能；否则回退到按项目类型（CLI、服务器、TUI、Electron、浏览器驱动、库）的内置模式。 +- init：初始化新的 CLAUDE.md 文件并添加代码库文档。 +- review：审查 GitHub 拉取请求；对于工作目录的 diff，请使用 /code-review。 +- security-review：完成对当前分支上待处理更改的安全审查。 +- slides：创建 PPTX 格式的演示文稿——包含基于数据驱动内容的图表、表格和图示。 +- pr：打开当前分支的 GitHub 比较页面以创建新的 PR。 +- doctor：诊断 Claude Code 设置问题并运行自检。 +- code-review：可作为斜杠命令 `/code-review` 使用。simplify 技能可作为 `/simplify` 使用。fewer-permission-prompts 技能可作为 `/fewer-permission-prompts` 使用。deep-research 技能可作为 `/deep-research` 使用。update-config 技能可作为 `/update-config` 使用。loop 技能可作为 `/loop` 使用。verify 技能可作为 `/verify` 使用。keybindings-help 技能可作为 `/keybindings-help` 使用。dataviz 技能可作为 `/dataviz` 使用。claude-api 技能可作为 `/claude-api` 使用。init 技能可作为 `/init` 使用。review 技能可作为 `/review` 使用。security-review 技能可作为 `/security-review` 使用。slides 技能可作为 `/slides` 使用。pr 技能可作为 `/pr` 使用。doctor 技能可作为 `/doctor` 使用。run 技能可作为 `/run` 使用。code-review 技能可作为 `/code-review` 使用。