MCP Server 开发实战

my-mcp-server/
├── server.py          # 主服务文件
├── tools/             # 工具实现
│   ├── __init__.py
│   ├── search.py      # 搜索工具
│   └── analytics.py   # 分析工具
├── resources/         # 资源实现
│   ├── __init__.py
│   └── data.py        # 数据资源
├── prompts/           # 提示词模板
│   ├── __init__.py
│   └── templates.py
├── requirements.txt
└── mcp-config.json    # MCP 配置文件

@server.list_tools()
async def list_tools() -> list[Tool]:
    """返回所有可用工具的列表"""
    return [
        Tool(
            name="query_data",
            description="查询数据分析结果。当需要获取特定指标或统计数据时使用此工具。",
            inputSchema={
                "type": "object",
                "properties": {
                    "metric": {
                        "type": "string",
                        "description": "要查询的指标名称，如 'revenue', 'users', 'conversion'",
                        "enum": ["revenue", "users", "conversion", "retention"]
                    },
                    "time_range": {
                        "type": "string",
                        "description": "时间范围",
                        "enum": ["today", "week", "month", "quarter", "year"]
                    },
                    "filters": {
                        "type": "object",
                        "description": "可选的过滤条件",
                        "properties": {
                            "region": {"type": "string"},
                            "product": {"type": "string"}
                        }
                    }
                },
                "required": ["metric", "time_range"]
            }
        ),
        Tool(
            name="generate_report",
            description="生成数据分析报告。支持多种格式和详细程度。",
            inputSchema={
                "type": "object",
                "properties": {
                    "report_type": {
                        "type": "string",
                        "enum": ["daily", "weekly", "monthly", "custom"]
                    },
                    "format": {
                        "type": "string",
                        "enum": ["text", "markdown", "json"],
                        "default": "markdown"
                    },
                    "include_charts": {
                        "type": "boolean",
                        "default": True
                    }
                },
                "required": ["report_type"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """处理工具调用请求"""

@server.list_resources()
async def list_resources() -> list[Resource]:
    """返回所有可用资源的列表"""
    return [
        Resource(
            uri="data://metrics/overview",
            name="数据概览",
            description="所有核心指标的概览数据",
            mimeType="application/json"
        ),
        Resource(
            uri="data://config/settings",
            name="系统配置",
            description="当前数据分析系统的配置信息",
            mimeType="application/json"
        )
    ]

@server.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
    """返回动态资源模板"""
    return [
        ResourceTemplate(
            uriTemplate="data://metrics/{metric_name}",
            name="指标详情",
            description="获取指定指标的详细数据",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    """读取资源内容"""
    if uri == "data://metrics/overview":
        return json.dumps(get_overview_data())
    elif uri.startswith("data://metrics/"):
        metric_name = uri.split("/")[-1]
        return json.dumps(get_metric_detail(metric_name))
    else:
        raise ValueError(f"未知资源: {uri}")

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="analyze_trend",
            description="分析数据趋势并提供洞察",
            arguments=[
                PromptArgument(
                    name="metric",
                    description="要分析的指标",
                    required=True
                ),
                PromptArgument(
                    name="period",
                    description="分析周期（如 7d, 30d, 90d）",
                    required=True
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> dict:
    if name == "analyze_trend":
        return {
            "messages": [
                PromptMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"""请分析 {arguments['metric']} 指标在最近 {arguments['period']} 的趋势：

async def handle_query_data(arguments: dict) -> str:
    """查询数据的实际实现"""
    metric = arguments["metric"]
    time_range = arguments["time_range"]
    filters = arguments.get("filters", {})

async def handle_generate_report(arguments: dict) -> str:
    """生成报告的实际实现"""
    report_type = arguments["report_type"]
    fmt = arguments.get("format", "markdown")

// 注册工具列表
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "query_data",
      description: "查询数据分析结果",
      inputSchema: {
        type: "object",
        properties: {
          metric: { type: "string", enum: ["revenue", "users", "conversion"] },
          time_range: { type: "string", enum: ["today", "week", "month"] },
        },
        required: ["metric", "time_range"],
      },
    },
  ],
}));

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

┌──────────────────────────────────────────────┐
│              Claude Desktop                  │
│                                              │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐     │
│  │ Client 1 │  │ Client 2 │  │ Client 3 │     │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘     │
└───────┼─────────────┼─────────────┼──────────┘
        │             │             │
   ┌────┴────┐  ┌────┴────┐  ┌────┴────┐
   │Database │  │ GitHub  │  │ File    │
   │ Server  │  │ Server  │  │ Server  │
   └─────────┘  └─────────┘  └─────────┘

Claude 可以跨 Server 工作：
"查看 GitHub 上的 Issue，查询数据库中的相关数据，把分析结果写入报告文件"
  → GitHub Server 获取 Issue
  → Database Server 查询数据
  → Filesystem Server 写入文件

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "long_running_analysis":
        # 发送进度通知
        await server.send_progress("正在连接数据库...")
        data = await fetch_data()