Day 6：MCP改造——让工具变成标准接口

昨天的问题

Day 5的Agent已经很强了——有记忆、会路由、能查订单、能答政策。但它的三个工具是硬编码在Agent内部的。如果微信小程序后端也想查订单，得把 tools.py 复制过去，改格式、适配不同LLM——重复劳动。

为什么需要标准化

前5天写的三个工具函数（lookup_order、check_return、calc_refund）现在是直接嵌在Agent代码里的。这意味着：

• 微信小程序后端想用这些工具 → 得复制粘贴代码
• 换了LLM提供商（从OpenAI换到Anthropic）→ 工具定义格式要重写
• 另一个Agent想复用这些工具 → 没有标准接口

MCP（Model Context Protocol）解决的就是这个问题——让工具定义一次，到处可用。 它就像USB-C标准：以前每个设备有自己的接口，现在一个标准，全通用。

今天做什么

把昨天的三个工具包装成MCP Server。完成之后，任何支持MCP的客户端（Claude Desktop、LangChain、Cursor编辑器）都能直接调用这些工具。

你需要安装

pip install mcp

⚠️ 注意：MCP Python SDK 仍在快速迭代。本书代码基于 0.x 版本 API。如果安装后报 ImportError，请尝试 pip install mcp==0.9.4 锁定版本。

📁 代码目录

customer_service_agent/
├── day1~5.py        (前5天代码不动)
├── tools.py
├── day6.py
└── tools_server.py  ← 新增！MCP Server，60行

代码

新建 tools_server.py——独立文件，不依赖前面的Agent代码：

# tools_server.py — MCP版本的客服工具

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from datetime import datetime

# ===== 工具逻辑和Day4完全一样 =====
ORDERS = {
    "ORD001": {"status": "运输中", "item": "无线耳机", "price": 299, "date": "2025-01-18"},
    "ORD002": {"status": "已签收", "item": "手机壳", "price": 39, "date": "2025-01-10"},
    "ORD003": {"status": "待发货", "item": "机械键盘", "price": 599, "date": "2025-01-21"},
}

def lookup_order(order_id: str) -> str:
    order = ORDERS.get(order_id.upper())
    if not order: return f"未找到订单 {order_id}"
    return f"📦 订单{order_id} | {order['item']} | ¥{order['price']} | {order['status']}"

def check_return(order_id: str) -> str:
    order = ORDERS.get(order_id.upper())
    if not order: return "订单不存在"
    if order["status"] != "已签收": return f"状态'{order['status']}'，请先确认收货"
    days = (datetime.now() - datetime.strptime(order["date"], "%Y-%m-%d")).days
    return f"✅ 可退货" if days <= 7 else f"❌ 已超7天({days}天)"

def calc_refund(order_id: str) -> str:
    order = ORDERS.get(order_id.upper())
    if not order: return "订单不存在"
    return f"💰 退款¥{order['price']}，3个工作日内到账"

# ===== MCP Server 包装层 =====
# 创建Server实例
server = Server("customer-service-tools")

@server.list_tools()
async def handle_list_tools() -> list:
    """
    客户端问"你有哪些工具？"时返回这个列表。
    每个工具包含：名称、描述、输入参数schema。
    LLM就是靠这些信息决定什么时候调哪个工具。
    """
    return [
        Tool(
            name="lookup_order",
            description="查询指定订单的状态、商品名称和金额",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号，如 ORD001"
                    }
                },
                "required": ["order_id"]
            }
        ),
        Tool(
            name="check_return",
            description="检查订单是否在7天退货期内",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "订单号"}
                },
                "required": ["order_id"]
            }
        ),
        Tool(
            name="calc_refund",
            description="计算订单的退款金额",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "订单号"}
                },
                "required": ["order_id"]
            }
        ),
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    """
    客户端说"调用这个工具，参数是这个"时执行。
    """
    tools = {
        "lookup_order": lookup_order,
        "check_return": check_return,
        "calc_refund": calc_refund,
    }
    
    func = tools.get(name)
    if not func:
        return [TextContent(type="text", text=f"未知工具: {name}")]
    
    try:
        result = func(**arguments)
        return [TextContent(type="text", text=result)]
    except Exception as e:
        return [TextContent(type="text", text=f"工具执行出错: {str(e)}")]

# 启动Server
async def main():
    """
    以stdio模式启动MCP Server。
    任何支持MCP的客户端都可以通过标准输入/输出和Server通信。
    """
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)
        print("✅ MCP Server 已启动，等待客户端连接...")

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

运行

python tools_server.py

Server启动后等待连接。在另一个终端，你可以用MCP客户端测试：

# test_mcp_client.py — 测试MCP Server
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
import asyncio

async def test():
    params = StdioServerParameters(command="python3", args=["tools_server.py"])
    
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 列出所有工具
            tools = await session.list_tools()
            print(f"可用工具: {[t.name for t in tools.tools]}")
            
            # 调用工具
            result = await session.call_tool("lookup_order", {"order_id": "ORD001"})
            print(f"查订单: {result.content[0].text}")

asyncio.run(test())

python test_mcp_client.py

你应该看到：

🔌 MCP Server 已启动，等待客户端连接...

--- 另一个终端运行测试 ---

🔌 可用工具: ['lookup_order', 'check_return', 'calc_refund']
📦 查订单: 📦 订单ORD001 | 无线耳机 | ¥299 | 运输中

你学到了什么

MCP让工具从"Agent的内置函数"变成了"独立的标准服务"。 工具逻辑（lookup_order等函数）没有变，只是加了一层标准的"插座"。

这个Server可以被任何支持MCP的客户端调用——不只是你自己的Agent，也可以是同事写的另一个Agent、微信小程序后端、甚至Claude Desktop直接连接。

分离关注点：工具开发者专注写工具逻辑，Agent开发者专注设计Agent流程，LLM厂商专注提供更好的模型。三者通过MCP这个"标准USB接口"连接，互不耦合。

明天的预告

最后一天——把7天的劳动成果部署上线。用FastAPI把Agent包装成HTTP API，让前端能通过URL调用。

---

Day 6完成。工具变成了标准MCP服务——一次定义，到处可用。