LongCat 官方 API 接入 OpenClaw：从非官方订阅到合规自动化的迁移路径

import os
import requests
import json
from datetime import datetime

# ---- 配置区 ----
# 从环境变量读取，避免硬编码凭证
LONGCAT_API_KEY = os.environ.get("LONGCAT_API_KEY")
LONGCAT_BASE_URL = "https://api.longcat.io/v1"  # 官方端点，以官方文档为准

if not LONGCAT_API_KEY:
    raise RuntimeError("请设置环境变量 LONGCAT_API_KEY，例如: export LONGCAT_API_KEY=your_key_here")

headers = {
    "Authorization": f"Bearer {LONGCAT_API_KEY}",
    "Content-Type": "application/json",
}


def submit_task(prompt: str, model: str = "openclaw-default") -> dict:
    """向 OpenClaw 提交一个自动化任务，返回任务 ID 与状态"""
    payload = {
        "model": model,
        "prompt": prompt,
        "timestamp": datetime.utcnow().isoformat(),
    }
    resp = requests.post(
        f"{LONGCAT_BASE_URL}/openclaw/tasks",
        headers=headers,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def poll_task(task_id: str, interval: int = 5, max_attempts: int = 20) -> dict:
    """轮询任务结果，直到完成或超时"""
    for attempt in range(max_attempts):
        resp = requests.get(
            f"{LONGCAT_BASE_URL}/openclaw/tasks/{task_id}",
            headers=headers,
            timeout=15,
        )
        resp.raise_for_status()
        data = resp.json()
        if data.get("status") in ("completed", "failed"):
            return data
        print(f"[{attempt+1}] 任务 {task_id} 状态: {data['status']}，等待 {interval}s...")
        import time
        time.sleep(interval)
    raise TimeoutError(f"任务 {task_id} 在 {max_attempts * interval}s 内未完成")


# ---- 主流程 ----
if __name__ == "__main__":
    # 提交任务
    result = submit_task("对最近一周的项目日志做摘要，输出 Markdown 格式")
    task_id = result["task_id"]
    print(f"任务已提交，ID: {task_id}")

    # 等待结果
    final = poll_task(task_id)
    if final["status"] == "completed":
        print("任务完成，输出内容：")
        print(final["output"])
    else:
        print(f"任务失败，原因: {final.get('error', '未知')}")

export LONGCAT_API_KEY=your_key_here
python longcat_openclaw_demo.py