Agent Skill开发实战：如何在Skill中高效调用MCP工具

1次阅读

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

MCP（Message Control Protocol）工具在 Agent Skill 中扮演着关键角色，它作为消息控制协议，负责 Skill 与外部系统或服务之间的通信。常见的应用场景包括：

跨系统数据查询（如用户信息验证）
第三方服务集成（如支付网关调用）
异步任务处理（如批量文件导出）

MCP 协议基于 HTTP/HTTPS，采用 JSON 格式传输数据。核心接口包括：

# MCP 基础请求结构示例
{
  "header": {
    "version": "1.0",
    "timestamp": "2023-08-20T14:30:00Z",
    "request_id": "uuidv4"
  },
  "body": {
    "operation": "query|update|delete",
    "payload": {...}
  }
}

以下是 Python 实现的完整封装类，包含重试机制和错误处理：

import requests
from retrying import retry

class MCPClient:
    def __init__(self, endpoint, api_key):
        self.endpoint = endpoint
        self.headers = {'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }

    @retry(stop_max_attempt_number=3, wait_fixed=2000)
    def call_mcp(self, operation, payload, timeout=5):
        """
        :param operation: MCP 操作类型
        :param payload: 请求数据体
        :param timeout: 超时时间 (秒)
        :return: (status_code, response)
        """
        try:
            data = {
                "header": {
                    "version": "1.0",
                    "timestamp": datetime.utcnow().isoformat()
                },
                "body": {"operation": operation, "payload": payload}
            }

            response = requests.post(
                self.endpoint,
                json=data,
                headers=self.headers,
                timeout=timeout
            )

            if response.status_code == 429:
                raise Exception("Rate limit exceeded")

            return response.status_code, response.json()

        except requests.exceptions.Timeout:
            # 自定义超时处理逻辑
            return 408, {"error": "request_timeout"}
        except Exception as e:
            # 记录详细错误日志
            logger.error(f"MCP call failed: {str(e)}")
            return 500, {"error": "internal_error"}

当处理高并发请求时，推荐采用以下模式：

使用连接池管理 HTTP 连接
实现请求队列控制并发量
采用令牌桶算法进行限流

from concurrent.futures import ThreadPoolExecutor

class ConcurrentMCPClient(MCPClient):
    def __init__(self, max_workers=10):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    async def batch_call(self, operations):
        """
        :param operations: [(operation1, payload1), ...]
        :return: 异步结果列表
        """
        futures = [
            self.executor.submit(
                self.call_mcp, 
                op, 
                payload
            ) for op, payload in operations
        ]
        return await asyncio.gather(*futures)

平均响应时间：控制在 300ms 内为佳
吞吐量：单节点建议维持 500-1000 QPS
错误率：保持在 1% 以下

连接复用：保持 HTTP 长连接
数据压缩：对大于 1KB 的 payload 启用 gzip
缓存策略：对查询类请求实现本地缓存
负载均衡：使用多 MCP 服务端点

# 缓存实现示例
from cachetools import TTLCache

class CachedMCPClient(MCPClient):
    def __init__(self, cache_ttl=300):
        self.cache = TTLCache(maxsize=1000, ttl=cache_ttl)

    def call_mcp(self, operation, payload):
        cache_key = f"{operation}:{hash(frozenset(payload.items()))}"

        if cache_key in self.cache:
            return 200, self.cache[cache_key]

        status, response = super().call_mcp(operation, payload)
        if status == 200:
            self.cache[cache_key] = response

        return status, response

传输安全：强制 HTTPS + TLS 1.2+
认证机制：JWT 令牌或 OAuth2.0
数据脱敏：敏感字段如 password 必须加密
权限最小化：按需分配操作权限

# security_config.yaml
mcp:
  security:
    ssl_verify: true
    cipher_list: "ECDHE-ECDSA-AES256-GCM-SHA384"
    token_rotation: 3600  # 令牌轮换时间 (秒)
    allowed_operations:  # 白名单控制
      - query
      - status_check

问题现象	可能原因	解决方案
响应超时	网络延迟或服务过载	增加超时阈值 + 实现重试
数据不一致	未处理幂等性问题	在 payload 中添加 request_id
内存泄漏	未关闭连接	使用 with 语句管理资源
认证失败	令牌过期	实现自动刷新机制