TUTORIALAI & dataFETCHED

Skills 与外部 API 集成

Skills 与外部 API 集成 Skill 脚本可以像普通 Python 程序一样调用外部 HTTP API,从而将 Claude 的能力与第三方服务连接起来。 调用外部 API 的基本模式 使用 requests 库发送 HTTP 请求,是 Skill 脚本中最常见的 API 调用方式。 实例 [mycode4 type='python'] # 文件路径:scripts/api_client.py i..

Updated 7/16/2026

Tutorial detail pages default to noindex.

Skills 与外部 API 集成

Skill 脚本可以像普通 Python 程序一样调用外部 HTTP API,从而将 Claude 的能力与第三方服务连接起来。


调用外部 API 的基本模式

使用 requests 库发送 HTTP 请求,是 Skill 脚本中最常见的 API 调用方式。

实例

# 文件路径:scripts/api_client.py
import requests
import json
import sys
import os

# API 密钥从环境变量读取,不硬编码到脚本中
API_KEY = os.environ.get("MY_API_KEY", "")
BASE_URL = "https://api.example.com/v1"

def call_api(endpoint: str, payload: dict, timeout: int = 30) -> dict:
    """
    通用 API 调用封装

    参数:
        endpoint: API 路径,如 "/analyze"
        payload:  请求体(JSON)
        timeout:  超时秒数,默认 30

    返回:
        解析后的 JSON 响应,或包含 error 的字典
    """

    if not API_KEY:
        return {"error": "未设置 API 密钥,请设置环境变量 MY_API_KEY"}

    url = BASE_URL + endpoint
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type":  "application/json"
    }

    try:
        response = requests.post(url, headers=headers,
                                 json=payload, timeout=timeout)
        response.raise_for_status()   # HTTP 4xx/5xx 时抛出异常
        return response.json()

    except requests.Timeout:
        return {"error": f"请求超时(超过 {timeout} 秒)"}
    except requests.HTTPError as e:
        return {"error": f"HTTP 错误 {e.response.status_code}: {e.response.text}"}
    except requests.ConnectionError:
        return {"error": "无法连接到 API 服务,请检查网络"}
    except Exception as e:
        return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    result = call_api("/analyze", {
        "text": "runoob 是国内知名的技术学习平台",
        "lang": "zh"
    })
    print(json.dumps(result, ensure_ascii=False, indent=2))

API 密钥的安全管理

API 密钥不能直接写在脚本文件中,应通过环境变量传递。

实例

# 在运行脚本前设置环境变量
export MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
python scripts/api_client.py

# 或在脚本执行时内联设置
MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx" python scripts/api_client.py

在 SKILL.md 中告知用户需要配置密钥:

## 前置配置

本 Skill 需要外部 API 密钥,首次使用前请完成以下配置:

1. 在 [API 服务商网站] 注册并获取 API 密钥
2. 在终端中设置环境变量:
   ```bash
   export MY_API_KEY="你的密钥"
   ```
3. 若需要永久生效,将上述命令添加到 ~/.bashrc 或 ~/.zshrc

若未配置,Skill 将提示"未设置 API 密钥"并退出。

永远不要将 API 密钥提交到 Git 仓库。可以在 Skill 目录下创建 .gitignore 文件,将 .env 文件排除在版本控制之外。


处理 API 限速(Rate Limiting)

大多数 API 有请求频率限制。当批量调用 API 时,需要在请求间加入延迟。

实例

# 文件路径:scripts/rate_limited_client.py
import time
import requests

class RateLimitedClient:
    """带限速控制的 API 客户端"""

    def __init__(self, base_url: str, api_key: str,
                 requests_per_minute: int = 60):
        self.base_url    = base_url
        self.api_key     = api_key
        self.min_interval = 60.0 / requests_per_minute  # 每次请求的最小间隔(秒)
        self.last_request = 0.0

    def _wait_if_needed(self):
        """若距上次请求时间不足,等待剩余时间"""
        elapsed = time.time() - self.last_request
        wait    = self.min_interval - elapsed
        if wait > 0:
            time.sleep(wait)
        self.last_request = time.time()

    def post(self, endpoint: str, payload: dict) -> dict:
        self._wait_if_needed()
        headers = {"Authorization": f"Bearer {self.api_key}"}
        try:
            resp = requests.post(
                self.base_url + endpoint,
                headers=headers, json=payload, timeout=30
            )
            # 处理 429 Too Many Requests:等待后重试一次
            if resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 5))
                print(f"触发限速,等待 {retry_after} 秒后重试...")
                time.sleep(retry_after)
                resp = requests.post(
                    self.base_url + endpoint,
                    headers=headers, json=payload, timeout=30
                )
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            return {"error": str(e)}

响应结果缓存

对于相同输入会返回相同结果的 API,可以将结果缓存到本地文件,避免重复请求,既省钱又提速。

实例

# 文件路径:scripts/cached_api.py
import hashlib
import json
import os

CACHE_DIR = "/home/claude/.api_cache"
os.makedirs(CACHE_DIR, exist_ok=True)

def _cache_key(endpoint: str, payload: dict) -> str:
    """根据请求内容生成缓存键(MD5 哈希)"""
    raw = json.dumps({"endpoint": endpoint, "payload": payload},
                     sort_keys=True)
    return hashlib.md5(raw.encode()).hexdigest()

def cached_call(endpoint: str, payload: dict, call_fn) -> dict:
    """
    带缓存的 API 调用

    参数:
        endpoint: API 路径
        payload:  请求体
        call_fn:  实际发出 HTTP 请求的函数

    返回:
        API 响应(命中缓存时直接返回缓存结果)
    """

    key       = _cache_key(endpoint, payload)
    cache_file = os.path.join(CACHE_DIR, f"{key}.json")

    # 命中缓存
    if os.path.exists(cache_file):
        with open(cache_file, "r") as f:
            print(f"缓存命中:{key[:8]}...")
            return json.load(f)

    # 未命中,发出真实请求
    result = call_fn(endpoint, payload)

    # 保存到缓存(只缓存成功结果)
    if "error" not in result:
        with open(cache_file, "w") as f:
            json.dump(result, f, ensure_ascii=False)

    return result