api-integration

API 集成

此技能可帮助 AI Agent 可靠地将应用与外部 API 集成。Agent 会选择合适的集成模式（REST 客户端、Webhook 消费者、轮询、SDK 封装），实现身份验证（API 密钥、OAuth、JWT），通过重试和熔断器处理错误，并遵守速率限制。最终生成可用于生产环境的集成代码，能够应对现实中的故障情况。

工作流

1. 分析目标 API： 查看 API 文档、OpenAPI 规范或 SDK 参考资料，了解可用端点、身份验证要求、速率限制和响应格式。确定 API 是否支持 Webhook 推送更新，还是需要轮询。留意非标准错误格式或分页方案等特殊情况。

2. 选择集成模式： 根据使用场景选择合适的模式。按需进行请求/响应交互时使用 REST 客户端。获取实时事件驱动数据时使用 Webhook 消费者。如果 API 不支持 Webhook，但需要近实时更新，则使用轮询。如果有官方 SDK，则对其进行封装，添加重试逻辑、日志记录和统一接口。

3. 实现身份验证： 配置正确的身份验证方式——在标头中使用 API 密钥、OAuth 2.0 Bearer 令牌、基于 JWT 的服务身份验证或 Basic Auth。使用环境变量或密钥管理器安全地存储凭据。对于 OAuth 流程，实现令牌刷新逻辑，避免长期运行的集成在访问令牌过期时中断。

4. 构建带错误处理的客户端： 编写包含结构化错误处理的集成代码。按状态码类别捕获 HTTP 错误：4xx 表示客户端错误（不重试），429 表示速率限制（采用退避策略重试），5xx 表示服务器错误（采用指数退避策略重试）。解析错误响应正文，提取可采取行动的信息。以调试级别记录所有请求和响应，以便排查问题。

5. 添加重试和熔断器逻辑： 针对暂时性故障实现带抖动的指数退避。设置最大重试次数（通常为 3-5 次）。实现熔断器，在连续失败后打开，并定期放行测试请求。这可以防止下游 API 性能下降时故障级联。

6. 测试和监控： 使用录制的 HTTP 固定数据（VCR 模式）编写集成测试，避免测试调用在线 API。通过请求延迟、错误率和速率限制余量等指标监控集成健康状况。针对持续高于阈值的错误率设置告警。

支持的技术

• HTTP 客户端： requests（Python）、httpx（Python 异步）、axios（Node.js）、fetch、HttpClient（.NET）、OkHttp（Java）
• 身份验证： API 密钥、OAuth 2.0、JWT、Basic Auth、HMAC 签名
• 弹性机制： tenacity（Python）、retry（Node.js）、Polly（.NET）、resilience4j（Java）
• 测试： responses（Python）、nock（Node.js）、VCR.py、WireMock
• GraphQL 客户端： gql（Python）、graphql-request（Node.js）、Apollo Client

用法

向 Agent 提供目标 API 的名称或文档 URL、需要执行的操作以及编程语言。注明身份验证方式和任何限制条件（速率限制、数据量）。Agent 将生成完整的集成模块，包含错误处理、重试和用法示例。

示例

示例 1：使用 Python 集成 Stripe API

import os
import time
import logging
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

class StripeClient:
    """Production-ready Stripe API client with retries and error handling."""

    BASE_URL = "https://api.stripe.com/v1"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ["STRIPE_SECRET_KEY"]
        self.session = self._build_session()

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/x-www-form-urlencoded",
            "Stripe-Version": "2024-06-20",
        })
        retry_strategy = Retry(
            total=4,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "DELETE"],
            respect_retry_after_header=True,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        return session

    def create_customer(self, email: str, name: str, metadata: Optional[dict] = None) -> dict:
        """Create a Stripe customer."""
        payload = {"email": email, "name": name}
        if metadata:
            for key, value in metadata.items():
                payload[f"metadata[{key}]"] = value
        response = self._request("POST", "/customers", data=payload)
        return response

    def create_payment_intent(self, amount_cents: int, currency: str = "usd",
                              customer_id: Optional[str] = None) -> dict:
        """Create a payment intent for a given amount."""
        payload = {"amount": amount_cents, "currency": currency}
        if customer_id:
            payload["customer"] = customer_id
        return self._request("POST", "/payment_intents", data=payload)

    def list_charges(self, customer_id: str, limit: int = 10) -> list:
        """List charges for a customer with automatic pagination."""
        charges = []
        params = {"customer": customer_id, "limit": limit}
        while True:
            data = self._request("GET", "/charges", params=params)
            charges.extend(data["data"])
            if not data["has_more"]:
                break
            params["starting_after"] = data["data"][-1]["id"]
        return charges

    def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{self.BASE_URL}{path}"
        try:
            resp = self.session.request(method, url, **kwargs)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.HTTPError as e:
            error_body = e.response.json().get("error", {})
            logger.error(
                "Stripe API error: type=%s code=%s message=%s",
                error_body.get("type"),
                error_body.get("code"),
                error_body.get("message"),
            )
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 5))
                logger.warning("Rate limited. Retry after %ds", retry_after)
            raise
        except requests.exceptions.ConnectionError:
            logger.error("Connection failed to Stripe API")
            raise

# Usage
client = StripeClient()
customer = client.create_customer("[email protected]", "Alice Smith")
payment = client.create_payment_intent(2500, customer_id=customer["id"])
print(f"Payment intent {payment['id']} for ${payment['amount']/100:.2f}")

示例 2：使用 requests 集成 GraphQL API

import requests
from typing import Any, Optional

class GraphQLClient:
    """Lightweight GraphQL client with error handling and variable support."""

    def __init__(self, endpoint: str, headers: Optional[dict] = None):
        self.endpoint = endpoint
        self.session = requests.Session()
        if headers:
            self.session.headers.update(headers)

    def execute(self, query: str, variables: Optional[dict] = None) -> dict:
        """Execute a GraphQL query or mutation."""
        payload = {"query": query}
        if variables:
            payload["variables"] = variables

        response = self.session.post(self.endpoint, json=payload, timeout=30)
        response.raise_for_status()
        result = response.json()

        if "errors" in result:
            error_messages = [e["message"] for e in result["errors"]]
            raise GraphQLError(error_messages, result.get("data"))

        return result["data"]

class GraphQLError(Exception):
    def __init__(self, messages: list, partial_data: Any = None):
        self.messages = messages
        self.partial_data = partial_data
        super().__init__(f"GraphQL errors: {'; '.join(messages)}")

# Usage: Query a GitHub-style GraphQL API
client = GraphQLClient(
    endpoint="https://api.github.com/graphql",
    headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
)

# Fetch repositories with pagination
query = """
query($owner: String!, $cursor: String) {
  user(login: $owner) {
    repositories(first: 10, after: $cursor, orderBy: {field: UPDATED_AT, direction: DESC}) {
      pageInfo {
        hasNextPage
        endCursor
      }
      nodes {
        name
        description
        stargazerCount
        primaryLanguage { name }
      }
    }
  }
}
"""

repos = []
cursor = None
while True:
    data = client.execute(query, {"owner": "octocat", "cursor": cursor})
    page = data["user"]["repositories"]
    repos.extend(page["nodes"])
    if not page["pageInfo"]["hasNextPage"]:
        break
    cursor = page["pageInfo"]["endCursor"]

for repo in repos:
    lang = repo["primaryLanguage"]["name"] if repo["primaryLanguage"] else "N/A"
    print(f"{repo['name']} ({lang}) - {repo['stargazerCount']} stars")

最佳实践

• 切勿硬编码凭据。 使用环境变量或密钥管理器。定期轮换 API 密钥，并在怀疑密钥泄露时立即轮换。
• 主动遵守速率限制。 解析 X-RateLimit-Remaining 标头，在达到限制前降低请求频率，而不是事后才对 429 响应作出反应。
• 对 POST 请求使用幂等键，尤其是面向支付和金融 API 的请求。在网络故障后重试时，这可以防止重复扣款或重复操作。
• 在调试级别记录请求和响应日志，并对敏感字段（令牌、卡号）进行脱敏。这对排查生产环境中的集成问题非常有帮助。
• 为所有 HTTP 请求设置明确的超时时间（连接超时为 5s，读取超时为 30s）。切勿使用无限超时——连接挂起会阻塞整个应用程序。
• 固定 API 集成所使用的版本，方法是发送明确的 API 版本标头（例如 Stripe-Version）。这样可以避免意外的破坏性变更影响集成。

边缘情况

• API 弃用： 监控响应中的 Deprecation 和 Sunset 标头。出现这些标头时记录警告，以便在端点移除前完成迁移。
• 部分响应： 某些 API 会返回包含 errors 数组的 200 OK 响应（GraphQL 中很常见）。即使 HTTP 状态码表示成功，也务必检查错误字段。
• 时钟偏差： 使用基于时间戳的签名（HMAC、JWT）的 API 可能会因服务器时钟偏差而拒绝请求。请使用 NTP，并设置一个较小的容差窗口。
• 大型负载： 获取大型数据集时，请使用流式响应和分页，而不是将所有内容都加载到内存中。注意那些会静默截断结果的 API。
• 网络分区： 请求可能已在服务器上成功处理，但响应却丢失了。这就是幂等键很重要的原因——你可以安全地重试，而不会造成重复的副作用。
• Webhook 与轮询的竞态条件： 如果对相同数据同时使用 webhook 和轮询，请实现去重，以避免重复处理同一事件。