MCP+A2A 从0到1：在腾讯云构建商业级多Agent全栈应用

import asyncio
from fastmcp import FastMCP
from pydantic import BaseModel

# 初始化MCP服务器
mcp = FastMCP("Ecommerce Inventory MCP")

class InventoryItem(BaseModel):
    product_id: str
    stock: int
    warehouse_location: str

@mcp.tool()
def get_inventory(product_id: str) -> str:
    """根据商品ID查询实时库存信息"""
    # 实际场景中，这里会通过SDK连接腾讯云数据库查询
    mock_db = {"P1001": InventoryItem(product_id="P1001", stock=150, warehouse_location="华北仓")}
    item = mock_db.get(product_id)
    if item:
        return f"商品 {product_id} 当前库存：{item.stock}，存放于 {item.warehouse_location}"
    return f"未找到商品 {product_id} 的库存信息"

@mcp.prompt()
def inventory_analysis_prompt(product_id: str, data: str) -> str:
    """生成库存分析的建议提示词"""
    return f"基于以下库存数据，为商品 {product_id} 生成补货建议：{data}"

if __name__ == "__main__":
    # 在腾讯云CVM或TKE中启动该服务
    asyncio.run(mcp.run_async(transport="streamable-http", host="0.0.0.0", port=8080))

import json
import uuid
from abc import ABC, abstractmethod

class BaseA2AAgent(ABC):
    def __init__(self, agent_id: str, capabilities: list):
        self.agent_id = agent_id
        self.capabilities = capabilities
        # 模拟发布A2A智能体卡片
        self.agent_card = {
            "id": agent_id,
            "capabilities": capabilities,
            "endpoint": f"http://tke-service.internal/{agent_id}/a2a"
        }

    @abstractmethod
    def handle_task(self, task_input: dict) -> dict:
        """处理接收到的任务"""
        pass

    def send_message(self, target_agent_endpoint: str, message: dict):
        """通过A2A协议发送标准化消息（实际可对接腾讯云Redis或HTTP请求）"""
        a2a_message = {
            "jsonrpc": "2.0",
            "id": str(uuid.uuid4()),
            "method": "message/send",
            "params": {
                "message": message,
                "metadata": {"sender": self.agent_id}
            }
        }
        print(f"[A2A通信] 向 {target_agent_endpoint} 发送消息: {json.dumps(a2a_message, ensure_ascii=False)}")
        # 此处可集成 httpx 或腾讯云SDK进行实际的RPC调用

class InventoryAgent(BaseA2AAgent):
    def handle_task(self, task_input: dict):
        product_id = task_input.get("product_id")
        # 调用MCP工具获取数据（此处模拟调用）
        result = f"商品 {product_id} 库存充足，建议维持现有发货节奏。"
        return {"status": "completed", "result": result}

# 实例化Agent并模拟协作
inventory_agent = InventoryAgent("agent-inventory-01", ["query_stock", "analyze_inventory"])
task_result = inventory_agent.handle_task({"product_id": "P1001"})
print(f"任务执行结果：{task_result}")