Claude Code MCP服务器开发从零到上线全实战

graph TB
    subgraph Host["宿主应用 (Host)"]
        CC["Claude Code / Claude Desktop"]
        Client["MCP Client"]
    end

    subgraph Servers["MCP Servers"]
        S1["📁 文件系统 Server"]
        S2["🗄️ 数据库 Server"]
        S3["🔧 自定义 Server"]
    end

    CC -->|"用户指令"| Client
    Client -->|"stdio / HTTP"| S1
    Client -->|"stdio / HTTP"| S2
    Client -->|"stdio / HTTP"| S3

    subgraph Primitives["MCP 三大原语"]
        T["🛠️ Tools<br/>执行操作"]
        R["📄 Resources<br/>暴露数据"]
        P["💬 Prompts<br/>提示模板"]
    end

    S1 --> T
    S2 --> R
    S3 --> P

# 检查Python版本
python --version  # 需要 >= 3.10

# 检查Node.js版本
node --version    # 需要 >= 18

# 检查Claude Code
claude --version

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

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装FastMCP（官方Python MCP框架）
pip install fastmcp httpx

from mcp.server.fastmcp import FastMCP

# 初始化MCP服务器，名称将显示在Claude Code中
mcp = FastMCP("weather-server")


@mcp.tool()
def get_current_weather(city: str) -> str:
    """获取指定城市的当前天气信息

    Args:
        city: 城市名称，如"北京"、"上海"

    Returns:
        天气信息描述字符串
    """
    # 这里使用模拟数据，实际项目中可接入天气API
    weather_data = {
        "北京": "☀️ 晴，温度 28°C，湿度 45%，北风3级",
        "上海": "🌧️ 小雨，温度 22°C，湿度 78%，东风2级",
        "深圳": "⛅ 多云，温度 30°C，湿度 65%，南风2级",
    }
    return weather_data.get(city, f"暂无{city}的天气数据")


@mcp.tool()
def get_weather_forecast(city: str, days: int = 3) -> str:
    """获取未来几天的天气预报

    Args:
        city: 城市名称
        days: 预报天数，默认3天，最多7天

    Returns:
        天气预报信息
    """
    if days > 7:
        return "最多支持7天预报"

    forecasts = []
    for i in range(1, days + 1):
        forecasts.append(f"  第{i}天: 晴转多云，温度 20-30°C")

    return f"{city}未来{days}天预报:\n" + "\n".join(forecasts)


# 运行服务器（默认使用stdio传输）
if __name__ == "__main__":
    mcp.run()

@mcp.resource("weather://cities/supported")
def get_supported_cities() -> str:
    """返回支持查询天气的城市列表"""
    cities = ["北京", "上海", "深圳", "广州", "成都", "杭州"]
    return "支持查询天气的城市：\n" + "\n".join(f"  - {c}" for c in cities)


@mcp.resource("weather://alert/{city}")
def get_weather_alert(city: str) -> str:
    """获取指定城市的天气预警信息

    Args:
        city: 城市名称
    """
    alerts = {
        "北京": "🟡 大风蓝色预警：预计未来24小时阵风可达7级",
        "上海": "🔴 暴雨橙色预警：预计未来6小时降雨量将达100mm以上",
    }
    return alerts.get(city, f"✅ {city}当前无天气预警")

@mcp.prompt()
def weather_report_prompt(city: str) -> str:
    """生成天气报告的提示模板

    Args:
        city: 要生成报告的城市
    """
    return f"""请为{city}生成一份详细的天气分析报告，包含以下内容：
1. 当前天气状况概述
2. 未来3天天气预报
3. 出行建议
4. 健康提示（如紫外线指数、空气质量等）

请使用专业但易懂的语言，适合公众阅读。"""

#!/usr/bin/env python3
"""天气查询 MCP 服务器 - 演示MCP三大原语的完整实现"""

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "weather-server",
    version="1.0.0",
    description="天气查询服务 - 提供城市天气查询、预报和预警功能"
)


# ============ Tools（工具）============

@mcp.tool()
def get_current_weather(city: str) -> str:
    """获取指定城市的当前天气信息

    Args:
        city: 城市名称，如"北京"、"上海"
    """
    weather_data = {
        "北京": "☀️ 晴，温度 28°C，湿度 45%，北风3级",
        "上海": "🌧️ 小雨，温度 22°C，湿度 78%，东风2级",
        "深圳": "⛅ 多云，温度 30°C，湿度 65%，南风2级",
        "广州": "🌤️ 晴间多云，温度 32°C，湿度 70%，南风2级",
        "成都": "☁️ 阴，温度 24°C，湿度 60%，微风",
        "杭州": "🌦️ 阵雨，温度 25°C，湿度 75%，东风3级",
    }
    return weather_data.get(city, f"暂无{city}的天气数据")


@mcp.tool()
def get_weather_forecast(city: str, days: int = 3) -> str:
    """获取未来几天的天气预报

    Args:
        city: 城市名称
        days: 预报天数，默认3天，最多7天
    """
    if days > 7:
        return "错误：最多支持7天预报"
    if days < 1:
        return "错误：天数必须大于0"

    forecasts = []
    conditions = ["☀️ 晴", "⛅ 多云", "🌤️ 晴间多云", "🌦️ 阵雨", "☁️ 阴"]
    for i in range(1, days + 1):
        cond = conditions[i % len(conditions)]
        high = 25 + (i % 5)
        low = 15 + (i % 3)
        forecasts.append(f"  第{i}天: {cond}，温度 {low}-{high}°C")

    return f"📍 {city}未来{days}天预报:\n" + "\n".join(forecasts)


@mcp.tool()
def compare_cities(city1: str, city2: str) -> str:
    """对比两个城市的天气

    Args:
        city1: 第一个城市
        city2: 第二个城市
    """
    w1 = get_current_weather(city1)
    w2 = get_current_weather(city2)
    return f"天气对比：\n\n📍 {city1}: {w1}\n📍 {city2}: {w2}"


# ============ Resources（资源）============

@mcp.resource("weather://cities/supported")
def get_supported_cities() -> str:
    """返回支持查询天气的城市列表"""
    cities = ["北京", "上海", "深圳", "广州", "成都", "杭州"]
    return "支持查询天气的城市：\n" + "\n".join(f"  - {c}" for c in cities)


@mcp.resource("weather://alert/{city}")
def get_weather_alert(city: str) -> str:
    """获取指定城市的天气预警信息"""
    alerts = {
        "北京": "🟡 大风蓝色预警：预计未来24小时阵风可达7级",
        "上海": "🔴 暴雨橙色预警：预计未来6小时降雨量将达100mm以上",
    }
    return alerts.get(city, f"✅ {city}当前无天气预警")


# ============ Prompts（提示模板）============

@mcp.prompt()
def weather_report_prompt(city: str) -> str:
    """生成天气分析报告的提示模板"""
    return f"""请为{city}生成一份详细的天气分析报告，包含：
1. 当前天气状况概述
2. 未来3天天气预报分析
3. 出行建议
4. 健康提示

请使用专业但易懂的语言。"""


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

# 创建项目
mkdir mcp-note-server && cd mcp-note-server
npm init -y

# 安装MCP 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: "note-server",
  version: "1.0.0",
  description: "笔记管理服务 - 轻量级笔记CRUD操作",
});

// 内存存储（实际项目中替换为数据库）
interface Note {
  id: string;
  title: string;
  content: string;
  tags: string[];
  createdAt: string;
  updatedAt: string;
}

const notes: Map<string, Note> = new Map();

function generateId(): string {
  return Date.now().toString(36) + Math.random().toString(36).slice(2);
}

// ============ Tools ============

server.tool(
  "create_note",
  "创建一条新笔记",
  {
    title: z.string().describe("笔记标题"),
    content: z.string().describe("笔记内容"),
    tags: z.array(z.string()).optional().describe("标签列表"),
  },
  async ({ title, content, tags = [] }) => {
    const id = generateId();
    const now = new Date().toISOString();
    const note: Note = { id, title, content, tags, createdAt: now, updatedAt: now };
    notes.set(id, note);
    return {
      content: [{ type: "text" as const, text: `✅ 笔记已创建\nID: ${id}\n标题: ${title}` }],
    };
  }
);

server.tool(
  "list_notes",
  "列出所有笔记（支持按标签筛选）",
  {
    tag: z.string().optional().describe("按标签筛选"),
  },
  async ({ tag }) => {
    let filtered = Array.from(notes.values());
    if (tag) {
      filtered = filtered.filter((n) => n.tags.includes(tag));
    }
    if (filtered.length === 0) {
      return {
        content: [{ type: "text" as const, text: "📭 没有找到笔记" }],
      };
    }
    const list = filtered
      .map((n) => `- [${n.id}] ${n.title} (${n.tags.join(", ") || "无标签"}) ${n.updatedAt.split("T")[0]}`)
      .join("\n");
    return {
      content: [{ type: "text" as const, text: `📝 笔记列表 (${filtered.length}条):\n${list}` }],
    };
  }
);

server.tool(
  "read_note",
  "读取指定ID的笔记详情",
  {
    id: z.string().describe("笔记ID"),
  },
  async ({ id }) => {
    const note = notes.get(id);
    if (!note) {
      return {
        content: [{ type: "text" as const, text: `❌ 未找到ID为 ${id} 的笔记` }],
        isError: true,
      };
    }
    return {
      content: [
        {
          type: "text" as const,
          text: `# ${note.title}\n\n${note.content}\n\n---\n标签: ${note.tags.join(", ") || "无"}\n创建: ${note.createdAt}\n更新: ${note.updatedAt}`,
        },
      ],
    };
  }
);

server.tool(
  "delete_note",
  "删除指定ID的笔记",
  {
    id: z.string().describe("笔记ID"),
  },
  async ({ id }) => {
    const deleted = notes.delete(id);
    return {
      content: [
        {
          type: "text" as const,
          text: deleted ? `✅ 已删除笔记 ${id}` : `❌ 未找到笔记 ${id}`,
        },
      ],
    };
  }
);

// ============ Resources ============

server.resource("notes://all", "notes://all", async (uri) => {
  const allNotes = Array.from(notes.values());
  return {
    contents: [
      {
        uri: uri.href,
        text: JSON.stringify(allNotes, null, 2),
      },
    ],
  };
});

// ============ Prompts ============

server.prompt(
  "summarize_notes",
  "生成所有笔记的摘要报告",
  [],
  async () => {
    const count = notes.size;
    const allTags = new Set(Array.from(notes.values()).flatMap((n) => n.tags));
    return {
      messages: [
        {
          role: "user" as const,
          content: {
            type: "text" as const,
            text: `当前共有 ${count} 条笔记，使用了 ${allTags.size} 个标签（${[...allTags].join("、")}）。\n请根据这些笔记内容生成一份结构化的知识摘要。`,
          },
        },
      ],
    };
  }
);

// ============ 启动服务器 ============

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  // 服务器已启动，通过stdio通信
}

main().catch((err) => {
  console.error("Server error:", err);
  process.exit(1);
});

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

npm run build

graph LR
    subgraph 方法1["CLI命令（推荐）"]
        A["claude mcp add"]
    end

    subgraph 方法2["项目级配置"]
        B[".mcp.json"]
    end

    subgraph 方法3["用户级配置"]
        C["~/.claude.json"]
    end

    A --> D["Claude Code"]
    B --> D
    C --> D

    style 方法1 fill:#4CAF50,color:#fff
    style 方法2 fill:#2196F3,color:#fff
    style 方法3 fill:#FF9800,color:#fff

# 在项目目录下执行（项目级）
claude mcp add weather-server -- python /path/to/mcp-weather-server/server.py

# 或添加到用户级（全局可用）
claude mcp add weather-server -s user -- python /path/to/mcp-weather-server/server.py

claude mcp add note-server -- node /path/to/mcp-note-server/dist/index.js

claude mcp add weather-server \
  -e OPENWEATHER_API_KEY=your_key \
  -- python /path/to/server.py

{
  "mcpServers": {
    "weather-server": {
      "command": "python",
      "args": ["/absolute/path/to/mcp-weather-server/server.py"],
      "env": {
        "OPENWEATHER_API_KEY": "your_api_key_here"
      }
    },
    "note-server": {
      "command": "node",
      "args": ["/absolute/path/to/mcp-note-server/dist/index.js"]
    }
  }
}

{
  "mcpServers": {
    "weather-server": {
      "command": "python",
      "args": ["/home/user/mcp-weather-server/server.py"],
      "env": {
        "OPENWEATHER_API_KEY": "sk-real-key-here"
      }
    }
  }
}

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

# 查看某个服务器详情
claude mcp get weather-server

# 删除服务器
claude mcp remove weather-server

# 以JSON格式添加（复杂配置）
claude mcp add-json weather-server '{
  "command": "python",
  "args": ["./server.py"],
  "env": {"API_KEY": "xxx"}
}'

# 启动Claude Code
claude

# 在交互界面中输入
> 帮我查一下北京现在的天气

# 预期结果：Claude会调用weather-server的get_current_weather工具

/MCP        # 查看已连接的MCP服务器列表
/tools      # 查看所有可用工具（包括MCP工具）

# 安装并启动Inspector
npx @modelcontextprotocol/inspector

# 指定Python服务器
npx @modelcontextprotocol/inspector python /path/to/server.py

# 指定Node服务器
npx @modelcontextprotocol/inspector node /path/to/dist/index.js

# 手动测试服务器是否能运行
python /path/to/server.py
# 如果报错，检查Python版本和依赖

# 查看Claude Code日志
claude --debug

# 检查MCP服务器是否被识别
claude mcp list

# 确认配置文件路径正确（使用绝对路径）
# 确认Python/node命令在PATH中可用
which python
which node

# 在server.py头部添加
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stdin = io.TextIOWrapper(sys.stdin.buffer, encoding='utf-8')

{
  "permissions": {
    "allow": ["mcp__weather-server__*"]
  }
}

# test_server.py
import pytest
from server import mcp

def test_get_current_weather():
    """测试天气查询工具"""
    # FastMCP提供test client用于测试
    from mcp.server.fastmcp import FastMCP

    result = mcp._tool_manager.call_tool("get_current_weather", {"city": "北京"})
    assert "28" in result or "北京" in str(result)

def test_get_forecast_validation():
    """测试预报参数校验"""
    result = mcp._tool_manager.call_tool(
        "get_weather_forecast", {"city": "北京", "days": 10}
    )
    assert "最多支持7天" in str(result)

mcp-code-reviewer/
├── server.py           # 主服务器文件
├── analyzers/          # 分析模块
│   ├── __init__.py
│   ├── complexity.py   # 圈复杂度分析
│   └── security.py     # 安全检查
├── requirements.txt
└── README.md

# server.py
import os
import subprocess
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "code-reviewer",
    version="1.0.0",
    description="代码审查助手 - 分析Git变更、检测安全和复杂度问题"
)


def run_git(cmd: list[str], cwd: str | None = None) -> str:
    """执行Git命令并返回输出"""
    result = subprocess.run(
        ["git"] + cmd,
        capture_output=True,
        text=True,
        cwd=cwd or os.getcwd()
    )
    return result.stdout.strip()


@mcp.tool()
def analyze_git_diff(repo_path: str = ".", base_branch: str = "main") -> str:
    """分析当前分支与基准分支的代码差异

    Args:
        repo_path: Git仓库路径
        base_branch: 基准分支名，默认为main

    Returns:
        变更文件列表和统计信息
    """
    # 获取差异统计
    diff_stat = run_git(
        ["diff", "--stat", f"origin/{base_branch}...HEAD"],
        cwd=repo_path
    )

    # 获取变更文件列表
    diff_files = run_git(
        ["diff", "--name-only", f"origin/{base_branch}...HEAD"],
        cwd=repo_path
    )

    files = diff_files.split("\n") if diff_files else []

    # 按文件类型分类
    categories = {
        "🐍 Python": [], "📄 TypeScript": [], "📄 JavaScript": [],
        "🏗️ Config": [], "📦 Other": []
    }

    ext_map = {
        ".py": "🐍 Python", ".ts": "📄 TypeScript", ".tsx": "📄 TypeScript",
        ".js": "📄 JavaScript", ".jsx": "📄 JavaScript",
        ".json": "🏗️ Config", ".yaml": "🏗️ Config", ".yml": "🏗️ Config",
    }

    for f in files:
        ext = Path(f).suffix
        cat = ext_map.get(ext, "📦 Other")
        categories[cat].append(f)

    result = f"📊 变更统计（对比 {base_branch}）:\n\n{diff_stat}\n\n"
    result += "📁 变更文件分类:\n"
    for cat, cat_files in categories.items():
        if cat_files:
            result += f"\n{cat}:\n"
            for f in cat_files:
                result += f"  - {f}\n"

    return result


@mcp.tool()
def check_python_complexity(file_path: str) -> str:
    """检查Python文件的圈复杂度

    Args:
        file_path: Python文件路径
    """
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except FileNotFoundError:
        return f"❌ 文件不存在: {file_path}"

    # 简单的复杂度估算（计算if/for/while/and/or数量）
    complexity_keywords = ["if ", "elif ", "for ", "while ", "and ", "or ", "except "]
    total_complexity = 1  # 基础复杂度
    issues = []

    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        if stripped.startswith("#"):
            continue
        for kw in complexity_keywords:
            if kw in stripped:
                total_complexity += 1

    if total_complexity > 15:
        level = "🔴 高复杂度"
    elif total_complexity > 8:
        level = "🟡 中等复杂度"
    else:
        level = "🟢 低复杂度"

    return (
        f"📋 复杂度分析: {file_path}\n"
        f"圈复杂度: {total_complexity} {level}\n"
        f"总行数: {len(lines)}\n"
        f"代码行数: {sum(1 for l in lines if l.strip() and not l.strip().startswith('#'))}"
    )


@mcp.tool()
def generate_review_summary(repo_path: str = ".", base_branch: str = "main") -> str:
    """生成代码审查摘要报告

    Args:
        repo_path: 仓库路径
        base_branch: 基准分支
    """
    diff_info = analyze_git_diff(repo_path, base_branch)

    # 获取提交信息
    commits = run_git(
        ["log", f"origin/{base_branch}...HEAD", "--oneline"],
        cwd=repo_path
    )

    return (
        f"📋 代码审查摘要报告\n"
        f"{'='*40}\n\n"
        f"{diff_info}\n\n"
        f"💬 提交记录:\n{commits}\n\n"
        f"建议审查重点:\n"
        f"  1. 检查新增文件的安全性和复杂度\n"
        f"  2. 确认配置文件变更是否符合预期\n"
        f"  3. 验证测试覆盖率"
    )


@mcp.resource("review://template")
def get_review_template() -> str:
    """代码审查模板"""
    return """## 代码审查清单

### 功能性
- [ ] 代码实现了预期功能
- [ ] 边界条件已处理
- [ ] 错误处理完善

### 安全性
- [ ] 无硬编码密钥/密码
- [ ] 输入已校验和清理
- [ ] 无SQL注入/XSS风险

### 可维护性
- [ ] 命名清晰有意义
- [ ] 复杂度可控
- [ ] 注释充分且准确

### 测试
- [ ] 单元测试覆盖关键路径
- [ ] 集成测试通过
- [ ] 性能测试（如适用）"""


@mcp.prompt()
def code_review_prompt(file_path: str) -> str:
    """对指定文件进行深度代码审查

    Args:
        file_path: 要审查的文件路径
    """
    return f"""请对文件 {file_path} 进行全面的代码审查。

审查维度：
1. **代码质量**：命名规范、代码风格、可读性
2. **安全性**：潜在漏洞、敏感信息泄露、输入校验
3. **性能**：不必要的计算、内存泄漏、N+1查询
4. **可维护性**：耦合度、复杂度、可扩展性
5. **测试**：测试覆盖、边界条件、错误路径

请用中文输出审查报告，按严重程度分级（🔴严重 🟡警告 🟢建议）。"""


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

# 添加服务器
claude mcp add code-reviewer \
  -s project \
  -- python /path/to/mcp-code-reviewer/server.py

# 验证
claude mcp list

请分析当前分支相对于main的代码变更

请检查 src/api/auth.py 的复杂度

请生成代码审查摘要报告

graph TB
    S["MCP服务器安全"] --> A["输入验证"]
    S --> B["权限最小化"]
    S --> C["敏感信息保护"]
    S --> D["错误处理"]

    A --> A1["使用Zod/Pydantic校验参数"]
    A --> A2["限制文件系统访问范围"]
    A --> A3["防止命令注入"]

    B --> B1["只暴露必要的工具"]
    B --> B2["不直接暴露数据库连接"]
    B --> B3["使用只读权限优先"]

    C --> C1["环境变量存储密钥"]
    C --> C2["不在日志中输出敏感数据"]
    C --> C3["响应中脱敏处理"]

    D --> D1["所有异常捕获并返回友好消息"]
    D --> D2["设置超时限制"]
    D --> D3["记录操作审计日志"]

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("async-example")

@mcp.tool()
async def fetch_api_data(url: str) -> str:
    """异步获取API数据

    Args:
        url: API端点URL
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(url)
            response.raise_for_status()
            return f"✅ 状态码: {response.status_code}\n数据长度: {len(response.text)} 字符"
        except httpx.TimeoutException:
            return "❌ 请求超时"
        except httpx.HTTPStatusError as e:
            return f"❌ HTTP错误: {e.response.status_code}"

# 1. 创建pyproject.toml
# 2. 添加console_scripts入口
# 3. 构建并发布
pip install build
python -m build
pip install twine
twine upload dist/*

# 1. 配置package.json的bin字段
# 2. 在入口文件添加shebang：#!/usr/bin/env node
# 3. 发布
npm publish

# Python包
claude mcp add my-tool -- my-tool-server

# npm包
claude mcp add my-tool -- npx -y my-mcp-server

flowchart TD
    A["🚀 需求分析"] --> B["选择技术栈"]
    B --> B1{"Python / TypeScript?"}
    B1 -->|"Python"| C1["安装FastMCP<br/>pip install fastmcp"]
    B1 -->|"TypeScript"| C2["安装MCP SDK<br/>npm install @modelcontextprotocol/sdk"]

    C1 --> D["实现三大原语"]
    C2 --> D

    D --> D1["🛠️ Tools"]
    D --> D2["📄 Resources"]
    D --> D3["💬 Prompts"]

    D1 --> E["本地测试"]
    D2 --> E
    D3 --> E

    E --> E1["MCP Inspector<br/>可视化调试"]
    E --> E2["单元测试"]

    E1 --> F["注册到Claude Code"]
    E2 --> F

    F --> F1["claude mcp add"]
    F --> F2["或 .mcp.json"]

    F1 --> G["集成测试"]
    F2 --> G

    G --> G1{"通过?"}
    G1 -->|"是"| H["📦 打包发布"]
    G1 -->|"否"| D

    H --> I["npm publish / PyPI upload"]
    I --> J["🎉 完成"]