Skills und externe API-Integration
Skill-Skripte können externe HTTP-APIs genau wie gewöhnliche Python-Programme aufrufen und dadurch die Fähigkeiten von Claude mit Diensten von Drittanbietern verbinden.
Grundlegendes Muster zum Aufrufen externer APIs
Die Verwendung der Bibliothekrequestszum Senden von HTTP-Anfragen ist die gebräuchlichste Methode zum Aufrufen von APIs in Skill-Skripten.
Beispiel
# Dateipfad: scripts/api_client.py
import requests
import json
import sys
import os
# Der API-Schlüssel wird aus Umgebungsvariablen gelesen, nicht im Skript hartcodiert
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:
"""
Generischer API-Aufruf-Wrapper
Parameter:
endpoint: API-Pfad, z. B. "/analyze"
payload: Anfragekörper (JSON)
timeout: Timeout in Sekunden, Standardwert 30
Rückgabe:
Geparste JSON-Antwort oder ein Dictionary, das einen Fehler enthält
"""
if not API_KEY:
return {"error": "API-Schlüssel nicht gesetzt, bitte setzen Sie die Umgebungsvariable 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() # Löst bei HTTP 4xx/5xx eine Ausnahme aus
return response.json()
except requests.Timeout:
return {"error": f"Zeitüberschreitung der Anfrage ({timeout} Sekunden überschritten)"}
except requests.HTTPError as e:
return {"error": f"HTTP-Fehler {e.response.status_code}: {e.response.text}"}
except requests.ConnectionError:
return {"error": "Verbindung zum API-Dienst nicht möglich, bitte überprüfen Sie das Netzwerk"}
except Exception as e:
return {"error": str(e)}
# Verwendungsbeispiel
if __name__ == "__main__":
result = call_api("/analyze", {
"text": "Beispiel ist eine bekannte technische Lernplattform in China",
"lang": "zh"
})
print(json.dumps(result, ensure_ascii=False, indent=2))
import requests
import json
import sys
import os
# Der API-Schlüssel wird aus Umgebungsvariablen gelesen, nicht im Skript hartcodiert
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:
"""
Generischer API-Aufruf-Wrapper
Parameter:
endpoint: API-Pfad, z. B. "/analyze"
payload: Anfragekörper (JSON)
timeout: Timeout in Sekunden, Standardwert 30
Rückgabe:
Geparste JSON-Antwort oder ein Dictionary, das einen Fehler enthält
"""
if not API_KEY:
return {"error": "API-Schlüssel nicht gesetzt, bitte setzen Sie die Umgebungsvariable 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() # Löst bei HTTP 4xx/5xx eine Ausnahme aus
return response.json()
except requests.Timeout:
return {"error": f"Zeitüberschreitung der Anfrage ({timeout} Sekunden überschritten)"}
except requests.HTTPError as e:
return {"error": f"HTTP-Fehler {e.response.status_code}: {e.response.text}"}
except requests.ConnectionError:
return {"error": "Verbindung zum API-Dienst nicht möglich, bitte überprüfen Sie das Netzwerk"}
except Exception as e:
return {"error": str(e)}
# Verwendungsbeispiel
if __name__ == "__main__":
result = call_api("/analyze", {
"text": "Beispiel ist eine bekannte technische Lernplattform in China",
"lang": "zh"
})
print(json.dumps(result, ensure_ascii=False, indent=2))
Sichere Verwaltung von API-Schlüsseln
API-Schlüssel sollten nicht direkt in Skriptdateien geschrieben werden; sie sollten über Umgebungsvariablen übergeben werden.
Beispiel
# Umgebungsvariablen vor dem Ausführen des Skripts setzen
export MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
python scripts/api_client.py
# Oder legen Sie sie inline fest, wenn Sie das Skript ausführen
MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx" python scripts/api_client.py
export MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
python scripts/api_client.py
# Oder legen Sie sie inline fest, wenn Sie das Skript ausführen
MY_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx" python scripts/api_client.py
Informieren Sie in SKILL.md die Benutzer darüber, dass sie den Schlüssel konfigurieren müssen:
## 前置配置 本 Skill 需要外部 API 密钥，首次使用前请完成以下配置： 1. 在 [API 服务商网站] 注册并获取 API 密钥 2. 在终端中设置环境变量： ```bash export MY_API_KEY="你的密钥" ``` 3. 若需要永久生效，将上述命令添加到 ~/.bashrc oder ~/.zshrc 若未配置，Skill 将提示"未设置 API 密钥"并退出。
Committen Sie niemals API-Schlüssel in ein Git-Repository. Sie können eine .gitignore-Datei im Skill-Verzeichnis erstellen, um die .env-Datei von der Versionskontrolle auszuschließen.
Umgang mit API-Ratenbegrenzung
Die meisten APIs haben Limits für die Anforderungsfrequenz. Wenn Sie APIs im Batch aufrufen, müssen Sie Verzögerungen zwischen den Anfragen hinzufügen.
Beispiel
# Dateipfad: scripts/rate_limited_client.py
import time
import requests
class RateLimitedClient:
"""API-Client mit Ratenbegrenzungssteuerung"""
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 # Mindestintervall zwischen Anfragen (Sekunden)
self.last_request = 0.0
def _wait_if_needed(self):
"""Wenn die Zeit seit der letzten Anfrage nicht ausreicht, warte die verbleibende Zeit ab"""
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 behandeln: warten und einmal erneut versuchen
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 5))
print(f"Rate-Limit ausgelöst, warte {retry_after} Sekunden, bevor erneut versucht wird...")
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-Client mit Ratenbegrenzungssteuerung"""
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 # Mindestintervall zwischen Anfragen (Sekunden)
self.last_request = 0.0
def _wait_if_needed(self):
"""Wenn die Zeit seit der letzten Anfrage nicht ausreicht, warte die verbleibende Zeit ab"""
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 behandeln: warten und einmal erneut versuchen
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 5))
print(f"Rate-Limit ausgelöst, warte {retry_after} Sekunden, bevor erneut versucht wird...")
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)}
Antwort-Caching
Für APIs, die für dieselbe Eingabe dasselbe Ergebnis zurückgeben, können Sie die Ergebnisse in lokalen Dateien zwischenspeichern, um doppelte Anfragen zu vermeiden, Geld zu sparen und die Geschwindigkeit zu verbessern.
Beispiel
# Dateipfad: 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:
"""Erzeuge einen Cache-Schlüssel basierend auf dem Inhalt der Anfrage (MD5-Hash)"""
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-Aufruf mit Caching
Parameter:
endpoint: API-Pfad
payload: Anfrage-Body
call_fn: Die Funktion, die tatsächlich die HTTP-Anfrage ausführt
Rückgabe:
API-Antwort (gibt bei einem Cache-Treffer das zwischengespeicherte Ergebnis direkt zurück)
"""
key = _cache_key(endpoint, payload)
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
# Cache-Treffer
if os.path.exists(cache_file):
with open(cache_file, "r") as f:
print(f"Cache-Treffer: {key[:8]}...")
return json.load(f)
# Cache-Miss, eine echte Anfrage stellen
result = call_fn(endpoint, payload)
# Im Cache speichern (nur erfolgreiche Ergebnisse cachen)
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:
"""Erzeuge einen Cache-Schlüssel basierend auf dem Inhalt der Anfrage (MD5-Hash)"""
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-Aufruf mit Caching
Parameter:
endpoint: API-Pfad
payload: Anfrage-Body
call_fn: Die Funktion, die tatsächlich die HTTP-Anfrage ausführt
Rückgabe:
API-Antwort (gibt bei einem Cache-Treffer das zwischengespeicherte Ergebnis direkt zurück)
"""
key = _cache_key(endpoint, payload)
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
# Cache-Treffer
if os.path.exists(cache_file):
with open(cache_file, "r") as f:
print(f"Cache-Treffer: {key[:8]}...")
return json.load(f)
# Cache-Miss, eine echte Anfrage stellen
result = call_fn(endpoint, payload)
# Im Cache speichern (nur erfolgreiche Ergebnisse cachen)
if "error" not in result:
with open(cache_file, "w") as f:
json.dump(result, f, ensure_ascii=False)
return result