如何为 Claude Code 构建自定义 MCP 服务器：手把手教程

┌─────────────┐     stdio/HTTP      ┌──────────────┐
│  Claude Code │ ◄──────────────────► │  你的 MCP     │
│  (客户端)    │   JSON-RPC 2.0      │  服务器       │
└─────────────┘                      └──────┬───────┘
                                            │
                                     ┌──────▼───────┐
                                     │  你的工具     │
                                     │  (Python 函数) │
                                     └──────────────┘

mkdir my-mcp-server && cd my-mcp-server
python3 -m venv .venv && source .venv/bin/activate
pip install mcp httpx

# !/usr/bin/env python3
"""自定义 MCP 服务器，包含三个工具：天气查询、文件摘要、待办管理。"""

import json
import os
from datetime import datetime
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server

# 内存中的待办列表（演示用；生产环境用 SQLite）
todos = []
next_id = 1

server = Server("my-tools-server")

# ── 工具 1：天气查询 ──────────────────────────────────
@server.list_tools()
async def list_tools() -> list:
    return [
        {
            "name": "get_weather",
            "description": "获取指定城市的当前天气。返回温度、天气状况和湿度。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名，例如 '北京' 或 '东京'"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位（默认：celsius）"
                    }
                },
                "required": ["city"]
            }
        },
        {
            "name": "summarize_file",
            "description": "摘要文本文件内容——返回行数、词数和前 500 个字符。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "要摘要的文件绝对路径"
                    }
                },
                "required": ["filepath"]
            }
        },
        {
            "name": "todo_manager",
            "description": "管理简单待办列表。支持 'add'、'list'、'done'、'clear' 操作。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": ["add", "list", "done", "clear"],
                        "description": "要对待办列表执行的操作"
                    },
                    "text": {
                        "type": "string",
                        "description": "待办事项文本（'add' 操作必填）"
                    },
                    "id": {
                        "type": "integer",
                        "description": "要标记为完成的待办 ID（'done' 操作必填）"
                    }
                },
                "required": ["action"]
            }
        }
    ]

# ── 工具执行 ──────────────────────────────────────────
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    global todos, next_id

    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")

# 模拟天气数据（替换为真实 API 调用）
        weather_data = {
            "北京": {"temp": 28, "condition": "晴", "humidity": 45},
            "上海": {"temp": 32, "condition": "多云", "humidity": 70},
            "东京": {"temp": 28, "condition": "晴", "humidity": 65},
            "纽约": {"temp": 24, "condition": "局部多云", "humidity": 55},
        }
        data = weather_data.get(city, {"temp": 22, "condition": "未知", "humidity": 60})
        return [{
            "type": "text",
            "text": f"📍 {city}: {data['temp']}°{'C' if units == 'celsius' else 'F'}, {data['condition']}, 湿度: {data['humidity']}%"
        }]

    elif name == "summarize_file":
        filepath = arguments["filepath"]
        if not os.path.exists(filepath):
            return [{"type": "text", "text": f"❌ 文件未找到: {filepath}"}]
        if not os.path.isfile(filepath):
            return [{"type": "text", "text": f"❌ 不是文件: {filepath}"}]
        try:
            with open(filepath, "r") as f:
                content = f.read()
            lines = content.split("\n")
            words = content.split()
            preview = content[:500]
            return [{
                "type": "text",
                "text": f"📄 {os.path.basename(filepath)}\n行数: {len(lines)}\n词数: {len(words)}\n字符数: {len(content)}\n\n--- 预览（前 500 字符）---\n{preview}"
            }]
        except Exception as e:
            return [{"type": "text", "text": f"❌ 读取文件出错: {str(e)}"}]

    elif name == "todo_manager":
        action = arguments["action"]
        if action == "add":
            text = arguments.get("text", "")
            if not text:
                return [{"type": "text", "text": "❌ 'add' 操作需要提供 'text' 参数"}]
            todos.append({"id": next_id, "text": text, "done": False, "created": datetime.now().isoformat()})
            next_id += 1
            return [{"type": "text", "text": f"✅ 已添加待办 #{next_id-1}: {text}"}]
        elif action == "list":
            if not todos:
                return [{"type": "text", "text": "📋 暂无待办事项。"}]
            lines = ["📋 待办列表:"]
            for t in todos:
                status = "✅" if t["done"] else "⬜"
                lines.append(f"  {status} #{t['id']}: {t['text']}")
            return [{"type": "text", "text": "\n".join(lines)}]
        elif action == "done":
            tid = arguments.get("id")
            if tid is None:
                return [{"type": "text", "text": "❌ 'done' 操作需要提供 'id' 参数"}]
            for t in todos:
                if t["id"] == tid:
                    t["done"] = True
                    return [{"type": "text", "text": f"✅ 已将 #{tid} 标记为完成: {t['text']}"}]
            return [{"type": "text", "text": f"❌ 未找到待办 #{tid}"}]
        elif action == "clear":
            count = len(todos)
            todos.clear()
            return [{"type": "text", "text": f"🗑️ 已清除 {count} 条待办"}]

    return [{"type": "text", "text": f"未知工具: {name}"}]

# ── 入口 ──────────────────────────────────────────────
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationCapabilities(
                sampling={},
                experimental={},
            ),
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

python3 server.py --help  # 应该不会崩溃

# 按 Ctrl+C 停止（它作为长驻进程运行）

{
  "mcpServers": {
    "my-tools": {
      "command": "python3",
      "args": ["/absolute/path/to/my-mcp-server/server.py"],
      "env": {}
    }
  }
}

claude

> 东京现在天气怎么样？

> 帮我摘要一下 /Users/you/project/README.md 文件

> 把"买菜"加到我的待办列表里

pip install httpx

import httpx

# 添加你的 API key（在 https://openweathermap.org/api 免费获取）
OPENWEATHER_API_KEY = "你的-key"

# ... 在 call_tool 中，替换 get_weather 代码块：

elif name == "get_weather":
    city = arguments["city"]
    units = arguments.get("units", "celsius")
    unit_param = "metric" if units == "celsius" else "imperial"

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": OPENWEATHER_API_KEY, "units": unit_param},
            timeout=10.0
        )

    if resp.status_code == 404:
        return [{"type": "text", "text": f"❌ 未找到城市: {city}"}]
    elif resp.status_code != 200:
        return [{"type": "text", "text": f"❌ API 错误: {resp.status_code}"}]

    data = resp.json()
    temp = data["main"]["temp"]
    condition = data["weather"][0]["description"]
    humidity = data["main"]["humidity"]
    symbol = "°C" if units == "celsius" else "°F"

    return [{
        "type": "text",
        "text": f"📍 {data['name']}: {temp}{symbol}, {condition}, 湿度: {humidity}%"
    }]

{
  "mcpServers": {
    "my-tools": {
      "command": "python3",
      "args": ["/absolute/path/to/my-mcp-server/server.py"],
      "env": {
        "OPENWEATHER_API_KEY": "你的真实API-key"
      }
    }
  }
}