外部 API 集成必须内置安全与可靠性

三个月前，你的团队在凌晨 3 点被告警短信轰醒——刚刚上线的那个“智能报表 Skill”因为没做密钥轮换，硬编码的 API Token 在 GitHub 上被爬虫抓到，一通调用后账单暴涨 1200 美元。更糟糕的是，下游 CRM 系统因为它的无限重试直接被打挂，核心业务中断了 47 分钟。事后复盘，大家达成了共识：外部 API 集成不是简单的 requests.get()，而是一门需要从第一天就内置安全与可靠性的工程手艺。

这一章，我们不聊抽象的原则，直接用一条真实链路走完：为你的 Skill 接入一个外部 REST 服务（以 OpenWeatherMap 天气接口为例），并把密钥管理、请求净化、熔断限流和优雅降级一层层补上。读完你不仅知道“该做什么”，还能当场把它改到生产级。

你需要什么

如果你的 Skill 运行在容器或 Serverless 环境中，本章的代码同样适用，只需注意环境变量的注入方式差异。

最终成果

你将得到一个名为 WeatherSkill 的类，它可以：

• 使用环境变量管理密钥，并支持无停机轮换
• 自动校验请求参数，对 API 返回进行格式化和输出净化
• 内置熔断器：连续失败 5 次自动熔断，60 秒后半开探测
• 内置限流：每秒最多 2 次请求，超出直接降级
• 当外部服务不可用时，返回缓存的最近一次结果，或预设的兜底信息

这套能力的骨架适用于任何 REST / GraphQL 集成，且可以通过 MCP 服务器进一步封装成安全沙箱（在进阶话题中会提到）。

步骤一：密钥管理与凭证轮换

目标

消除硬编码凭据，用环境变量 + 热加载实现凭证轮换，并降低泄露面。

预期效果

API Key 不出现于代码仓库、日志或报错信息中；更换密钥时无需重启服务。

实施

1. 在项目根目录创建 .env 文件（仅本地开发使用，必须加入 .gitignore）：

# .env
WEATHER_API_KEY=your-real-key-here

2. 编写一个 Config 类负责加载和刷新配置。为了支持热轮换，每次调用都从数据源获取，而不在内存中缓存过久的 Key。

import os
from dotenv import load_dotenv

load_dotenv()  # 本地开发加载 .env

class Config:
    @staticmethod
    def get_api_key() -> str:
        # 生产环境可从 Vault、AWS Secrets Manager 等读取
        # 此处使用环境变量作为统一入口
        key = os.getenv("WEATHER_API_KEY")
        if not key:
            raise RuntimeError("WEATHER_API_KEY is not set")
        return key

3. 轮换策略：在密钥管理服务中创建两个版本，通过切换版本号实现滚动切换。Config 可以增加一个刷新触发器，每隔 5 分钟检查一次环境变量（或远程配置）的更新，但本章为保留清晰度，直接每次调用读取。踩坑经验：不要使用 os.environ.get() 后就存为类属性，那会让你重启进程才能轮换，与我们追求的无停机目标相悖。

⚠️ 注意：如果你在 Skill 里打印日志，务必过滤掉请求头中的 Authorization 或 x-api-key，推荐使用日志脱敏库，或者直接 repr() 前截断。

步骤二：请求构造、校验与输出净化

目标

防止非法参数触达外部 API，同时对返回内容进行格式化与过滤，避免被注入或篡改下游展示。

预期效果

调用方传入错误参数时立即抛出明确异常，不向 API 发出无效请求；返回结果由 Skill 统一清洗后再交付给用户或下一个节点。

实施

1. 输入校验：以天气查询为例，我们需要城市名不能为空，且只能是字母、空格和破折号。

import re
from typing import Optional

def validate_city(city: str) -> str:
    if not city or not city.strip():
        raise ValueError("city name cannot be empty")
    if not re.match(r"^[a-zA-Z\u0080-\u024F\u1E00-\u1EFF \-]+$", city.strip()):
        raise ValueError("city contains illegal characters")
    return city.strip()

对于复杂参数（如 GraphQL 的 JSON 体），建议加上 schema 校验库（如 cerberus）并限制深度与长度。

2. 发起安全的 HTTP 请求：使用 httpx.AsyncClient，强制设置超时，并在请求头中加入认证信息。

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class WeatherAPIClient:
    def __init__(self):
        self.base_url = "https://api.openweathermap.org/data/2.5/weather"
        self.timeout = httpx.Timeout(5.0, connect=2.0)

    async def _request(self, city: str) -> dict:
        key = Config.get_api_key()
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.get(
                self.base_url,
                params={"q": city, "appid": key, "units": "metric"},
                headers={"Accept": "application/json"}
            )
            resp.raise_for_status()
            return resp.json()

3. 输出净化：外面拿到的原始 JSON 可能包含大量字段，甚至有 HTML 片段或未知结构。Skill 应该只提取需要的字段，并确保值类型符合预期。

from typing import Dict, Any

def sanitize_weather_response(raw: Dict[str, Any]) -> Dict[str, Any]:
    try:
        return {
            "city": str(raw["name"]),
            "temperature": float(raw["main"]["temp"]),
            "description": str(raw["weather"][0]["description"]),
        }
    except (KeyError, TypeError, ValueError) as e:
        raise ValueError(f"Unexpected API response structure: {e}")

混在描述里的 HTML 标签可以用简单正则去除，或使用 html.unescape() 处理。这能防止某些第三方 API 返回的脏数据被渲染到你的前端时引发 XSS 或页面错乱。

踩坑实录：有一次，某天气预报 API 返回的城市名中意外包含了单引号，直接拼进 SQL 查询导致语法错误。输出净化不仅是安全需要，也是稳定性的保障。

步骤三：熔断器与限流

目标

当外部 API 故障或响应极慢时，快速失败而不是雪崩；同时遵守对方的速率限制。

预期效果

连续失败达到阈值后，熔断器打开，后续请求直接抛 CircuitBreakerError；限流器保证每秒最多 N 个请求，溢出部分可配置等待或降级。

实施

1. 熔断器：使用 pybreaker。

import pybreaker

weather_breaker = pybreaker.CircuitBreaker(
    fail_max=5,           # 5 次连续失败后熔断
    timeout_duration=60,  # 熔断后 60 秒转入半开状态
    reset_timeout=30      # 半开状态下成功 30 秒后恢复
)

2. 限流器：用 token bucket 实现，这里采用 asyncio 配合 time 简单做一个。

import asyncio, time

class RateLimiter:
    def __init__(self, rate: int):
        self.rate = rate          # 每秒允许次数
        self.tokens = rate
        self.updated_at = time.monotonic()

    async def acquire(self):
        while True:
            now = time.monotonic()
            elapsed = now - self.updated_at
            self.tokens = min(self.rate, self.tokens + elapsed * self.rate)
            self.updated_at = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            await asyncio.sleep(0.1 / self.rate)

weather_limiter = RateLimiter(rate=2)  # 每秒 2 次

3. 组合进请求函数：在 WeatherAPIClient 中构建一个核心调用方法。

class WeatherAPIClient:
    # ... 省略初始化

    @weather_breaker
    async def get_current_weather(self, city: str) -> Dict[str, Any]:
        await weather_limiter.acquire()
        raw = await self._request(city)
        return sanitize_weather_response(raw)

当持续失败时，@weather_breaker 装饰器会自动打开熔断，后续调用直接抛出 pybreaker.CircuitBreakerError，避免消耗资源。

微调建议：根据你使用的实际 API 的 SLA 调整 fail_max 和 timeout_duration。有些 API 间歇性 500 是常态，阈值设得太低反而导致过多熔断。

步骤四：优雅降级与缓存

目标

即使外部 API 挂掉，Skill 依然能返回有意义的信息，而不是抛出 5xx 中断用户流程。

预期效果

响应被缓存一定时间，当实时请求因超时、熔断或限流失败时，自动回退到缓存数据；若缓存也失效，则返回预置静态兜底信息。

实施

1. 缓存层：使用 cachetools.TTLCache 实现内存缓存。

from cachetools import TTLCache

weather_cache = TTLCache(maxsize=100, ttl=600)  # 单条缓存 10 分钟

2. 降级调用逻辑：在 Skill 的主要入口中，捕获所有可能的异常，逐级降级。

class WeatherSkill:
    def __init__(self):
        self.api = WeatherAPIClient()

    async def query(self, city: str) -> str:
        # 输入校验
        city = validate_city(city)

        # 尝试实时查询
        try:
            data = await self.api.get_current_weather(city)
            weather_cache[city] = data
        except (pybreaker.CircuitBreakerError, httpx.TimeoutException,
                httpx.HTTPStatusError, Exception) as e:
            # 回退到缓存
            cached = weather_cache.get(city)
            if cached:
                return f"[来自缓存] {cached['city']}: {cached['temperature']}°C, {cached['description']}"
            # 缓存也无效，返回预设降级
            return f"天气服务暂时不可用，请稍后重试。（{city}）"
        else:
            # 正常返回
            return f"{data['city']}: {data['temperature']}°C, {data['description']}"

3. 为 GraphQL 加点料：若你的 Skill 调用的是 GraphQL 端点，降级策略同样适用。区别在于请求体是 JSON 的 query 和 variables，可以在缓存 key 中加入 query 的哈希值。

警示：降级消息也可能带有业务含义，注意不要随意暴露内部 IP、堆栈等敏感信息。

回顾

你刚才在 40 分钟内，为一个简单的天气 Skill 添加了生产级防护：

1. 密钥管理：从硬编码迁移到环境变量 + 热加载，支持动态轮换。
2. 请求净化：输入校验和输出净化，杜绝无效调用和脏数据。
3. 熔断与限流：用 pybreaker 和 token bucket 实现了自动隔离和频率控制。
4. 降级与缓存：TTL 缓存加多级回退，让 Skill 在外部崩溃时依然“体面”。

如果你正在构建更复杂的多工具 Skill 体系，可以考虑将这些能力封装进一个 MCP 服务器，让凭证管理、连接池和熔断完全在沙箱内完成，Skill 仅通过本地协议调用工具——这正是下一章我们要探讨的基础：自定义工具开发与注册，把今天啃下的安全能力标准化，让整个团队都能安全地复用它。

行动清单

• [ ] 将所有外部 API 的密钥从代码中移至环境变量或密钥管理服务，并设置轮换流程。
• [ ] 为每个外部调用编写统一客户端，内置超时、重试和熔断。
• [ ] 对所有对外输入参数做严格校验，输出结果做字段白名单与净化。
• [ ] 配置 TTL 缓存和降级兜底策略，在 API 故障时保证 Skill 有体面的输出。
• [ ] 如果 Skill 需同时连接多个有状态服务，考虑引入 MCP 服务器隔离凭证和连接池。

下一章 《自定义工具开发与注册让 Skills 能力无限扩展》，我们将把本章构建的安全调用能力进一步抽象成可发现、可复用的标准化工具，让你的 Skill 生态真正从“单打独斗”走向“协作共生”。