MCP：连接外部工具

目标：使用 Model Context Protocol 连接外部工具，扩展 Claude 的能力边界
预计时间：35 分钟
对应官方文档：MCP、MCP Quickstart

什么是 MCP？

MCP（Model Context Protocol） 是 Anthropic 推出的开放协议，让 Claude 可以连接各种外部工具：

🗄️ 数据库（PostgreSQL、MySQL、MongoDB）
🔍 搜索引擎（Google、Bing）
📊 监控平台（Datadog、Grafana）
🛠️ 开发工具（Jira、GitHub、Docker）
☁️ 云服务（AWS、GCP、Azure）

类比： MCP 就像是 AI 的 USB 接口，任何支持 MCP 的工具都能即插即用。

快速开始

1. 查看可用 MCP Servers

> /mcp list

Available MCP servers:
  - filesystem     文件系统操作
  - fetch          HTTP 请求
  - postgres       PostgreSQL 数据库
  - sqlite         SQLite 数据库
  - git            Git 操作

2. 添加 MCP Server

> /mcp add postgres

Connection string: postgresql://user:pass@localhost/dbname

3. 使用 MCP 工具

> 查询数据库中最近 7 天的活跃用户数量

[Claude 使用 postgres MCP 执行 SQL 查询]

常用 MCP Server 配置

PostgreSQL

// ~/.claude/mcp.json
{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/mydb"
      }
    }
  }
}

SQLite

{
  "mcpServers": {
    "sqlite": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-sqlite", "/path/to/db.sqlite"]
    }
  }
}

GitHub

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxx"
      }
    }
  }
}

实战：数据库查询

场景：分析用户数据

配置了 PostgreSQL MCP 后：

> 帮我分析最近 30 天的用户注册趋势，按天统计

Claude: 我将查询用户注册数据...

SQL: SELECT DATE(created_at) as date, COUNT(*) 
     FROM users 
     WHERE created_at >= NOW() - INTERVAL '30 days'
     GROUP BY DATE(created_at)
     ORDER BY date

结果：
2025-05-20 | 45
2025-05-21 | 52
...

趋势：注册量稳步上升，周末有小幅下降。

场景：结合代码和数据库

> 用户报告订单状态显示错误。
> 
> 相关代码：@src/orders.py
> 相关表：orders, order_status
> 
> 请诊断问题并修复。

Claude:
1. 读取代码逻辑
2. 查询数据库 schema
3. 发现 status 字段映射错误
4. 修复代码
5. 验证修复

自建 MCP Server

如果现有工具不满足需求，可以自己开发：

Python 示例

# my_mcp_server.py
from mcp.server import Server
from mcp.types import TextContent
 
app = Server("my-tool")
 
@app.call_tool()
async def query_internal_api(name: str, arguments: dict):
    if name == "get_user":
        user_id = arguments["user_id"]
        # 调用内部 API
        result = await internal_api.get_user(user_id)
        return [TextContent(type="text", text=str(result))]
 
if __name__ == "__main__":
    app.run()

配置使用

{
  "mcpServers": {
    "my-tool": {
      "command": "python",
      "args": ["/path/to/my_mcp_server.py"]
    }
  }
}

MCP 安全注意事项

风险	防护措施
数据库误操作	使用只读账号
敏感信息泄露	环境变量管理 token
恶意工具	只安装可信来源的 MCP
权限过大	企业用 managed-mcp 统一管控

企业配置参考：Managed MCP

补充：MCP 架构详解与实战

MCP 连接架构图

完整配置示例

示例 1：多 Server 完整 `~/.claude/mcp.json`

{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://readonly_user:pass@localhost:5432/app_db"
      }
    },
    "sqlite": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-sqlite", "/home/user/data/app.db"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "fetch": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-fetch"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    },
    "my-company-api": {
      "command": "python",
      "args": ["/home/user/tools/mcp-company-server.py"],
      "env": {
        "API_BASE_URL": "https://internal.api.company.com",
        "API_KEY": "${COMPANY_API_KEY}"
      }
    }
  }
}

提示：使用 ${VAR} 引用环境变量，避免密钥硬编码。

示例 2：Python 自建 MCP Server（完整可运行代码）

# mcp-company-server.py
# 连接公司内部用户管理 API 的 MCP Server
 
import os
import json
import asyncio
from urllib import request, error
from mcp.server import Server
from mcp.types import TextContent, Tool
 
API_BASE = os.environ["API_BASE_URL"]
API_KEY = os.environ["API_KEY"]
 
app = Server("company-api")
 
@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_user_by_id",
            description="根据用户 ID 查询用户信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "string", "description": "用户唯一标识"}
                },
                "required": ["user_id"]
            }
        ),
        Tool(
            name="search_users",
            description="按姓名或邮箱搜索用户",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "limit": {"type": "integer", "default": 10}
                },
                "required": ["query"]
            }
        )
    ]
 
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    if name == "get_user_by_id":
        user_id = arguments["user_id"]
        req = request.Request(
            f"{API_BASE}/users/{user_id}",
            headers=headers
        )
        try:
            with request.urlopen(req) as resp:
                data = json.loads(resp.read().decode())
                return [TextContent(type="text", text=json.dumps(data, indent=2, ensure_ascii=False))]
        except error.HTTPError as e:
            return [TextContent(type="text", text=f"Error: {e.code} - {e.reason}")]
    
    elif name == "search_users":
        query = arguments["query"]
        limit = arguments.get("limit", 10)
        req = request.Request(
            f"{API_BASE}/users/search?q={query}&limit={limit}",
            headers=headers
        )
        with request.urlopen(req) as resp:
            data = json.loads(resp.read().decode())
            return [TextContent(type="text", text=json.dumps(data, indent=2, ensure_ascii=False))]
    
    return [TextContent(type="text", text=f"Unknown tool: {name}")]
 
if __name__ == "__main__":
    asyncio.run(app.run())

安装依赖：

pip install mcp

配置引用：

{
  "mcpServers": {
    "company-api": {
      "command": "python",
      "args": ["/home/user/tools/mcp-company-server.py"],
      "env": {
        "API_BASE_URL": "https://internal.api.company.com",
        "API_KEY": "sk-xxx"
      }
    }
  }
}

示例 3：带权限控制的企业 MCP 配置

{
  "mcpServers": {
    "postgres-readonly": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://readonly:[email protected]/analytics"
      }
    }
  },
  "mcpPolicies": {
    "default": {
      "requireApproval": ["postgres-readonly"],
      "maxRowsPerQuery": 1000,
      "allowedCommands": ["SELECT", "EXPLAIN", "SHOW"],
      "deniedCommands": ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
    }
  }
}

实战场景

场景一：全栈调试（数据库 + GitHub + 代码联动）

背景：用户反馈订单状态显示异常，需要同时排查代码逻辑、数据库实际状态、以及近期相关代码变更。

步骤：

开发者已配置 postgres、github、filesystem 三个 MCP Server

在 Claude Code 中提问：

> 用户反馈订单 #12345 状态显示为"待支付"，但实际已扣款。
> 相关代码：@src/orders/service.py
> 相关表：orders, payments
> 请诊断问题。

Claude 自动执行：
- 读取 src/orders/service.py 理解状态流转逻辑
- 通过 postgres MCP 查询订单 #12345 的 orders.status 和 payments.status
- 通过 github MCP 查看最近 7 天谁修改过 src/orders/service.py
发现代码中状态判断条件写反，且 3 天前的 PR 引入了该 Bug

结果：

原本需要切换 3 个工具（IDE、DB 客户端、GitHub Web）的工作，在一个会话内完成
诊断时间从平均 30 分钟缩短到 5 分钟
Claude 直接给出修复代码，并通过 postgres MCP 验证修复后的状态判断逻辑
同时自动生成了该 Bug 的回归测试用例

场景二：内部 API 集成（自建 MCP 连接公司服务）

背景：公司有一套内部用户管理系统（非标准 SaaS），开发者在写客服后台功能时，经常需要人工查询用户状态、订单历史，效率低下。

步骤：

后端工程师按示例 2 编写 mcp-company-server.py，封装内部 API 的 get_user_by_id 和 search_users
配置到团队共享的 mcp.json，通过环境变量注入 API Key

客服后台开发者在 Claude Code 中直接提问：

> 用户 [email protected] 反馈无法登录，请查询该用户状态并分析可能原因。
> 相关代码：@src/auth/login.py

Claude 通过自建 MCP 查询用户状态，发现 is_active=false，再查代码发现是密码错误次数超限导致的自动锁定

结果：

开发者无需离开编码环境即可查询内部系统数据
减少了在多个内部平台之间切换的上下文中断
自建 MCP 一次开发，全团队复用，后续新功能开发均可直接调用
通过 MCP 的标准化接口，未来替换底层内部 API 时，上层调用代码无需改动

下一步

→ 06. 子代理：分工协作处理复杂任务