Skip to content

7.6 MCP Builder Skill - 构建 MCP 服务器

概述

MCP Builder Skill 是一个专门用于指导开发高质量 MCP (Model Context Protocol) 服务器的技能。MCP 是 Anthropic 推出的模型上下文协议,它允许 LLM 通过精心设计的工具与外部服务进行交互。这个 Skill 帮助开发者创建能够让 AI 真正完成实际任务的 MCP 服务器。

核心理念

MCP 服务器的质量不在于它实现了多少 API 端点,而在于它能多好地帮助 LLM 完成真实世界的任务。一个优秀的 MCP 服务器应该:

  • 提供清晰、描述性的工具名称
  • 返回简洁、相关的数据
  • 给出可操作的错误消息
  • 支持分页和过滤

适用场景

MCP Builder Skill 适用于以下场景:

  1. 集成外部 API:将第三方服务(如 GitHub、Slack、Jira)封装为 MCP 工具
  2. 构建本地工具服务器:创建文件操作、数据库查询等本地工具
  3. 开发远程服务:部署可多客户端访问的 MCP 服务
  4. 扩展 AI 能力:为 Claude 等 AI 模型添加新的工具能力

开发流程四阶段

MCP Builder Skill 定义了一个标准化的四阶段开发流程:

+------------------+     +------------------+     +------------------+     +------------------+
|   Phase 1        |     |   Phase 2        |     |   Phase 3        |     |   Phase 4        |
|   深度研究与规划  | --> |   实现           | --> |   审查与测试     | --> |   创建评估       |
+------------------+     +------------------+     +------------------+     +------------------+
| - 理解MCP设计    |     | - 项目结构       |     | - 代码质量审查   |     | - 创建10个问题   |
| - 学习协议文档   |     | - 核心基础设施   |     | - 构建测试       |     | - 验证答案       |
| - 研究框架文档   |     | - 实现工具       |     | - MCP Inspector  |     | - 输出XML格式    |
| - 规划实现方案   |     | - 错误处理       |     | - 质量检查表     |     | - 运行评估       |
+------------------+     +------------------+     +------------------+     +------------------+

Phase 1: 深度研究与规划

在开始编码之前,需要深入理解以下概念:

API 覆盖 vs 工作流工具

  • 全面的 API 端点覆盖给予代理灵活性
  • 专门的工作流工具对特定任务更方便
  • 当不确定时,优先选择全面的 API 覆盖

工具命名与可发现性

  • 使用清晰、描述性的工具名称
  • 使用一致的前缀(如 github_create_issue, slack_send_message
  • 使用动作导向的命名

上下文管理

  • 设计返回聚焦、相关数据的工具
  • 支持过滤和分页结果
  • 保持工具描述简洁

Phase 2: 实现

支持两种主要的编程语言:

特性TypeScriptPython
框架MCP TypeScript SDKFastMCP
验证库ZodPydantic
命名约定service-mcp-serverservice_mcp
示例github-mcp-servergithub_mcp

Phase 3: 审查与测试

使用 MCP Inspector 测试服务器:

bash
# TypeScript
npx @modelcontextprotocol/inspector

# Python
python -m py_compile your_server.py

Phase 4: 创建评估

评估是验证 MCP 服务器质量的关键步骤,通过创建 10 个复杂问题来测试 LLM 是否能有效使用你的工具。

TypeScript 实现指南

项目结构

service-mcp-server/
|-- package.json
|-- tsconfig.json
|-- README.md
|-- src/
|   |-- index.ts          # 主入口,McpServer 初始化
|   |-- types.ts          # TypeScript 类型定义
|   |-- tools/            # 工具实现(每个领域一个文件)
|   |-- services/         # API 客户端和共享工具
|   |-- schemas/          # Zod 验证模式
|   |-- constants.ts      # 共享常量
|-- dist/                 # 构建输出目录

服务器初始化

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

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

工具注册模式

typescript
// Zod schema 用于输入验证
const UserSearchInputSchema = z.object({
  query: z.string()
    .min(2, "查询至少2个字符")
    .max(200, "查询不超过200字符")
    .describe("搜索字符串,匹配名称/邮箱"),
  limit: z.number()
    .int()
    .min(1)
    .max(100)
    .default(20)
    .describe("最大返回结果数"),
  offset: z.number()
    .int()
    .min(0)
    .default(0)
    .describe("分页偏移量"),
  response_format: z.nativeEnum(ResponseFormat)
    .default(ResponseFormat.MARKDOWN)
    .describe("输出格式: 'markdown' 或 'json'")
}).strict();

// 注册工具
server.registerTool(
  "example_search_users",
  {
    title: "搜索示例用户",
    description: `在 Example 系统中按名称、邮箱或团队搜索用户。

    此工具搜索所有用户资料,支持部分匹配和各种搜索过滤器。
    它不会创建或修改用户,只搜索现有用户。

    Args:
      - query (string): 搜索字符串
      - limit (number): 最大结果数 1-100 (默认: 20)
      - offset (number): 分页偏移量 (默认: 0)
      - response_format: 输出格式 (默认: 'markdown')

    Returns:
      JSON 格式包含 total, count, offset, users, has_more 等字段`,
    inputSchema: UserSearchInputSchema,
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params) => {
    // 实现逻辑
    return {
      content: [{ type: "text", text: result }],
      structuredContent: output
    };
  }
);

传输选项

typescript
// stdio 传输(本地工具)
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

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

// Streamable HTTP 传输(远程服务器)
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";

const app = express();
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);
});

Python 实现指南

服务器初始化

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

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

Pydantic 模型定义

python
class ResponseFormat(str, Enum):
    """输出格式选项"""
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    """用户搜索操作的输入模型"""
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True,
        extra='forbid'
    )

    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()

工具注册模式

python
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "搜索示例用户",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    """在 Example 系统中按名称、邮箱或团队搜索用户。

    此工具搜索所有用户资料,支持部分匹配和各种搜索过滤器。

    Args:
        params: 验证后的输入参数

    Returns:
        str: JSON 格式的搜索结果
    """
    try:
        data = await _make_api_request("users/search", params={
            "q": params.query,
            "limit": params.limit,
            "offset": params.offset
        })

        if params.response_format == ResponseFormat.MARKDOWN:
            return format_markdown(data)
        else:
            return json.dumps(data, indent=2)

    except Exception as e:
        return _handle_api_error(e)

高级特性:上下文注入

python
from mcp.server.fastmcp import FastMCP, Context

@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
    """带上下文访问的高级工具"""

    # 报告进度
    await ctx.report_progress(0.25, "开始搜索...")

    # 记录日志
    await ctx.log_info("处理查询", {"query": query})

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

    return format_results(results)

最佳实践

工具命名约定

约定示例
使用 snake_casesearch_users, create_project
包含服务前缀slack_send_message, github_create_issue
动作导向get, list, search, create, update, delete
避免冲突不要用 send_message,用 slack_send_message

响应格式

支持两种格式以适应不同场景:

JSON 格式 (response_format="json")

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

Markdown 格式 (response_format="markdown")

  • 人类可读的格式化文本
  • 使用标题、列表等格式
  • 将时间戳转换为可读格式
  • 显示带ID的显示名称

分页实现

python
# 分页响应结构
{
    "total": 150,        # 总数
    "count": 20,         # 当前页数量
    "offset": 0,         # 当前偏移
    "items": [...],      # 数据项
    "has_more": True,    # 是否有更多
    "next_offset": 20    # 下一页偏移
}

错误处理

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

python
def _handle_api_error(e: Exception) -> str:
    """一致的错误格式化"""
    if isinstance(e, httpx.HTTPStatusError):
        status = e.response.status_code
        if status == 404:
            return "错误: 资源未找到。请检查 ID 是否正确。"
        elif status == 403:
            return "错误: 权限被拒绝。您没有访问此资源的权限。"
        elif status == 429:
            return "错误: 速率限制。请稍后再试。"
        return f"错误: API 请求失败,状态码 {status}"
    return f"错误: 发生意外错误: {type(e).__name__}"

工具注解

注解类型默认值描述
readOnlyHintbooleanfalse工具不修改环境
destructiveHintbooleantrue工具可能执行破坏性更新
idempotentHintbooleanfalse重复调用无额外效果
openWorldHintbooleantrue工具与外部实体交互

传输选择指南

+------------------+------------------+
|     stdio        |  Streamable HTTP |
+------------------+------------------+
| 本地集成         | 远程服务器       |
| 命令行工具       | Web 服务         |
| 单用户           | 多客户端         |
| 简单设置         | 中等复杂度       |
| 无网络配置       | 需要网络配置     |
+------------------+------------------+

选择建议:

  • stdio:本地开发环境、桌面应用集成、单用户场景
  • Streamable HTTP:云服务部署、多客户端访问、Web 应用集成

评估系统

评估目的

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

创建评估问题的要求

  1. 独立性:每个问题不依赖其他问题
  2. 只读:只需要非破坏性操作
  3. 复杂性:需要多个工具调用
  4. 真实性:基于真实用例
  5. 可验证:单一、明确的答案
  6. 稳定性:答案不会随时间变化

评估文件格式

xml
<evaluation>
   <qa_pair>
      <question>找到2024年Q2创建的任务完成数最多的项目。
      项目名称是什么?</question>
      <answer>Website Redesign</answer>
   </qa_pair>
   <qa_pair>
      <question>搜索2024年3月关闭的标记为"bug"的问题。
      哪个用户关闭的问题最多?提供用户名。</question>
      <answer>sarah_dev</answer>
   </qa_pair>
</evaluation>

运行评估

bash
# 安装依赖
pip install anthropic mcp

# 设置 API Key
export ANTHROPIC_API_KEY=your_api_key

# 运行 stdio 评估
python scripts/evaluation.py \
  -t stdio \
  -c python \
  -a my_mcp_server.py \
  -e API_KEY=abc123 \
  evaluation.xml

# 运行 HTTP 评估
python scripts/evaluation.py \
  -t http \
  -u https://example.com/mcp \
  -H "Authorization: Bearer token123" \
  evaluation.xml

评估报告内容

  • 准确率(正确/总数)
  • 平均任务时长
  • 平均工具调用次数
  • 每个任务的详细结果

质量检查表

策略设计

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

实现质量

  • [ ] 实现最重要和有价值的工具
  • [ ] 所有工具使用 registerTool/@mcp.tool 注册
  • [ ] 所有工具包含 title、description、inputSchema、annotations
  • [ ] 使用 Zod/Pydantic 进行运行时输入验证
  • [ ] 描述包含返回值示例和完整模式文档
  • [ ] 错误消息清晰、可操作

代码质量

  • [ ] 分页正确实现
  • [ ] 大响应检查字符限制并截断
  • [ ] 提供过滤选项
  • [ ] 网络操作正确处理超时和连接错误
  • [ ] 公共功能提取为可复用函数

测试

  • [ ] 构建成功完成
  • [ ] 服务器可以运行
  • [ ] 示例工具调用按预期工作

实际应用示例

示例 1: GitHub MCP 服务器

typescript
// github-mcp-server

server.registerTool(
  "github_search_issues",
  {
    title: "搜索 GitHub Issues",
    description: "在仓库中搜索 issues...",
    inputSchema: z.object({
      repo: z.string().describe("仓库名 owner/repo"),
      query: z.string().describe("搜索查询"),
      state: z.enum(["open", "closed", "all"]).default("open"),
      limit: z.number().int().min(1).max(100).default(20)
    }),
    annotations: {
      readOnlyHint: true,
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true
    }
  },
  async (params) => {
    // 实现搜索逻辑
  }
);

示例 2: Slack MCP 服务器

python
# slack_mcp

@mcp.tool(
    name="slack_send_message",
    annotations={
        "title": "发送 Slack 消息",
        "readOnlyHint": False,
        "destructiveHint": False,
        "idempotentHint": False,
        "openWorldHint": True
    }
)
async def slack_send_message(params: SendMessageInput) -> str:
    """向指定频道发送消息。

    Args:
        params: 包含 channel_id 和 text 的输入参数

    Returns:
        发送结果,包含消息 ID 和时间戳
    """
    # 实现发送逻辑

总结

MCP Builder Skill 提供了一套完整的方法论来构建高质量的 MCP 服务器:

  1. 四阶段流程:研究规划 -> 实现 -> 审查测试 -> 评估
  2. 双语言支持:TypeScript(推荐)和 Python
  3. 最佳实践:命名约定、响应格式、分页、错误处理
  4. 评估系统:通过复杂问题验证服务器质量

关键要记住的是:MCP 服务器的质量不在于功能的数量,而在于它能多好地帮助 AI 完成真实世界的任务。设计时要始终从用户(AI 和最终用户)的角度出发,创建直观、高效、可靠的工具。

基于 MIT 许可证发布。内容版权归作者所有。