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))
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
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)}
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
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