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 和凭据）并显示 Toast 反馈：

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）