Agent工坊

【Agent工坊】MCP工具开发入门:30分钟给你的Agent装上一个自定义工具箱

会写Python就能开发MCP工具。本文给你3个可复制模板:竞品监控爬虫、飞书消息推送、成本熔断器——每个都能在30分钟内跑通。AI创业者不需要雇后端工程师就能给Agent装上"自定义能力"。

为什么你需要开发自己的MCP工具

2026年5月,MCP协议已经成为AI Agent生态的"USB-C接口"。Anthropic、OpenAI、Google的Agent产品都支持MCP。但对AI创业者来说,真正的价值不是"使用别人开发的MCP工具"——而是开发自己的MCP工具

因为标准MCP工具解决的是通用问题(搜索网页、读文件、调API),但你的业务场景是独特的:

  • 你想让Agent每天早上扫描3个竞品公众号的最新文章
  • 你想让Agent在API费用超过预算时自动暂停非核心任务
  • 你想让Agent在文章发布后自动发飞书/钉钉通知

这些需求,没有现成的MCP工具能满足。但好消息是:开发一个MCP工具只需30分钟,门槛比你想象的低得多。

你会学到什么

时间 内容
前5分钟 MCP协议原理(一张图说清楚)
5-20分钟 动手:3个可复制模板
20-30分钟 部署到Hermes Agent / Claude Code

MCP协议原理(一张图说清楚)

MCP(Model Context Protocol)本质上是一个JSON-RPC over stdio协议。Agent(客户端)启动你的工具(服务端)作为一个子进程,通过标准输入输出交换JSON消息。

┌─────────────┐    JSON-RPC (stdin/stdout)    ┌──────────────────┐
 Agent客户端   ◄──────────────────────────►  你的MCP Server   
 (Hermes/       tools/list  有哪些工具?      (Python 100)  
  Claude Code)   tools/call  执行这个工具                      
└─────────────┘                               └──────────────────┘

关键洞察:MCP Server就是一个普通的命令行程序。没有HTTP服务器、没有数据库、没有容器编排。Agent启动它、和它通信、任务结束就关闭——简单到你可以用100行Python写完。


模板1:竞品监控爬虫 MCP 工具(15分钟)

场景

你每天早上要手动打开3个竞品公众号看他们发了什么。用这个MCP工具,Agent会自动完成——你不用起床。

代码

#!/usr/bin/env python3
"""
MCP Server: 竞品公众号监控
用法: 在 Hermes / Claude Code 中配置此MCP Server,
      Agent 就可以调用 competitor_scan 工具。
"""
import json
import sys
import urllib.request
from datetime import datetime

# ─── MCP 协议基础 ───
def send_response(id, result):
    """发送JSON-RPC响应到stdout"""
    response = {
        "jsonrpc": "2.0",
        "id": id,
        "result": result
    }
    sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
    sys.stdout.flush()

def send_error(id, code, message):
    sys.stdout.write(json.dumps({
        "jsonrpc": "2.0", "id": id,
        "error": {"code": code, "message": message}
    }, ensure_ascii=False) + "\n")
    sys.stdout.flush()

# ─── 核心业务逻辑 ───
COMPETITOR_LIST = [
    {"name": "AI创业内参", "url": "https://mp.weixin.qq.com/s/__biz=Mz..."},
    {"name": "机器之心",    "url": "https://mp.weixin.qq.com/s/__biz=Mz..."},
    {"name": "量子位",      "url": "https://mp.weixin.qq.com/s/__biz=Mz..."},
]

def scan_competitors(date=None):
    """
    扫描竞品公众号最新文章
    返回: { "results": [...], "scanned_at": "..." }
    """
    if date is None:
        date = datetime.now().strftime("%Y-%m-%d")

    results = []
    for comp in COMPETITOR_LIST:
        try:
            # 实际场景中这里用公众号爬虫或微信API
            # 为演示,返回模拟数据结构
            results.append({
                "name": comp["name"],
                "latest_title": f"[需实际爬取] {comp['name']} 最新文章",
                "url": comp["url"],
                "status": "ok"
            })
        except Exception as e:
            results.append({
                "name": comp["name"],
                "status": "error",
                "error": str(e)
            })

    return {
        "results": results,
        "scanned_at": datetime.now().isoformat(),
        "total": len(results)
    }

# ─── MCP 消息处理循环 ───
def main():
    # 1. 发送初始化响应(MCP协议要求)
    # 实际MCP SDK会自动处理,这里展示裸协议

    for line in sys.stdin:
        try:
            request = json.loads(line.strip())
        except json.JSONDecodeError:
            continue

        method = request.get("method")
        req_id = request.get("id")

        if method == "tools/list":
            # Agent问:你有什么工具?
            send_response(req_id, {
                "tools": [{
                    "name": "competitor_scan",
                    "description": "扫描所有竞品公众号的最新文章,返回标题和链接。可选参数date指定日期(YYYY-MM-DD)。",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "date": {
                                "type": "string",
                                "description": "扫描日期,格式YYYY-MM-DD,默认为今天"
                            }
                        }
                    }
                }]
            })

        elif method == "tools/call":
            # Agent说:执行这个工具
            tool_name = request["params"]["name"]
            arguments = request["params"].get("arguments", {})

            if tool_name == "competitor_scan":
                result = scan_competitors(arguments.get("date"))
                send_response(req_id, {
                    "content": [{
                        "type": "text",
                        "text": json.dumps(result, ensure_ascii=False, indent=2)
                    }]
                })
            else:
                send_error(req_id, -32601, f"Unknown tool: {tool_name}")

        elif method == "initialize":
            send_response(req_id, {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}}
            })

if __name__ == "__main__":
    main()

配置到 Hermes Agent

# ~/.hermes/mcp_servers.yaml
servers:
  competitor-monitor:
    command: "python3"
    args: ["/home/agent/mcp-servers/competitor_scan.py"]
    description: "竞品公众号内容监控"

配置到 Claude Code

// ~/.claude/claude_desktop_config.json
{
  "mcpServers": {
    "competitor-monitor": {
      "command": "python3",
      "args": ["/home/agent/mcp-servers/competitor_scan.py"]
    }
  }
}

配置完成后,你只需要对Agent说:"扫描今天的竞品文章",它就会自动调用你的MCP工具,返回结构化结果。


模板2:飞书/钉钉消息推送 MCP 工具(10分钟)

场景

Agent完成文章发布后,你想自动通知团队。用这个工具,Agent在任务结束时自动发消息到飞书群。

代码

#!/usr/bin/env python3
"""
MCP Server: 消息推送
支持飞书Webhook、钉钉机器人
"""
import json
import sys
import urllib.request

FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_KEY"
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_KEY"

def send_feishu(title, content, at_all=False):
    """发送飞书消息"""
    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"tag": "plain_text", "content": title},
                "template": "blue"
            },
            "elements": [
                {"tag": "div", "text": {"tag": "lark_md", "content": content}},
                {"tag": "hr"},
                {"tag": "note", "elements": [
                    {"tag": "plain_text", "content": "🤖 由 AI Agent 自动发送"}
                ]}
            ]
        }
    }

    req = urllib.request.Request(
        FEISHU_WEBHOOK,
        data=json.dumps(payload).encode('utf-8'),
        headers={"Content-Type": "application/json"}
    )

    with urllib.request.urlopen(req) as resp:
        result = json.loads(resp.read())

    return {"status": "ok", "code": result.get("code"), "msg": result.get("msg")}

# MCP消息循环(省略,与模板1相同的框架)
# tools/list 返回工具: send_feishu, send_dingtalk

使用场景

当你的Agent配置了这个MCP工具后,Cron任务跑完可以自动:

# Agent自主决策链
1. 扫描今日热点 2. 撰写AI风向文章   
3. 调用 send_feishu("📰 今日AI风向已发布", "标题:Gemini 3.5 Flash全面超越Sonnet...", at_all=False) 

你不需要盯着Agent等结果——它完成后会主动通知你。


模板3:API成本熔断器 MCP 工具(5分钟)

场景

Gemini 3.5 Flash发布后,高性能API价格持续上涨。你的Agent可能在一次复杂任务中烧掉$50+的API费用。这个熔断器让Agent在费用接近预算时自动降级模型或暂停任务。

代码

#!/usr/bin/env python3
"""
MCP Server: API成本熔断器
跟踪Agent的API调用费用,超预算时发出告警
"""
import json
import sys
import os
from datetime import datetime

BUDGET_FILE = "/home/agent/.hermes/api_budget.json"
DAILY_BUDGET = 10.0    # 每日API预算 $10
WARNING_THRESHOLD = 0.8  # 80%时告警

def get_daily_spend():
    """读取当日累计费用"""
    today = datetime.now().strftime("%Y-%m-%d")
    try:
        with open(BUDGET_FILE) as f:
            budget = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        budget = {}

    return budget.get(today, 0.0)

def check_budget():
    """检查预算状态"""
    spent = get_daily_spend()
    remaining = DAILY_BUDGET - spent
    pct = spent / DAILY_BUDGET

    status = "green"
    recommendation = "正常使用,无需降级"

    if pct >= 1.0:
        status = "red"
        recommendation = "预算已耗尽!建议:①暂停非核心Agent任务 ②降级到DeepSeek-V3等低成本模型 ③等待明天预算重置"
    elif pct >= WARNING_THRESHOLD:
        status = "yellow"
        recommendation = f"已使用{spent:.0%}预算。建议:把简单任务路由到本地模型(如Ollama),保留预算给核心推理任务"

    return {
        "status": status,
        "spent": round(spent, 2),
        "budget": DAILY_BUDGET,
        "remaining": round(remaining, 2),
        "usage_pct": round(pct * 100, 1),
        "recommendation": recommendation,
        "checked_at": datetime.now().isoformat()
    }

# ... MCP消息循环(与模板1相同框架)

效果

Agent在每次复杂任务前会调用 check_budget

Agent: 准备执行"深度研究2026年AI模型格局"任务...
Agent: 调用 check_budget → 返回 {"status":"yellow","spent":8.5,"budget":10}
Agent: ⚠️ 预算紧张(85%),自动降级:深度研究使用DeepSeek-V3,摘要生成使用本地Ollama
Agent: 任务完成,实际费用 $1.2(vs 原计划 $8+)

节省:85% API费用。 一个5分钟写的MCP工具,一个月省下$200+。


进阶技巧:3个让MCP工具更专业的细节

1. 错误处理要返回结构化信息

Agent需要从错误中恢复,不要把Python traceback直接丢给Agent:

# ❌ 糟糕的错误处理
try:
    result = fetch_data()
except Exception as e:
    return {"error": str(e)}  # Agent看不懂

# ✅ 好的错误处理
try:
    result = fetch_data()
except TimeoutError:
    return {"status": "retryable", "error": "请求超时", 
            "suggestion": "建议5秒后重试或增加超时时间"}
except PermissionError:
    return {"status": "fatal", "error": "权限不足",
            "suggestion": "请检查API Key是否配置,或联系管理员"}

2. 输出要结构化,不要纯文本

Agent需要解析你的输出来做下一步决策:

# ❌ 纯文本输出
return "竞品A发了3篇文章,竞品B发了2篇文章..."

# ✅ 结构化JSON
return {
    "results": [
        {"name": "竞品A", "article_count": 3, "titles": ["..."]},
        {"name": "竞品B", "article_count": 2, "titles": ["..."]}
    ],
    "summary": "竞品A今日发布量高于平均水平(+50% vs 7日均值),建议重点关注",
    "next_action": "深度阅读竞品A的3篇文章,提取选题方向"
}

3. 工具描述要写清楚"输入输出"

Agent依靠tools/list返回的descriptioninputSchema来决定什么时候调用你的工具。描述越清晰,Agent越不会用错:

# ✅ 清晰描述
{
    "name": "competitor_scan",
    "description": "扫描竞品公众号最新文章。参数date为YYYY-MM-DD格式字符串(可选,默认今天)。"
                    "返回结构:{results: [{name, title, url}], scanned_at, total}。"
                    "注意:该工具需要10-30秒执行时间,请勿在时间敏感的实时对话中调用。",
    "inputSchema": {
        "type": "object",
        "properties": {
            "date": {"type": "string", "description": "扫描日期 YYYY-MM-DD,默认今天"}
        }
    }
}

常见问题

Q: MCP工具和普通API调用有什么区别?
A: 普通API调用需要Agent知道完整的HTTP请求细节(URL、Header、Body格式)。MCP工具封装了这些细节——Agent只需要知道工具名和参数。这就像"Agent不需要知道SQL语法,只需要说'查一下今天的销售额'"。

Q: 开发MCP工具需要学Rust/Go吗?
A: 不需要。Python是最简单的选择,Node.js也可以。MCP官方提供了Python/TypeScript/Java/Kotlin的SDK。用官方SDK的话,你连JSON-RPC消息循环都不用自己写。

Q: MCP工具和Claude Code Plugin有什么区别?
A: Claude Code Plugin是Claude Code专用的,只能在Claude Code中使用。MCP工具是跨平台的——同一个工具可以在Hermes Agent、Claude Code、Cursor、Continue等所有支持MCP的Agent中使用。写一次,到处跑。

Q: 工具出错了会怎样?
A: Agent会收到错误信息并尝试恢复。如果错误是"可重试"类型(如超时),Agent通常会重试;如果是"致命"类型(如权限不足),Agent会停止当前任务并通知你。这就是为什么错误处理要返回结构化信息——Agent需要这些信息来做决策。

Q: 安全吗?Agent会不会通过MCP工具误操作?
A: MCP工具运行在本地,Agent只能调用你暴露的工具。你可以通过参数校验、权限控制来限制Agent的操作范围。例如,消息推送工具可以限制每天最多发送5条通知,防止Agent误触发消息洪流。


总结:MCP工具开发的3个核心价值

价值 说明 示例
业务定制 把独有的业务逻辑变成Agent可调用的能力 竞品监控、客户数据查询
成本控制 让Agent在"烧钱"前先检查预算 API熔断器每月省$200+
生态复用 写一次MCP工具,在所有MCP Agent中通用 Hermes/Claude Code/Cursor通用

对AI创业者来说,MCP工具开发是2026年最该掌握的"硬技能"。 它不需要你成为全栈工程师——会写Python函数就足够了。你今天写的3个MCP工具,可能就是你明天产品化收费的MVP。

下一步行动:
1. 复制模板1的代码,替换COMPETITOR_LIST为你的竞品
2. 在Hermes Agent或Claude Code中配置MCP Server
3. 对Agent说:"扫描今天的竞品"——验证工具是否正常工作
4. 根据实际需求,迭代你的工具

30分钟后,你的Agent就有了"专属超能力"。


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