中文译文

MCP 服务器开发指南

概述

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

流程

🚀 高层工作流

创建高质量的 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、stdio）
工具、资源和提示词定义

1.3 研究框架文档

推荐技术栈：

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

加载框架文档：

MCP 最佳实践：📋 查看最佳实践 - 核心指南

对于 TypeScript（推荐）：

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

对于 Python：

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

1.4 规划你的实现

理解 API： 查看服务的 API 文档，以确定关键端点、认证要求和数据模型。根据需要运用网络搜索和 WebFetch。

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

阶段 2：实现

2.1 搭建项目结构

参见针对具体语言的项目搭建指南：

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

2.2 实现核心基础设施

创建共享工具：

带身份验证的 API 客户端
错误处理辅助函数
响应格式化（JSON/Markdown）
分页支持

2.3 实现工具

对于每个工具：

输入 Schema：

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

输出 Schema：

尽可能为结构化数据定义 outputSchema
在工具响应中使用 structuredContent（TypeScript SDK 特性）
帮助客户端理解和处理工具输出

工具描述：

功能简介
参数描述
返回类型 schema

实现：

对 I/O 操作使用 async/await
使用可操作的消息进行适当的错误处理
在适用的情况下支持分页
使用现代 SDK 时同时返回文本内容和结构化数据

注解：

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

阶段 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 服务器之后，创建全面的评估来测试其有效性。

加载 ✅ 评估指南以获取完整的评估指南。

4.1 理解评估目的

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

4.2 创建 10 个评估问题

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

工具检查：列出可用工具并理解其能力
内容探索：使用只读操作来探索可用数据
问题生成：创建 10 个复杂、真实的问题
答案验证：自己解决每个问题以验证答案

4.3 评估要求

确保每个问题：

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

4.4 输出格式

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

<evaluation>
  <qa_pair>
    <question>Find discussions about AI model launches with animal codenames. One model needed a specific safety designation that uses the format ASL-X. What number X was being determined for the model named after a spotted wild cat?</question>
    <answer>3</answer>
  </qa_pair>
<!-- More qa_pairs... -->
</evaluation>

参考文件

📚 文档库

在开发过程中按需加载这些资源：

核心 MCP 文档（首先加载）

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

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 schema 模式
- 使用 server.registerTool 注册工具
- 完整可运行示例
- 质量检查清单

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

✅ 评估指南 - 完整的评估创建指南，包括：
- 问题创建指南
- 答案验证策略
- XML 格式规范
- 示例问题与答案
- 使用提供的脚本运行评估 FILE:reference/mcp_best_practices.md

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 项

传输方式

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

名称应当通用、能描述所集成的服务、易于从任务描述中推断，且不含版本号。

工具命名与设计

工具命名

使用 snake_case：search_users、create_project、get_channel_info
包含服务前缀：预判你的 MCP 服务器可能与其他 MCP 服务器一起使用
- 使用 slack_send_message 而不是仅用 send_message
- 使用 github_create_issue 而不是仅用 create_issue
以动作为导向：以动词开头（get、list、search、create 等）
具体明确：避免可能与其他服务器冲突的通用名称

工具设计

工具描述必须狭义且无歧义地描述功能
描述必须与实际功能精确匹配
提供工具注解（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
}

传输选项

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 重绑定防护

对于在本地运行的可流式 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 错误码
在结果对象中报告工具错误（而非协议级错误）
提供有帮助的、具体的错误消息，并附上建议的后续步骤
不要暴露内部实现细节
出错时正确清理资源

错误处理示例：

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 个）
记录安全注意事项
明确所需的权限和访问级别
记录速率限制和性能特征 FILE:reference/evaluation.md

MCP 服务器评估指南

概述

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

快速参考

评估要求

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

输出格式

<evaluation>
   <qa_pair>
      <question>Your question here</question>
      <answer>Single verifiable answer</answer>
   </qa_pair>
</evaluation>

评估的目的

衡量一个 MCP 服务器质量的标准，不是它实现工具的程度有多好或多全面，而是这些实现（输入/输出 schema、文档字符串/描述、功能）在 LLM 没有任何其他上下文、且只能访问这些 MCP 服务器的情况下，能在多大程度上帮助 LLM 回答真实且困难的问题。

评估概览

创建 10 个人类可读的问题，这些问题仅需只读、独立、非破坏性且幂等的操作即可回答。每个问题都应：

真实
清晰简洁
无歧义
复杂，可能需要数十次工具调用或步骤
能用一个你事先确定的、可验证的单一值来回答

问题指南

核心要求

问题必须相互独立
- 每个问题都不应依赖于任何其他问题的答案
- 不应假定因处理另一个问题而产生过先前的写操作
问题必须仅要求非破坏性且幂等的工具使用
- 不应指示或要求修改状态才能得出正确答案
问题必须真实、清晰、简洁且复杂
- 必须要求另一个 LLM 使用多个（可能是数十个）工具或步骤才能回答

复杂性与深度

问题必须要求深入探索
- 考虑需要多个子问题和顺序工具调用的多跳问题
- 每一步都应受益于在前面的问题中发现的信息
问题可能需要大量分页
- 可能需要翻阅多页结果
- 可能需要查询旧数据（1-2 年前的过时数据）以找到小众信息
- 问题必须足够困难
问题必须要求深入理解
- 而非表面层次的知识
- 可以提出复杂观点作为真/假问题，要求提供证据
- 可以使用多项选择形式，让 LLM 必须搜索不同的假设
问题绝不能通过直接的关键词搜索就能解决
- 不要包含目标内容中的具体关键词
- 使用同义词、相关概念或改述
- 要求进行多次搜索、分析多个相关条目、提取上下文，然后推导出答案

工具测试

问题应对工具的返回值进行压力测试
- 可能促使工具返回大型 JSON 对象或列表，从而使 LLM 不堪重负
- 应要求理解多种模态的数据：
  - ID 和名称
  - 时间戳和日期时间（月、日、年、秒）
  - 文件 ID、名称、扩展名和 mimetype
  - URL、GID 等
- 应探查工具返回所有有用数据形式的能力
问题应 MOSTLY 反映真实的人类使用场景
- 即 HUMANS 在 LLM 辅助下会关心的那类信息检索任务
问题可能需要数十次工具调用
- 这对上下文有限的 LLM 构成挑战
- 鼓励 MCP 服务器工具减少返回的信息量
包含模糊的问题
- 可能具有歧义，或者需要在调用哪些工具上做出困难决策
- 迫使 LLM 有可能犯错或产生误解
- 确保尽管存在 AMBIGUITY，仍然有 STILL A SINGLE VERIFIABLE ANSWER

稳定性

问题的设计必须使答案 DOES NOT CHANGE
- 不要提出依赖动态「当前状态」的问题
- 例如，不要统计：
  - 某条帖子的反应数
  - 某个帖子的回复数
  - 某个频道的成员数
DO NOT 让 MCP 服务器 RESTRICT 你创建的问题种类
- 创建具有挑战性和复杂性的问题
- 有些可能无法用现有的 MCP 服务器工具解决
- 问题可能要求特定的输出格式（datetime 与 epoch time、JSON 与 MARKDOWN）
- 问题可能需要数十次工具调用才能完成

答案指南

可验证性

答案必须能通过直接字符串比较进行验证
- 如果答案可以用多种格式重写，请在问题中明确指定输出格式
- 示例："使用 YYYY/MM/DD 格式。"、"回答 True 或 False。"、"只回答 A、B、C 或 D，不要有其他内容。"
- 答案应当是一个可验证的单一值，例如：
  - 用户 ID、用户名、显示名称、名、姓
  - 频道 ID、频道名称
  - 消息 ID、字符串
  - URL、标题
  - 数值
  - 时间戳、日期时间
  - 布尔值（用于 True/False 问题）
  - 电子邮件地址、电话号码
  - 文件 ID、文件名、文件扩展名
  - 多项选择题答案
- 答案不得要求特殊格式或复杂的结构化输出
- 答案将使用直接字符串比较进行验证

可读性

答案通常应优先采用人类可读的格式
- 示例：姓名、名、姓、日期时间、文件名、消息字符串、URL、是/否、true/false、a/b/c/d
- 而不是不透明的 ID（尽管 ID 也可以接受）
- 绝大多数答案应当是人类可读的

稳定性

答案必须是稳定/不变的
- 查看旧内容（例如已结束的对话、已上线的项目、已回答的问题）
- 基于"已封闭"的概念创建问题，使其始终返回相同的答案
- 问题可以要求考虑固定的时间窗口，以隔离非稳定性答案
- 依赖不太可能发生变化的上下文
- 示例：如果查找论文名称，要足够具体，以免答案与后来发表的论文混淆
答案必须清晰且无歧义
- 问题的设计必须确保只有一个清晰明确的答案
- 答案可以通过使用 MCP 服务器工具推导得出

多样性

答案必须具有多样性
- 答案应当是单一的可验证值，涵盖多样的模态和格式
- 用户概念：用户 ID、用户名、显示名称、名字、姓氏、电子邮件地址、电话号码
- 频道概念：频道 ID、频道名称、频道主题
- 消息概念：消息 ID、消息字符串、时间戳、月、日、年
答案绝不能是复杂结构
- 不是值的列表
- 不是复杂对象
- 不是 ID 或字符串的列表
- 不是自然语言文本
- 除非该答案可以通过直接字符串比较被直接验证
- 并且可以被真实地复现
- 应当不太可能出现 LLM 以任何其他顺序或格式返回相同列表的情况

评估流程

步骤 1：文档检查

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

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

步骤 2：工具检查

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

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

步骤 3：建立理解

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

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

步骤 4：只读内容检查

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

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

第 5 步：任务生成

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

LLM 应当能够借助 MCP 服务器回答这些问题
遵循上述所有问题和答案准则

输出格式

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

<evaluation>
   <qa_pair>
      <question>Find the project created in Q2 2024 with the highest number of completed tasks. What is the project name?</question>
      <answer>Website Redesign</answer>
   </qa_pair>
   <qa_pair>
      <question>Search for issues labeled as "bug" that were closed in March 2024. Which user closed the most issues? Provide their username.</question>
      <answer>sarah_dev</answer>
   </qa_pair>
   <qa_pair>
      <question>Look for pull requests that modified files in the /api directory and were merged between January 1 and January 31, 2024. How many different contributors worked on these PRs?</question>
      <answer>7</answer>
   </qa_pair>
   <qa_pair>
      <question>Find the repository with the most stars that was created before 2023. What is the repository name?</question>
      <answer>data-pipeline</answer>
   </qa_pair>
</evaluation>

评估示例

好问题

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

<qa_pair>
   <question>Find the repository that was archived in Q3 2023 and had previously been the most forked project in the organization. What was the primary programming language used in that repository?</question>
   <answer>Python</answer>
</qa_pair>

这个问题很好，因为：

需要多次搜索才能找到已归档的仓库
需要确定哪个仓库在归档前拥有最多的 fork
需要查看仓库详情以确定语言
答案是一个简单、可验证的值
基于不会改变的历史（已关闭）数据

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

<qa_pair>
   <question>Locate the initiative focused on improving customer onboarding that was completed in late 2023. The project lead created a retrospective document after completion. What was the lead's role title at that time?</question>
   <answer>Product Manager</answer>
</qa_pair>

这个问题很好，因为：

不使用具体的项目名称（“专注于改进客户入职体验的倡议”）
需要从特定时间范围内找到已完成的项目
需要识别项目负责人及其角色
需要从回顾文档中理解上下文
答案易于人类阅读且稳定
基于已完成的工作（不会变化）

示例 3：需要多个步骤的复杂聚合（Issue Tracker MCP）

<qa_pair>
   <question>Among all bugs reported in January 2024 that were marked as critical priority, which assignee resolved the highest percentage of their assigned bugs within 48 hours? Provide the assignee's username.</question>
   <answer>alex_eng</answer>
</qa_pair>

这个问题很好，因为：

需要按日期、优先级和状态筛选缺陷
需要按负责人分组并计算解决率
需要理解时间戳以确定 48 小时窗口
测试分页（可能需要处理大量缺陷）
答案是单个用户名
基于特定时间段的历史数据

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

<qa_pair>
   <question>Find the account that upgraded from the Starter to Enterprise plan in Q4 2023 and had the highest annual contract value. What industry does this account operate in?</question>
   <answer>Healthcare</answer>
</qa_pair>

这个问题很好，因为：

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

差问题

示例 1：答案随时间变化

<qa_pair>
   <question>How many open issues are currently assigned to the engineering team?</question>
   <answer>47</answer>
</qa_pair>

这个问题不好，因为：

答案会随着 issue 的创建、关闭或重新分配而变化
不是基于稳定/固定的数据
依赖于动态的“当前状态”

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

<qa_pair>
   <question>Find the pull request with title "Add authentication feature" and tell me who created it.</question>
   <answer>developer123</answer>
</qa_pair>

这个问题不好，因为：

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

示例 3：答案格式模糊

<qa_pair>
   <question>List all the repositories that have Python as their primary language.</question>
   <answer>repo1, repo2, repo3, data-pipeline, ml-tools</answer>
</qa_pair>

这个问题不好，因为：

答案是一个列表，可能以任意顺序返回
难以通过直接字符串比较来验证
LLM 的格式可能不同（JSON 数组、逗号分隔、换行分隔）
更好的做法是要求一个特定的聚合值（计数）或最高级（最多 stars）

验证流程

创建评估之后：

检查 XML 文件以理解其 schema
加载每个任务指令，并使用 MCP 服务器和工具并行地通过尝试自己解决任务来识别正确答案
标记任何需要 WRITE 或 DESTRUCTIVE 操作的操作
汇总所有 CORRECT 答案，并替换文档中任何不正确的答案
移除任何需要 WRITE 或 DESTRUCTIVE 操作的 <qa_pair>

记住要并行化解决任务，以避免上下文耗尽，然后在最后汇总所有答案并对文件进行修改。

创建高质量评估的技巧

深入思考并提前规划，然后再生成任务
在有机会时并行化，以加快流程并管理上下文
专注于现实的使用场景，即人类实际想要完成的事情
创建具有挑战性的问题，以测试 MCP 服务器能力的极限
确保稳定性，通过使用历史数据和封闭概念
验证答案，通过使用 MCP 服务器工具自己解决这些问题
迭代并改进，基于你在过程中学到的东西

运行评估

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

设置

安装依赖

pip install -r scripts/requirements.txt

或手动安装：

pip install anthropic mcp

设置 API 密钥

export ANTHROPIC_API_KEY=your_api_key_here

评估文件格式

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

<evaluation>
   <qa_pair>
      <question>Find the project created in Q2 2024 with the highest number of completed tasks. What is the project name?</question>
      <answer>Website Redesign</answer>
   </qa_pair>
   <qa_pair>
      <question>Search for issues labeled as "bug" that were closed in March 2024. Which user closed the most issues? Provide their username.</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             Path to evaluation XML file

optional arguments:
  -h, --help            Show help message
  -t, --transport       Transport type: stdio, sse, or http (default: stdio)
  -m, --model           Claude model to use (default: claude-3-7-sonnet-20250219)
  -o, --output          Output file for report (default: print to stdout)

stdio options:
  -c, --command         Command to run MCP server (e.g., python, node)
  -a, --args            Arguments for the command (e.g., server.py)
  -e, --env             Environment variables in KEY=VALUE format

sse/http options:
  -u, --url             MCP server URL
  -H, --header          HTTP headers in 'Key: Value' format

输出

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

汇总统计：
- 准确率（正确/总数）
- 平均任务时长
- 每个任务的平均工具调用次数
- 工具调用总次数
逐任务结果：
- 提示词与预期响应
- 智能体的实际响应
- 答案是否正确（✅/❌）
- 时长与工具调用详情
- 智能体对其方法的总结
- 智能体对工具的反馈

将报告保存到文件

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

完整示例工作流

以下是一个创建并运行评估的完整示例：

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

<evaluation>
   <qa_pair>
      <question>Find the user who created the most issues in January 2024. What is their username?</question>
      <answer>alice_developer</answer>
   </qa_pair>
   <qa_pair>
      <question>Among all pull requests merged in Q1 2024, which repository had the highest number? Provide the repository name.</question>
      <answer>backend-api</answer>
   </qa_pair>
   <qa_pair>
      <question>Find the project that was completed in December 2023 and had the longest duration from start to finish. How many days did it take?</question>
      <answer>127</answer>
   </qa_pair>
</evaluation>

安装依赖：

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

运行评估：

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

查看报告，在 github_eval_report.md 中：
- 查看哪些问题通过/失败
- 阅读 agent 对你工具的反馈
- 找出需要改进的地方
- 迭代你的 MCP 服务器设计

故障排查

连接错误

如果你遇到连接错误：

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

准确率低

如果许多评估失败：

查看 agent 对每个任务的反馈
检查工具描述是否清晰且全面
验证输入参数是否有完善的文档
考虑工具返回的数据是否过多或过少
确保错误信息具有可操作性

超时问题

如果任务超时：

使用能力更强的模型（例如 claude-3-7-sonnet-20250219）
检查工具是否返回了过多数据
验证分页是否正常工作
考虑简化复杂问题 FILE:reference/node_mcp_server.md

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: "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 // Modern pattern for structured data
    };
  }
);

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          # Main entry point with McpServer initialization
│   ├── types.ts          # TypeScript type definitions and interfaces
│   ├── tools/            # Tool implementations (one file per domain)
│   ├── services/         # API clients and shared utilities
│   ├── schemas/          # Zod validation schemas
│   └── constants.ts      # Shared constants (API_URL, CHARACTER_LIMIT, etc.)
└── dist/                 # Built JavaScript files (entry point: 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）
显式标注所有参数和返回值的类型

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 for input validation
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "Query must be at least 2 characters")
    .max(200, "Query must not exceed 200 characters")
    .describe("Search string to match against names/emails"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip for pagination"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();

// Type definition from Zod schema
type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

server.registerTool(
  "example_search_users",
  {
    title: "Search Example Users",
    description: `Search for users in the Example system by name, email, or team.

This tool searches across all user profiles in the Example platform, supporting partial matches and various search filters. It does NOT create or modify users, only searches existing ones.

Args:
  - query (string): Search string to match against names/emails
  - limit (number): Maximum results to return, between 1-100 (default: 20)
  - offset (number): Number of results to skip for pagination (default: 0)
  - response_format ('markdown' | 'json'): Output format (default: 'markdown')

Returns:
  For JSON format: Structured data with schema:
  {
    "total": number,           // Total number of matches found
    "count": number,           // Number of results in this response
    "offset": number,          // Current pagination offset
    "users": [
      {
        "id": string,          // User ID (e.g., "U123456789")
        "name": string,        // Full name (e.g., "John Doe")
        "email": string,       // Email address
        "team": string,        // Team name (optional)
        "active": boolean      // Whether user is active
      }
    ],
    "has_more": boolean,       // Whether more results are available
    "next_offset": number      // Offset for next page (if has_more is true)
  }

Examples:
  - Use when: "Find all marketing team members" -> params with query="team:marketing"
  - Use when: "Search for John's account" -> params with query="john"
  - Don't use when: You need to create a user (use example_create_user instead)

Error Handling:
  - Returns "Error: Rate limit exceeded" if too many requests (429 status)
  - Returns "No users found matching '<query>'" if search returns empty`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    try {
      // Input validation is handled by Zod schema
      // Make API request using validated parameters
      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: `No users found matching '${params.query}'`
          }]
        };
      }

      // Prepare structured output
      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
        } : {})
      };

      // Format text representation based on requested format
      let textContent: string;
      if (params.response_format === ResponseFormat.MARKDOWN) {
        const lines = [`# User Search Results: '${params.query}'`, "",
          `Found ${total} users (showing ${users.length})`, ""];
        for (const user of users) {
          lines.push(`## ${user.name} (${user.id})`);
          lines.push(`- **Email**: ${user.email}`);
          if (user.team) lines.push(`- **Team**: ${user.team}`);
          lines.push("");
        }
        textContent = lines.join("\n");
      } else {
        textContent = JSON.stringify(output, null, 2);
      }

      return {
        content: [{ type: "text", text: textContent }],
        structuredContent: output // Modern pattern for structured data
      };
    } catch (error) {
      return {
        content: [{
          type: "text",
          text: handleApiError(error)
        }]
      };
    }
  }
);

用于输入校验的 Zod Schema

Zod 提供运行时类型校验：

import { z } from "zod";

// Basic schema with validation
const CreateUserSchema = z.object({
  name: z.string()
    .min(1, "Name is required")
    .max(100, "Name must not exceed 100 characters"),
  email: z.string()
    .email("Invalid email format"),
  age: z.number()
    .int("Age must be a whole number")
    .min(0, "Age cannot be negative")
    .max(150, "Age cannot be greater than 150")
}).strict();  // Use .strict() to forbid extra fields

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

const SearchSchema = z.object({
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format")
});

// Optional fields with defaults
const PaginationSchema = z.object({
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip")
});

响应格式选项

支持多种输出格式以提供灵活性：

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

const inputSchema = z.object({
  query: z.string(),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
});

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 常量，以防止响应内容过多：

// At module level in constants.ts
export const CHARACTER_LIMIT = 25000;  // Maximum response size in characters

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

  // Check character limit and truncate if needed
  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 =
      `Response truncated from ${data.length} to ${truncatedData.length} items. ` +
      `Use 'offset' parameter or add filters to see more results.`;
    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 "Error: Resource not found. Please check the ID is correct.";
        case 403:
          return "Error: Permission denied. You don't have access to this resource.";
        case 429:
          return "Error: Rate limit exceeded. Please wait before making more requests.";
        default:
          return `Error: API request failed with status ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "Error: Request timed out. Please try again.";
    }
  }
  return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}

共享工具

将通用功能提取为可复用的函数：

// Shared API request function
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;
  }
}

Async/Await 最佳实践

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

// Good: Async network request
async function fetchData(resourceId: string): Promise<ResourceData> {
  const response = await axios.get(`${API_URL}/resource/${resourceId}`);
  return response.data;
}

// Bad: Promise chains
function fetchData(resourceId: string): Promise<ResourceData> {
  return axios.get(`${API_URL}/resource/${resourceId}`)
    .then(response => response.data);  // Harder to read and maintain
}

TypeScript 最佳实践

使用严格模式 TypeScript：在 tsconfig.json 中启用严格模式
定义接口：为所有数据结构创建清晰的接口定义
避免 any：使用合适的类型或 unknown 代替 any
用 Zod 做运行时校验：使用 Zod schema 校验外部数据
类型守卫：为复杂类型检查创建类型守卫函数
错误处理：始终使用 try-catch 并配合正确的错误类型检查
空值安全：使用可选链（?.）和空值合并（??）

// Good: Type-safe with Zod and interfaces
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);  // Runtime validation
}

// Bad: Using any
async function getUser(id: string): Promise<any> {
  return await apiCall(`/users/${id}`);  // No type safety
}

包配置

package.json

{
  "name": "{service}-mcp-server",
  "version": "1.0.0",
  "description": "MCP server for {Service} API integration",
  "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 Server for Example Service.
 *
 * This server provides tools to interact with Example API, including user search,
 * project management, and data export capabilities.
 */

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";

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

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

// Zod schemas
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "Query must be at least 2 characters")
    .max(200, "Query must not exceed 200 characters")
    .describe("Search string to match against names/emails"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip for pagination"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();

type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

// Shared utility functions
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 "Error: Resource not found. Please check the ID is correct.";
        case 403:
          return "Error: Permission denied. You don't have access to this resource.";
        case 429:
          return "Error: Rate limit exceeded. Please wait before making more requests.";
        default:
          return `Error: API request failed with status ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "Error: Request timed out. Please try again.";
    }
  }
  return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}

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

// Register tools
server.registerTool(
  "example_search_users",
  {
    title: "Search Example Users",
    description: `[Full description as shown above]`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    // Implementation as shown above
  }
);

// Main function
// For stdio (local):
async function runStdio() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
    process.exit(1);
  }

  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP server running via stdio");
}

// For streamable HTTP (remote):
async function runHTTP() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
    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 server running on http://localhost:${port}/mcp`);
  });
}

// Choose transport based on environment
const transport = process.env.TRANSPORT || 'stdio';
if (transport === 'http') {
  runHTTP().catch(error => {
    console.error("Server error:", error);
    process.exit(1);
  });
} else {
  runStdio().catch(error => {
    console.error("Server error:", error);
    process.exit(1);
  });
}

高级 MCP 功能

资源注册

将数据暴露为资源，以实现高效的、基于 URI 的访问：

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

// Register a resource with URI template
server.registerResource(
  {
    uri: "file://documents/{name}",
    name: "Document Resource",
    description: "Access documents by name",
    mimeType: "text/plain"
  },
  async (uri: string) => {
    // Extract parameter from URI
    const match = uri.match(/^file:\/\/documents\/(.+)$/);
    if (!match) {
      throw new Error("Invalid URI format");
    }

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

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

// List available resources dynamically
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
    }))
  };
});

何时使用资源（Resources）与工具（Tools）：

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

传输选项

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) => {
  // Create new transport for each request (stateless, prevents request ID collisions)
  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);

传输方式选择：

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

通知支持

当服务器状态发生变化时通知客户端：

// Notify when tools list changes
server.notification({
  method: "notifications/tools/list_changed"
});

// Notify when resources change
server.notification({
  method: "notifications/resources/list_changed"
});

谨慎使用通知——仅当服务器能力真正发生变化时。

代码最佳实践

代码可组合性与可复用性

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

提取通用功能：
- 为多个工具中使用的操作创建可复用的辅助函数
- 为 HTTP 请求构建共享的 API 客户端，而不是重复代码
- 将错误处理逻辑集中到工具函数中
- 将业务逻辑提取到可组合的专用函数中
- 提取共享的 markdown 或 JSON 字段选择与格式化功能
避免重复：
- 绝不在工具之间复制粘贴相似的代码
- 如果你发现自己两次编写相似的逻辑，就把它提取成一个函数
- 分页、过滤、字段选择和格式化等常见操作应当共享
- 认证/授权逻辑应当集中管理

构建与运行

运行前务必先构建你的 TypeScript 代码：

# Build the project
npm run build

# Run the server
npm start

# Development with auto-reload
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
所有导入均能正确解析
示例工具调用按预期工作 FILE:reference/python_mcp_server.md

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:
    # Implementation
    pass

MCP Python SDK 与 FastMCP

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

根据函数签名和文档字符串自动生成描述和 inputSchema
集成 Pydantic 模型进行输入验证
通过 @mcp.tool 实现基于装饰器的工具注册

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

服务器命名规范

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

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

名称应当：

通用（不绑定特定功能）
能描述所集成的服务/API
易于从任务描述中推断
不含版本号或日期

工具实现

工具命名

工具名称使用 snake_case（例如 "search_users"、"create_project"、"get_channel_info"），名称应清晰且以动作导向。

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

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

使用 FastMCP 的工具结构

工具通过 @mcp.tool 装饰器定义，并使用 Pydantic 模型进行输入验证：

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

# Initialize the MCP server
mcp = FastMCP("example_mcp")

# Define Pydantic model for input validation
class ServiceToolInput(BaseModel):
    '''Input model for service tool operation.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,  # Auto-strip whitespace from strings
        validate_assignment=True,    # Validate on assignment
        extra='forbid'              # Forbid extra fields
    )

    param1: str = Field(..., description="First parameter description (e.g., 'user123', 'project-abc')", min_length=1, max_length=100)
    param2: Optional[int] = Field(default=None, description="Optional integer parameter with constraints", ge=0, le=1000)
    tags: Optional[List[str]] = Field(default_factory=list, description="List of tags to apply", max_items=10)

@mcp.tool(
    name="service_tool_name",
    annotations={
        "title": "Human-Readable Tool Title",
        "readOnlyHint": True,     # Tool does not modify environment
        "destructiveHint": False,  # Tool does not perform destructive operations
        "idempotentHint": True,    # Repeated calls have no additional effect
        "openWorldHint": False     # Tool does not interact with external entities
    }
)
async def service_tool_name(params: ServiceToolInput) -> str:
    '''Tool description automatically becomes the 'description' field.

    This tool performs a specific operation on the service. It validates all inputs
    using the ServiceToolInput Pydantic model before processing.

    Args:
        params (ServiceToolInput): Validated input parameters containing:
            - param1 (str): First parameter description
            - param2 (Optional[int]): Optional parameter with default
            - tags (Optional[List[str]]): List of tags

    Returns:
        str: JSON-formatted response containing operation results
    '''
    # Implementation here
    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="User's full name", min_length=1, max_length=100)
    email: str = Field(..., description="User's email address", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., description="User's age", ge=0, le=150)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Email cannot be empty")
        return v.lower()

响应格式选项

支持多种输出格式以提升灵活性：

from enum import Enum

class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    query: str = Field(..., description="Search query")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
    )

Markdown 格式：

使用标题、列表和格式以保持清晰
将时间戳转换为人类可读格式（例如，使用 "2024-01-15 10:30:00 UTC" 而不是 epoch）
显示名称时在括号中附上 ID（例如，"@john.doe (U123456)"）
省略冗长的元数据（例如，只显示一个头像图片 URL，而不是所有尺寸）
按逻辑对相关信息进行分组

JSON 格式：

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

分页实现

对于列出资源的工具：

class ListInput(BaseModel):
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)

async def list_items(params: ListInput) -> str:
    # Make API request with pagination
    data = await api_request(limit=params.limit, offset=params.offset)

    # Return pagination info
    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:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"

共享工具

将通用功能提取为可复用的函数：

# Shared API request function
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    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：

# Good: Async network request
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()

# Bad: Synchronous request
def fetch_data(resource_id: str) -> dict:
    response = requests.get(f"{API_URL}/resource/{resource_id}")  # Blocks
    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:
    '''
    Search for users in the Example system by name, email, or team.

    This tool searches across all user profiles in the Example platform,
    supporting partial matches and various search filters. It does NOT
    create or modify users, only searches existing ones.

    Args:
        params (UserSearchInput): Validated input parameters containing:
            - query (str): Search string to match against names/emails (e.g., "john", "@example.com", "team:marketing")
            - limit (Optional[int]): Maximum results to return, between 1-100 (default: 20)
            - offset (Optional[int]): Number of results to skip for pagination (default: 0)

    Returns:
        str: JSON-formatted string containing search results with the following schema:

        Success response:
        {
            "total": int,           # Total number of matches found
            "count": int,           # Number of results in this response
            "offset": int,          # Current pagination offset
            "users": [
                {
                    "id": str,      # User ID (e.g., "U123456789")
                    "name": str,    # Full name (e.g., "John Doe")
                    "email": str,   # Email address (e.g., "john@example.com")
                    "team": str     # Team name (e.g., "Marketing") - optional
                }
            ]
        }

        Error response:
        "Error: <error message>" or "No users found matching '<query>'"

    Examples:
        - Use when: "Find all marketing team members" -> params with query="team:marketing"
        - Use when: "Search for John's account" -> params with query="john"
        - Don't use when: You need to create a user (use example_create_user instead)
        - Don't use when: You have a user ID and need full details (use example_get_user instead)

    Error Handling:
        - Input validation errors are handled by Pydantic model
        - Returns "Error: Rate limit exceeded" if too many requests (429 status)
        - Returns "Error: Invalid API authentication" if API key is invalid (401 status)
        - Returns formatted list of results or "No users found matching 'query'"
    '''

完整示例

完整的 Python MCP 服务器示例见下文：

#!/usr/bin/env python3
'''
MCP Server for Example Service.

This server provides tools to interact with Example API, including user search,
project management, and data export capabilities.
'''

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

# Initialize the MCP server
mcp = FastMCP("example_mcp")

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

# Enums
class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

# Pydantic Models for Input Validation
class UserSearchInput(BaseModel):
    '''Input model for user search operations.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    query: str = Field(..., description="Search string to match against names/emails", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Query cannot be empty or whitespace only")
        return v.strip()

# Shared utility functions
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    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:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"

# Tool definitions
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Example Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''Search for users in the Example system by name, email, or team.

    [Full docstring as shown above]
    '''
    try:
        # Make API request using validated parameters
        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"No users found matching '{params.query}'"

        # Format response based on requested format
        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [f"# User Search Results: '{params.query}'", ""]
            lines.append(f"Found {total} users (showing {len(users)})")
            lines.append("")

            for user in users:
                lines.append(f"## {user['name']} ({user['id']})")
                lines.append(f"- **Email**: {user['email']}")
                if user.get('team'):
                    lines.append(f"- **Team**: {user['team']}")
                lines.append("")

            return "\n".join(lines)

        else:
            # Machine-readable JSON format
            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:
    '''Advanced tool with context access for logging and progress.'''

    # Report progress for long operations
    await ctx.report_progress(0.25, "Starting search...")

    # Log information for debugging
    await ctx.log_info("Processing query", {"query": query, "timestamp": datetime.now()})

    # Perform search
    results = await search_api(query)
    await ctx.report_progress(0.75, "Formatting results...")

    # Access server configuration
    server_name = ctx.fastmcp.name

    return format_results(results)

@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
    '''Tool that can request additional input from users.'''

    # Request sensitive information when needed
    api_key = await ctx.elicit(
        prompt="Please provide your API key:",
        input_type="password"
    )

    # Use the provided key
    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:
    '''Expose documents as MCP resources.

    Resources are useful for static or semi-static data that doesn't
    require complex parameters. They use URI templates for flexible access.
    '''
    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:
    '''Expose configuration as resources with context.'''
    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 for structured returns
class UserData(TypedDict):
    id: str
    name: str
    email: str

@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
    '''Returns structured data - FastMCP handles serialization.'''
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# Pydantic models for complex validation
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:
    '''Returns Pydantic model - automatically generates schema.'''
    user = await fetch_user(user_id)
    return DetailedUser(**user)

生命周期管理

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

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan():
    '''Manage resources that live for the server's lifetime.'''
    # Initialize connections, load config, etc.
    db = await connect_to_database()
    config = load_configuration()

    # Make available to all tools
    yield {"db": db, "config": config}

    # Cleanup on shutdown
    await db.close()

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

@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
    '''Access lifespan resources through context.'''
    db = ctx.request_context.lifespan_state["db"]
    results = await db.query(query)
    return format_results(results)

传输选项

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

# stdio transport (for local tools) - default
if __name__ == "__main__":
    mcp.run()

# Streamable HTTP transport (for remote servers)
if __name__ == "__main__":
    mcp.run(transport="streamable_http", port=8000)

传输方式选择：

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

代码最佳实践

代码可组合性与可复用性

你的实现必须优先考虑可组合性与代码复用：

提取通用功能：
- 为多个工具中使用的操作创建可复用的辅助函数
- 为 HTTP 请求构建共享的 API 客户端，而不是重复代码
- 将错误处理逻辑集中到工具函数中
- 将业务逻辑提取到可组合的专用函数中
- 提取共享的 markdown 或 JSON 字段选择与格式化功能
避免重复：
- 绝不在工具之间复制粘贴相似的代码
- 如果你发现自己两次编写相似的逻辑，就把它提取成一个函数
- 分页、过滤、字段选择和格式化等常见操作应当共享
- 认证/授权逻辑应当集中管理

Python 专属最佳实践

使用类型提示：始终为函数参数和返回值添加类型注解
Pydantic 模型：为所有输入校验定义清晰的 Pydantic 模型
避免手动校验：让 Pydantic 通过约束来处理输入校验
规范的导入：对导入进行分组（标准库、第三方、本地）
错误处理：使用具体的异常类型（httpx.HTTPStatusError，而不是泛化的 Exception）
异步上下文管理器：对需要清理的资源使用 async with
常量：在模块级别用 UPPER_CASE 定义常量

质量检查清单

在最终确定你的 Python MCP 服务器实现之前，确保：

战略设计

工具能够支撑完整的工作流，而不仅仅是 API 端点的封装
工具名称反映自然的任务划分
响应格式针对智能体上下文效率进行优化
在适当的地方使用人类可读的标识符
错误消息引导智能体走向正确的用法

实现质量

聚焦实现：实现了最重要、最有价值的工具
所有工具都有描述性名称和文档
相似操作的返回类型保持一致
所有外部调用都实现了错误处理
服务器名称遵循格式：{service}_mcp
所有网络操作都使用 async/await
通用功能已提取为可复用函数
错误消息清晰、可操作且具有指导性
输出经过正确验证和格式化

工具配置

所有工具都在装饰器中实现了 'name' 和 'annotations'
注解设置正确（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
所有工具都使用 Pydantic BaseModel 进行输入验证，并带有 Field() 定义
所有 Pydantic Field 都有显式类型和带约束的描述
所有工具都有全面的 docstring，并明确标注输入/输出类型
Docstring 包含 dict/JSON 返回值的完整 schema 结构
Pydantic 模型处理输入验证（无需手动验证）

高级功能（适用时）

使用上下文注入进行日志记录、进度报告或引导
为适当的数据端点注册了 Resources
为持久连接实现了 Lifespan 管理
使用了结构化输出类型（TypedDict、Pydantic 模型）
配置了适当的传输方式（stdio 或可流式 HTTP）

代码质量

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

测试

服务器成功运行：python your_server.py --help
所有导入正确解析
示例工具调用按预期工作
错误场景得到优雅处理 FILE:scripts/connections.py """MCP 服务器的轻量级连接处理。"""

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): """MCP 服务器连接的基类。"""

def __init__(self):
    self.session = None
    self._stack = None

@abstractmethod
def _create_context(self):
    """根据连接类型创建连接上下文。"""

async def __aenter__(self):
    """初始化 MCP 服务器连接。"""
    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):
    """清理 MCP 服务器连接资源。"""
    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]]: """从 MCP 服务器检索可用工具。""" 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:
    """使用提供的参数调用 MCP 服务器上的工具。"""
    result = await self.session.call_tool(tool_name, arguments=arguments)
    return result.content

class MCPConnectionStdio(MCPConnection): """使用标准输入/输出的 MCP 连接。"""

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 连接。"""

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): """使用可流式 HTTP 的 MCP 连接。"""

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: """用于创建相应 MCP 连接的工厂函数。

参数：
    transport: 连接类型（"stdio"、"sse" 或 "http"）
    command: 要运行的命令（仅 stdio）
    args: 命令参数（仅 stdio）
    env: 环境变量（仅 stdio）
    url: 服务器 URL（仅 sse 和 http）
    headers: HTTP 请求头（仅 sse 和 http）

返回：
    MCPConnection 实例
"""
transport = transport.lower()

if transport == "stdio":
    if not command:
        raise ValueError("stdio 传输方式需要提供命令")
    return MCPConnectionStdio(command=command, args=args, env=env)

elif transport == "sse":
    if not url:
        raise ValueError("sse 传输方式需要提供 URL")
    return MCPConnectionSSE(url=url, headers=headers)

elif transport in ["http", "streamable_http", "streamable-http"]:
    if not url:
        raise ValueError("http 传输方式需要提供 URL")
    return MCPConnectionHTTP(url=url, headers=headers)

else:
    raise ValueError(f"不支持的传输类型：{transport}。请使用 'stdio'、'sse' 或 'http'")

FILE:scripts/evaluation.py """MCP 服务器评估测试框架

此脚本通过使用 Claude 对 MCP 服务器运行测试问题来对其进行评估。 """

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 = """你是一个可以使用工具的 AI 助手。当被给定一项任务时，你必须：

使用可用的工具来完成任务
在
标签中提供你方法中每个步骤的摘要
在标签中提供对所提供工具的反馈
在标签中提供你的最终回复

摘要要求：

在你的
标签中，你必须说明：
- 你为完成任务所采取的步骤
- 你使用了哪些工具、以什么顺序使用以及为什么
- 你提供给每个工具的输入
- 你从每个工具收到的输出
- 关于你如何得出该回复的摘要

反馈要求：

在你的标签中，提供关于这些工具的建设性反馈：
- 评论工具名称：它们是否清晰且具有描述性？
- 评论输入参数：它们是否有良好的文档说明？必需参数与可选参数是否清晰？
- 评论描述：它们是否准确描述了工具的功能？
- 评论工具使用过程中遇到的任何错误：工具是否执行失败？工具是否返回了过多的 token？
- 指出具体的改进领域，并解释为什么它们会有帮助
- 在你的建议中要具体且可操作

回复要求：

你的回复应简洁，并直接针对所提出的问题
始终将你的最终回复包裹在标签中
如果你无法解决该任务，返回 NOT_FOUND
对于数字回复，只提供数字
对于 ID，只提供 ID
对于名称或文本，提供所要求的确切文本
你的回复应放在最后"""

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: """从 XML 标签中提取内容。""" pattern = rf"<{tag}>(.*?)</{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]]: """使用 MCP 工具运行智能体循环。""" 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]: """使用给定的工具评估单个问答对。""" 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 = """

评估报告

摘要

准确率：{correct}/{total}（{accuracy:.1f}%）
平均任务耗时：{average_duration_s:.2f}s
每任务平均工具调用次数：{average_tool_calls:.2f}
工具调用总次数：{total_tool_calls}

"""

TASK_TEMPLATE = """

任务 {task_num}

问题：{question} 标准答案：{expected_answer} 实际答案：{actual_answer} 是否正确：{correct_indicator} 耗时：{total_duration:.2f}s 工具调用：{tool_calls}

摘要 {summary}

反馈 {feedback}

"""

async def run_evaluation( eval_path: Path, connection: Any, model: str = "claude-3-7-sonnet-20250219", ) -> str: """使用 MCP 服务器工具运行评估。""" print("🚀 开始评估")

client = Anthropic()

tools = await connection.list_tools()
print(f"📋 已从 MCP 服务器加载 {len(tools)} 个工具")

qa_pairs = parse_evaluation_file(eval_path)
print(f"📋 已加载 {len(qa_pairs)} 个评估任务")

results = []
for i, qa_pair in enumerate(qa_pairs):
    print(f"正在处理任务 {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]: """将格式为 'Key: Value' 的请求头字符串解析为字典。""" 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"警告：忽略格式错误的请求头：{header}")
return headers

def parse_env_vars(env_list: list[str]) -> dict[str, str]: """将格式为 'KEY=VALUE' 的环境变量字符串解析为字典。""" 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"警告：忽略格式错误的环境变量：{env_var}")
return env

async def main(): parser = argparse.ArgumentParser( description="使用测试问题评估 MCP 服务器", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例：

评估本地 stdio MCP 服务器

python evaluation.py -t stdio -c python -a my_server.py eval.xml

评估一个 SSE MCP 服务器

python evaluation.py -t sse -u https://example.com/mcp -H "Authorization: Bearer token" eval.xml

使用自定义模型评估一个 HTTP MCP 服务器

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="评估 XML 文件的路径")
parser.add_argument("-t", "--transport", choices=["stdio", "sse", "http"], default="stdio", help="传输类型（默认：stdio）")
parser.add_argument("-m", "--model", default="claude-3-7-sonnet-20250219", help="要使用的 Claude 模型（默认：claude-3-7-sonnet-20250219）")

stdio_group = parser.add_argument_group("stdio 选项")
stdio_group.add_argument("-c", "--command", help="运行 MCP 服务器的命令（仅限 stdio）")
stdio_group.add_argument("-a", "--args", nargs="+", help="命令的参数（仅限 stdio）")
stdio_group.add_argument("-e", "--env", nargs="+", help="KEY=VALUE 格式的环境变量（仅限 stdio）")

remote_group = parser.add_argument_group("sse/http 选项")
remote_group.add_argument("-u", "--url", help="MCP 服务器 URL（仅限 sse/http）")
remote_group.add_argument("-H", "--header", nargs="+", dest="headers", help="'Key: Value' 格式的 HTTP 头（仅限 sse/http）")

parser.add_argument("-o", "--output", type=Path, help="评估报告的输出文件（默认：stdout）")

args = parser.parse_args()

if not args.eval_file.exists():
    print(f"错误：未找到评估文件：{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"错误：{e}")
    sys.exit(1)

print(f"🔗 正在通过 {args.transport} 连接到 MCP 服务器...")

async with connection:
    print("✅ 连接成功")
    report = await run_evaluation(args.eval_file, connection, args.model)

    if args.output:
        args.output.write_text(report)
        print(f"\n✅ 报告已保存至 {args.output}")
    else:
        print("\n" + report)

if name == "main": asyncio.run(main()) FILE:scripts/example_evaluation.xml 计算以 5% 年利率投资 10,000 美元、按月复利、为期 3 年的复利。最终金额是多少美元（四舍五入到小数点后 2 位）？ 11614.72 一枚抛射物以 45 度角发射，初速度为 50 m/s。计算它在 2 秒后从发射点起已行进的总距离（以米为单位），假设 g=9.8 m/s²。四舍五入到小数点后 2 位。 87.25 一个球体的体积为 500 立方米。计算其表面积，以平方米为单位。四舍五入到小数点后 2 位。 304.65 计算以下数据集的总体标准差：[12, 15, 18, 22, 25, 30, 35]。四舍五入到小数点后 2 位。 7.61 计算氢离子浓度为 3.5 × 10^-5 M 的溶液的 pH 值。四舍五入到小数点后 2 位。 4.46 FILE:scripts/requirements.txt anthropic>=0.39.0 mcp>=1.1.0

English original

MCP Server Development Guide

Overview

Create MCP (Model Context Protocol) servers that enable LLMs to interact with external services through well-designed tools. The quality of an MCP server is measured by how well it enables LLMs to accomplish real-world tasks.

Process

🚀 High-Level Workflow

Creating a high-quality MCP server involves four main phases:

Phase 1: Deep Research and Planning

1.1 Understand Modern MCP Design

API Coverage vs. Workflow Tools: Balance comprehensive API endpoint coverage with specialized workflow tools. Workflow tools can be more convenient for specific tasks, while comprehensive coverage gives agents flexibility to compose operations. Performance varies by client—some clients benefit from code execution that combines basic tools, while others work better with higher-level workflows. When uncertain, prioritize comprehensive API coverage.

Tool Naming and Discoverability: Clear, descriptive tool names help agents find the right tools quickly. Use consistent prefixes (e.g., github_create_issue, github_list_repos) and action-oriented naming.

Context Management: Agents benefit from concise tool descriptions and the ability to filter/paginate results. Design tools that return focused, relevant data. Some clients support code execution which can help agents filter and process data efficiently.

Actionable Error Messages: Error messages should guide agents toward solutions with specific suggestions and next steps.

1.2 Study MCP Protocol Documentation

Navigate the MCP specification:

Start with the sitemap to find relevant pages: https://modelcontextprotocol.io/sitemap.xml

Then fetch specific pages with .md suffix for markdown format (e.g., https://modelcontextprotocol.io/specification/draft.md).

Key pages to review:

Specification overview and architecture
Transport mechanisms (streamable HTTP, stdio)
Tool, resource, and prompt definitions

1.3 Study Framework Documentation

Recommended stack:

Language: TypeScript (high-quality SDK support and good compatibility in many execution environments e.g. MCPB. Plus AI models are good at generating TypeScript code, benefiting from its broad usage, static typing and good linting tools)
Transport: Streamable HTTP for remote servers, using stateless JSON (simpler to scale and maintain, as opposed to stateful sessions and streaming responses). stdio for local servers.

Load framework documentation:

MCP Best Practices: 📋 View Best Practices - Core guidelines

For TypeScript (recommended):

TypeScript SDK: Use WebFetch to load https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md
⚡ TypeScript Guide - TypeScript patterns and examples

For Python:

Python SDK: Use WebFetch to load https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md
🐍 Python Guide - Python patterns and examples

1.4 Plan Your Implementation

Understand the API: Review the service's API documentation to identify key endpoints, authentication requirements, and data models. Use web search and WebFetch as needed.

Tool Selection: Prioritize comprehensive API coverage. List endpoints to implement, starting with the most common operations.

Phase 2: Implementation

2.1 Set Up Project Structure

See language-specific guides for project setup:

⚡ TypeScript Guide - Project structure, package.json, tsconfig.json
🐍 Python Guide - Module organization, dependencies

2.2 Implement Core Infrastructure

Create shared utilities:

API client with authentication
Error handling helpers
Response formatting (JSON/Markdown)
Pagination support

2.3 Implement Tools

For each tool:

Input Schema:

Use Zod (TypeScript) or Pydantic (Python)
Include constraints and clear descriptions
Add examples in field descriptions

Output Schema:

Define outputSchema where possible for structured data
Use structuredContent in tool responses (TypeScript SDK feature)
Helps clients understand and process tool outputs

Tool Description:

Concise summary of functionality
Parameter descriptions
Return type schema

Implementation:

Async/await for I/O operations
Proper error handling with actionable messages
Support pagination where applicable
Return both text content and structured data when using modern SDKs

Annotations:

readOnlyHint: true/false
destructiveHint: true/false
idempotentHint: true/false
openWorldHint: true/false

Phase 3: Review and Test

3.1 Code Quality

Review for:

No duplicated code (DRY principle)
Consistent error handling
Full type coverage
Clear tool descriptions

3.2 Build and Test

TypeScript:

Run npm run build to verify compilation
Test with MCP Inspector: npx @modelcontextprotocol/inspector

Python:

Verify syntax: python -m py_compile your_server.py
Test with MCP Inspector

See language-specific guides for detailed testing approaches and quality checklists.

Phase 4: Create Evaluations

After implementing your MCP server, create comprehensive evaluations to test its effectiveness.

Load ✅ Evaluation Guide for complete evaluation guidelines.

4.1 Understand Evaluation Purpose

Use evaluations to test whether LLMs can effectively use your MCP server to answer realistic, complex questions.

4.2 Create 10 Evaluation Questions

To create effective evaluations, follow the process outlined in the evaluation guide:

Tool Inspection: List available tools and understand their capabilities
Content Exploration: Use READ-ONLY operations to explore available data
Question Generation: Create 10 complex, realistic questions
Answer Verification: Solve each question yourself to verify answers

4.3 Evaluation Requirements

Ensure each question is:

Independent: Not dependent on other questions
Read-only: Only non-destructive operations required
Complex: Requiring multiple tool calls and deep exploration
Realistic: Based on real use cases humans would care about
Verifiable: Single, clear answer that can be verified by string comparison
Stable: Answer won't change over time

4.4 Output Format

Create an XML file with this structure:

<evaluation>
  <qa_pair>
    <question>Find discussions about AI model launches with animal codenames. One model needed a specific safety designation that uses the format ASL-X. What number X was being determined for the model named after a spotted wild cat?</question>
    <answer>3</answer>
  </qa_pair>
<!-- More qa_pairs... -->
</evaluation>

Reference Files

📚 Documentation Library

Load these resources as needed during development:

Core MCP Documentation (Load First)

MCP Protocol: Start with sitemap at https://modelcontextprotocol.io/sitemap.xml, then fetch specific pages with .md suffix
📋 MCP Best Practices - Universal MCP guidelines including:
- Server and tool naming conventions
- Response format guidelines (JSON vs Markdown)
- Pagination best practices
- Transport selection (streamable HTTP vs stdio)
- Security and error handling standards

SDK Documentation (Load During Phase 1/2)

Python SDK: Fetch from https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md
TypeScript SDK: Fetch from https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md

Language-Specific Implementation Guides (Load During Phase 2)

🐍 Python Implementation Guide - Complete Python/FastMCP guide with:
- Server initialization patterns
- Pydantic model examples
- Tool registration with @mcp.tool
- Complete working examples
- Quality checklist
⚡ TypeScript Implementation Guide - Complete TypeScript guide with:
- Project structure
- Zod schema patterns
- Tool registration with server.registerTool
- Complete working examples
- Quality checklist

Evaluation Guide (Load During Phase 4)

✅ Evaluation Guide - Complete evaluation creation guide with:
- Question creation guidelines
- Answer verification strategies
- XML format specifications
- Example questions and answers
- Running an evaluation with the provided scripts FILE:reference/mcp_best_practices.md

MCP Server Best Practices

Quick Reference

Server Naming

Python: {service}_mcp (e.g., slack_mcp)
Node/TypeScript: {service}-mcp-server (e.g., slack-mcp-server)

Tool Naming

Use snake_case with service prefix
Format: {service}_{action}_{resource}
Example: slack_send_message, github_create_issue

Response Formats

Support both JSON and Markdown formats
JSON for programmatic processing
Markdown for human readability

Pagination

Always respect limit parameter
Return has_more, next_offset, total_count
Default to 20-50 items

Transport

Streamable HTTP: For remote servers, multi-client scenarios
stdio: For local integrations, command-line tools
Avoid SSE (deprecated in favor of streamable HTTP)

Server Naming Conventions

Follow these standardized naming patterns:

Python: Use format {service}_mcp (lowercase with underscores)

Examples: slack_mcp, github_mcp, jira_mcp

Node/TypeScript: Use format {service}-mcp-server (lowercase with hyphens)

Examples: slack-mcp-server, github-mcp-server, jira-mcp-server

The name should be general, descriptive of the service being integrated, easy to infer from the task description, and without version numbers.

Tool Naming and Design

Tool Naming

Use snake_case: search_users, create_project, get_channel_info
Include service prefix: Anticipate that your MCP server may be used alongside other MCP servers
- Use slack_send_message instead of just send_message
- Use github_create_issue instead of just create_issue
Be action-oriented: Start with verbs (get, list, search, create, etc.)
Be specific: Avoid generic names that could conflict with other servers

Tool Design

Tool descriptions must narrowly and unambiguously describe functionality
Descriptions must precisely match actual functionality
Provide tool annotations (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
Keep tool operations focused and atomic

Response Formats

All tools that return data should support multiple formats:

JSON Format (`response_format="json"`)

Machine-readable structured data
Include all available fields and metadata
Consistent field names and types
Use for programmatic processing

Markdown Format (`response_format="markdown"`, typically default)

Human-readable formatted text
Use headers, lists, and formatting for clarity
Convert timestamps to human-readable format
Show display names with IDs in parentheses
Omit verbose metadata

Pagination

For tools that list resources:

Always respect the limit parameter
Implement pagination: Use offset or cursor-based pagination
Return pagination metadata: Include has_more, next_offset/next_cursor, total_count
Never load all results into memory: Especially important for large datasets
Default to reasonable limits: 20-50 items is typical

Example pagination response:

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

Transport Options

Streamable HTTP

Best for: Remote servers, web services, multi-client scenarios

Characteristics:

Bidirectional communication over HTTP
Supports multiple simultaneous clients
Can be deployed as a web service
Enables server-to-client notifications

Use when:

Serving multiple clients simultaneously
Deploying as a cloud service
Integration with web applications

stdio

Best for: Local integrations, command-line tools

Characteristics:

Standard input/output stream communication
Simple setup, no network configuration needed
Runs as a subprocess of the client

Use when:

Building tools for local development environments
Integrating with desktop applications
Single-user, single-session scenarios

Note: stdio servers should NOT log to stdout (use stderr for logging)

Transport Selection

Criterion	stdio	Streamable HTTP
Deployment	Local	Remote
Clients	Single	Multiple
Complexity	Low	Medium
Real-time	No	Yes

Security Best Practices

Authentication and Authorization

OAuth 2.1:

Use secure OAuth 2.1 with certificates from recognized authorities
Validate access tokens before processing requests
Only accept tokens specifically intended for your server

API Keys:

Store API keys in environment variables, never in code
Validate keys on server startup
Provide clear error messages when authentication fails

Input Validation

Sanitize file paths to prevent directory traversal
Validate URLs and external identifiers
Check parameter sizes and ranges
Prevent command injection in system calls
Use schema validation (Pydantic/Zod) for all inputs

Error Handling

Don't expose internal errors to clients
Log security-relevant errors server-side
Provide helpful but not revealing error messages
Clean up resources after errors

DNS Rebinding Protection

For streamable HTTP servers running locally:

Enable DNS rebinding protection
Validate the Origin header on all incoming connections
Bind to 127.0.0.1 rather than 0.0.0.0

Tool Annotations

Provide annotations to help clients understand tool behavior:

Annotation	Type	Default	Description
`readOnlyHint`	boolean	false	Tool does not modify its environment
`destructiveHint`	boolean	true	Tool may perform destructive updates
`idempotentHint`	boolean	false	Repeated calls with same args have no additional effect
`openWorldHint`	boolean	true	Tool interacts with external entities

Important: Annotations are hints, not security guarantees. Clients should not make security-critical decisions based solely on annotations.

Error Handling

Use standard JSON-RPC error codes
Report tool errors within result objects (not protocol-level errors)
Provide helpful, specific error messages with suggested next steps
Don't expose internal implementation details
Clean up resources properly on errors

Example error handling:

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.`
    }]
  };
}

Testing Requirements

Comprehensive testing should cover:

Functional testing: Verify correct execution with valid/invalid inputs
Integration testing: Test interaction with external systems
Security testing: Validate auth, input sanitization, rate limiting
Performance testing: Check behavior under load, timeouts
Error handling: Ensure proper error reporting and cleanup

Documentation Requirements

Provide clear documentation of all tools and capabilities
Include working examples (at least 3 per major feature)
Document security considerations
Specify required permissions and access levels
Document rate limits and performance characteristics FILE:reference/evaluation.md

MCP Server Evaluation Guide

Overview

This document provides guidance on creating comprehensive evaluations for MCP servers. Evaluations test whether LLMs can effectively use your MCP server to answer realistic, complex questions using only the tools provided.

Quick Reference

Evaluation Requirements

Create 10 human-readable questions
Questions must be READ-ONLY, INDEPENDENT, NON-DESTRUCTIVE
Each question requires multiple tool calls (potentially dozens)
Answers must be single, verifiable values
Answers must be STABLE (won't change over time)

Output Format

<evaluation>
   <qa_pair>
      <question>Your question here</question>
      <answer>Single verifiable answer</answer>
   </qa_pair>
</evaluation>

Purpose of Evaluations

The measure of quality of an MCP server is NOT how well or comprehensively the server implements tools, but how well these implementations (input/output schemas, docstrings/descriptions, functionality) enable LLMs with no other context and access ONLY to the MCP servers to answer realistic and difficult questions.

Evaluation Overview

Create 10 human-readable questions requiring ONLY READ-ONLY, INDEPENDENT, NON-DESTRUCTIVE, and IDEMPOTENT operations to answer. Each question should be:

Realistic
Clear and concise
Unambiguous
Complex, requiring potentially dozens of tool calls or steps
Answerable with a single, verifiable value that you identify in advance

Question Guidelines

Core Requirements

Questions MUST be independent
- Each question should NOT depend on the answer to any other question
- Should not assume prior write operations from processing another question
Questions MUST require ONLY NON-DESTRUCTIVE AND IDEMPOTENT tool use
- Should not instruct or require modifying state to arrive at the correct answer
Questions must be REALISTIC, CLEAR, CONCISE, and COMPLEX
- Must require another LLM to use multiple (potentially dozens of) tools or steps to answer

Complexity and Depth

Questions must require deep exploration
- Consider multi-hop questions requiring multiple sub-questions and sequential tool calls
- Each step should benefit from information found in previous questions
Questions may require extensive paging
- May need paging through multiple pages of results
- May require querying old data (1-2 years out-of-date) to find niche information
- The questions must be DIFFICULT
Questions must require deep understanding
- Rather than surface-level knowledge
- May pose complex ideas as True/False questions requiring evidence
- May use multiple-choice format where LLM must search different hypotheses
Questions must not be solvable with straightforward keyword search
- Do not include specific keywords from the target content
- Use synonyms, related concepts, or paraphrases
- Require multiple searches, analyzing multiple related items, extracting context, then deriving the answer

Tool Testing

Questions should stress-test tool return values
- May elicit tools returning large JSON objects or lists, overwhelming the LLM
- Should require understanding multiple modalities of data:
  - IDs and names
  - Timestamps and datetimes (months, days, years, seconds)
  - File IDs, names, extensions, and mimetypes
  - URLs, GIDs, etc.
- Should probe the tool's ability to return all useful forms of data
Questions should MOSTLY reflect real human use cases
- The kinds of information retrieval tasks that HUMANS assisted by an LLM would care about
Questions may require dozens of tool calls
- This challenges LLMs with limited context
- Encourages MCP server tools to reduce information returned
Include ambiguous questions
- May be ambiguous OR require difficult decisions on which tools to call
- Force the LLM to potentially make mistakes or misinterpret
- Ensure that despite AMBIGUITY, there is STILL A SINGLE VERIFIABLE ANSWER

Stability

Questions must be designed so the answer DOES NOT CHANGE
- Do not ask questions that rely on "current state" which is dynamic
- For example, do not count:
  - Number of reactions to a post
  - Number of replies to a thread
  - Number of members in a channel
DO NOT let the MCP server RESTRICT the kinds of questions you create
- Create challenging and complex questions
- Some may not be solvable with the available MCP server tools
- Questions may require specific output formats (datetime vs. epoch time, JSON vs. MARKDOWN)
- Questions may require dozens of tool calls to complete

Answer Guidelines

Verification

Answers must be VERIFIABLE via direct string comparison
- If the answer can be re-written in many formats, clearly specify the output format in the QUESTION
- Examples: "Use YYYY/MM/DD.", "Respond True or False.", "Answer A, B, C, or D and nothing else."
- Answer should be a single VERIFIABLE value such as:
  - User ID, user name, display name, first name, last name
  - Channel ID, channel name
  - Message ID, string
  - URL, title
  - Numerical quantity
  - Timestamp, datetime
  - Boolean (for True/False questions)
  - Email address, phone number
  - File ID, file name, file extension
  - Multiple choice answer
- Answers must not require special formatting or complex, structured output
- Answer will be verified using DIRECT STRING COMPARISON

Readability

Answers should generally prefer HUMAN-READABLE formats
- Examples: names, first name, last name, datetime, file name, message string, URL, yes/no, true/false, a/b/c/d
- Rather than opaque IDs (though IDs are acceptable)
- The VAST MAJORITY of answers should be human-readable

Stability

Answers must be STABLE/STATIONARY
- Look at old content (e.g., conversations that have ended, projects that have launched, questions answered)
- Create QUESTIONS based on "closed" concepts that will always return the same answer
- Questions may ask to consider a fixed time window to insulate from non-stationary answers
- Rely on context UNLIKELY to change
- Example: if finding a paper name, be SPECIFIC enough so answer is not confused with papers published later
Answers must be CLEAR and UNAMBIGUOUS
- Questions must be designed so there is a single, clear answer
- Answer can be derived from using the MCP server tools

Diversity

Answers must be DIVERSE
- Answer should be a single VERIFIABLE value in diverse modalities and formats
- User concept: user ID, user name, display name, first name, last name, email address, phone number
- Channel concept: channel ID, channel name, channel topic
- Message concept: message ID, message string, timestamp, month, day, year
Answers must NOT be complex structures
- Not a list of values
- Not a complex object
- Not a list of IDs or strings
- Not natural language text
- UNLESS the answer can be straightforwardly verified using DIRECT STRING COMPARISON
- And can be realistically reproduced
- It should be unlikely that an LLM would return the same list in any other order or format

Evaluation Process

Step 1: Documentation Inspection

Read the documentation of the target API to understand:

Available endpoints and functionality
If ambiguity exists, fetch additional information from the web
Parallelize this step AS MUCH AS POSSIBLE
Ensure each subagent is ONLY examining documentation from the file system or on the web

Step 2: Tool Inspection

List the tools available in the MCP server:

Inspect the MCP server directly
Understand input/output schemas, docstrings, and descriptions
WITHOUT calling the tools themselves at this stage

Step 3: Developing Understanding

Repeat steps 1 & 2 until you have a good understanding:

Iterate multiple times
Think about the kinds of tasks you want to create
Refine your understanding
At NO stage should you READ the code of the MCP server implementation itself
Use your intuition and understanding to create reasonable, realistic, but VERY challenging tasks

Step 4: Read-Only Content Inspection

After understanding the API and tools, USE the MCP server tools:

Inspect content using READ-ONLY and NON-DESTRUCTIVE operations ONLY
Goal: identify specific content (e.g., users, channels, messages, projects, tasks) for creating realistic questions
Should NOT call any tools that modify state
Will NOT read the code of the MCP server implementation itself
Parallelize this step with individual sub-agents pursuing independent explorations
Ensure each subagent is only performing READ-ONLY, NON-DESTRUCTIVE, and IDEMPOTENT operations
BE CAREFUL: SOME TOOLS may return LOTS OF DATA which would cause you to run out of CONTEXT
Make INCREMENTAL, SMALL, AND TARGETED tool calls for exploration
In all tool call requests, use the limit parameter to limit results (<10)
Use pagination

Step 5: Task Generation

After inspecting the content, create 10 human-readable questions:

An LLM should be able to answer these with the MCP server
Follow all question and answer guidelines above

Output Format

Each QA pair consists of a question and an answer. The output should be an XML file with this structure:

<evaluation>
   <qa_pair>
      <question>Find the project created in Q2 2024 with the highest number of completed tasks. What is the project name?</question>
      <answer>Website Redesign</answer>
   </qa_pair>
   <qa_pair>
      <question>Search for issues labeled as "bug" that were closed in March 2024. Which user closed the most issues? Provide their username.</question>
      <answer>sarah_dev</answer>
   </qa_pair>
   <qa_pair>
      <question>Look for pull requests that modified files in the /api directory and were merged between January 1 and January 31, 2024. How many different contributors worked on these PRs?</question>
      <answer>7</answer>
   </qa_pair>
   <qa_pair>
      <question>Find the repository with the most stars that was created before 2023. What is the repository name?</question>
      <answer>data-pipeline</answer>
   </qa_pair>
</evaluation>

Evaluation Examples

Good Questions

Example 1: Multi-hop question requiring deep exploration (GitHub MCP)

<qa_pair>
   <question>Find the repository that was archived in Q3 2023 and had previously been the most forked project in the organization. What was the primary programming language used in that repository?</question>
   <answer>Python</answer>
</qa_pair>

This question is good because:

Requires multiple searches to find archived repositories
Needs to identify which had the most forks before archival
Requires examining repository details for the language
Answer is a simple, verifiable value
Based on historical (closed) data that won't change

Example 2: Requires understanding context without keyword matching (Project Management MCP)

<qa_pair>
   <question>Locate the initiative focused on improving customer onboarding that was completed in late 2023. The project lead created a retrospective document after completion. What was the lead's role title at that time?</question>
   <answer>Product Manager</answer>
</qa_pair>

This question is good because:

Doesn't use specific project name ("initiative focused on improving customer onboarding")
Requires finding completed projects from specific timeframe
Needs to identify the project lead and their role
Requires understanding context from retrospective documents
Answer is human-readable and stable
Based on completed work (won't change)

Example 3: Complex aggregation requiring multiple steps (Issue Tracker MCP)

<qa_pair>
   <question>Among all bugs reported in January 2024 that were marked as critical priority, which assignee resolved the highest percentage of their assigned bugs within 48 hours? Provide the assignee's username.</question>
   <answer>alex_eng</answer>
</qa_pair>

This question is good because:

Requires filtering bugs by date, priority, and status
Needs to group by assignee and calculate resolution rates
Requires understanding timestamps to determine 48-hour windows
Tests pagination (potentially many bugs to process)
Answer is a single username
Based on historical data from specific time period

Example 4: Requires synthesis across multiple data types (CRM MCP)

<qa_pair>
   <question>Find the account that upgraded from the Starter to Enterprise plan in Q4 2023 and had the highest annual contract value. What industry does this account operate in?</question>
   <answer>Healthcare</answer>
</qa_pair>

This question is good because:

Requires understanding subscription tier changes
Needs to identify upgrade events in specific timeframe
Requires comparing contract values
Must access account industry information
Answer is simple and verifiable
Based on completed historical transactions

Poor Questions

Example 1: Answer changes over time

<qa_pair>
   <question>How many open issues are currently assigned to the engineering team?</question>
   <answer>47</answer>
</qa_pair>

This question is poor because:

The answer will change as issues are created, closed, or reassigned
Not based on stable/stationary data
Relies on "current state" which is dynamic

Example 2: Too easy with keyword search

<qa_pair>
   <question>Find the pull request with title "Add authentication feature" and tell me who created it.</question>
   <answer>developer123</answer>
</qa_pair>

This question is poor because:

Can be solved with a straightforward keyword search for exact title
Doesn't require deep exploration or understanding
No synthesis or analysis needed

Example 3: Ambiguous answer format

<qa_pair>
   <question>List all the repositories that have Python as their primary language.</question>
   <answer>repo1, repo2, repo3, data-pipeline, ml-tools</answer>
</qa_pair>

This question is poor because:

Answer is a list that could be returned in any order
Difficult to verify with direct string comparison
LLM might format differently (JSON array, comma-separated, newline-separated)
Better to ask for a specific aggregate (count) or superlative (most stars)

Verification Process

After creating evaluations:

Examine the XML file to understand the schema
Load each task instruction and in parallel using the MCP server and tools, identify the correct answer by attempting to solve the task YOURSELF
Flag any operations that require WRITE or DESTRUCTIVE operations
Accumulate all CORRECT answers and replace any incorrect answers in the document
Remove any <qa_pair> that require WRITE or DESTRUCTIVE operations

Remember to parallelize solving tasks to avoid running out of context, then accumulate all answers and make changes to the file at the end.

Tips for Creating Quality Evaluations

Think Hard and Plan Ahead before generating tasks
Parallelize Where Opportunity Arises to speed up the process and manage context
Focus on Realistic Use Cases that humans would actually want to accomplish
Create Challenging Questions that test the limits of the MCP server's capabilities
Ensure Stability by using historical data and closed concepts
Verify Answers by solving the questions yourself using the MCP server tools
Iterate and Refine based on what you learn during the process

Running Evaluations

After creating your evaluation file, you can use the provided evaluation harness to test your MCP server.

Setup

Install Dependencies

pip install -r scripts/requirements.txt

Or install manually:

pip install anthropic mcp

Set API Key

export ANTHROPIC_API_KEY=your_api_key_here

Evaluation File Format

Evaluation files use XML format with <qa_pair> elements:

<evaluation>
   <qa_pair>
      <question>Find the project created in Q2 2024 with the highest number of completed tasks. What is the project name?</question>
      <answer>Website Redesign</answer>
   </qa_pair>
   <qa_pair>
      <question>Search for issues labeled as "bug" that were closed in March 2024. Which user closed the most issues? Provide their username.</question>
      <answer>sarah_dev</answer>
   </qa_pair>
</evaluation>

Running Evaluations

The evaluation script (scripts/evaluation.py) supports three transport types:

Important:

stdio transport: The evaluation script automatically launches and manages the MCP server process for you. Do not run the server manually.
sse/http transports: You must start the MCP server separately before running the evaluation. The script connects to the already-running server at the specified URL.

1. Local STDIO Server

For locally-run MCP servers (script launches the server automatically):

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

With environment variables:

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)

For SSE-based MCP servers (you must start the server first):

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)

For HTTP-based MCP servers (you must start the server first):

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

Command-Line Options

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             Path to evaluation XML file

optional arguments:
  -h, --help            Show help message
  -t, --transport       Transport type: stdio, sse, or http (default: stdio)
  -m, --model           Claude model to use (default: claude-3-7-sonnet-20250219)
  -o, --output          Output file for report (default: print to stdout)

stdio options:
  -c, --command         Command to run MCP server (e.g., python, node)
  -a, --args            Arguments for the command (e.g., server.py)
  -e, --env             Environment variables in KEY=VALUE format

sse/http options:
  -u, --url             MCP server URL
  -H, --header          HTTP headers in 'Key: Value' format

Output

The evaluation script generates a detailed report including:

Summary Statistics:
- Accuracy (correct/total)
- Average task duration
- Average tool calls per task
- Total tool calls
Per-Task Results:
- Prompt and expected response
- Actual response from the agent
- Whether the answer was correct (✅/❌)
- Duration and tool call details
- Agent's summary of its approach
- Agent's feedback on the tools

Save Report to File

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

Complete Example Workflow

Here's a complete example of creating and running an evaluation:

Create your evaluation file (my_evaluation.xml):

<evaluation>
   <qa_pair>
      <question>Find the user who created the most issues in January 2024. What is their username?</question>
      <answer>alice_developer</answer>
   </qa_pair>
   <qa_pair>
      <question>Among all pull requests merged in Q1 2024, which repository had the highest number? Provide the repository name.</question>
      <answer>backend-api</answer>
   </qa_pair>
   <qa_pair>
      <question>Find the project that was completed in December 2023 and had the longest duration from start to finish. How many days did it take?</question>
      <answer>127</answer>
   </qa_pair>
</evaluation>

Install dependencies:

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

Run evaluation:

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

Review the report in github_eval_report.md to:
- See which questions passed/failed
- Read the agent's feedback on your tools
- Identify areas for improvement
- Iterate on your MCP server design

Troubleshooting

Connection Errors

If you get connection errors:

STDIO: Verify the command and arguments are correct
SSE/HTTP: Check the URL is accessible and headers are correct
Ensure any required API keys are set in environment variables or headers

Low Accuracy

If many evaluations fail:

Review the agent's feedback for each task
Check if tool descriptions are clear and comprehensive
Verify input parameters are well-documented
Consider whether tools return too much or too little data
Ensure error messages are actionable

Timeout Issues

If tasks are timing out:

Use a more capable model (e.g., claude-3-7-sonnet-20250219)
Check if tools are returning too much data
Verify pagination is working correctly
Consider simplifying complex questions FILE:reference/node_mcp_server.md

Node/TypeScript MCP Server Implementation Guide

Overview

This document provides Node/TypeScript-specific best practices and examples for implementing MCP servers using the MCP TypeScript SDK. It covers project structure, server setup, tool registration patterns, input validation with Zod, error handling, and complete working examples.

Quick Reference

Key Imports

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";

Server Initialization

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

Tool Registration Pattern

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 // Modern pattern for structured data
    };
  }
);

MCP TypeScript SDK

The official MCP TypeScript SDK provides:

McpServer class for server initialization
registerTool method for tool registration
Zod schema integration for runtime input validation
Type-safe tool handler implementations

IMPORTANT - Use Modern APIs Only:

DO use: server.registerTool(), server.registerResource(), server.registerPrompt()
DO NOT use: Old deprecated APIs such as server.tool(), server.setRequestHandler(ListToolsRequestSchema, ...), or manual handler registration
The register* methods provide better type safety, automatic schema handling, and are the recommended approach

See the MCP SDK documentation in the references for complete details.

Server Naming Convention

Node/TypeScript MCP servers must follow this naming pattern:

Format: {service}-mcp-server (lowercase with hyphens)
Examples: github-mcp-server, jira-mcp-server, stripe-mcp-server

The name should be:

General (not tied to specific features)
Descriptive of the service/API being integrated
Easy to infer from the task description
Without version numbers or dates

Project Structure

Create the following structure for Node/TypeScript MCP servers:

{service}-mcp-server/
├── package.json
├── tsconfig.json
├── README.md
├── src/
│   ├── index.ts          # Main entry point with McpServer initialization
│   ├── types.ts          # TypeScript type definitions and interfaces
│   ├── tools/            # Tool implementations (one file per domain)
│   ├── services/         # API clients and shared utilities
│   ├── schemas/          # Zod validation schemas
│   └── constants.ts      # Shared constants (API_URL, CHARACTER_LIMIT, etc.)
└── dist/                 # Built JavaScript files (entry point: dist/index.js)

Tool Implementation

Tool Naming

Use snake_case for tool names (e.g., "search_users", "create_project", "get_channel_info") with clear, action-oriented names.

Avoid Naming Conflicts: Include the service context to prevent overlaps:

Use "slack_send_message" instead of just "send_message"
Use "github_create_issue" instead of just "create_issue"
Use "asana_list_tasks" instead of just "list_tasks"

Tool Structure

Tools are registered using the registerTool method with the following requirements:

Use Zod schemas for runtime input validation and type safety
The description field must be explicitly provided - JSDoc comments are NOT automatically extracted
Explicitly provide title, description, inputSchema, and annotations
The inputSchema must be a Zod schema object (not a JSON schema)
Type all parameters and return values explicitly

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 for input validation
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "Query must be at least 2 characters")
    .max(200, "Query must not exceed 200 characters")
    .describe("Search string to match against names/emails"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip for pagination"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();

// Type definition from Zod schema
type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

server.registerTool(
  "example_search_users",
  {
    title: "Search Example Users",
    description: `Search for users in the Example system by name, email, or team.

This tool searches across all user profiles in the Example platform, supporting partial matches and various search filters. It does NOT create or modify users, only searches existing ones.

Args:
  - query (string): Search string to match against names/emails
  - limit (number): Maximum results to return, between 1-100 (default: 20)
  - offset (number): Number of results to skip for pagination (default: 0)
  - response_format ('markdown' | 'json'): Output format (default: 'markdown')

Returns:
  For JSON format: Structured data with schema:
  {
    "total": number,           // Total number of matches found
    "count": number,           // Number of results in this response
    "offset": number,          // Current pagination offset
    "users": [
      {
        "id": string,          // User ID (e.g., "U123456789")
        "name": string,        // Full name (e.g., "John Doe")
        "email": string,       // Email address
        "team": string,        // Team name (optional)
        "active": boolean      // Whether user is active
      }
    ],
    "has_more": boolean,       // Whether more results are available
    "next_offset": number      // Offset for next page (if has_more is true)
  }

Examples:
  - Use when: "Find all marketing team members" -> params with query="team:marketing"
  - Use when: "Search for John's account" -> params with query="john"
  - Don't use when: You need to create a user (use example_create_user instead)

Error Handling:
  - Returns "Error: Rate limit exceeded" if too many requests (429 status)
  - Returns "No users found matching '<query>'" if search returns empty`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    try {
      // Input validation is handled by Zod schema
      // Make API request using validated parameters
      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: `No users found matching '${params.query}'`
          }]
        };
      }

      // Prepare structured output
      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
        } : {})
      };

      // Format text representation based on requested format
      let textContent: string;
      if (params.response_format === ResponseFormat.MARKDOWN) {
        const lines = [`# User Search Results: '${params.query}'`, "",
          `Found ${total} users (showing ${users.length})`, ""];
        for (const user of users) {
          lines.push(`## ${user.name} (${user.id})`);
          lines.push(`- **Email**: ${user.email}`);
          if (user.team) lines.push(`- **Team**: ${user.team}`);
          lines.push("");
        }
        textContent = lines.join("\n");
      } else {
        textContent = JSON.stringify(output, null, 2);
      }

      return {
        content: [{ type: "text", text: textContent }],
        structuredContent: output // Modern pattern for structured data
      };
    } catch (error) {
      return {
        content: [{
          type: "text",
          text: handleApiError(error)
        }]
      };
    }
  }
);

Zod Schemas for Input Validation

Zod provides runtime type validation:

import { z } from "zod";

// Basic schema with validation
const CreateUserSchema = z.object({
  name: z.string()
    .min(1, "Name is required")
    .max(100, "Name must not exceed 100 characters"),
  email: z.string()
    .email("Invalid email format"),
  age: z.number()
    .int("Age must be a whole number")
    .min(0, "Age cannot be negative")
    .max(150, "Age cannot be greater than 150")
}).strict();  // Use .strict() to forbid extra fields

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

const SearchSchema = z.object({
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format")
});

// Optional fields with defaults
const PaginationSchema = z.object({
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip")
});

Response Format Options

Support multiple output formats for flexibility:

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

const inputSchema = z.object({
  query: z.string(),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
});

Markdown format:

Use headers, lists, and formatting for clarity
Convert timestamps to human-readable format
Show display names with IDs in parentheses
Omit verbose metadata
Group related information logically

JSON format:

Return complete, structured data suitable for programmatic processing
Include all available fields and metadata
Use consistent field names and types

Pagination Implementation

For tools that list resources:

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 Limits and Truncation

Add a CHARACTER_LIMIT constant to prevent overwhelming responses:

// At module level in constants.ts
export const CHARACTER_LIMIT = 25000;  // Maximum response size in characters

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

  // Check character limit and truncate if needed
  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 =
      `Response truncated from ${data.length} to ${truncatedData.length} items. ` +
      `Use 'offset' parameter or add filters to see more results.`;
    result = JSON.stringify(response, null, 2);
  }

  return result;
}

Error Handling

Provide clear, actionable error messages:

import axios, { AxiosError } from "axios";

function handleApiError(error: unknown): string {
  if (error instanceof AxiosError) {
    if (error.response) {
      switch (error.response.status) {
        case 404:
          return "Error: Resource not found. Please check the ID is correct.";
        case 403:
          return "Error: Permission denied. You don't have access to this resource.";
        case 429:
          return "Error: Rate limit exceeded. Please wait before making more requests.";
        default:
          return `Error: API request failed with status ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "Error: Request timed out. Please try again.";
    }
  }
  return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}

Shared Utilities

Extract common functionality into reusable functions:

// Shared API request function
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;
  }
}

Async/Await Best Practices

Always use async/await for network requests and I/O operations:

// Good: Async network request
async function fetchData(resourceId: string): Promise<ResourceData> {
  const response = await axios.get(`${API_URL}/resource/${resourceId}`);
  return response.data;
}

// Bad: Promise chains
function fetchData(resourceId: string): Promise<ResourceData> {
  return axios.get(`${API_URL}/resource/${resourceId}`)
    .then(response => response.data);  // Harder to read and maintain
}

TypeScript Best Practices

Use Strict TypeScript: Enable strict mode in tsconfig.json
Define Interfaces: Create clear interface definitions for all data structures
Avoid any: Use proper types or unknown instead of any
Zod for Runtime Validation: Use Zod schemas to validate external data
Type Guards: Create type guard functions for complex type checking
Error Handling: Always use try-catch with proper error type checking
Null Safety: Use optional chaining (?.) and nullish coalescing (??)

// Good: Type-safe with Zod and interfaces
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);  // Runtime validation
}

// Bad: Using any
async function getUser(id: string): Promise<any> {
  return await apiCall(`/users/${id}`);  // No type safety
}

Package Configuration

package.json

{
  "name": "{service}-mcp-server",
  "version": "1.0.0",
  "description": "MCP server for {Service} API integration",
  "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"]
}

Complete Example

#!/usr/bin/env node
/**
 * MCP Server for Example Service.
 *
 * This server provides tools to interact with Example API, including user search,
 * project management, and data export capabilities.
 */

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";

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

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

// Zod schemas
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "Query must be at least 2 characters")
    .max(200, "Query must not exceed 200 characters")
    .describe("Search string to match against names/emails"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("Maximum results to return"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("Number of results to skip for pagination"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();

type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

// Shared utility functions
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 "Error: Resource not found. Please check the ID is correct.";
        case 403:
          return "Error: Permission denied. You don't have access to this resource.";
        case 429:
          return "Error: Rate limit exceeded. Please wait before making more requests.";
        default:
          return `Error: API request failed with status ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "Error: Request timed out. Please try again.";
    }
  }
  return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}

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

// Register tools
server.registerTool(
  "example_search_users",
  {
    title: "Search Example Users",
    description: `[Full description as shown above]`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    // Implementation as shown above
  }
);

// Main function
// For stdio (local):
async function runStdio() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
    process.exit(1);
  }

  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP server running via stdio");
}

// For streamable HTTP (remote):
async function runHTTP() {
  if (!process.env.EXAMPLE_API_KEY) {
    console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
    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 server running on http://localhost:${port}/mcp`);
  });
}

// Choose transport based on environment
const transport = process.env.TRANSPORT || 'stdio';
if (transport === 'http') {
  runHTTP().catch(error => {
    console.error("Server error:", error);
    process.exit(1);
  });
} else {
  runStdio().catch(error => {
    console.error("Server error:", error);
    process.exit(1);
  });
}

Advanced MCP Features

Resource Registration

Expose data as resources for efficient, URI-based access:

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

// Register a resource with URI template
server.registerResource(
  {
    uri: "file://documents/{name}",
    name: "Document Resource",
    description: "Access documents by name",
    mimeType: "text/plain"
  },
  async (uri: string) => {
    // Extract parameter from URI
    const match = uri.match(/^file:\/\/documents\/(.+)$/);
    if (!match) {
      throw new Error("Invalid URI format");
    }

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

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

// List available resources dynamically
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
    }))
  };
});

When to use Resources vs Tools:

Resources: For data access with simple URI-based parameters
Tools: For complex operations requiring validation and business logic
Resources: When data is relatively static or template-based
Tools: When operations have side effects or complex workflows

Transport Options

The TypeScript SDK supports two main transport mechanisms:

Streamable HTTP (Recommended for Remote Servers)

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) => {
  // Create new transport for each request (stateless, prevents request ID collisions)
  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 (For Local Integrations)

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

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

Transport selection:

Streamable HTTP: Web services, remote access, multiple clients
stdio: Command-line tools, local development, subprocess integration

Notification Support

Notify clients when server state changes:

// Notify when tools list changes
server.notification({
  method: "notifications/tools/list_changed"
});

// Notify when resources change
server.notification({
  method: "notifications/resources/list_changed"
});

Use notifications sparingly - only when server capabilities genuinely change.

Code Best Practices

Code Composability and Reusability

Your implementation MUST prioritize composability and code reuse:

Extract Common Functionality:
- Create reusable helper functions for operations used across multiple tools
- Build shared API clients for HTTP requests instead of duplicating code
- Centralize error handling logic in utility functions
- Extract business logic into dedicated functions that can be composed
- Extract shared markdown or JSON field selection & formatting functionality
Avoid Duplication:
- NEVER copy-paste similar code between tools
- If you find yourself writing similar logic twice, extract it into a function
- Common operations like pagination, filtering, field selection, and formatting should be shared
- Authentication/authorization logic should be centralized

Building and Running

Always build your TypeScript code before running:

# Build the project
npm run build

# Run the server
npm start

# Development with auto-reload
npm run dev

Always ensure npm run build completes successfully before considering the implementation complete.

Quality Checklist

Before finalizing your Node/TypeScript MCP server implementation, ensure:

Strategic Design

Tools enable complete workflows, not just API endpoint wrappers
Tool names reflect natural task subdivisions
Response formats optimize for agent context efficiency
Human-readable identifiers used where appropriate
Error messages guide agents toward correct usage

Implementation Quality

FOCUSED IMPLEMENTATION: Most important and valuable tools implemented
All tools registered using registerTool with complete configuration
All tools include title, description, inputSchema, and annotations
Annotations correctly set (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
All tools use Zod schemas for runtime input validation with .strict() enforcement
All Zod schemas have proper constraints and descriptive error messages
All tools have comprehensive descriptions with explicit input/output types
Descriptions include return value examples and complete schema documentation
Error messages are clear, actionable, and educational

TypeScript Quality

TypeScript interfaces are defined for all data structures
Strict TypeScript is enabled in tsconfig.json
No use of any type - use unknown or proper types instead
All async functions have explicit Promise return types
Error handling uses proper type guards (e.g., axios.isAxiosError, z.ZodError)

Advanced Features (where applicable)

Resources registered for appropriate data endpoints
Appropriate transport configured (stdio or streamable HTTP)
Notifications implemented for dynamic server capabilities
Type-safe with SDK interfaces

Project Configuration

Package.json includes all necessary dependencies
Build script produces working JavaScript in dist/ directory
Main entry point is properly configured as dist/index.js
Server name follows format: {service}-mcp-server
tsconfig.json properly configured with strict mode

Code Quality

Pagination is properly implemented where applicable
Large responses check CHARACTER_LIMIT constant and truncate with clear messages
Filtering options are provided for potentially large result sets
All network operations handle timeouts and connection errors gracefully
Common functionality is extracted into reusable functions
Return types are consistent across similar operations

Testing and Build

npm run build completes successfully without errors
dist/index.js created and executable
Server runs: node dist/index.js --help
All imports resolve correctly
Sample tool calls work as expected FILE:reference/python_mcp_server.md

Python MCP Server Implementation Guide

Overview

This document provides Python-specific best practices and examples for implementing MCP servers using the MCP Python SDK. It covers server setup, tool registration patterns, input validation with Pydantic, error handling, and complete working examples.

Quick Reference

Key Imports

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

Server Initialization

mcp = FastMCP("service_mcp")

Tool Registration Pattern

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

MCP Python SDK and FastMCP

The official MCP Python SDK provides FastMCP, a high-level framework for building MCP servers. It provides:

Automatic description and inputSchema generation from function signatures and docstrings
Pydantic model integration for input validation
Decorator-based tool registration with @mcp.tool

For complete SDK documentation, use WebFetch to load: https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md

Server Naming Convention

Python MCP servers must follow this naming pattern:

Format: {service}_mcp (lowercase with underscores)
Examples: github_mcp, jira_mcp, stripe_mcp

The name should be:

General (not tied to specific features)
Descriptive of the service/API being integrated
Easy to infer from the task description
Without version numbers or dates

Tool Implementation

Tool Naming

Use snake_case for tool names (e.g., "search_users", "create_project", "get_channel_info") with clear, action-oriented names.

Avoid Naming Conflicts: Include the service context to prevent overlaps:

Use "slack_send_message" instead of just "send_message"
Use "github_create_issue" instead of just "create_issue"
Use "asana_list_tasks" instead of just "list_tasks"

Tool Structure with FastMCP

Tools are defined using the @mcp.tool decorator with Pydantic models for input validation:

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

# Initialize the MCP server
mcp = FastMCP("example_mcp")

# Define Pydantic model for input validation
class ServiceToolInput(BaseModel):
    '''Input model for service tool operation.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,  # Auto-strip whitespace from strings
        validate_assignment=True,    # Validate on assignment
        extra='forbid'              # Forbid extra fields
    )

    param1: str = Field(..., description="First parameter description (e.g., 'user123', 'project-abc')", min_length=1, max_length=100)
    param2: Optional[int] = Field(default=None, description="Optional integer parameter with constraints", ge=0, le=1000)
    tags: Optional[List[str]] = Field(default_factory=list, description="List of tags to apply", max_items=10)

@mcp.tool(
    name="service_tool_name",
    annotations={
        "title": "Human-Readable Tool Title",
        "readOnlyHint": True,     # Tool does not modify environment
        "destructiveHint": False,  # Tool does not perform destructive operations
        "idempotentHint": True,    # Repeated calls have no additional effect
        "openWorldHint": False     # Tool does not interact with external entities
    }
)
async def service_tool_name(params: ServiceToolInput) -> str:
    '''Tool description automatically becomes the 'description' field.

    This tool performs a specific operation on the service. It validates all inputs
    using the ServiceToolInput Pydantic model before processing.

    Args:
        params (ServiceToolInput): Validated input parameters containing:
            - param1 (str): First parameter description
            - param2 (Optional[int]): Optional parameter with default
            - tags (Optional[List[str]]): List of tags

    Returns:
        str: JSON-formatted response containing operation results
    '''
    # Implementation here
    pass

Pydantic v2 Key Features

Use model_config instead of nested Config class
Use field_validator instead of deprecated validator
Use model_dump() instead of deprecated dict()
Validators require @classmethod decorator
Type hints are required for validator methods

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="User's full name", min_length=1, max_length=100)
    email: str = Field(..., description="User's email address", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., description="User's age", ge=0, le=150)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Email cannot be empty")
        return v.lower()

Response Format Options

Support multiple output formats for flexibility:

from enum import Enum

class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    query: str = Field(..., description="Search query")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
    )

Markdown format:

Use headers, lists, and formatting for clarity
Convert timestamps to human-readable format (e.g., "2024-01-15 10:30:00 UTC" instead of epoch)
Show display names with IDs in parentheses (e.g., "@john.doe (U123456)")
Omit verbose metadata (e.g., show only one profile image URL, not all sizes)
Group related information logically

JSON format:

Return complete, structured data suitable for programmatic processing
Include all available fields and metadata
Use consistent field names and types

Pagination Implementation

For tools that list resources:

class ListInput(BaseModel):
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)

async def list_items(params: ListInput) -> str:
    # Make API request with pagination
    data = await api_request(limit=params.limit, offset=params.offset)

    # Return pagination info
    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)

Error Handling

Provide clear, actionable error messages:

def _handle_api_error(e: Exception) -> str:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"

Shared Utilities

Extract common functionality into reusable functions:

# Shared API request function
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    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 Best Practices

Always use async/await for network requests and I/O operations:

# Good: Async network request
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()

# Bad: Synchronous request
def fetch_data(resource_id: str) -> dict:
    response = requests.get(f"{API_URL}/resource/{resource_id}")  # Blocks
    return response.json()

Type Hints

Use type hints throughout:

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"]}

Tool Docstrings

Every tool must have comprehensive docstrings with explicit type information:

async def search_users(params: UserSearchInput) -> str:
    '''
    Search for users in the Example system by name, email, or team.

    This tool searches across all user profiles in the Example platform,
    supporting partial matches and various search filters. It does NOT
    create or modify users, only searches existing ones.

    Args:
        params (UserSearchInput): Validated input parameters containing:
            - query (str): Search string to match against names/emails (e.g., "john", "@example.com", "team:marketing")
            - limit (Optional[int]): Maximum results to return, between 1-100 (default: 20)
            - offset (Optional[int]): Number of results to skip for pagination (default: 0)

    Returns:
        str: JSON-formatted string containing search results with the following schema:

        Success response:
        {
            "total": int,           # Total number of matches found
            "count": int,           # Number of results in this response
            "offset": int,          # Current pagination offset
            "users": [
                {
                    "id": str,      # User ID (e.g., "U123456789")
                    "name": str,    # Full name (e.g., "John Doe")
                    "email": str,   # Email address (e.g., "john@example.com")
                    "team": str     # Team name (e.g., "Marketing") - optional
                }
            ]
        }

        Error response:
        "Error: <error message>" or "No users found matching '<query>'"

    Examples:
        - Use when: "Find all marketing team members" -> params with query="team:marketing"
        - Use when: "Search for John's account" -> params with query="john"
        - Don't use when: You need to create a user (use example_create_user instead)
        - Don't use when: You have a user ID and need full details (use example_get_user instead)

    Error Handling:
        - Input validation errors are handled by Pydantic model
        - Returns "Error: Rate limit exceeded" if too many requests (429 status)
        - Returns "Error: Invalid API authentication" if API key is invalid (401 status)
        - Returns formatted list of results or "No users found matching 'query'"
    '''

Complete Example

See below for a complete Python MCP server example:

#!/usr/bin/env python3
'''
MCP Server for Example Service.

This server provides tools to interact with Example API, including user search,
project management, and data export capabilities.
'''

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

# Initialize the MCP server
mcp = FastMCP("example_mcp")

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

# Enums
class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

# Pydantic Models for Input Validation
class UserSearchInput(BaseModel):
    '''Input model for user search operations.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    query: str = Field(..., description="Search string to match against names/emails", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Query cannot be empty or whitespace only")
        return v.strip()

# Shared utility functions
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    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:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"

# Tool definitions
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Example Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''Search for users in the Example system by name, email, or team.

    [Full docstring as shown above]
    '''
    try:
        # Make API request using validated parameters
        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"No users found matching '{params.query}'"

        # Format response based on requested format
        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [f"# User Search Results: '{params.query}'", ""]
            lines.append(f"Found {total} users (showing {len(users)})")
            lines.append("")

            for user in users:
                lines.append(f"## {user['name']} ({user['id']})")
                lines.append(f"- **Email**: {user['email']}")
                if user.get('team'):
                    lines.append(f"- **Team**: {user['team']}")
                lines.append("")

            return "\n".join(lines)

        else:
            # Machine-readable JSON format
            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()

Advanced FastMCP Features

Context Parameter Injection

FastMCP can automatically inject a Context parameter into tools for advanced capabilities like logging, progress reporting, resource reading, and user interaction:

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("example_mcp")

@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
    '''Advanced tool with context access for logging and progress.'''

    # Report progress for long operations
    await ctx.report_progress(0.25, "Starting search...")

    # Log information for debugging
    await ctx.log_info("Processing query", {"query": query, "timestamp": datetime.now()})

    # Perform search
    results = await search_api(query)
    await ctx.report_progress(0.75, "Formatting results...")

    # Access server configuration
    server_name = ctx.fastmcp.name

    return format_results(results)

@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
    '''Tool that can request additional input from users.'''

    # Request sensitive information when needed
    api_key = await ctx.elicit(
        prompt="Please provide your API key:",
        input_type="password"
    )

    # Use the provided key
    return await api_call(resource_id, api_key)

Context capabilities:

ctx.report_progress(progress, message) - Report progress for long operations
ctx.log_info(message, data) / ctx.log_error() / ctx.log_debug() - Logging
ctx.elicit(prompt, input_type) - Request input from users
ctx.fastmcp.name - Access server configuration
ctx.read_resource(uri) - Read MCP resources

Resource Registration

Expose data as resources for efficient, template-based access:

@mcp.resource("file://documents/{name}")
async def get_document(name: str) -> str:
    '''Expose documents as MCP resources.

    Resources are useful for static or semi-static data that doesn't
    require complex parameters. They use URI templates for flexible access.
    '''
    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:
    '''Expose configuration as resources with context.'''
    settings = await load_settings()
    return json.dumps(settings.get(key, {}))

When to use Resources vs Tools:

Resources: For data access with simple parameters (URI templates)
Tools: For complex operations with validation and business logic

Structured Output Types

FastMCP supports multiple return types beyond strings:

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

# TypedDict for structured returns
class UserData(TypedDict):
    id: str
    name: str
    email: str

@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
    '''Returns structured data - FastMCP handles serialization.'''
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# Pydantic models for complex validation
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:
    '''Returns Pydantic model - automatically generates schema.'''
    user = await fetch_user(user_id)
    return DetailedUser(**user)

Lifespan Management

Initialize resources that persist across requests:

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan():
    '''Manage resources that live for the server's lifetime.'''
    # Initialize connections, load config, etc.
    db = await connect_to_database()
    config = load_configuration()

    # Make available to all tools
    yield {"db": db, "config": config}

    # Cleanup on shutdown
    await db.close()

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

@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
    '''Access lifespan resources through context.'''
    db = ctx.request_context.lifespan_state["db"]
    results = await db.query(query)
    return format_results(results)

Transport Options

FastMCP supports two main transport mechanisms:

# stdio transport (for local tools) - default
if __name__ == "__main__":
    mcp.run()

# Streamable HTTP transport (for remote servers)
if __name__ == "__main__":
    mcp.run(transport="streamable_http", port=8000)

Transport selection:

stdio: Command-line tools, local integrations, subprocess execution
Streamable HTTP: Web services, remote access, multiple clients

Code Best Practices

Code Composability and Reusability

Your implementation MUST prioritize composability and code reuse:

Extract Common Functionality:
- Create reusable helper functions for operations used across multiple tools
- Build shared API clients for HTTP requests instead of duplicating code
- Centralize error handling logic in utility functions
- Extract business logic into dedicated functions that can be composed
- Extract shared markdown or JSON field selection & formatting functionality
Avoid Duplication:
- NEVER copy-paste similar code between tools
- If you find yourself writing similar logic twice, extract it into a function
- Common operations like pagination, filtering, field selection, and formatting should be shared
- Authentication/authorization logic should be centralized

Python-Specific Best Practices

Use Type Hints: Always include type annotations for function parameters and return values
Pydantic Models: Define clear Pydantic models for all input validation
Avoid Manual Validation: Let Pydantic handle input validation with constraints
Proper Imports: Group imports (standard library, third-party, local)
Error Handling: Use specific exception types (httpx.HTTPStatusError, not generic Exception)
Async Context Managers: Use async with for resources that need cleanup
Constants: Define module-level constants in UPPER_CASE

Quality Checklist

Before finalizing your Python MCP server implementation, ensure:

Strategic Design

Tools enable complete workflows, not just API endpoint wrappers
Tool names reflect natural task subdivisions
Response formats optimize for agent context efficiency
Human-readable identifiers used where appropriate
Error messages guide agents toward correct usage

Implementation Quality

FOCUSED IMPLEMENTATION: Most important and valuable tools implemented
All tools have descriptive names and documentation
Return types are consistent across similar operations
Error handling is implemented for all external calls
Server name follows format: {service}_mcp
All network operations use async/await
Common functionality is extracted into reusable functions
Error messages are clear, actionable, and educational
Outputs are properly validated and formatted

Tool Configuration

All tools implement 'name' and 'annotations' in the decorator
Annotations correctly set (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
All tools use Pydantic BaseModel for input validation with Field() definitions
All Pydantic Fields have explicit types and descriptions with constraints
All tools have comprehensive docstrings with explicit input/output types
Docstrings include complete schema structure for dict/JSON returns
Pydantic models handle input validation (no manual validation needed)

Advanced Features (where applicable)

Context injection used for logging, progress, or elicitation
Resources registered for appropriate data endpoints
Lifespan management implemented for persistent connections
Structured output types used (TypedDict, Pydantic models)
Appropriate transport configured (stdio or streamable HTTP)

Code Quality

File includes proper imports including Pydantic imports
Pagination is properly implemented where applicable
Filtering options are provided for potentially large result sets
All async functions are properly defined with async def
HTTP client usage follows async patterns with proper context managers
Type hints are used throughout the code
Constants are defined at module level in UPPER_CASE

Testing

Server runs successfully: python your_server.py --help
All imports resolve correctly
Sample tool calls work as expected
Error scenarios handled gracefully FILE:scripts/connections.py """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)

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'")

FILE:scripts/evaluation.py """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:

Use the available tools to complete the task
Provide summary of each step in your approach, wrapped in
tags
Provide feedback on the tools provided, wrapped in tags
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}>(.*?)</{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()) FILE:scripts/example_evaluation.xml 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 FILE:scripts/requirements.txt anthropic>=0.39.0 mcp>=1.1.0

MCP 服务器开发指南

概述

流程

🚀 高层工作流

阶段 1：深入调研与规划

1.1 理解现代 MCP 设计

1.2 研究 MCP 协议文档

1.3 研究框架文档

1.4 规划你的实现

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

阶段 2：实现

2.1 搭建项目结构

2.2 实现核心基础设施

2.3 实现工具

阶段 3：审查与测试

3.1 代码质量

3.2 构建与测试

参见针对具体语言的指南，了解详细的测试方法和质量检查清单。

第 4 阶段：创建评估

4.1 理解评估目的

4.2 创建 10 个评估问题

4.3 评估要求

4.4 输出格式

参考文件

📚 文档库

核心 MCP 文档（首先加载）

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

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

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

MCP 服务器最佳实践

快速参考

服务器命名

工具命名

响应格式

分页

传输方式

服务器命名约定

工具命名与设计

工具命名

工具设计

响应格式

JSON 格式（response_format="json"）

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

分页

传输选项

Streamable HTTP

stdio

传输方式选择

安全最佳实践

身份验证与授权

输入验证

错误处理

DNS 重绑定防护

工具注解

错误处理

测试要求

文档要求

MCP 服务器评估指南

概述

快速参考

评估要求

输出格式

评估的目的

评估概览

问题指南

核心要求

复杂性与深度

工具测试

稳定性

答案指南

可验证性

可读性

稳定性

多样性

评估流程

步骤 1：文档检查

步骤 2：工具检查

步骤 3：建立理解

步骤 4：只读内容检查

第 5 步：任务生成

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

JSON 格式（`response_format="json"`）

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