你的AI Agent能写代码、能搜网页、能发公众号——但就是不能查你自己的数据库。因为Agent的工具箱是厂商预设的，你的内部API、业务看板、私有数据源——它一个都不认识。MCP（Model Context Protocol）就是解决这个问题的协议：你写一个服务，Agent就能调用它。今天带你从零构建第一个MCP Server，给Agent装上你自己的工具。

你肯定遇到过这个场景

你用Claude Code或Hermes处理日常工作——写代码、查文档、做数据分析。一切都很流畅，直到你遇到这个问题：

"帮我查一下过去7天的用户注册量，对比上周同期，如果下降超过10%就发预警到企业微信。"

Agent能做什么？
- ✅ 它能写Python脚本
- ✅ 它能调API（如果你告诉它endpoint和token）
- ❌ 但它不知道你的数据库在哪
- ❌ 它不知道你的内部API鉴权方式
- ❌ 它不知道企业微信的webhook地址

你只好自己先查数据库 → 手动做对比 → 人工判断要不要发预警 → 再把结果喂给Agent。

这不是Agent能力不够，是你的工具没有"注册"到Agent的工具箱里。

MCP Server 就是解决这个问题的：你写一个轻量级服务，把内部API包装成Agent能调用的"工具"，Agent看到它就像看到内置的 web_search 一样自然。

MCP 是什么？（30秒速览）

MCP（Model Context Protocol）是 Anthropic 在2024年底发布的开放协议，本质是一个标准化的"Agent工具接口"：

┌──────────────┐      MCP协议(JSON-RPC)     ┌──────────────┐
│              │ ◄─────────────────────────► │              │
│  AI Agent    │   tools/list → 有哪些工具？  │  MCP Server  │
│  (Claude/Hermes│  tools/call → 请执行这个工具 │  (你写的服务)  │
│   /Cursor等)  │   resources/read → 读数据    │              │
│              │                            │              │
└──────────────┘                            └──────┬───────┘
                                                   │
                                            ┌──────▼───────┐
                                            │  你的内部系统   │
                                            │ (DB/API/看板)  │
                                            └──────────────┘

MCP Server 做的事情非常简单：接收JSON-RPC请求，执行对应的函数，返回结果。它可以用任何语言实现——Python、Node.js、Go、Rust都行。

三大核心能力：

能力	说明	典型场景
Tools	Agent可以调用的函数	查数据库、调API、发消息
Resources	Agent可以读取的数据	文档、配置、实时指标
Prompts	预定义的提示词模板	标准化任务流程

今天这篇教程聚焦在 Tools 上——因为这是90%场景的核心需求。

实战：从零构建「业务数据查询」MCP Server

我们来写一个真实可用的 MCP Server，暴露3个工具给Agent：

get_daily_metrics — 查询某日的业务指标（模拟数据库查询）
compare_metrics — 对比两个时间段的指标变化
send_alert — 向企业微信发送预警消息

第一步：安装依赖

# Python 3.10+ 项目
mkdir my-mcp-server && cd my-mcp-server
python3 -m venv venv && source venv/bin/activate
pip install mcp fastmcp

fastmcp 是 MCP 官方推荐的 Python 高级封装，比原始 mcp 低层 API 简洁10倍。但如果你需要对协议有完全控制，也可以直接用 mcp。

第二步：编写 Server（核心代码）

# server.py — 完整可运行的 MCP Server
from fastmcp import FastMCP
from datetime import datetime, timedelta
import random
import json

# 初始化 MCP Server
mcp = FastMCP("业务数据查询助手", port=8000)

# ============ 工具1: 查询单日指标 ============
@mcp.tool()
def get_daily_metrics(date: str) -> dict:
    """
    查询指定日期的业务指标。

    Args:
        date: 日期，格式 YYYY-MM-DD（如 2026-07-24）

    Returns:
        包含注册数、活跃用户数、付费金额的字典
    """
    # 模拟数据库查询（生产环境替换为真实 DB 查询）
    base = {
        "date": date,
        "new_users": random.randint(80, 200),
        "active_users": random.randint(5000, 8000),
        "revenue": round(random.uniform(3000, 15000), 2),
        "conversion_rate": round(random.uniform(0.03, 0.12), 4)
    }
    return base

# ============ 工具2: 对比两个时间段 ============
@mcp.tool()
def compare_metrics(date_a: str, date_b: str) -> dict:
    """
    对比两个日期的业务指标，计算变化率。

    Args:
        date_a: 基准日期，格式 YYYY-MM-DD
        date_b: 对比日期，格式 YYYY-MM-DD

    Returns:
        各项指标的变化率和原始数据
    """
    a = get_daily_metrics(date_a)
    b = get_daily_metrics(date_b)

    def pct_change(old, new):
        if old == 0:
            return None
        return round((new - old) / old * 100, 2)

    return {
        "baseline": a,
        "comparison": b,
        "changes": {
            "new_users": f"{pct_change(a['new_users'], b['new_users'])}%",
            "active_users": f"{pct_change(a['active_users'], b['active_users'])}%",
            "revenue": f"{pct_change(a['revenue'], b['revenue'])}%",
            "conversion_rate": f"{pct_change(a['conversion_rate'], b['conversion_rate'])}%"
        }
    }

# ============ 工具3: 发送企业微信预警 ============
@mcp.tool()
def send_alert(message: str, level: str = "warning") -> dict:
    """
    向企业微信群机器人发送预警消息。

    Args:
        message: 预警消息内容（支持 Markdown）
        level: 预警级别，可选 info/warning/critical

    Returns:
        发送结果
    """
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

    # 根据级别选择颜色和前缀
    emoji_map = {
        "info": "ℹ️",
        "warning": "⚠️", 
        "critical": "🚨"
    }
    prefix = emoji_map.get(level, "📢")

    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"{prefix} **{level.upper()}**\n{message}"
        }
    }

    # 生产环境中取消注释：
    # import requests
    # resp = requests.post(webhook_url, json=payload)
    # return {"status": resp.status_code, "body": resp.json()}

    # 演示模式：模拟发送
    return {
        "status": "simulated",
        "payload": payload,
        "note": "生产环境请替换 webhook_url 并取消 requests.post 注释"
    }

# 启动 Server
if __name__ == "__main__":
    mcp.run()

第三步：配置 Agent 连接 MCP Server

Claude Code（.claude/settings.json）：

{
  "mcpServers": {
    "business-metrics": {
      "command": "python3",
      "args": ["/path/to/my-mcp-server/server.py"],
      "env": {
        "WECHAT_WEBHOOK_KEY": "your_actual_key_here"
      }
    }
  }
}

Hermes Agent（config.yaml）：

mcp_servers:
  - name: business-metrics
    command: python3
    args:
      - /path/to/my-mcp-server/server.py
    description: 查询业务指标和发送预警

配置完成后重启 Agent，它就能看到这3个新工具了。

第四步：测试效果

启动 Server 后，在 Agent 对话中直接说：

"查一下2026-07-24和2026-07-17的指标对比，如果付费转化率下降超过10%，发一条critical预警到企业微信。"

Agent 会自动：
1. 调用 compare_metrics("2026-07-24", "2026-07-17")
2. 分析返回的 changes 数据
3. 如果 conversion_rate 下降超过10%，调用 send_alert(message="...", level="critical")

全程不需要你告诉它 API endpoint、不需要你写查询 SQL、不需要你手动复制粘贴结果。

进阶：生产环境 3 个必做优化

1. 真实数据库集成

把 get_daily_metrics 中的随机数据换成真实查询：

import sqlite3

@mcp.tool()
def get_daily_metrics(date: str) -> dict:
    conn = sqlite3.connect("/data/metrics.db")
    cursor = conn.cursor()

    cursor.execute("""
        SELECT new_users, active_users, revenue, conversion_rate
        FROM daily_metrics WHERE date = ?
    """, (date,))

    row = cursor.fetchone()
    conn.close()

    if row:
        return {
            "date": date,
            "new_users": row[0],
            "active_users": row[1],
            "revenue": row[2],
            "conversion_rate": row[3]
        }
    return {"error": f"No data for {date}"}

2. 鉴权与安全

不要让 Agent 直接拿到数据库连接——加一层鉴权中间件：

from fastmcp import FastMCP, Context

mcp = FastMCP("secure-metrics")

@mcp.tool()
def get_daily_metrics(date: str, ctx: Context) -> dict:
    # 从请求上下文中获取鉴权信息
    api_key = ctx.request_context.headers.get("x-api-key")
    if api_key != os.environ.get("MCP_API_KEY"):
        return {"error": "Unauthorized"}
    # ... 实际查询逻辑

3. 错误处理与重试

Agent 调用工具时可能遇到超时或临时故障，做好错误处理：

@mcp.tool()
def get_daily_metrics(date: str) -> dict:
    max_retries = 3
    for attempt in range(max_retries):
        try:
            # 查询逻辑
            result = query_db(date)
            return result
        except TimeoutError:
            if attempt == max_retries - 1:
                return {"error": "Database timeout after 3 retries"}
            time.sleep(1)
        except Exception as e:
            return {"error": str(e)}

对比：MCP Server vs 传统 API 集成

维度	传统 API 集成	MCP Server
Agent 发现工具	手动告诉 Agent API 文档	`tools/list` 自动发现
参数校验	Agent 需要猜测格式	JSON Schema 自动校验
错误处理	靠 Agent 自己解析 HTTP 状态码	结构化错误返回
工具组合	Agent 无法知道哪些工具相关	MCP Server 内聚相关工具
多 Agent 复用	每个 Agent 要单独配置	启动一个 Server，所有 Agent 共用
版本管理	改 API 要通知所有 Agent 用户	更新 Server，Agent 自动感知

一句话总结：MCP Server 让"内部工具"变得像"内置工具"一样好用。

小结

今天我们完成了：

✅ 理解 MCP 的核心价值——让 Agent 能调用你的内部系统
✅ 从零构建了一个 business-metrics MCP Server（3个工具）
✅ 配置 Claude Code / Hermes Agent 连接 MCP Server
✅ 测试端到端：Agent 自动查指标 + 判断 + 发预警
✅ 3个生产环境优化技巧：真实数据库、鉴权、错误重试

下一步行动：
- 把你最常用的3个内部查询封装成 MCP Tools
- 配置到你的主力 Agent 中
- 对比一下"之前手动操作"和"现在 Agent 自动执行"的效率差距

你可能会发现：Agent 的瓶颈从来不是模型能力，而是它能连接多少真实系统。

本文基于 MCP 协议官方文档和 fastmcp 0.5+ 版本。生产环境建议使用 systemd 或 Docker 管理 MCP Server 进程，确保 Agent 重启后自动恢复连接。

AI创业 #Agent工坊 #MCP #ClaudeCode #HermesAgent #一人公司