MCP服务构建指南

---
name: mcp-builder
description: 创建高质量 MCP（模型上下文协议）服务器的指南，使 LLM 能够通过精心设计的工具与外部服务交互。在构建 MCP 服务器以集成外部 API 或或服务时使用，无论是使用 Python (FastMCP) 还是 Node/TypeScript (MCP SDK)。
license: 完整条款请参见 LICENSE.txt
---

# MCP 服务器开发指南

## 概述

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

---

# 流程

## 🚀 高级工作流程

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

### 阶段 1：深入研究和规划

#### 1.1 理解现代 MCP 设计

**API 覆盖与工作流工具：**
平衡全面的 API 端点覆盖与专业的工作流工具。工作流工具对于特定任务可能更方便，而全面的覆盖则赋予代理组合操作的灵活性。性能因客户端而异——有些客户端受益于结合基本工具的代码执行，而另一些则更适合使用更高级别的工作流。不确定时，优先考虑全面的 API 覆盖。

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

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

**可操作的错误消息：**
错误消息应通过具体的建议和后续步骤引导代理解决问题。

#### 1.2 研究 MCP 协议文档

**浏览 MCP 规范：**

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

然后获取带有 `.md` 后缀的特定页面以获取 Markdown 格式（例如，`https://modelcontextprotocol.io/specification/draft.md`）。

需要查看的关键页面：
- 规范概述和架构
- 传输机制（可流式 HTTP、stdio）
- 工具、资源和提示定义

#### 1.3 研究框架文档

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

**加载框架文档：**

- **MCP 最佳实践**：[📋 查看最佳实践](./reference/mcp_best_practices.md) - 核心指南

**对于 TypeScript（推荐）：**
- **TypeScript SDK**：使用 WebFetch 加载 `https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md`
- [⚡ TypeScript 指南](./reference/node_mcp_server.md) - TypeScript 模式和示例

**对于 Python：**
- **Python SDK**：使用 WebFetch 加载 `https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`
- [🐍 Python 指南](./reference/python_mcp_server.md) - Python 模式和示例

#### 1.4 规划你的实现

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

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

---

### 阶段 2：实施

#### 2.1 设置项目结构

请参阅特定语言的项目设置指南：
- [⚡ TypeScript 指南](./reference/node_mcp_server.md) - 项目结构、package.json、tsconfig.json
- [🐍 Python 指南](./reference/python_mcp_server.md) - 模块组织、依赖项

#### 2.2 实施核心基础设施

创建共享实用程序：
- 带有身份验证的 API 客户端
- 错误处理助手
- 响应格式化（JSON/Markdown）
- 分页支持

#### 2.3 实施工具

对于每个工具：

**输入 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 服务器后，创建全面的评估以测试其有效性。

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

#### 4.1 理解评估目的

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

#### 4.2 创建 10 个评估问题

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

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

#### 4.3 评估要求

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

#### 4.4 输出格式

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

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

---

# 参考文件

## 📚 文档库

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

### 核心 MCP 文档（首先加载）
- **MCP 协议**：从 `https://modelcontextprotocol.io/sitemap.xml` 的站点地图开始，然后获取带有 `.md` 后缀的特定页面
- [📋 MCP 最佳实践](./reference/mcp_best_practices.md) - 通用 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 实现指南](./reference/python_mcp_server.md) - 完整的 Python/FastMCP 指南，包括：
  - 服务器初始化模式
  - Pydantic 模型示例
  - 使用 `@mcp.tool` 进行工具注册
  - 完整的运行示例
  - 质量检查清单

- [⚡ TypeScript 实现指南](./reference/node_mcp_server.md) - 完整的 TypeScript 指南，包含：
  - 项目结构
  - Zod 模式
  - 使用 `server.registerTool` 进行工具注册
  - 完整的运行示例
  - 质量检查清单

### 评估指南（在第四阶段加载）
- [✅ 评估指南](./reference/evaluation.md) - 完整的评估创建指南，包含：
  - 问题创建指南
  - 答案验证策略
  - XML 格式规范
  - 示例问题和答案
  - 使用提供的脚本运行评估
 文件：reference/mcp_best_practices.md
# MCP 服务器最佳实践

## 快速参考

### 服务器命名
- **Python**：`{服务}_mcp`（例如，`slack_mcp`）
- **Node/TypeScript**：`{服务}-mcp-server`（例如，`slack-mcp-server`）

### 工具命名
- 使用 snake_case 并带有服务前缀
- 格式：`{服务}_{动作}_{资源}`
- 示例：`slack_send_message`，`github_create_issue`

### 响应格式
- 支持 JSON 和 Markdown 格式
- JSON 用于程序化处理
- Markdown 用于人类可读性

### 分页
- 始终遵守 `limit` 参数
- 返回 `has_more`、`next_offset`、`total_count`
- 默认返回 20-50 项

### 传输
- **可流式 HTTP**：适用于远程服务器、多客户端场景
- **stdio**：适用于本地集成、命令行工具
- 避免 SSE（已弃用，转而使用可流式 HTTP）

---

## 服务器命名约定

遵循以下标准化命名模式：

**Python**：使用格式 `{服务}_mcp`（小写带下划线）
- 示例：`slack_mcp`、`github_mcp`、`jira_mcp`

**Node/TypeScript**：使用格式 `{服务}-mcp-server`（小写带连字符）
- 示例：`slack-mcp-server`、`github-mcp-server`、`jira-mcp-server`

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

---

## 工具命名与设计

### 工具命名

1. **使用 snake_case**：`search_users`、`create_project`、`get_channel_info`
2. **包含服务前缀**：预料您的 MCP 服务器可能会与其他 MCP 服务器一起使用
   - 使用 `slack_send_message` 而不是 `send_message`
   - 使用 `github_create_issue` 而不是 `create_issue`
3. **面向动作**：以动词开头（get、list、search、create 等）
4. **具体明确**：避免可能与其他服务器冲突的通用名称

### 工具设计

- 工具描述必须狭义且明确地描述功能
- 描述必须精确匹配实际功能
- 提供工具注解（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
- 保持工具操作专注且原子化

---

## 响应格式

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

### JSON 格式 (`response_format="json"`)
- 机器可读的结构化数据
- 包含所有可用字段和元数据
- 字段名称和类型一致
- 用于程序化处理

### Markdown 格式 (`response_format="markdown"`, 通常为默认)
- 人类可读的格式化文本
- 使用标题、列表和格式化以提高清晰度
- 将时间戳转换为人类可读格式
- 显示带有括号中 ID 的显示名称
- 省略冗长的元数据

---

## 分页

对于列出资源的工具：

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

分页响应示例：
```json
{
  "total": 150,
  "count": 20,
  "offset": 0,
  "items": [...],
  "has_more": true,
  "next_offset": 20
}
```

---

## 传输选项

### 可流式 HTTP

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

**特点**：
- 基于 HTTP 的双向通信
- 支持多个并发客户端
- 可部署为 Web 服务
- 支持服务器到客户端通知

**使用场景**：
- 同时服务多个客户端
- 部署为云服务
- 与 Web 应用程序集成

### stdio

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

**特点**：
- 标准输入/输出流通信
- 设置简单，无需网络配置
- 作为客户端的子进程运行

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

**注意**：stdio 服务器不应将日志输出到 stdout（请使用 stderr 进行日志记录）

### 传输选择

| 标准 | stdio | 可流式 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` | 布尔值 | false | 工具不修改其环境 |
| `destructiveHint` | 布尔值 | true | 工具可能执行破坏性更新 |
| `idempotentHint` | 布尔值 | false | 使用相同参数重复调用没有额外效果 |
| `openWorldHint` | 布尔值 | true | 工具与外部实体交互 |

**重要提示**：注解是提示，不是安全保证。客户端不应仅根据注解做出安全关键的决策。

---

## 错误处理

- 使用标准 JSON-RPC 错误码
- 在结果对象中报告工具错误（而非协议级错误）
- 提供有帮助、具体的错误消息，并给出建议的下一步操作
- 不要暴露内部实现细节
- 错误时正确清理资源

错误处理示例：
```typescript
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 个可读性强的问题
- 问题必须是只读、独立、非破坏性的
- 每个问题需要多次工具调用（可能多达几十次）
- 答案必须是单一的、可验证的值
- 答案必须是稳定的（不会随时间变化）

### 输出格式
```xml
<evaluation>
   <qa_pair>
      <question>你的问题在这里</question>
      <answer>单一可验证的答案</answer>
   </qa_pair>
</evaluation>
```

---

## 评估目的

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

## 评估概述

创建 10 个可读性强的问题，这些问题仅需只读、独立、非破坏性且幂等的操作即可回答。每个问题应：
- 真实
- 清晰简洁
- 无歧义
- 复杂，可能需要几十次工具调用或步骤
- 可用你预先确定的单一可验证值回答

## 问题指南

### 核心要求

1. **问题必须独立**
   - 每个问题不应依赖于任何其他问题的答案
   - 不应假设处理另一个问题时有先前的写入操作

2. **问题必须仅要求非破坏性且幂等的工具使用**
   - 不应指示或要求修改状态以得出正确答案

3. **问题必须真实、清晰、简洁且复杂**
   - 必须要求另一个 LLM 使用多个（可能多达几十个）工具或步骤来回答

### 复杂性和深度

4. **问题必须需要深度探索**
   - 考虑需要多个子问题和顺序工具调用的多跳问题
   - 每一步都应受益于前一个问题中找到的信息

5. **问题可能需要大量翻页**
   - 可能需要翻阅多页结果
   - 可能需要查询旧数据（1-2年前的过时数据）以查找小众信息
   - 问题必须是**困难的**

6. **问题必须需要深度理解**
   - 而不是表面知识
   - 可以将复杂概念作为需要证据的真/假问题提出
   - 可以使用多项选择格式，其中LLM必须搜索不同的假设

7. **问题不能通过直接的关键词搜索解决**
   - 不要包含目标内容中的特定关键词
   - 使用同义词、相关概念或释义
   - 需要多次搜索，分析多个相关项，提取上下文，然后得出答案

### 工具测试

8. **问题应压力测试工具的返回值**
   - 可能引发工具返回大型JSON对象或列表，使LLM不堪重负
   - 应要求理解多种数据模态：
     - ID和名称
     - 时间戳和日期时间（月、日、年、秒）
     - 文件ID、名称、扩展名和MIME类型
     - URL、GID等
   - 应探究工具返回所有有用数据形式的能力

9. **问题应**主要**反映真实的人类使用场景**
   - 人类在LLM辅助下会关心的信息检索任务类型

10. **问题可能需要数十次工具调用**
    - 这对上下文有限的LLM构成挑战
    - 鼓励MCP服务器工具减少返回的信息

11. **包含模糊问题**
    - 可能模糊不清，或者需要对调用哪些工具做出艰难决定
    - 迫使 LLM 可能会犯错或误解
    - 确保尽管存在模糊性，但仍然只有一个可验证的答案

### 稳定性

12. **问题必须设计成答案不会改变**
    - 不要问依赖于动态“当前状态”的问题
    - 例如，不要统计：
      - 帖子收到的反应数量
      - 线程的回复数量
      - 频道中的成员数量

13. **不要让 MCP 服务器限制你创建的问题类型**
    - 创建具有挑战性和复杂性的问题
    - 有些可能无法通过可用的 MCP 服务器工具解决
    - 问题可能需要特定的输出格式（日期时间 vs. epoch 时间，JSON vs. MARKDOWN）
    - 问题可能需要数十次工具调用才能完成

## 答案指南

### 验证

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

### 可读性

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

### 稳定性

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

4. **答案必须清晰且明确**
   - 问题必须设计成只有一个明确的答案
   - 答案可以通过使用 MCP 服务器工具得出

### 多样性

5. **答案必须是多样的**
   - 答案应该是一个单一的可验证值，以多种模态和格式呈现
   - 用户概念：用户 ID、用户名、显示名称、名字、姓氏、电子邮件地址、电话号码
   - 频道概念：频道 ID、频道名称、频道主题
   - 消息概念：消息 ID、消息字符串、时间戳、月份、日期、年份

6. **答案不得是复杂结构**
   - 不是值的列表
   - 不是复杂的对象
   - 不是 ID 或字符串的列表
   - 不是自然语言文本
   - 除非答案可以通过直接字符串比较轻松验证
   - 并且可以实际重现
   - LLM 不太可能以任何其他顺序或格式返回相同的列表

## 评估过程

### 步骤 1：文档检查

阅读目标 API 的文档以理解：
- 可用的端点和功能
- 如果存在歧义，从网上获取额外信息
- 尽可能并行化此步骤
- 确保每个子代理只检查文件系统或网络上的文档

### 步骤 2：工具检查

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

### 步骤 3：建立理解

重复步骤 1 和 2，直到你有了很好的理解：
- 多次迭代
- 思考你想创建的任务类型
- 完善你的理解
- 在任何阶段都不应阅读 MCP 服务器实现的代码本身
- 运用你的直觉和理解来创建合理、现实但极具挑战性的任务

### 步骤 4：只读内容检查

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

### 步骤 5：任务生成

检查内容后，创建10个可读的问题：
- LLM应该能够使用MCP服务器回答这些问题
- 遵循上述所有问答指南

## 输出格式

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

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

## 评估示例

### 好的问题

**示例1：需要深度探索的多跳问题（GitHub MCP）**
```xml
<qa_pair>
   <question>找到在2023年第三季度归档的、此前是组织中被fork最多的项目。该存储库使用的主要编程语言是什么？</question>
   <answer>Python</answer>
</qa_pair>
```

这个问题很好，因为：
- 需要多次搜索才能找到已归档的仓库
- 需要识别在归档前哪个仓库的 fork 最多
- 需要检查仓库详情以获取语言信息
- 答案是一个简单、可验证的值
- 基于历史（已关闭）数据，不会改变

**示例 2：需要理解上下文而非关键词匹配（项目管理 MCP）**
```xml
<qa_pair>
   <question>找到那个专注于改进客户入职流程、于 2023 年底完成的倡议。项目负责人完成后创建了一份回顾性文档。当时该负责人的职位是什么？</question>
   <answer>产品经理</answer>
</qa_pair>
```

这个问题很好，因为：
- 没有使用具体的项目名称（“专注于改进客户入职流程的倡议”）
- 需要找到特定时间段内已完成的项目
- 需要识别项目负责人及其角色
- 需要从回顾性文档中理解上下文
- 答案是人类可读且稳定的
- 基于已完成的工作（不会改变）

**示例 3：需要多步骤的复杂聚合（问题追踪器 MCP）**
```xml
<qa_pair>
   <question>在 2024 年 1 月报告的所有被标记为“严重”优先级的 bug 中，哪个经办人在 48 小时内解决了其分配到的 bug 的最高百分比？请提供该经办人的用户名。</question>
   <answer>alex_eng</qa_pair>
```

这个问题很好，因为：
- 需要按日期、优先级和状态筛选 bug
- 需要按经办人分组并计算解决率
- 需要理解时间戳以确定 48 小时窗口
- 测试分页（可能需要处理大量 bug）
- 答案是单个用户名
- 基于特定时间段的历史数据

**示例 4：需要跨多种数据类型（CRM MCP）进行综合分析**
```xml
<qa_pair>
   <question>找出在 2023 年第四季度从 Starter 计划升级到 Enterprise 计划且年度合同价值最高的客户。该客户属于哪个行业？</question>
   <answer>医疗保健</answer>
</qa_pair>
```

这个问题很好，因为：
- 需要理解订阅层级变化
- 需要识别特定时间范围内的升级事件
- 需要比较合同价值
- 必须获取客户行业信息
- 答案简单且可验证
- 基于已完成的历史交易

### 差的问题

**示例 1：答案随时间变化**
```xml
<qa_pair>
   <question>目前分配给工程团队的未解决问题有多少个？</question>
   <answer>47</answer>
</qa_pair>
```

这个问题很差，因为：
- 随着问题的创建、关闭或重新分配，答案会发生变化
- 不是基于稳定/静态数据
- 依赖于动态的“当前状态”

**示例 2：通过关键词搜索太容易**
```xml
<qa_pair>
   <question>找出标题为“添加身份验证功能”的拉取请求，并告诉我它的创建者是谁。</question>
   <answer>developer123</answer>
</qa_pair>
```

这个问题很差，因为：
- 可以通过对精确标题进行直接关键词搜索来解决
- 不需要深入探索或理解
- 不需要综合或分析

**示例 3：答案格式模糊**
```xml
<qa_pair>
   <question>列出所有以 Python 作为主要语言的仓库。</question>
   <answer>repo1, repo2, repo3, data-pipeline, ml-tools</answer>
</qa_pair>
```

这个问题很糟糕，因为：
- 答案是一个列表，可以以任何顺序返回
- 难以通过直接字符串比较进行验证
- LLM 可能会以不同方式格式化（JSON 数组、逗号分隔、换行符分隔）
- 最好询问特定的聚合（计数）或最高级（最多星级）

## 验证过程

创建评估后：

1. **检查 XML 文件**以了解其架构
2. **加载每个任务指令**，并使用 MCP 服务器和工具并行地尝试自己解决任务，从而确定正确答案
3. **标记任何需要写入或破坏性操作**的操作
4. **累积所有正确答案**并替换文档中的任何错误答案
5. **删除任何需要写入或破坏性操作**的 `<qa_pair>`

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

## 创建高质量评估的技巧

1. 在生成任务之前**认真思考并提前计划**
2. **在机会出现时进行并行处理**以加快流程并管理上下文
3. **专注于人类实际想要完成的现实用例**
4. **创建具有挑战性的问题**来测试 MCP 服务器能力的极限
5. 通过使用历史数据和封闭概念**确保稳定性**
6. 通过使用 MCP 服务器工具自己解决问题来**验证答案**
7. 根据在此过程中学到的知识**迭代和完善**

---

# 运行评估

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

## 设置

1. **安装依赖项**

   ```bash
   pip install -r scripts/requirements.txt
   ```

或者手动安装：
```bash
pip install anthropic mcp
```

2. **设置 API 密钥**

```bash
export ANTHROPIC_API_KEY=your_api_key_here
```

## 评估文件格式

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

```xml
<evaluation>
   <qa_pair>
      <question>查找 2024 年第二季度创建的、已完成任务数量最多的项目。项目名称是什么？</question>
      <answer>网站重新设计</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索 2024 年 3 月关闭的、标记为“bug”的问题。哪个用户关闭的问题最多？提供他们的用户名。</question>
      <answer>sarah_dev</answer>
   </qa_pair>
</evaluation>
```

## 运行评估

评估脚本 (`scripts/evaluation.py`) 支持三种传输类型：

**重要提示：**
- **stdio 传输**：评估脚本会自动为您启动和管理 MCP 服务器进程。请勿手动运行服务器。
- **sse/http 传输**：您必须在运行评估之前单独启动 MCP 服务器。脚本会连接到指定 URL 处已运行的服务器。

### 1. 本地 STDIO 服务器

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

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

使用环境变量：
```bash
python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a my_mcp_server.py \
  -e API_KEY=abc123 \
  -e DEBUG=true \
  evaluation.xml
```

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

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

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

### 3. HTTP (可流式 HTTP)

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

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

## 命令行选项

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

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

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

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

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

## 输出

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

- **摘要统计**:
  - 准确率 (正确/总数)
  - 平均任务持续时间
  - 每个任务的平均工具调用次数
  - 总工具调用次数

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

### 将报告保存到文件

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

## 完整示例工作流

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

1. **创建评估文件** (`my_evaluation.xml`)：

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

2. **安装依赖项**：

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

3. **运行评估**：

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

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

## 故障排除

### 连接错误

如果遇到连接错误：
- **STDIO**：验证命令和参数是否正确
- **SSE/HTTP**：检查 URL 是否可访问以及请求头是否正确
- 确保所有必需的 API 密钥已在环境变量或请求头中设置

### 准确率低

如果许多评估失败：
- 审查代理对每个任务的反馈
- 检查工具描述是否清晰全面
- 验证输入参数是否文档齐全
- 考虑工具返回的数据是过多还是过少
- 确保错误消息是可操作的

### 超时问题

如果任务超时：
- 使用更强大的模型（例如 `claude-3-7-sonnet-20250219`）
- 检查工具是否返回过多数据
- 验证分页是否正常工作
- 考虑简化复杂问题
 FILE:reference/node_mcp_server.md 
# Node/TypeScript MCP 服务器实现指南

## 概述

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

---

## 快速参考

### 关键导入
```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import express from "express";
import { z } from "zod";
```

### 服务器初始化
```typescript
const server = new McpServer({
  name: "service-mcp-server",
  version: "1.0.0"
});
```

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

---

## MCP TypeScript SDK

官方 MCP TypeScript SDK 提供：
- `McpServer` 类用于服务器初始化
- `registerTool` 方法用于工具注册
- Zod schema 集成用于运行时输入验证
- 类型安全的工具处理程序实现

**重要提示 - 仅使用现代 API：**
- **请使用**：`server.registerTool()`、`server.registerResource()`、`server.registerPrompt()`
- **请勿使用**：旧的已弃用 API，例如 `server.tool()`、`server.setRequestHandler(ListToolsRequestSchema, ...)` 或手动处理程序注册
- `register*` 方法提供更好的类型安全性、自动 schema 处理，并且是推荐的方法

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

## 服务器命名约定

Node/TypeScript MCP 服务器必须遵循以下命名模式：
- **格式**：`{service}-mcp-server`（小写，带连字符）
- **示例**：`github-mcp-server`、`jira-mcp-server`、`stripe-mcp-server`

名称应：
- 具有通用性（不与特定功能绑定）
- 描述所集成的服务/API
- 易于从任务描述中推断
- 不包含版本号或日期

## 项目结构

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

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

## 工具实现

### 工具命名

工具名称使用 snake_case（例如，“search_users”、“create_project”、“get_channel_info”），名称要清晰、面向动作。

**避免命名冲突**：包含服务上下文以防止重叠：
- 使用 "slack_send_message" 而不是 "send_message"
- 使用 "github_create_issue" 而不是 "create_issue"
- 使用 "asana_list_tasks" 而不是 "list_tasks"

### 工具结构

工具使用 `registerTool` 方法注册，并满足以下要求：
- 使用 Zod 模式进行运行时输入验证和类型安全
- 必须明确提供 `description` 字段 - JSDoc 注释不会自动提取
- 明确提供 `title`、`description`、`inputSchema` 和 `annotations`
- `inputSchema` 必须是 Zod 模式对象（而不是 JSON 模式）
- 明确指定所有参数和返回值的类型

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

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

```

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

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

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

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

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

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

示例：
  - 使用场景："查找所有市场团队成员" -> 参数中 query="team:marketing"
  - 使用场景："搜索 John 的账户" -> 参数中 query="john"
  - 不使用场景：你需要创建一个用户 (请改用 example_create_user)

错误处理：
  - 如果请求过多 (429 状态)，返回 "Error: Rate limit exceeded"
  - 如果搜索返回空，返回 "No users found matching '<query>'"`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params: UserSearchInput) => {
    try {
      // 输入验证由 Zod 架构处理
      // 使用已验证的参数发起 API 请求
      const data = await makeApiRequest<any>(
        "users/search",
        "GET",
        undefined,
        {
          q: params.query,
          limit: params.limit,
          offset: params.offset
        }
      );

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

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

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

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

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

## Zod 输入验证模式

Zod 提供运行时类型验证：

```typescript
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")
});
```

## 响应格式选项

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

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

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

**Markdown 格式**：
- 使用标题、列表和格式化以提高清晰度
- 将时间戳转换为人类可读的格式
- 显示带有括号中 ID 的显示名称
- 省略冗长的元数据
- 逻辑地分组相关信息

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

## 分页实现

对于列出资源的工具：

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

async function listItems(params: z.infer<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 常量以防止响应过大：

```typescript
// 在 constants.ts 模块级别
export const CHARACTER_LIMIT = 25000;  // 最大响应大小（字符数）

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

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

  return result;
}
```

## 错误处理

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

```typescript
import axios, { AxiosError } from "axios";

```

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

## 共享工具函数

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

```typescript
// 共享的 API 请求函数
async function makeApiRequest<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：

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

```

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

## TypeScript 最佳实践

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

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

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

type User = z.infer<typeof UserSchema>;

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

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

## 包配置

### package.json

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

### tsconfig.json

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

## 完整示例

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

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

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

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

```

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

type UserSearchInput = z.infer<typeof UserSearchInputSchema>;

// 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 "错误：未找到资源。请检查 ID 是否正确。";
        case 403:
          return "错误：权限被拒绝。您无权访问此资源。";
        case 429:
          return "错误：超出速率限制。请稍后再发起请求。";
        default:
          return `错误：API 请求失败，状态码为 ${error.response.status}`;
      }
    } else if (error.code === "ECONNABORTED") {
      return "错误：请求超时。请重试。";
    }
  }
  return `错误：发生意外错误：${error instanceof Error ? error.message : String(error)}`;
}

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

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

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

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

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

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

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

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

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

---

## 高级 MCP 功能

### 资源注册

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

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

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

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

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

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

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

### 传输选项

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

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

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

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

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

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

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

app.listen(3000);
```

#### stdio（用于本地集成）

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

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

**传输选择：**
- **可流式 HTTP**：Web 服务、远程访问、多客户端
- **stdio**：命令行工具、本地开发、子进程集成

### 通知支持

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

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

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

谨慎使用通知——仅当服务器功能确实改变时才使用。

---

## 代码最佳实践

### 代码可组合性和可重用性

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

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

2. **避免重复**：
   - 绝不要在工具之间复制粘贴相似的代码
   - 如果你发现自己编写了两次相似的逻辑，请将其提取到一个函数中
   - 分页、过滤、字段选择和格式化等常见操作应共享
   - 认证/授权逻辑应集中化

## 构建和运行

在运行之前，请务必构建你的 TypeScript 代码：

```bash
# 构建项目
npm run build

# 运行服务器
npm start

# 带有自动重载的开发模式
npm run dev
```

在认为实现完成之前，请务必确保 `npm run build` 成功完成。

## 质量清单

在最终确定你的 Node/TypeScript MCP 服务器实现之前，请确保：

### 战略设计
- [ ] 工具能够实现完整的工作流，而不仅仅是 API 端点封装
- [ ] 工具名称反映自然的任务细分
- [ ] 响应格式优化了代理上下文效率
- [ ] 在适当的地方使用人类可读的标识符
- [ ] 错误消息引导代理正确使用

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

### TypeScript 质量
- [ ] 为所有数据结构定义了 TypeScript 接口
- [ ] tsconfig.json 中启用了严格的 TypeScript
- [ ] 不使用 `any` 类型 - 改用 `unknown` 或适当的类型
- [ ] 所有异步函数都具有明确的 Promise<T> 返回类型
- [ ] 错误处理使用适当的类型守卫（例如，`axios.isAxiosError`, `z.ZodError`）

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

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

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

### 测试和构建
- [ ] `npm run build` 成功完成，无错误
- [ ] dist/index.js 已创建且可执行
- [ ] 服务器运行：`node dist/index.js --help`
- [ ] 所有导入都正确解析
- [ ] 示例工具调用按预期工作
文件：reference/python_mcp_server.md
# Python MCP 服务器实现指南

## 概述

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

---

## 快速参考

### 关键导入
```python
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
```

### 服务器初始化
```python
mcp = FastMCP("service_mcp")
```

### 工具注册模式
```python
@mcp.tool(name="tool_name", annotations={...})
async def tool_function(params: InputModel) -> str:
    # 实现
    pass
```

---

## MCP Python SDK 和 FastMCP

官方 MCP Python SDK 提供了 FastMCP，一个用于构建 MCP 服务器的高级框架。它提供：
- 从函数签名和文档字符串自动生成描述和 inputSchema
- Pydantic 模型集成用于输入验证
- 基于装饰器的工具注册，使用 `@mcp.tool`

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

## 服务器命名约定

Python MCP 服务器必须遵循以下命名模式：
- **格式**：`{service}_mcp`（小写，带下划线）
- **示例**：`github_mcp`、`jira_mcp`、`stripe_mcp`

名称应：
- 泛化（不与特定功能绑定）
- 描述所集成的服务/API
- 易于从任务描述中推断
- 不包含版本号或日期

## 工具实现

### 工具命名

工具名称使用 snake_case（例如，“search_users”、“create_project”、“get_channel_info”），名称应清晰、面向动作。

**避免命名冲突**：包含服务上下文以防止重叠：
- 使用“slack_send_message”而不是“send_message”
- 使用“github_create_issue”而不是“create_issue”
- 使用“asana_list_tasks”而不是“list_tasks”

### 使用 FastMCP 的工具结构

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

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

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

```

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

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

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

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

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

返回：
        str：JSON 格式的响应，包含操作结果
    '''
    # 在此实现
    pass
```

## Pydantic v2 主要特性

- 使用 `model_config` 代替嵌套的 `Config` 类
- 使用 `field_validator` 代替已弃用的 `validator`
- 使用 `model_dump()` 代替已弃用的 `dict()`
- 验证器需要 `@classmethod` 装饰器
- 验证器方法需要类型提示

```python
from pydantic import BaseModel, Field, field_validator, ConfigDict

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

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

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

## 响应格式选项

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

```python
from enum import Enum

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

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

**Markdown 格式**:
- 使用标题、列表和格式以提高清晰度
- 将时间戳转换为人类可读的格式（例如，“2024-01-15 10:30:00 UTC”而不是 epoch）
- 显示带有 ID 的显示名称（例如，“@john.doe (U123456)”）
- 省略详细的元数据（例如，只显示一个个人资料图片 URL，而不是所有尺寸）
- 逻辑地分组相关信息

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

## 分页实现

对于列出资源的工具：

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

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

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

## 错误处理

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

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

## 共享工具函数

将通用功能提取到可复用函数中：

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

## Async/Await 最佳实践

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

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

```

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

## 类型提示

全程使用类型提示：

```python
from typing import Optional, List, Dict, Any

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

## 工具 Docstrings

每个工具都必须有全面的 docstrings，并包含明确的类型信息：

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

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

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

    Returns:
        str: 包含搜索结果的 JSON 格式字符串，其 schema 如下：

```

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

        错误响应：
        "Error: <错误信息>" 或 "No users found matching '<查询>'"

    示例：
        - 使用时机："查找所有市场团队成员" -> params 中 query="team:marketing"
        - 使用时机："搜索 John 的账户" -> params 中 query="john"
        - 不使用时机：你需要创建一个用户（请改用 example_create_user）
        - 不使用时机：你有一个用户ID并需要完整详细信息（请改用 example_get_user）

    错误处理：
        - 输入验证错误由 Pydantic 模型处理
        - 如果请求过多（429 状态），返回 "Error: Rate limit exceeded"
        - 如果 API 密钥无效（401 状态），返回 "Error: Invalid API authentication"
        - 返回格式化的结果列表或 "No users found matching 'query'"
    '''
```

## 完整示例

请参阅下面的完整 Python MCP 服务器示例：

```python
#!/usr/bin/env python3
'''
Example 服务的 MCP 服务器。

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

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

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

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

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

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

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

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("查询不能为空或只包含空格")
        return v.strip()

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

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

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

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

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

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

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

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

            return "\n".join(lines)

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

    except Exception as e:
        return _handle_api_error(e)

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

---

## FastMCP 高级功能

### 上下文参数注入

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

```python
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("example_mcp")

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

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

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

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

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

    return format_results(results)

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

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

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

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

### 资源注册

将数据作为资源公开，以实现高效的、基于模板的访问：

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

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

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

**何时使用资源（Resources）与工具（Tools）：**
- **资源（Resources）**：用于通过简单参数（URI 模板）进行数据访问
- **工具（Tools）**：用于包含验证和业务逻辑的复杂操作

### 结构化输出类型

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

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

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

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

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

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

### 生命周期管理

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

```python
from contextlib import asynccontextmanager

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

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

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

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

```

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

### 传输选项

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

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

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

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

---

## 代码最佳实践

### 代码可组合性和可重用性

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

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

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

### Python 特定的最佳实践

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

## 质量清单

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

### 战略设计
- [ ] 工具能够实现完整的工作流，而不仅仅是 API 端点封装。
- [ ] 工具名称反映自然的任务划分。
- [ ] 响应格式优化了代理上下文效率。
- [ ] 在适当的地方使用人类可读的标识符。
- [ ] 错误消息能够引导代理正确使用。

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

### 工具配置
- [ ] 所有工具都在装饰器中实现了“name”和“annotations”
- [ ] 注解设置正确（readOnlyHint, destructiveHint, idempotentHint, openWorldHint）
- [ ] 所有工具都使用 Pydantic BaseModel 进行输入验证，并带有 Field() 定义
- [ ] 所有 Pydantic Field 都具有明确的类型和带有约束的描述
- [ ] 所有工具都具有全面的文档字符串，并带有明确的输入/输出类型
- [ ] 文档字符串包含 dict/JSON 返回的完整 schema 结构
- [ ] Pydantic 模型处理输入验证（无需手动验证）

### 高级功能（如适用）
- [ ] 上下文注入用于日志记录、进度或信息获取
- [ ] 为适当的数据端点注册资源
- [ ] 为持久连接实现生命周期管理
- [ ] 使用结构化输出类型（TypedDict, Pydantic 模型）
- [ ] 配置适当的传输方式（stdio 或可流式 HTTP）

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

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

from abc import ABC, abstractmethod
from contextlib import AsyncExitStack
from typing import Any

```python
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"意外的上下文结果: {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):
    """使用服务器发送事件 (Server-Sent Events) 的 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 连接的工厂函数。

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

    Returns:
        MCPConnection 实例
    """
    transport = transport.lower()

    if transport == "stdio":
        if not command:
            raise ValueError("stdio 传输需要 command")
        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'")
 文件: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助手。

当你接到任务时，你必须：
1. 使用可用的工具完成任务
2. 提供你方法中每个步骤的摘要，用<summary>标签包裹
3. 提供对所提供工具的反馈，用<feedback>标签包裹
4. 提供你的最终回复，用<response>标签包裹

摘要要求：
- 在你的<summary>标签中，你必须解释：
  - 你为完成任务所采取的步骤
  - 你使用了哪些工具，按什么顺序，以及为什么
  - 你提供给每个工具的输入
  - 你从每个工具收到的输出
  - 你是如何得出回复的总结

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

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

def parse_evaluation_file(file_path: Path) -> list[dict[str, Any]]:
    """解析包含 qa_pair 元素的 XML 评估文件。"""
    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"解析评估文件 {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"执行工具 {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_index + 1}: 正在运行任务，问题是: {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}秒
- **每个任务的平均工具调用次数**: {average_tool_calls:.2f}
- **总工具调用次数**: {total_tool_calls}

---
"""

TASK_TEMPLATE = """
### 任务 {task_num}

**问题**: {question}
**参考答案**: `{expected_answer}`
**实际答案**: `{actual_answer}`
**正确**: {correct_indicator}
**时长**: {total_duration:.2f}秒
**工具调用**: {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' 格式的 header 字符串解析为字典。"""
    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_group.add_argument("-c", "--command", help="运行 MCP 服务器的命令（仅限标准输入输出）")
    stdio_group.add_argument("-a", "--args", nargs="+", help="命令的参数（仅限标准输入输出）")
    stdio_group.add_argument("-e", "--env", nargs="+", help="环境变量，格式为 KEY=VALUE（仅限标准输入输出）")

    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="HTTP 头，格式为 'Key: Value'（仅限 SSE/HTTP）")

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

    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 
<evaluation>
   <qa_pair>
      <question>计算以5%年利率投资10,000美元，每月复利，为期3年的复利。最终金额（美元，四舍五入到小数点后2位）是多少？</question>
      <answer>11614.72</answer>
   </qa_pair>
   <qa_pair>
      <question>一个弹丸以45度角发射，初始速度为50米/秒。假设g=9.8米/秒²，计算2秒后它从发射点移动的总距离（米）。四舍五入到小数点后2位。</question>
      <answer>87.25</answer>
   </qa_pair>
   <qa_pair>
      <question>一个球体的体积是500立方米。计算其表面积（平方米）。四舍五入到小数点后2位。</question>
      <answer>304.65</answer>
   </qa_pair>
   <qa_pair>
      <question>计算此数据集的总体标准差：[12, 15, 18, 22, 25, 30, 35]。四舍五入到小数点后2位。</question>
      <answer>7.61</answer>
   </qa_pair>
   <qa_pair>
      <question>计算氢离子浓度为3.5 × 10^-5 M的溶液的pH值。四舍五入到小数点后2位。</question>
      <answer>4.46</answer>
   </qa_pair>
</evaluation>
 FILE:scripts/requirements.txt 
anthropic>=0.39.0
mcp>=1.1.0