别再只会用现成的 MCP Server 了——2026 年 AI 创业者的分水岭，是能不能自己写 MCP 工具让 Agent 调用你的业务系统。本文从零带你构建一个完整 MCP Server，含可复制代码、测试方法和生产部署清单。
为什么「用 MCP」和「写 MCP」是两个完全不同的层级
打开任何一个 AI Agent 工具（Claude Code、Hermes Agent、Cursor），你都会看到 MCP 配置选项。大多数人的做法：去 GitHub 搜「awesome-mcp-servers」，找到现成的配置，复制粘贴，完事。
这没问题——直到你需要 Agent 操作你自己的业务系统。
这时候你会发现：
- ❌ 没有现成的 MCP Server 能访问你公司内部的 CRM 数据
- ❌ 没有人写过你那个小众 SaaS 工具的 MCP 接口
- ❌ 你需要的「查本月收入」「列出待处理订单」这些操作，只有你自己知道数据库结构
这就是「用 MCP」和「写 MCP」的分水岭。
2026 年 6 月，MCP 生态已经有 200+ 个公开 Server，但真正让 Agent 产生商业价值的，永远是那些连接你私有业务系统的自定义 MCP 工具。
今天这篇文章，我用一个完整案例带你走完全流程：用 Python 写一个「创业公司数据看板」MCP Server，让 Claude Code 能直接回答「这个月收入多少」「哪个渠道转化最好」这类业务问题。
MCP 协议速览：你需要知道的 3 个概念
写 MCP Server 之前，先理解协议核心。MCP 本质上是一个 JSON-RPC 2.0 协议，Server 通过标准输入/输出（stdio）与 Client（你的 Agent）通信。
核心概念 1：Tools（工具）
Tool 是 MCP 最基本的能力单元。每个 Tool 就是 Agent 可以调用的一个函数：
{
"name": "get_monthly_revenue",
"description": "获取指定月份的营业收入",
"inputSchema": {
"type": "object",
"properties": {
"month": {"type": "string", "description": "月份，格式 YYYY-MM"}
},
"required": ["month"]
}
}
当 Agent 决定调用这个工具时，MCP Client 会发送一个 tools/call 请求，Server 返回执行结果。
核心概念 2：Resources（资源）
Resource 是 Agent 可以读取的数据源。和 Tool 的区别：
| Tool | Resource | |
|---|---|---|
| 触发方式 | Agent 主动调用 | Agent 订阅/读取 |
| 有副作用？ | 是（可能修改数据） | 否（只读） |
| 典型场景 | 发邮件、创建订单 | 读取配置文件、查日志 |
| 类比 | REST API 的 POST/PUT | REST API 的 GET |
核心概念 3：Prompts（提示模板）
Prompts 是预定义的提示词模板，Agent 可以按需加载。比如你可以定义一个「数据分析师」角色 Prompt，让 Agent 在需要时切换到数据分析模式。
实战：构建「创业公司数据看板」MCP Server
场景设定
你经营一家 SaaS 创业公司，核心数据在 PostgreSQL 数据库里。你希望 Claude Code 能直接回答：
- 「这个月 MRR 是多少？环比增长多少？」
- 「上周新增了多少付费用户？」
- 「哪个渠道的转化率最高？」
这些查询逻辑你都清楚，但没有现成的 MCP Server 能对接你的数据库结构。
第一步：项目初始化
mkdir mcp-dashboard && cd mcp-dashboard
python3 -m venv venv && source venv/bin/activate
pip install mcp psycopg2-binary python-dotenv
创建项目结构：
mcp-dashboard/
├── server.py # MCP Server 主入口
├── db.py # 数据库查询逻辑
├── tools/
│ ├── revenue.py # 收入查询工具
│ ├── users.py # 用户指标工具
│ └── channels.py # 渠道分析工具
├── requirements.txt
└── .env # 数据库连接信息
第二步：编写 MCP Server 主入口
# server.py
"""
创业公司数据看板 MCP Server
让 AI Agent 能直接查询你的业务数据
"""
import asyncio
import os
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 导入工具模块
from tools.revenue import register_revenue_tools
from tools.users import register_user_tools
from tools.channels import register_channel_tools
# 创建 Server 实例
server = Server("startup-dashboard")
@server.list_tools()
async def list_tools():
"""注册所有工具"""
tools = []
tools.extend(register_revenue_tools())
tools.extend(register_user_tools())
tools.extend(register_channel_tools())
return tools
@server.call_tool()
async def call_tool(name: str, arguments: dict):
"""路由工具调用到对应的处理函数"""
# 工具路由表
handlers = {
"get_monthly_revenue": handle_revenue,
"get_user_growth": handle_user_growth,
"get_channel_conversion": handle_channel_conversion,
"get_active_subscriptions": handle_active_subscriptions,
}
handler = handlers.get(name)
if not handler:
raise ValueError(f"未知工具: {name}")
result = await handler(**arguments)
# 返回结果（MCP 要求返回 ContentBlock 列表）
from mcp.types import TextContent
return [TextContent(type="text", text=result)]
async def main():
"""启动 MCP Server"""
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationCapabilities(
sampling={},
experimental={},
),
notification_options=NotificationOptions(
tools_changed=False,
resources_changed=False,
prompts_changed=False,
),
)
if __name__ == "__main__":
asyncio.run(main())
第三步：实现收入查询工具
# tools/revenue.py
"""
收入查询工具 — Agent 可以通过此工具查询公司营收数据
"""
from mcp.types import Tool
import json
def register_revenue_tools():
"""注册收入相关工具"""
return [
Tool(
name="get_monthly_revenue",
description="查询指定月份的营业收入（MRR、一次性收入、总收入），返回JSON格式数据",
inputSchema={
"type": "object",
"properties": {
"month": {
"type": "string",
"description": "查询月份，格式 YYYY-MM，例如 2026-06。不填默认为当前月份"
}
},
"required": []
}
),
Tool(
name="get_revenue_trend",
description="查询最近N个月的收入趋势，包含环比增长率",
inputSchema={
"type": "object",
"properties": {
"months": {
"type": "integer",
"description": "查询最近几个月的趋势，默认6个月",
"default": 6
}
},
"required": []
}
)
]
async def handle_revenue(month: str = None):
"""处理收入查询"""
import psycopg2
import os
from datetime import datetime
if month is None:
month = datetime.now().strftime("%Y-%m")
conn = psycopg2.connect(os.getenv("DATABASE_URL"))
cur = conn.cursor()
# 查询 MRR 和一次性收入
cur.execute("""
SELECT
SUM(CASE WHEN plan_type = 'recurring' THEN amount ELSE 0 END) as mrr,
SUM(CASE WHEN plan_type = 'one_time' THEN amount ELSE 0 END) as one_time,
COUNT(DISTINCT user_id) as paying_users
FROM revenue
WHERE to_char(created_at, 'YYYY-MM') = %s
""", (month,))
row = cur.fetchone()
cur.close()
conn.close()
mrr, one_time, users = row
total = (mrr or 0) + (one_time or 0)
return json.dumps({
"month": month,
"mrr": round(mrr or 0, 2),
"one_time_revenue": round(one_time or 0, 2),
"total_revenue": round(total, 2),
"paying_users": users or 0,
"arpu": round(total / (users or 1), 2)
}, ensure_ascii=False, indent=2)
async def handle_revenue_trend(months: int = 6):
"""处理收入趋势查询"""
import psycopg2
import os
conn = psycopg2.connect(os.getenv("DATABASE_URL"))
cur = conn.cursor()
cur.execute("""
SELECT
to_char(created_at, 'YYYY-MM') as month,
SUM(amount) as total_revenue
FROM revenue
WHERE created_at >= date_trunc('month', now()) - interval '%s months'
GROUP BY 1
ORDER BY 1
""", (months,))
rows = cur.fetchall()
cur.close()
conn.close()
# 计算环比增长率
result = []
prev = None
for month, rev in rows:
growth = None
if prev and prev > 0:
growth = round((rev - prev) / prev * 100, 1)
result.append({
"month": month,
"revenue": round(rev, 2),
"mom_growth_pct": growth
})
prev = rev
return json.dumps(result, ensure_ascii=False, indent=2)
第四步：用户增长工具
# tools/users.py
from mcp.types import Tool
import json
def register_user_tools():
return [
Tool(
name="get_user_growth",
description="查询指定时间段内的用户增长数据（新增、活跃、流失）",
inputSchema={
"type": "object",
"properties": {
"period": {
"type": "string",
"enum": ["week", "month", "quarter"],
"description": "统计周期"
}
},
"required": ["period"]
}
),
Tool(
name="get_active_subscriptions",
description="查询当前活跃订阅数及分计划统计",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
]
async def handle_user_growth(period: str):
"""处理用户增长查询"""
import psycopg2, os
period_days = {"week": 7, "month": 30, "quarter": 90}
days = period_days[period]
conn = psycopg2.connect(os.getenv("DATABASE_URL"))
cur = conn.cursor()
cur.execute("""
SELECT
COUNT(*) FILTER (WHERE status = 'active' AND created_at >= now() - interval '%s days') as new_users,
COUNT(*) FILTER (WHERE status = 'active') as active_users,
COUNT(*) FILTER (WHERE status = 'churned' AND churned_at >= now() - interval '%s days') as churned_users
FROM users
""", (days, days))
row = cur.fetchone()
cur.close()
conn.close()
new, active, churned = row
churn_rate = round(churned / (active + churned) * 100, 1) if (active + churned) > 0 else 0
return json.dumps({
"period": f"最近{days}天",
"new_users": new,
"active_users": active,
"churned_users": churned,
"churn_rate_pct": churn_rate
}, ensure_ascii=False, indent=2)
连接到 Claude Code：一键配置
写完 Server，接下来让 Claude Code 能调用它。在项目根目录创建 .claude/mcp.json：
{
"mcpServers": {
"startup-dashboard": {
"command": "python3",
"args": ["server.py"],
"cwd": "/path/to/mcp-dashboard",
"env": {
"DATABASE_URL": "postgresql://user:pass@localhost:5432/startup_db"
}
}
}
}
Hermes Agent 用户，配置方式类似，在 ~/.hermes/config.yaml 中：
mcp_servers:
startup-dashboard:
command: python3
args: ["server.py"]
cwd: /path/to/mcp-dashboard
配置完成后，重启 Claude Code，在对话中试试：
> 查一下这个月的收入情况
Claude Code 会自动调用 get_monthly_revenue，返回：
📊 2026年6月收入概览：
• MRR: ¥48,500
• 一次性收入: ¥12,300
• 总收入: ¥60,800
• 付费用户: 127人
• ARPU: ¥478.74
本地测试：不接 Agent 也能验证
在接入 Claude Code 之前，可以用 MCP 官方提供的 Inspector 工具本地测试：
# 安装 MCP Inspector
npx @modelcontextprotocol/inspector python3 server.py
这会在浏览器打开一个调试界面，你可以：
- 查看所有注册的 Tool 列表
- 手动调用每个 Tool 并查看返回结果
- 检查 JSON-RPC 通信日志
或者更简单的方式——写一个测试脚本：
# test_server.py
"""不依赖 Agent 的 MCP Server 单元测试"""
import asyncio
import json
async def test_tools():
# 直接导入工具函数测试
from tools.revenue import handle_revenue
from tools.users import handle_user_growth
print("=== 测试收入查询 ===")
result = await handle_revenue(month="2026-06")
print(json.loads(result))
print("\n=== 测试用户增长 ===")
result = await handle_user_growth(period="month")
print(json.loads(result))
asyncio.run(test_tools())
生产部署清单：从开发到上线的 7 个检查点
写完代码只是第一步。以下是让 MCP Server 在生产环境稳定运行的检查清单：
✅ 1. 超时控制
# 在 call_tool 中添加超时
import asyncio
result = await asyncio.wait_for(handler(**arguments), timeout=30.0)
✅ 2. 错误处理
# 统一错误返回格式
try:
result = await handler(**arguments)
except Exception as e:
return [TextContent(type="text", text=json.dumps({
"error": str(e),
"type": type(e).__name__
}))]
✅ 3. 连接池复用
# db.py — 不要每次查询都新建连接
from psycopg2.pool import SimpleConnectionPool
_pool = None
def get_pool():
global _pool
if _pool is None:
_pool = SimpleConnectionPool(1, 5, os.getenv("DATABASE_URL"))
return _pool
✅ 4. 敏感数据脱敏
# 查询结果自动脱敏邮箱/手机号
import re
def mask_pii(data: dict) -> dict:
"""脱敏个人身份信息"""
text = json.dumps(data)
text = re.sub(r'[\w\.-]+@[\w\.-]+', '***@***.***', text)
text = re.sub(r'1[3-9]\d{9}', '1**********', text)
return json.loads(text)
✅ 5. 日志记录
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-dashboard")
# 在 call_tool 中记录
logger.info(f"Tool called: {name}, args: {arguments}")
✅ 6. 工具权限分级
# 在 MCP Server 配置中分组
tools:
read_only:
- get_monthly_revenue
- get_user_growth
- get_channel_conversion
read_write:
- update_subscription_status # 生产环境谨慎开放
✅ 7. Docker 化部署
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python3", "server.py"]
总结：2026 年 Agent 创业者的核心竞争力
回顾这一年 AI Agent 工具的发展，有一个趋势越来越清晰：
会用 AI Agent 是基本功，能让 AI Agent 调用你的业务系统才是护城河。
MCP Server 开发不是「高级话题」——它是每个 AI 创业者都应该掌握的基础技能。因为：
- 现成的 MCP Server 解决通用问题（GitHub、Slack、数据库），但只有你自己写的 MCP Server 能解决你的特殊业务逻辑
- Agent 的价值 = 它能接触的数据 × 它能执行的操作，自定义 MCP Server 直接扩大这个乘积
- 每次重复性人工操作被 MCP Tool 替代，你就离「一人公司」更近一步
今天这篇文章的完整代码可以在你的项目中直接使用——替换数据库查询逻辑为你的业务 SQL，10 分钟内你就能让 Claude Code 或 Hermes Agent 变成你的业务数据分析师。
下一步行动：
1. 列出你每天要手动查询的 3 个业务指标
2. 把查询逻辑写成 MCP Tool
3. 接入 Claude Code / Hermes Agent，以后直接对话获取答案