api-integration

API 統合スキル

有効化するタイミング

次の場合にこのスキルを有効化します。

• 外部 API を統合するとき
• API クライアントやラッパーを構築するとき
• API 認証を処理するとき
• レート制限を実装するとき
• API レスポンスをキャッシュするとき

基本原則

1. レート制限を遵守する - API は共有リソースです
2. 認証を安全に扱う - キーは secrets.json に保存し、コードには記述しません
3. エラーを適切に処理する - リトライとバックオフを実装します
4. レスポンスをキャッシュする - 重複したリクエストを減らします

認証の設定

secrets.json

{
  "github_token": "ghp_your_token_here",
  "openweather_api_key": "your_key_here",
  "comment": "Never commit this file"
}

Python での読み込み

import os
import json
from pathlib import Path

def load_secrets():
    secrets_path = Path(__file__).parent / "secrets.json"
    try:
        with open(secrets_path) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}

secrets = load_secrets()
API_KEY = secrets.get("github_token", os.getenv("GITHUB_TOKEN", ""))

if not API_KEY:
    raise ValueError("No API key found")

リクエストのパターン

基本的な GET（Python）

import requests

def api_request(url: str, api_key: str) -> dict:
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json"
    }
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()

リトライとバックオフを使用

import time
from typing import Optional

def api_request_with_retry(
    url: str,
    api_key: str,
    max_retries: int = 3
) -> Optional[dict]:
    headers = {"Authorization": f"Bearer {api_key}"}
    wait_time = 1

    for attempt in range(max_retries):
        try:
            response = requests.get(url, headers=headers, timeout=10)

            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:
                print(f"Rate limited. Waiting {wait_time}s...")
                time.sleep(wait_time)
                wait_time *= 2
            else:
                print(f"Error: HTTP {response.status_code}")
                return None
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
            time.sleep(wait_time)
            wait_time *= 2

    return None

Bash リクエスト

#!/bin/bash
API_KEY=$(python3 -c "import json; print(json.load(open('secrets.json'))['github_token'])")

curl -s -H "Authorization: Bearer $API_KEY" \
  -H "Accept: application/json" \
  "https://api.github.com/user" | jq '.'

エラー処理

try:
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    data = response.json()
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 429:
        print("Rate limited - waiting")
    elif e.response.status_code == 401:
        print("Unauthorized - check API key")
    else:
        print(f"HTTP error: {e}")
except requests.exceptions.ConnectionError:
    print("Connection error")
except requests.exceptions.Timeout:
    print("Request timeout")

HTTP ステータスコード

コード	意味	対応
200	成功	レスポンスを処理する
401	認証されていません	API キーを確認する
403	アクセス禁止	権限を確認する
404	見つかりません	エンドポイントを確認する
429	レート制限中	待機して再試行する
5xx	サーバーエラー	バックオフして再試行する

キャッシュ

import time

cache = {}
CACHE_TTL = 3600  # 1 hour

def cached_request(url: str, api_key: str) -> dict:
    now = time.time()

    if url in cache:
        data, timestamp = cache[url]
        if now - timestamp < CACHE_TTL:
            return data

    data = api_request(url, api_key)
    cache[url] = (data, now)
    return data

レート制限

ヘッダーを確認する

curl -I -H "Authorization: Bearer $API_KEY" "https://api.github.com/user" | grep -i rate
# x-ratelimit-limit: 5000
# x-ratelimit-remaining: 4999

待機時間を実装する

import time

def bulk_requests(urls: list, api_key: str, delay: float = 1.0):
    results = []
    for url in urls:
        result = api_request(url, api_key)
        results.append(result)
        time.sleep(delay)
    return results

ページネーション

def fetch_all_pages(base_url: str, api_key: str) -> list:
    all_items = []
    page = 1

    while True:
        url = f"{base_url}?page={page}&per_page=100"
        data = api_request(url, api_key)

        if not data:
            break

        all_items.extend(data)
        page += 1
        time.sleep(1)  # Respect rate limits

    return all_items

ベストプラクティス

推奨 ✅

• キーを secrets.json に保存する
• 指数バックオフを伴うリトライを実装する
• 適切な場合はレスポンスをキャッシュする
• レート制限を遵守する
• エラーを適切に処理する
• リクエストをログに記録する（機密データを含めない）

非推奨 ❌

• API キーをハードコードする
• レート制限を無視する
• エラー処理を省略する
• 短い間隔でリクエストを繰り返す
• API キーをログに記録する

API 利用時のチェックリスト

• API ドキュメントと利用規約を読む
• レート制限を確認する
• キーを安全に保存する
• レート制限を実装する
• エラー処理を追加する
• 適切にキャッシュする
• 使用状況を監視する

Grove API エラーレスポンス（必須）

Grove アプリケーションで API ルートを構築する場合、すべてのエラーレスポンスで Signpost エラーコードを使用する必要があります。独自形式の JSON エラー構造は決して返さないでください。

import {
  API_ERRORS,
  buildErrorJson,
  logGroveError,
} from "@autumnsgrove/lattice/errors";
import { json } from "@sveltejs/kit";

export const POST: RequestHandler = async ({ request, locals }) => {
  if (!locals.user) {
    logGroveError("Engine", API_ERRORS.UNAUTHORIZED, { path: "/api/resource" });
    return json(buildErrorJson(API_ERRORS.UNAUTHORIZED), { status: 401 });
  }

  const body = schema.safeParse(await request.json());
  if (!body.success) {
    return json(buildErrorJson(API_ERRORS.INVALID_REQUEST_BODY), {
      status: 400,
    });
  }

  // ... business logic
};

クライアント側では、apiRequest()（CSRF と認証情報を処理します）を使用し、トーストでフィードバックを表示します。

import { toast } from "@autumnsgrove/lattice/ui";

try {
  await apiRequest("/api/resource", { method: "POST", body });
  toast.success("Created!");
} catch (err) {
  toast.error(err instanceof Error ? err.message : "Something went wrong");
}

Signpost エラーコードの完全なリファレンスについては、AgentUsage/error_handling.md を参照してください。

関連リソース

以下を含む完全なドキュメントについては、AgentUsage/api_usage.md を参照してください。

• Bash リクエストのパターン
• 条件付きリクエスト（ETag）
• 高度なキャッシュ戦略
• 個別の API の例（GitHub、OpenWeather）