🧩 MCP生态

MCP协议实战:Python为AI Agent集成工具,统一调用标准

发布时间:2026-05-06 分类: MCP生态
摘要:MCP协议实战:用Python给Agent装“外挂”,告别工具调用混乱工具调用协议混乱,正在吃掉你的开发时间。你刚写完一个Agent,想给它接上天气查询、数据库操作、第三方API。结果发现:OpenAI的Function Calling一套格式,Claude的Tool Use另一套,国内大模型又有自己的“函数调用”规范。每个工具都得写适配层,代码重复率高,维护起来简直是噩梦。这不是你一个人的...

封面

MCP协议实战:用Python给Agent装“外挂”,告别工具调用混乱

工具调用协议混乱,正在吃掉你的开发时间。

你刚写完一个Agent,想给它接上天气查询、数据库操作、第三方API。结果发现:OpenAI的Function Calling一套格式,Claude的Tool Use另一套,国内大模型又有自己的“函数调用”规范。每个工具都得写适配层,代码重复率高,维护起来简直是噩梦。

这不是你一个人的痛点。协议碎片化,是当前AI Agent开发最大的效率杀手。

MCP(Model Context Protocol)就是为解决这个问题而生的。它提供了一个标准化的“外挂”接口,让你的Agent能以统一方式调用任何工具。本文将用Python实战,带你从零搭建一个基于MCP的Agent工具集成方案,并附上一个可复制的自动化赚钱案例。


一、MCP核心概念:三句话讲明白

MCP本质上是一个客户端-服务器协议,定义了AI模型(客户端)如何发现和调用外部工具(服务器)。

关键就三点:

  1. 统一描述:每个工具(无论天气查询还是数据库操作)都用同一个JSON Schema描述自己的功能和参数。
  2. 统一调用:Agent通过一个标准的call_tool接口调用任何工具,无需关心工具内部实现。
  3. 即插即用:工具以“服务器”形式独立运行,Agent可以动态发现和连接新工具。

类比一下:MCP就像是USB-C接口。你的手机(Agent)只需要一个USB-C口,就能连接显示器(工具A)、硬盘(工具B)、充电器(工具C),而不用为每个设备准备不同的线缆。


二、Python实战:5步搭建MCP工具服务器

我们以一个“自动监控竞品价格并发送告警”的工具为例,这是电商和跨境从业者高频需求。

步骤1:安装MCP Python SDK

pip install mcp

步骤2:定义工具描述(核心)

创建一个price_monitor.py文件:

from mcp.server import Server
from mcp.types import Tool, TextContent
import json

# 初始化MCP服务器
server = Server("price-monitor")

# 定义工具:这是你的“外挂”说明书
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="monitor_price",
            description="监控指定商品价格,当价格低于阈值时发送告警",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_url": {
                        "type": "string",
                        "description": "商品链接(支持Amazon、淘宝等)"
                    },
                    "threshold_price": {
                        "type": "number",
                        "description": "价格阈值,低于此价格触发告警"
                    },
                    "alert_channel": {
                        "type": "string",
                        "enum": ["email", "webhook", "dingtalk"],
                        "description": "告警渠道"
                    }
                },
                "required": ["product_url", "threshold_price"]
            }
        )
    ]

步骤3:实现工具逻辑

# 实现工具调用(这里放你的核心业务代码)
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "monitor_price":
        url = arguments["product_url"]
        threshold = arguments["threshold_price"]
        channel = arguments.get("alert_channel", "webhook")
        
        # 模拟价格抓取(实际项目中用requests/BeautifulSoup)
        current_price = 89.99  # 假设抓取到的价格
        
        if current_price <= threshold:
            # 发送告警(这里简化为返回消息)
            alert_msg = f"🚨 低价告警!商品 {url} 当前价格 ${current_price},低于阈值 ${threshold}"
            # 实际项目中:send_alert(alert_msg, channel)
            return [TextContent(type="text", text=alert_msg)]
        else:
            return [TextContent(type="text", text=f"价格正常:${current_price} > ${threshold}")]
    

![配图](https://yitb.com/usr/uploads/covers/cover_mcp_20260506_081244.jpg)

    raise ValueError(f"未知工具: {name}")

步骤4:启动服务器

import asyncio
from mcp.server.stdio import stdio_server

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

步骤5:Agent端调用示例

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def agent_call_example():
    # 连接到你的工具服务器
    server_params = StdioServerParameters(
        command="python",
        args=["price_monitor.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 1. 发现工具:Agent自动知道有哪些“外挂”可用
            tools = await session.list_tools()
            print(f"可用工具: {[t.name for t in tools.tools]}")
            
            # 2. 调用工具:统一格式,无需适配
            result = await session.call_tool(
                "monitor_price",
                arguments={
                    "product_url": "https://amazon.com/dp/B08N5WRWNW",
                    "threshold_price": 100.0,
                    "alert_channel": "webhook"
                }
            )
            print(result.content[0].text)

部署步骤

  1. price_monitor.py部署为常驻服务(可用Docker)。
  2. 在你的Agent配置中,添加该MCP服务器的连接信息。
  3. Agent启动时自动发现并加载monitor_price工具。

三、赚钱案例:自动化竞品监控服务,月入$2000+

场景:为亚马逊卖家提供竞品价格监控SaaS。

具体方法

  1. 工具开发:用上述MCP方案,开发一个支持多平台(Amazon、Shopify、独立站)的价格抓取工具服务器。
  2. Agent封装:创建一个“监控Agent”,用户只需说“监控这个链接,低于$50告诉我”,Agent自动调用MCP工具。
  3. 商业化

    • 基础版:$29/月,监控10个商品,邮件告警。
    • 专业版:$99/月,监控100个商品,支持Webhook/钉钉/Slack告警,生成价格趋势报告。
    • 企业版:$299/月,无限监控,API接入,专属客服。

可复制路径

  1. 第一周:用本文代码搭建价格监控MCP服务器,测试稳定。
  2. 第二周:开发一个简单的Web界面(Flask/Vue),让用户提交监控任务。
  3. 第三周:接入支付(Stripe/Paddle),在跨境卖家论坛(如创蓝论坛)推广。
  4. 第四周:根据用户反馈增加工具(如库存监控、Review评分监控),形成工具矩阵。

关键数字

  • 开发成本:主要时间成本,服务器月费$20(低配云主机)。
  • 客户获取:通过内容营销(写“如何用AI监控竞品”干货),单个客户获取成本可控制在$50以内。
  • 利润空间:若发展100个专业版客户,月收入$9900,毛利率超80%。

核心优势:MCP让你的工具成为“乐高积木”。今天做价格监控,明天可以快速拼接“库存监控”、“舆情监控”工具,形成竞争壁垒。


四、下一步行动:今天就能做的3件事

  1. 跑通Demo:复制上面的代码,本地运行一个“天气查询”或“汇率计算”的MCP工具服务器(10分钟搞定)。
  2. 找到痛点:列出你或你所在行业最重复、最耗时的3个信息收集/处理任务(如报表生成、数据核对、内容抓取)。
  3. 设计工具:为其中一个任务设计MCP工具Schema(参考步骤2的JSON结构),思考其商业价值(能为谁节省多少时间/多少钱?)。

协议统一的时代已经到来。 别再为每个大模型写一遍适配代码了。用MCP把你的工具做成标准件,一次开发,处处运行。这才是AI Agent该有的开发效率。

你的第一个MCP工具,准备做什么? 欢迎在龙虾官网(yitb.com)社区分享你的设计。

返回首页