<!-- title: Claude Code MCP服务器开发实战指南 -->

# Claude Code MCP服务器开发实战：从协议原理到生产部署

## 概述

MCP（Model Context Protocol）是 Anthropic 推出的开放协议，标准化了 AI 应用与外部工具的交互方式。本教程面向有基础编程经验的开发者，系统讲解 MCP 协议架构、三大原语（Tools/Resources/Prompts）的设计哲学、TypeScript 与 Python 双语言实现路径，以及如何将自定义 MCP 服务器接入 Claude Code。学完本教程，你将能独立构建生产级 MCP 工具，扩展 Claude Code 的能力边界。

**适用人群**：有 Node.js 或 Python 基础、使用过 Claude Code 并希望深度定制其工具链的开发者。

**学习目标**：
- 理解 MCP 协议核心概念与通信机制
- 掌握 Tools、Resources、Prompts 三大原语的适用场景
- 用 TypeScript SDK 和 Python FastMCP 分别实现 MCP 服务器
- 将自建服务器配置到 Claude Code 并实际使用
- 了解 OAuth 2.1 认证、错误处理等生产级实践

---

## 前置条件

| 项目 | 要求 |
|------|------|
| Node.js | v18+（TypeScript 方案） |
| Python | 3.10+（Python 方案） |
| Claude Code | 最新版 CLI |
| 包管理器 | npm/pnpm 或 uv/pip |
| 基础知识 | REST API、JSON Schema、命令行操作 |

---

## 第一部分：MCP 协议核心架构

### 1.1 为什么需要 MCP

在 MCP 出现之前，每个 AI 工具集成都是定制开发——让 Claude 调用 GitHub API 需要 GitHub 插件，调用数据库需要数据库插件，调用内部服务需要专用适配器。MCP 的核心价值是**统一接口**：

```
┌──────────────┐    统一协议     ┌──────────────┐
│              │◄──────────────►│              │
│  AI 应用     │   MCP 协议     │  MCP 服务器   │
│ (Claude Code)│  JSON-RPC 2.0  │  (你的工具)   │
│              │◄──────────────►│              │
└──────────────┘                └──────────────┘
```

一次开发，所有 MCP 兼容客户端（Claude Code、Claude Desktop、Cursor 等）都能使用。

### 1.2 通信机制

MCP 基于 **JSON-RPC 2.0** 协议，支持两种传输层：


![图表1](images/claude-code-tutorial-r3-1630-mermaid-0.png)

> *图表 1*


**stdio 传输**：Claude Code 将 MCP 服务器作为子进程启动，通过标准输入输出交换 JSON-RPC 消息。适合本地工具。

**HTTP Streamable 传输**：MCP 服务器作为独立的 HTTP 服务运行，客户端通过 POST 请求发送指令，通过 SSE（Server-Sent Events）接收响应。适合远程服务和需要 OAuth 认证的场景。

### 1.3 三大原语详解

MCP 定义了三种核心能力，每种有不同的控制模型：


![图表2](images/claude-code-tutorial-r3-1630-mermaid-1.png)

> *图表 2*


| 原语 | 控制方 | 类比 | 典型用途 |
|------|--------|------|----------|
| **Tools** | LLM（模型决定何时调用） | POST 请求 | 查询数据库、调用 API、执行脚本 |
| **Resources** | 应用（按需读取） | GET 请求 | 返回文件内容、配置数据、日志 |
| **Prompts** | 用户（主动触发） | 模板引擎 | 代码审查模板、文档生成框架 |

关键设计哲学：**Tools 是核心**。Claude Code 与 MCP 服务器的主要交互方式就是调用 Tool。Resources 和 Prompts 是辅助能力。

---

## 第二部分：TypeScript SDK 实现

### 2.1 项目初始化

```bash
# 创建项目目录
mkdir weather-mcp-server && cd weather-mcp-server

# 初始化 npm 项目
npm init -y

# 安装 MCP TypeScript SDK
npm install @modelcontextprotocol/sdk zod

# 安装开发依赖
npm install -D typescript @types/node

# 初始化 TypeScript 配置
npx tsc --init
```

修改 `tsconfig.json` 关键配置：

```json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true
  },
  "include": ["src/**/*"]
}
```

### 2.2 最小可运行服务器

创建 `src/index.ts`：

```typescript
#!/usr/bin/env node

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

// 创建 MCP 服务器实例
const server = new McpServer({
  name: "weather-server",
  version: "1.0.0",
  description: "天气查询 MCP 服务器",
});

// 注册第一个 Tool
server.tool(
  "get_weather",                              // 工具名称
  "获取指定城市的当前天气信息",                   // 工具描述（LLM 看到这段文字决定是否调用）
  {
    city: z.string().describe("城市名称，如 Beijing、Tokyo"),
    unit: z.enum(["celsius", "fahrenheit"]).optional().default("celsius").describe("温度单位"),
  },
  async ({ city, unit }) => {
    // 实际项目中这里调用真实 API
    const mockData = {
      city,
      temperature: unit === "celsius" ? 22 : 71.6,
      condition: "晴朗",
      humidity: "45%",
      wind: "东北风 3级",
    };

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(mockData, null, 2),
        },
      ],
    };
  }
);

// 注册第二个 Tool：天气预报
server.tool(
  "get_forecast",
  "获取未来几天的天气预报",
  {
    city: z.string().describe("城市名称"),
    days: z.number().min(1).max(7).describe("预报天数，1-7天"),
  },
  async ({ city, days }) => {
    const forecast = Array.from({ length: days }, (_, i) => ({
      day: `第${i + 1}天`,
      high: `${20 + Math.floor(Math.random() * 10)}°C`,
      low: `${10 + Math.floor(Math.random() * 8)}°C`,
      condition: ["晴", "多云", "小雨", "阴"][Math.floor(Math.random() * 4)],
    }));

    return {
      content: [
        {
          type: "text",
          text: `${city}未来${days}天预报：\n${JSON.stringify(forecast, null, 2)}`,
        },
      ],
    };
  }
);

// 注册 Resource：返回支持的城市列表
server.resource(
  "supported-cities",
  "weather://cities",
  async (uri) => {
    const cities = [
      "Beijing", "Shanghai", "Guangzhou", "Shenzhen",
      "Tokyo", "Seoul", "New York", "London", "Paris",
    ];
    return {
      contents: [
        {
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify(cities, null, 2),
        },
      ],
    };
  }
);

// 注册 Prompt：天气报告生成模板
server.prompt(
  "weather-report",
  "生成结构化天气报告的提示词模板",
  {
    city: z.string().describe("目标城市"),
    format: z.enum(["brief", "detailed"]).optional().default("brief"),
  },
  async ({ city, format }) => {
    return {
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `请为 ${city} 生成一份${format === "detailed" ? "详细" : "简要"}天气报告。` +
                  `包含温度、湿度、风力等信息，并给出穿衣建议。`,
          },
        },
      ],
    };
  }
);

// 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Weather MCP Server 已启动（stderr 日志）");
}

main().catch((error) => {
  console.error("启动失败:", error);
  process.exit(1);
});
```

### 2.3 构建与配置

在 `package.json` 中添加构建和启动脚本：

```json
{
  "name": "weather-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js"
  },
  "bin": {
    "weather-mcp": "./dist/index.js"
  }
}
```

构建项目：

```bash
npm run build
```

将 MCP 服务器添加到 Claude Code：

```bash
# 方式一：直接指定构建产物
claude mcp add weather-server -- node /path/to/weather-mcp-server/dist/index.js

# 方式二：使用 npx 运行
claude mcp add weather-server -- npx -y /path/to/weather-mcp-server/dist/index.js

# 验证配置
claude mcp list
```

预期输出：

```
weather-server
  Scope: project
  Type: stdio
  Command: node /path/to/weather-mcp-server/dist/index.js
  Tools: get_weather, get_forecast
  Resources: weather://cities
  Prompts: weather-report
```

### 2.4 常见问题排查

**问题：`claude mcp list` 显示服务器但 Tools 数量为 0**

可能原因：`package.json` 缺少 `"type": "module"`。MCP SDK 使用 ESM 模块，必须在 package.json 中声明。

**问题：服务器启动后立即退出**

排查方法：查看 Claude Code 的 MCP 日志：

```bash
# 日志通常位于
cat ~/.claude/logs/mcp-weather-server.log
```

确保 `index.ts` 中的 `import` 路径以 `.js` 结尾（TypeScript ESM 约定）。

**问题：Tool 调用返回 undefined**

检查 `server.tool()` 的回调函数是否正确返回了 `{ content: [{ type: "text", text: "..." }] }` 结构。这是 MCP Tool 响应的固定格式。

---

## 第三部分：Python FastMCP 实现

### 3.1 为什么选择 FastMCP

Python 生态有两种 MCP 开发方式：

| 方案 | 安装包 | 特点 |
|------|--------|------|
| `mcp`（官方 SDK） | `pip install mcp` | 底层控制、灵活 |
| `fastmcp` | `pip install fastmcp` | 装饰器驱动、极简 API |

FastMCP 是目前 Python 社区最流行的 MCP 框架，一个装饰器即可注册 Tool/Resource/Prompt，自动生成 JSON Schema。

### 3.2 项目搭建

```bash
# 创建项目
mkdir db-query-mcp && cd db-query-mcp

# 使用 uv 创建虚拟环境（推荐）
uv init
uv add fastmcp

# 或使用 pip
python -m venv .venv
source .venv/bin/activate
pip install fastmcp
```

### 3.3 实战：数据库查询 MCP 服务器

创建 `server.py`：

```python
#!/usr/bin/env python3
"""数据库查询 MCP 服务器 - 支持只读SQL查询和表结构浏览"""

import json
import sqlite3
from pathlib import Path
from fastmcp import FastMCP

# 创建 MCP 服务器
mcp = FastMCP(
    name="db-query-server",
    version="1.0.0",
)

# 内存数据库，用示例数据初始化
DB_PATH = Path(__file__).parent / "sample.db"


def get_connection() -> sqlite3.Connection:
    """获取数据库连接"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn


def init_sample_db():
    """初始化示例数据库"""
    conn = get_connection()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE,
            department TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER REFERENCES users(id),
            product TEXT NOT NULL,
            amount REAL NOT NULL,
            status TEXT DEFAULT 'pending',
            order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        INSERT OR IGNORE INTO users (id, name, email, department) VALUES
            (1, '张三', 'zhang@example.com', '工程部'),
            (2, '李四', 'li@example.com', '产品部'),
            (3, '王五', 'wang@example.com', '设计部'),
            (4, '赵六', 'zhao@example.com', '工程部');

        INSERT OR IGNORE INTO orders (id, user_id, product, amount, status) VALUES
            (101, 1, 'MacBook Pro', 14999.00, 'completed'),
            (102, 1, 'TypeScript高级编程', 89.00, 'completed'),
            (103, 2, 'Figma企业版', 1200.00, 'pending'),
            (104, 3, '显示器4K', 3200.00, 'shipped'),
            (105, 4, '机械键盘', 899.00, 'pending');
    """)
    conn.commit()
    conn.close()


# ============ Tools ============

@mcp.tool()
def execute_query(sql: str) -> str:
    """执行只读SQL查询语句。仅支持SELECT语句，禁止写入操作。

    Args:
        sql: SQL查询语句，仅支持SELECT
    """
    # 安全校验：只允许 SELECT
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return json.dumps({"error": "安全限制：仅允许 SELECT 查询"}, ensure_ascii=False)

    # 禁止危险操作
    forbidden = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE"]
    for keyword in forbidden:
        if keyword in normalized:
            return json.dumps({"error": f"安全限制：禁止使用 {keyword} 操作"}, ensure_ascii=False)

    try:
        conn = get_connection()
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
        conn.close()

        return json.dumps({
            "columns": columns,
            "row_count": len(rows),
            "data": rows
        }, ensure_ascii=False, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)}, ensure_ascii=False)


@mcp.tool()
def list_tables() -> str:
    """列出数据库中所有表及其行数"""
    conn = get_connection()
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    tables = [row[0] for row in cursor.fetchall()]

    result = []
    for table in tables:
        count = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
        result.append({"table": table, "rows": count})

    conn.close()
    return json.dumps(result, ensure_ascii=False, indent=2)


@mcp.tool()
def describe_table(table_name: str) -> str:
    """查看指定表的列结构（字段名、类型、是否可空、默认值）

    Args:
        table_name: 表名称
    """
    conn = get_connection()
    cursor = conn.execute(f"PRAGMA table_info({table_name})")
    columns = [
        {
            "cid": row[0],
            "name": row[1],
            "type": row[2],
            "notnull": bool(row[3]),
            "default": row[4],
            "primary_key": bool(row[5]),
        }
        for row in cursor.fetchall()
    ]
    conn.close()
    return json.dumps(columns, ensure_ascii=False, indent=2)


# ============ Resources ============

@mcp.resource("db://schema")
def get_schema() -> str:
    """返回完整数据库 schema"""
    conn = get_connection()
    cursor = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    schemas = [row[0] for row in cursor.fetchall() if row[0]]
    conn.close()
    return "\n\n".join(schemas)


@mcp.resource("db://stats")
def get_stats() -> str:
    """返回数据库统计信息"""
    conn = get_connection()
    tables = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    ).fetchall()

    stats = {}
    for (table_name,) in tables:
        count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
        stats[table_name] = {"row_count": count}

    conn.close()
    return json.dumps(stats, ensure_ascii=False, indent=2)


# ============ Prompts ============

@mcp.prompt()
def analyze_data(table_name: str, question: str) -> str:
    """数据分析提示词模板 - 引导LLM对指定表进行数据分析

    Args:
        table_name: 要分析的表名
        question: 用户想了解的数据问题
    """
    return f"""你是一个数据分析专家。请分析 {table_name} 表的数据来回答以下问题：

问题：{question}

请按以下步骤进行：
1. 先用 list_tables 了解数据库结构
2. 用 describe_table 查看 {table_name} 的列定义
3. 构造合适的 SQL 查询并用 execute_query 执行
4. 基于查询结果给出分析结论

注意：只能使用 SELECT 查询，不能修改数据。"""


# 初始化并启动
if __name__ == "__main__":
    init_sample_db()
    mcp.run()
```

### 3.4 注册到 Claude Code

```bash
# Python MCP 服务器使用 stdio 传输
claude mcp add db-query -- python /path/to/db-query-mcp/server.py

# 如果使用虚拟环境，指定完整路径
claude mcp add db-query -- /path/to/db-query-mcp/.venv/bin/python /path/to/db-query-mcp/server.py

# 使用 uv run（推荐，自动处理虚拟环境）
claude mcp add db-query -- uv run --directory /path/to/db-query-mcp server.py
```

### 3.5 FastMCP 装饰器速查

```python
from fastmcp import FastMCP

mcp = FastMCP("my-server")

# Tool：LLM 可调用的函数
@mcp.tool()
def my_tool(param: str) -> str:
    """描述文字会被 LLM 看到"""
    return f"结果: {param}"

# Resource：应用层读取的数据端点
@mcp.resource("config://app")
def get_config() -> str:
    return '{"theme": "dark"}'

# Resource Template：带参数的资源
@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    return f"用户 {user_id} 的档案数据"

# Prompt：用户触发的提示词模板
@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    return f"请审查以下 {language} 代码：\n{code}"
```

---

## 第四部分：在 Claude Code 中使用自建 MCP

### 4.1 MCP 管理命令速查

```bash
# 添加服务器
claude mcp add <名称> -- <命令> [参数...]

# 列出所有已配置服务器
claude mcp list

# 查看服务器详情
claude mcp get <名称>

# 删除服务器
claude mcp remove <名称>

# 重置服务器（重新连接）
claude mcp reset <名称>
```

### 4.2 配置作用域

Claude Code 支持三个配置作用域：

```bash
# 项目级：仅在当前项目中可用（推荐）
claude mcp add my-server -s project -- node ./server.js

# 用户级：所有项目共享
claude mcp add my-server -s user -- node ~/servers/my-server.js

# 局部级
claude mcp add my-server -s local -- node ./server.js
```

**推荐实践**：项目级配置将 MCP 服务器信息存储在 `.claude/settings.json`，可以提交到 Git 仓库，团队成员共享。

### 4.3 环境变量与敏感信息

```bash
# 传递环境变量（如 API Key）
claude mcp add api-server \
  -e API_KEY=sk-xxxx \
  -e BASE_URL=https://api.example.com \
  -- node ./api-server.js

# 在服务器代码中读取
# process.env.API_KEY  // Node.js
# os.environ["API_KEY"]  # Python
```

### 4.4 实际对话示例

配置完成后，在 Claude Code 中对话：

```
> 帮我查一下北京今天的天气

Claude 调用: get_weather(city="Beijing", unit="celsius")
返回:
{
  "city": "Beijing",
  "temperature": 22,
  "condition": "晴朗",
  "humidity": "45%",
  "wind": "东北风 3级"
}

北京今天天气晴朗，气温 22°C，湿度 45%，东北风 3 级。建议穿着薄外套或长袖。
```

```
> 分析一下 orders 表，看看哪个部门消费最多

Claude 会自动调用:
1. list_tables() → 了解数据库结构
2. describe_table("orders") → 查看字段
3. execute_query("SELECT u.department, SUM(o.amount) ...") → 执行分析查询
```

---

## 第五部分：实战项目——Git 仓库分析 MCP 服务器

将前面学到的知识整合为一个完整的、有实际用途的 MCP 服务器。这个服务器能分析 Git 仓库的提交历史、代码统计和贡献者信息。

### 5.1 项目结构

```
git-analyzer-mcp/
├── package.json
├── tsconfig.json
├── src/
│   └── index.ts        # 主入口
└── .claude/
    └── settings.json   # MCP 配置（项目级）
```

### 5.2 完整代码

`src/index.ts`：

```typescript
#!/usr/bin/env node

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { execSync } from "child_process";

const server = new McpServer({
  name: "git-analyzer",
  version: "1.0.0",
  description: "Git 仓库分析工具 - 提交历史、代码统计、贡献者分析",
});

// 安全执行 git 命令
function gitExec(args: string, cwd?: string): string {
  try {
    return execSync(`git ${args}`, {
      encoding: "utf-8",
      maxBuffer: 10 * 1024 * 1024,
      cwd: cwd || process.cwd(),
      timeout: 30000,
    }).trim();
  } catch (error: any) {
    return `ERROR: ${error.message}`;
  }
}

// Tool 1: 提交历史分析
server.tool(
  "git_commit_stats",
  "分析 Git 仓库的提交统计信息，包括提交频率、活跃时段",
  {
    repo_path: z.string().optional().describe("仓库路径，默认为当前目录"),
    since: z.string().optional().describe("起始日期，如 2024-01-01"),
    until: z.string().optional().describe("截止日期"),
    author: z.string().optional().describe("按作者过滤"),
  },
  async ({ repo_path, since, until, author }) => {
    let cmd = "log --pretty=format:'%H|%an|%ae|%ai' --no-merges";
    if (since) cmd += ` --since='${since}'`;
    if (until) cmd += ` --until='${until}'`;
    if (author) cmd += ` --author='${author}'`;

    const output = gitExec(cmd, repo_path);
    if (output.startsWith("ERROR")) {
      return { content: [{ type: "text", text: output }] };
    }

    const lines = output.split("\n").filter(Boolean);
    const commits = lines.map((line) => {
      const [hash, name, email, date] = line.split("|");
      return { hash: hash?.slice(0, 8), name, email, date };
    });

    // 按作者统计
    const authorStats: Record<string, number> = {};
    commits.forEach((c) => {
      authorStats[c.name] = (authorStats[c.name] || 0) + 1;
    });

    const summary = {
      total_commits: commits.length,
      unique_authors: Object.keys(authorStats).length,
      author_breakdown: Object.entries(authorStats)
        .sort(([, a], [, b]) => b - a)
        .reduce((obj, [k, v]) => ({ ...obj, [k]: v }), {}),
      first_commit: commits[commits.length - 1]?.date,
      latest_commit: commits[0]?.date,
    };

    return {
      content: [{ type: "text", text: JSON.stringify(summary, null, 2) }],
    };
  }
);

// Tool 2: 代码变更统计
server.tool(
  "git_code_changes",
  "统计代码变更量——新增行、删除行、修改文件数",
  {
    repo_path: z.string().optional().describe("仓库路径"),
    since: z.string().optional().describe("起始日期"),
  },
  async ({ repo_path, since }) => {
    const period = since ? `--since='${since}'` : "--all";
    const output = gitExec(
      `log ${period} --pretty=tformat: --numstat`,
      repo_path
    );

    if (output.startsWith("ERROR") || !output) {
      return {
        content: [
          { type: "text", text: output || "无变更记录" },
        ],
      };
    }

    let totalAdded = 0;
    let totalDeleted = 0;
    let totalFiles = 0;
    const fileChanges: Record<string, { added: number; deleted: number }> = {};

    output.split("\n").forEach((line) => {
      const parts = line.trim().split(/\s+/);
      if (parts.length >= 3) {
        const added = parseInt(parts[0]) || 0;
        const deleted = parseInt(parts[1]) || 0;
        const file = parts[2];
        totalAdded += added;
        totalDeleted += deleted;
        totalFiles++;
        if (!fileChanges[file]) {
          fileChanges[file] = { added: 0, deleted: 0 };
        }
        fileChanges[file].added += added;
        fileChanges[file].deleted += deleted;
      }
    });

    // 变更最多的文件 Top 10
    const topFiles = Object.entries(fileChanges)
      .sort(([, a], [, b]) => b.added + b.deleted - (a.added + a.deleted))
      .slice(0, 10)
      .map(([file, stats]) => ({ file, ...stats }));

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(
            {
              total_added: totalAdded,
              total_deleted: totalDeleted,
              net_change: totalAdded - totalDeleted,
              files_changed: totalFiles,
              top_changed_files: topFiles,
            },
            null,
            2
          ),
        },
      ],
    };
  }
);

// Tool 3: 分支对比
server.tool(
  "git_compare_branches",
  "对比两个分支的差异——提交数、文件变更",
  {
    base: z.string().describe("基准分支名，如 main"),
    head: z.string().describe("对比分支名，如 feature-xyz"),
    repo_path: z.string().optional().describe("仓库路径"),
  },
  async ({ base, head, repo_path }) => {
    const aheadBehind = gitExec(
      `rev-list --left-right --count ${base}...${head}`,
      repo_path
    );
    const diffStat = gitExec(`diff --stat ${base}...${head}`, repo_path);

    const [behind, ahead] = aheadBehind
      .split(/\s+/)
      .map((n) => parseInt(n) || 0);

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(
            {
              head_branch: head,
              base_branch: base,
              commits_ahead: ahead,
              commits_behind: behind,
              diff_stat: diffStat,
            },
            null,
            2
          ),
        },
      ],
    };
  }
);

// Resource: 仓库基本信息
server.resource("repo-info", "git://repo/info", async (uri) => {
  const branch = gitExec("rev-parse --abbrev-ref HEAD");
  const remote = gitExec("remote get-url origin");
  const lastCommit = gitExec("log -1 --pretty=format:'%h %s (%cr)'");

  return {
    contents: [
      {
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(
          { current_branch: branch, remote_url: remote, last_commit: lastCommit },
          null,
          2
        ),
      },
    ],
  };
});

// 启动
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Git Analyzer MCP Server 已启动");
}

main().catch(console.error);
```

### 5.3 注册到 Claude Code

```bash
# 构建
npm run build

# 注册（项目级）
claude mcp add git-analyzer -s project -- node /absolute/path/to/git-analyzer-mcp/dist/index.js
```

### 5.4 使用场景

```bash
# 在 Claude Code 中
> 分析这个仓库最近30天的提交情况

# Claude 会自动调用 git_commit_stats(since="30 days ago")
# 返回提交数、贡献者排名、时间范围等信息

> 对比 main 和 feature-auth 分支的差异

# Claude 会调用 git_compare_branches(base="main", head="feature-auth")

> 哪些文件变更最频繁？

# Claude 会调用 git_code_changes() 并返回 Top 10 变更文件
```

---

## 第六部分：HTTP Streamable 传输与远程部署

### 6.1 从 stdio 切换到 HTTP

当 MCP 服务器需要远程访问或团队共享时，使用 HTTP Streamable 传输：

**TypeScript 方案：**

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

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

const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: undefined, // 无状态模式
});

// 将 MCP 服务器连接到 HTTP 传输
server.connect(transport);

// 处理 MCP 请求
app.post("/mcp", async (req, res) => {
  await transport.handleRequest(req, res);
});

app.listen(3000, () => {
  console.log("MCP HTTP Server 运行在 http://localhost:3000/mcp");
});
```

**Python FastMCP 方案：**

```python
# FastMCP 原生支持 HTTP 模式
mcp = FastMCP("my-server", host="0.0.0.0", port=3000)

if __name__ == "__main__":
    # 默认是 stdio，改为 sse 或 streamable-http
    mcp.run(transport="streamable-http")
```

### 6.2 在 Claude Code 中添加远程服务器

```bash
# HTTP 传输
claude mcp add my-remote-server --transport http -- http://your-server.com/mcp

# 带认证头
claude mcp add my-remote-server \
  --transport http \
  -H "Authorization: Bearer your-token" \
  -- http://your-server.com/mcp
```

### 6.3 OAuth 2.1 认证

MCP 规范要求远程服务器支持 OAuth 2.1 with PKCE。完整流程：


![图表3](images/claude-code-tutorial-r3-1630-mermaid-2.png)

> *图表 3*


---

## 第七部分：调试与测试

### 7.1 MCP Inspector

官方提供的调试工具，可以独立测试 MCP 服务器：

```bash
# 安装并启动 Inspector
npx @modelcontextprotocol/inspector node ./dist/index.js

# 或测试 Python 服务器
npx @modelcontextprotocol/inspector python ./server.py
```

Inspector 启动后会打开一个 Web 界面（默认 `http://localhost:5173`），可以：

- 查看服务器注册的所有 Tools/Resources/Prompts
- 手动调用 Tool 并查看响应
- 检查 JSON-RPC 通信日志
- 测试错误处理

### 7.2 日志输出

MCP 服务器通过 stdio 与 Claude Code 通信，所以**不能用 `console.log`/`print`** 输出调试信息（会干扰协议通信）。正确方式：

```typescript
// TypeScript - 输出到 stderr
console.error("调试信息:", data);

// 或者写入文件
import { appendFileSync } from "fs";
appendFileSync("/tmp/mcp-debug.log", `${JSON.stringify(data)}\n`);
```

```python
# Python - 输出到 stderr
import sys
print("调试信息", file=sys.stderr)

# 或者使用 logging
import logging
logging.basicConfig(filename="/tmp/mcp-debug.log", level=logging.DEBUG)
```

### 7.3 单元测试 Tool 函数

Tool 本质上是普通函数，可以独立测试：

```typescript
// 提取 Tool 逻辑为独立函数
export function validateSql(sql: string): boolean {
  const normalized = sql.trim().toUpperCase();
  if (!normalized.startsWith("SELECT")) return false;
  const forbidden = ["DROP", "DELETE", "INSERT", "UPDATE"];
  return !forbidden.some((k) => normalized.includes(k));
}

// 测试文件
import { validateSql } from "./index";
import assert from "assert";

assert.strictEqual(validateSql("SELECT * FROM users"), true);
assert.strictEqual(validateSql("DROP TABLE users"), false);
assert.strictEqual(validateSql("INSERT INTO users VALUES (1,2)"), false);
```

---

## 第八部分：最佳实践与架构模式

### 8.1 Tool 设计原则


![图表4](images/claude-code-tutorial-r3-1630-mermaid-3.png)

> *图表 4*


**关键原则：`description` 是给 LLM 看的"使用说明书"。** 写得越精确，LLM 越能准确判断何时调用、传什么参数。

```typescript
// 好的描述
server.tool("search_users",
  "在用户数据库中按姓名或邮箱搜索用户。支持模糊匹配，返回最多50条结果。",
  { query: z.string().min(1).max(200).describe("搜索关键词") },
  async ({ query }) => { /* ... */ }
);

// 差的描述
server.tool("search", "搜索", { q: z.string() }, async ({ q }) => { /* ... */ });
```

### 8.2 安全清单

| 风险 | 防御措施 |
|------|----------|
| SQL 注入 | 参数化查询 + 白名单校验 |
| 命令注入 | 避免 `eval()` / `exec()` 拼接用户输入 |
| 路径遍历 | 校验路径在允许的目录内 |
| 信息泄露 | 日志脱敏、错误信息不暴露内部细节 |
| 权限提升 | Tool 按最小权限运行 |

### 8.3 性能优化

- **异步操作**：Tool handler 必须返回 Promise，不要阻塞事件循环
- **超时控制**：为耗时操作设置超时，避免 Claude Code 长时间等待
- **结果缓存**：对频繁调用的只读数据使用内存缓存
- **分页返回**：大数据集分页返回，避免一次性传输过多数据

---

## 完整工作流程


![图表5](images/claude-code-tutorial-r3-1630-mermaid-4.png)

> *图表 5*


---

## 延伸阅读

- [MCP 官方规范](https://spec.modelcontextprotocol.io) — 协议完整定义
- [MCP TypeScript SDK](https://github.com/modelcontextprotocol/typescript-sdk) — GitHub 仓库与示例
- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) — 官方 Python 实现
- [FastMCP 文档](https://github.com/PrefectHQ/fastmcp) — Python 装饰器框架
- [Claude Code MCP 文档](https://code.claude.com/docs/en/mcp) — 配置与管理指南
- [MCP Inspector](https://github.com/modelcontextprotocol/inspector) — 可视化调试工具
- [MCP 服务器示例集](https://github.com/modelcontextprotocol/servers) — 官方参考实现

---

## 训练营适用性评估

| 维度 | 评估 |
|------|------|
| 适合训练营类型 | AI 工具开发、LLM 应用工程、DevOps 自动化 |
| 预计教学时长 | 4-6 小时（含实操） |
| 学员基础要求 | JavaScript/Python 基础、命令行操作、API 概念 |
| 实操比重 | 70% 编码 + 30% 讲解 |
| 建议分组 | 2-3 人一组，每组完成一个 MCP 服务器 |
| 考核方式 | 现场演示 MCP 服务器在 Claude Code 中运行 |

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)