MCP 工具开发：手把手做一个自己的 MCP server

L4-07 讲过 MCP 是 “AI 工具的 USB”—— 这一篇我们实际做一个。

读完你能：

• 写一个能让 Claude Desktop / Cursor 等使用的 MCP server
• 暴露任何你需要的工具能力
• 加入 MCP 生态

选择语言

MCP 支持多种语言：

• TypeScript / Node（最成熟）
• Python（也很好）
• Go / Rust（小众但能用）

本文用 Python—— 最适合 AI / 数据场景。

安装

pip install mcp

最简 MCP Server

让我们做一个简单的”天气查询”工具：

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 创建 server
server = Server("weather-tool")

# 声明工具
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name (e.g., 'Beijing', 'San Francisco')"
                    }
                },
                "required": ["city"]
            }
        )
    ]

# 工具调用处理
@server.call_tool()
async def call_tool(name, arguments):
    if name == "get_weather":
        city = arguments["city"]
        # 这里就是你的真实业务逻辑
        weather = await fetch_weather_api(city)
        return [
            TextContent(
                type="text",
                text=f"Current weather in {city}: {weather}"
            )
        ]
    raise ValueError(f"Unknown tool: {name}")

# 实际查天气的函数
async def fetch_weather_api(city):
    # 调真实 weather API，比如 OpenWeatherMap
    # 这里简化
    return "Sunny, 22°C"

# 启动 server
if __name__ == "__main__":
    import asyncio
    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream, server.create_initialization_options())
    asyncio.run(main())

就这样——一个完整的 MCP server。

让 Claude Desktop 用它

1. 找到 Claude Desktop 配置文件

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json

2. 添加 server 配置

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

3. 重启 Claude Desktop

4. 用！

在 Claude Desktop 里： “What’s the weather in Beijing?”

Claude 自动调用你的 weather server—— 返回真实天气。

这就是 MCP 的魔力—— 工具 + LLM = 任意组合。

添加更复杂的工具

让我们加更多工具：

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_forecast",
            description="Get 5-day forecast",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "days": {"type": "integer", "default": 5, "maximum": 10}
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="compare_cities",
            description="Compare weather between multiple cities",
            inputSchema={
                "type": "object",
                "properties": {
                    "cities": {
                        "type": "array",
                        "items": {"type": "string"}
                    }
                },
                "required": ["cities"]
            }
        ),
    ]

@server.call_tool()
async def call_tool(name, arguments):
    if name == "get_weather":
        return await get_weather_handler(arguments)
    elif name == "get_forecast":
        return await get_forecast_handler(arguments)
    elif name == "compare_cities":
        return await compare_cities_handler(arguments)
    raise ValueError(f"Unknown tool: {name}")

每个工具一个 handler——清晰分工。

添加 Resources

不只是工具——也暴露可读资源：

from mcp.types import Resource

@server.list_resources()
async def list_resources():
    return [
        Resource(
            uri="weather://config",
            name="Weather Service Config",
            mimeType="application/json"
        ),
        Resource(
            uri="weather://history/beijing",
            name="Beijing Weather History",
            mimeType="text/csv"
        ),
    ]

@server.read_resource()
async def read_resource(uri):
    if uri == "weather://config":
        return json.dumps({"api_key": "...", "units": "celsius"})
    elif uri.startswith("weather://history/"):
        city = uri.split("/")[-1]
        return load_weather_history(city)
    raise ValueError(f"Unknown resource: {uri}")

LLM 能 “读” 这些 resource—— 让它了解上下文。

添加 Prompts

预定义的工作流：

from mcp.types import Prompt

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="weather_report",
            description="Generate a daily weather report",
            arguments=[
                {"name": "city", "description": "City name", "required": True}
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name, arguments):
    if name == "weather_report":
        city = arguments["city"]
        return {
            "description": f"Weather report for {city}",
            "messages": [
                {
                    "role": "user",
                    "content": {
                        "type": "text",
                        "text": f"""Please generate a daily weather report for {city}:
1. Use the get_weather tool to fetch current conditions
2. Use the get_forecast tool for the next 5 days
3. Format as: Current → Trend → Recommendations
"""
                    }
                }
            ]
        }

用户点 “weather_report” 预设 → Claude 自动按流程做。

生产级实现要点

1. 错误处理

@server.call_tool()
async def call_tool(name, arguments):
    try:
        if name == "get_weather":
            city = arguments.get("city")
            if not city:
                return [TextContent(type="text", text="Error: city is required")]
            return await get_weather_handler(arguments)
    except Exception as e:
        # 把错误返回给 LLM，不要崩
        return [TextContent(type="text", text=f"Error: {str(e)}")]

LLM 能看到错误 → 知道怎么调整。

2. 限速

from datetime import datetime
import collections

call_history = collections.defaultdict(list)

async def rate_limit_check(tool_name, max_per_minute=10):
    now = datetime.now()
    history = call_history[tool_name]
    # 清理 1 分钟前的记录
    history[:] = [t for t in history if (now - t).seconds < 60]
    if len(history) >= max_per_minute:
        raise Exception(f"Rate limit exceeded for {tool_name}")
    history.append(now)

3. 安全

关键工具加权限检查：

ALLOWED_PATHS = ["/home/user/projects", "/tmp"]

async def file_tool_handler(arguments):
    path = arguments["path"]
    # 不允许的路径直接拒
    if not any(path.startswith(p) for p in ALLOWED_PATHS):
        return [TextContent(type="text", text="Error: path not allowed")]
    # 不允许危险操作
    if "rm -rf" in arguments.get("command", ""):
        return [TextContent(type="text", text="Error: dangerous command rejected")]
    # 执行
    ...

4. 日志

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@server.call_tool()
async def call_tool(name, arguments):
    logger.info(f"Tool called: {name} with {arguments}")
    result = await actual_handler(name, arguments)
    logger.info(f"Tool returned: {result[:200]}")
    return result

5. 配置管理

import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("WEATHER_API_KEY")
ALLOWED_USERS = os.getenv("ALLOWED_USERS", "").split(",")

部署

本地（Claude Desktop / Cursor）

stdio transport——本地调用：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["server.py"],
      "env": {"WEATHER_API_KEY": "..."}
    }
  }
}

远程（HTTP）

MCP 也支持 HTTP transport：

from mcp.server.fastmcp import FastMCP

app = FastMCP("weather-tool")

@app.tool()
def get_weather(city: str) -> str:
    return f"Weather in {city}: sunny, 22°C"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

部署到 Cloudflare Workers / AWS Lambda / 自家服务器—— 让 MCP server 全球可用。

发布到 MCP 生态

如果你的 server 通用—— 发布给所有人用！

npm（TypeScript）

npm publish @your-org/mcp-server-weather

PyPI（Python）

pip install build twine
python -m build
twine upload dist/*

加入 MCP Servers 索引

提交 PR 到 modelcontextprotocol/servers —— 你的 server 出现在官方索引里。

全球开发者都能用。

一些灵感

社区 MCP server 例子：

Server	干啥
mcp-server-filesystem	读写本地文件
mcp-server-git	Git 操作
mcp-server-github	GitHub API
mcp-server-postgres	数据库查询
mcp-server-slack	Slack 消息
mcp-server-spotify	音乐控制
mcp-server-playwright	浏览器自动化

几乎任何能 API 化的服务，都已经有 / 应该有 MCP server。

一些建议

做什么

问自己：

• 你公司 / 项目里 LLM 想”做”什么但做不了？
• 你最常用的工具（Slack / Notion / GitHub）—— 有 MCP 包装吗？
• 你的内部 tool—— 包成 MCP 后能让 AI 用吗？

做不存在的 + 通用的 —— 容易广泛使用。

起步建议

1. 读官方文档：modelcontextprotocol.io
2. 看例子：mcp servers GitHub
3. Fork 改一改 —— 比从 0 写快
4. 在 Claude Desktop 上测
5. 发布 + 写文档

避免

• 不要做权限太宽的（“任意命令执行” 危险）
• 不要 hardcode 凭证（用 env var）
• 不要返回大量数据（截断、分页）
• 不要 ignore 错误（返回给 LLM 让它知道）

下一篇推荐：L4-12 LLM 成本优化 或 L4-13 Agent 在生产。