从零实现Agent Skill中调用MCP工具：新手避坑指南

1次阅读

共计 2631 个字符，预计需要花费 7 分钟才能阅读完成。

Agent Skill 是现代智能对话系统中的核心组件，负责处理特定领域的任务（如查询天气、订餐服务等）。MCP（Message Control Protocol）工具则是一套高效的消息处理中间件，常用于技能间的数据交换和系统集成。两者的结合可以显著提升 Agent Skill 的扩展性和复用能力。

新手在集成过程中常遇到以下问题：

认证配置复杂：MCP 通常需要多层安全认证，新手容易遗漏步骤
异步调用混乱：不熟悉回调机制导致消息丢失或重复处理
性能瓶颈：频繁调用 MCP 接口引发系统过载
错误处理缺失：未考虑网络波动或服务不可用场景
文档版本滞后：官方示例与实际 API 存在差异

在项目依赖中添加 MCP 客户端库（以 Python 为例）：
```
# requirements.txt
mcp-client>=2.3.0
```

创建配置文件mcp_config.yaml：

endpoint: https://api.mcp.example.com/v3
auth:
  type: oauth2
  client_id: your_client_id
  secret: your_secret
  token_url: /oauth/token

from mcp_client import MCPAuth, MCPClient

# 初始化认证模块
auth = MCPAuth(
    config_path='mcp_config.yaml',
    token_cache='./token_cache.dat'  # 避免频繁获取 token
)

# 创建客户端实例
client = MCPClient(auth)

推荐使用 asyncio 实现非阻塞调用：

import asyncio

async def process_message(msg):
    try:
        response = await client.async_call(
            service='weather',
            payload={'city': msg['location']},
            timeout=5  # 超时控制
        )
        return response['data']
    except Exception as e:
        logger.error(f"MCP 调用失败: {str(e)}")
        return None

# skill_mcp_integration.py
import logging
from typing import Optional
from mcp_client import MCPClient, MCPAuth

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MCPSkillIntegrator:
    def __init__(self):
        self._client = None

    def initialize(self):
        """初始化 MCP 连接"""
        try:
            auth = MCPAuth(config_path='mcp_config.yaml')
            self._client = MCPClient(auth)
            return True
        except Exception as e:
            logger.exception("MCP 初始化失败")
            return False

    async def query_service(self, service: str, params: dict) -> Optional[dict]:
        """
        调用 MCP 服务
        :param service: 服务名称
        :param params: 请求参数
        :return: 响应数据或 None
        """
        if not self._client:
            logger.warning("MCP 客户端未初始化")
            return None

        try:
            # 添加请求 ID 便于追踪
            params['request_id'] = str(uuid.uuid4())

            response = await self._client.async_call(
                service=service,
                payload=params,
                timeout=3
            )

            # 验证响应格式
            if not response.get('success', False):
                logger.warning(f"服务调用失败: {response.get('error')}")
                return None

            return response['data']

        except asyncio.TimeoutError:
            logger.warning(f"MCP 服务 {service} 调用超时")
        except Exception as e:
            logger.error(f"MCP 调用异常: {str(e)}")

        return None

批处理机制：合并短周期内的同类请求

# 批量查询城市天气
aasync def batch_query_weather(cities):
    return await client.batch_call(
        service='weather',
        payloads=[{'city': city} for city in cities],
        batch_size=10  # 每批最大请求数
    )

本地缓存：对静态数据使用 Redis 缓存

from redis import Redis

cache = Redis(host='localhost', port=6379)

def get_cached_data(key):
    if cache.exists(key):
        return cache.get(key)
    data = query_service(...)
    cache.setex(key, 3600, data)  # 缓存 1 小时
    return data