api-integration

API 統合

このスキルを使用すると、AI エージェントはアプリケーションを外部 API と確実に統合できます。エージェントは適切な統合パターン（REST クライアント、Webhook コンシューマー、ポーリング、SDK ラッパー）を選択し、認証（API キー、OAuth、JWT）を実装して、再試行やサーキットブレーカーでエラーに対処し、レート制限を順守します。その結果、現実の障害にも対応できる、本番環境向けの統合コードが生成されます。

ワークフロー

1. 対象 API を分析する： API ドキュメント、OpenAPI 仕様、または SDK リファレンスを確認し、利用可能なエンドポイント、認証要件、レート制限、レスポンス形式を把握します。API がプッシュ型の更新に対応する Webhook をサポートしているか、ポーリングが必要かを特定します。標準外のエラー形式やページネーション方式など、特有の仕様にも注意します。

2. 統合パターンを選択する： ユースケースに基づいて適切なパターンを選択します。必要に応じたリクエストとレスポンスのやり取りには REST クライアントを使用します。リアルタイムのイベント駆動型データには Webhook コンシューマーを使用します。API が Webhook をサポートしていないものの、ほぼリアルタイムの更新が必要な場合はポーリングを使用します。公式 SDK がある場合は、それをラップして再試行ロジック、ロギング、一貫したインターフェースを追加します。

3. 認証を実装する： ヘッダーでの API キー、OAuth 2.0 のベアラートークン、JWT ベースのサービス認証、基本認証など、適切な認証方法を設定します。環境変数またはシークレットマネージャーを使用して、認証情報を安全に保管します。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、基本認証、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

使用方法

対象 API の名前またはドキュメント URL、実行する操作、プログラミング言語をエージェントに伝えます。認証方法と、レート制限やデータ量などの制約も指定します。エージェントは、エラー処理、再試行、使用例を含む完全な統合モジュールを生成します。

例

例 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 レスポンスが返ってから対処するのではなく、制限に達する前にリクエストを減速してください。
• 決済 API や金融 API への POST リクエストには、冪等性キーを使用してください。 これにより、ネットワーク障害の後に再試行しても、二重請求や操作の重複を防げます。
• リクエストとレスポンスのログを実装し、 機密フィールド（トークン、カード番号）をマスキングしたうえで、デバッグレベルで記録してください。これは、本番環境での統合に関する問題のデバッグに非常に役立ちます。
• すべての HTTP リクエストに明示的なタイムアウトを設定してください（接続タイムアウトは 5s、読み取りタイムアウトは 30s）。タイムアウトを無限に設定しないでください。接続がハングすると、アプリケーション全体がブロックされます。
• 明示的な API バージョンヘッダー（例：Stripe-Version）を送信して、API 統合のバージョンを固定してください。これにより、予期しない破壊的変更が統合に影響するのを防げます。

エッジケース

• API の非推奨化： レスポンスの Deprecation ヘッダーと Sunset ヘッダーを監視してください。これらが現れたら警告をログに記録し、エンドポイントが削除される前に移行できるようにしてください。
• 部分的なレスポンス： 一部の API は、errors 配列を含む 200 OK を返します（GraphQL でよく見られます）。HTTP ステータスコードが成功を示していても、必ずエラーフィールドを確認してください。
• クロックスキュー： タイムスタンプベースの署名（HMAC、JWT）を使用する API では、サーバーの時計がずれているとリクエストが拒否されることがあります。NTP を使用し、小さな許容時間枠を設けてください。
• 大きなペイロード： 大規模なデータセットを取得する場合は、すべてをメモリに読み込むのではなく、ストリーミングレスポンスとページネーションを使用してください。API が結果を通知なしに切り詰める場合があるため、注意してください。
• ネットワークの分断： サーバーでリクエストが成功しても、レスポンスが失われることがあります。これが冪等性キーが重要な理由です。副作用を重複させずに安全に再試行できます。
• Webhook とポーリングの競合状態： 同じデータに対して Webhook とポーリングの両方を使用する場合は、同じイベントを二重に処理しないよう、重複排除を実装してください。