Skip to content

Skills 与外部 API 集成

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


调用外部 API 的基本模式

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

实例

python

# 文件路径: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 密钥不能直接写在脚本文件中,应通过环境变量传递。

实例

bash

# 在运行脚本前设置环境变量

export MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"

python scripts/api_client.py

# 或在脚本执行时内联设置

MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx" python scripts/api_client.py

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

bash
## 前置配置

 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 时,需要在请求间加入延迟。

实例

python

# 文件路径: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,可以将结果缓存到本地文件,避免重复请求,既省钱又提速。

实例

bash

# 文件路径: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

AI 思考中...

Skills 组合与编排

自定义 Skills 生态系统搭建

基于 VitePress 构建,部署于 GitHub Pages