REST API 参考

基础 URL

http://localhost:8080/api/v1

端点摘要

Shannon 提供 40+ 个 REST 端点，分为 8 个类别：

任务操作（10 个端点）

端点	描述
`POST /tasks`	提交新任务
`POST /tasks/stream`	提交任务并接收流 URL
`GET /tasks`	使用过滤器列出任务（limit、offset、status、session_id）
`GET /tasks/{id}`	获取任务状态和结果
`GET /tasks/{id}/events`	获取持久化事件历史
`GET /tasks/{id}/timeline`	获取确定性 Temporal 时间线
`POST /tasks/{id}/pause`	在工作流检查点暂停运行中的任务
`POST /tasks/{id}/resume`	恢复之前暂停的任务
`GET /tasks/{id}/control-state`	获取任务的暂停/取消控制状态
`POST /tasks/{id}/cancel`	取消排队或运行中的任务

流式传输（2 个端点）

端点	描述
`GET /stream/sse?workflow_id={id}`	服务器发送事件流
`GET /stream/ws?workflow_id={id}`	WebSocket 流

会话管理（6 个端点）

端点	描述
`GET /sessions`	列出会话（分页）
`GET /sessions/{sessionId}`	获取会话详情
`GET /sessions/{sessionId}/history`	获取会话对话历史
`GET /sessions/{sessionId}/events`	获取会话事件（按轮次分组）
`PATCH /sessions/{sessionId}`	更新会话标题（UUID 或 external_id）
`DELETE /sessions/{sessionId}`	删除会话（软删除；幂等 204）

定时任务（8 个端点）

端点	描述
`POST /schedules`	创建新的定时任务
`GET /schedules`	获取定时任务列表
`GET /schedules/{id}`	获取定时任务详情
`GET /schedules/{id}/runs`	获取定时任务执行历史
`PUT /schedules/{id}`	更新定时任务配置
`POST /schedules/{id}/pause`	暂停定时任务
`POST /schedules/{id}/resume`	恢复定时任务
`DELETE /schedules/{id}`	删除定时任务

认证（8 个端点）

端点	描述
`POST /auth/register`	注册新用户
`POST /auth/login`	登录并获取令牌
`POST /auth/refresh`	刷新访问令牌
`GET /auth/me`	获取当前用户信息
`POST /auth/api-keys/refresh`	刷新 API 密钥
`GET /auth/api-keys`	列出 API 密钥
`POST /auth/api-keys`	创建新的 API 密钥
`DELETE /auth/api-keys/{id}`	撤销 API 密钥

审批与人工审核（3 个端点）

端点	描述
`POST /approvals/decision`	提交人工审批决策
`GET /tasks/{id}/review`	获取人工审核详情
`POST /tasks/{id}/review`	提交人工审核决策

OpenAI 兼容接口（3 个端点）

端点	描述
`POST /v1/chat/completions`	聊天补全（支持流式传输）
`GET /v1/models`	获取可用模型列表
`GET /v1/models/{model}`	获取模型详情

健康检查和可观测性（3 个端点）

端点	需要认证	描述
`GET /health`	否	健康检查探针
`GET /readiness`	否	就绪探针（检查 orchestrator）
`GET /openapi.json`	否	OpenAPI 3.0 规范

认证

开发环境：认证默认禁用（GATEWAY_SKIP_AUTH=1）。生产环境：设置 GATEWAY_SKIP_AUTH=0 并使用 API 密钥。

API 密钥头部

curl -H "X-API-Key: sk_test_123456" \
  http://localhost:8080/api/v1/tasks

SSE 流式传输认证

浏览器 EventSource 无法发送自定义请求头。生产环境请通过后端代理转发 SSE，并注入 X-API-Key 或 Bearer 头。SSE 端点支持 api_key 查询参数作为回退（如 ?api_key=sk_...）。

const eventSource = new EventSource(
  `/api/v1/stream/sse?workflow_id=${id}`
);

常用请求头

请求头

头部	必需	描述
`X-API-Key`	是*	API 认证密钥
`Content-Type`	是（POST）	`application/json`
`Idempotency-Key`	否	防止重复提交（24 小时缓存）
`traceparent`	否	W3C 跟踪上下文，用于分布式追踪

*如果 GATEWAY_SKIP_AUTH=1 则不需要

响应头

头部	描述
`X-Workflow-ID`	Temporal 工作流标识符，用于追踪
`X-Session-ID`	会话标识符
`X-RateLimit-Limit`	每分钟允许的请求数
`X-RateLimit-Remaining`	当前窗口剩余的请求数
`X-RateLimit-Reset`	限制重置的 Unix 时间戳

速率限制

默认：每个 API 密钥 60 请求/分钟
状态码：超出时返回 429 Too Many Requests
响应头：Retry-After 指示等待秒数

{
  "error": "超出速率限制",
  "retry_after": 60
}

幂等性

使用 Idempotency-Key 头部防止重复任务提交：

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000" \
  -H "Content-Type: application/json" \
  -d '{"query": "分析数据"}'

行为：

24 小时内使用相同密钥返回缓存响应
重复请求的请求体被忽略
使用 Redis 进行分布式缓存

错误处理

HTTP 状态码

代码	含义	常见原因
`200`	成功	请求完成
`201`	已创建	任务提交成功
`400`	错误请求	无效的 JSON、缺少必需字段
`401`	未授权	缺少或无效的 API 密钥
`404`	未找到	任务/会话未找到
`429`	请求过多	超出速率限制
`500`	内部服务器错误	服务器错误
`502`	网关错误	Orchestrator 不可用
`503`	服务不可用	系统过载

错误响应格式

{
  "error": "任务未找到"
}

CORS

为开发启用 CORS：

Access-Control-Allow-Origin: *
Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS
Access-Control-Allow-Headers: Content-Type, X-API-Key, Idempotency-Key, traceparent

生产环境：将 Access-Control-Allow-Origin 配置为特定域名。

超时设置

操作	超时
HTTP 读取	30 秒
HTTP 写入	30 秒
HTTP 空闲	60 秒
代理执行	10 分钟（通过 `AGENT_TIMEOUT_SECONDS` 可配置）

最佳实践

1. 始终使用幂等性键

import uuid
import requests

response = requests.post(
    "http://localhost:8080/api/v1/tasks",
    headers={
        "X-API-Key": "sk_test_123456",
        "Idempotency-Key": str(uuid.uuid4())
    },
    json={"query": "分析数据"}
)

2. 实现指数退避

import time

def submit_with_retry(query, max_retries=3):
    for attempt in range(max_retries):
        response = requests.post(url, json={"query": query})

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            time.sleep(retry_after)
            continue

        return response

3. 对长任务使用流式传输事件

import httpx

with httpx.stream(
    "GET",
    f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}",
    headers={"X-API-Key": "sk_test_123456"}
) as response:
    for line in response.iter_lines():
        if line.startswith("data:"):
            event = json.loads(line[5:])
            print(event["type"])

4. 从响应中提取工作流 ID

response = requests.post(
    "http://localhost:8080/api/v1/tasks",
    headers={"X-API-Key": "sk_test_123456"},
    json={"query": "分析数据"}
)

workflow_id = response.headers.get("X-Workflow-ID")
# 使用 workflow_id 进行流式传输或 Temporal UI

快速示例

#!/bin/bash
API_KEY="sk_test_123456"
BASE="http://localhost:8080/api/v1"

# 提交任务
RESPONSE=$(curl -s -X POST "$BASE/tasks" \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"query": "2+2 等于多少？"}')

TASK_ID=$(echo $RESPONSE | jq -r '.task_id')
echo "任务: $TASK_ID"

# 获取状态
curl -s "$BASE/tasks/$TASK_ID" -H "X-API-Key: $API_KEY" | jq

# 流式传输事件
curl -N "$BASE/stream/sse?workflow_id=$TASK_ID" -H "X-API-Key: $API_KEY"

下一步

提交任务

任务提交端点

流式传输

SSE 和 WebSocket 流

会话

会话管理

Python SDK

使用 Python SDK

​基础 URL

​端点摘要

​任务操作（10 个端点）

​流式传输（2 个端点）

​会话管理（6 个端点）

​定时任务（8 个端点）

​认证（8 个端点）

​审批与人工审核（3 个端点）

​OpenAI 兼容接口（3 个端点）

​健康检查和可观测性（3 个端点）

​认证

​API 密钥头部

​SSE 流式传输认证

​常用请求头

​请求头

​响应头

​速率限制

​幂等性

​错误处理

​HTTP 状态码

​错误响应格式

​CORS

​超时设置

​最佳实践

​1. 始终使用幂等性键

​2. 实现指数退避

​3. 对长任务使用流式传输事件

​4. 从响应中提取工作流 ID

​快速示例

​下一步

提交任务

流式传输

会话

Python SDK

基础 URL

端点摘要

任务操作（10 个端点）

流式传输（2 个端点）

会话管理（6 个端点）

定时任务（8 个端点）

认证（8 个端点）

审批与人工审核（3 个端点）

OpenAI 兼容接口（3 个端点）

健康检查和可观测性（3 个端点）

认证

API 密钥头部

SSE 流式传输认证

常用请求头

请求头

响应头

速率限制

幂等性

错误处理

HTTP 状态码

错误响应格式

CORS

超时设置

最佳实践

1. 始终使用幂等性键

2. 实现指数退避

3. 对长任务使用流式传输事件

4. 从响应中提取工作流 ID

快速示例

下一步