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

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

# 创建项目目录
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

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

#!/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);
});

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

npm run build

# 方式一：直接指定构建产物
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

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

# 创建项目
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

#!/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, '张三', '[email protected]', '工程部'),
            (2, '李四', '[email protected]', '产品部'),
            (3, '王五', '[email protected]', '设计部'),
            (4, '赵六', '[email protected]', '工程部');

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

# 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

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 mcp add <名称> -- <命令> [参数...]

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

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

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

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

# 项目级：仅在当前项目中可用（推荐）
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

# 传递环境变量（如 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

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

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-analyzer-mcp/
├── package.json
├── tsconfig.json
├── src/
│   └── index.ts        # 主入口
└── .claude/
    └── settings.json   # MCP 配置（项目级）

#!/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);

# 构建
npm run build

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

# 在 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 变更文件

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

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

# 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

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

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

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

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

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

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

// 提取 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);

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

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