Skip to main content

​
基础 URL

http://localhost:8080/api/v1

​
端点摘要

Shannon 提供 19 个 REST 端点，分为 4 个类别：

​
任务操作（8 个端点）

端点描述
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}/cancel取消排队或运行中的任务
POST /approvals/decision提交人工审批决策

​
流式传输（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）

​
健康检查和可观测性（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 头。不要通过 URL 查询参数传递 API 密钥。
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-IDTemporal 工作流标识符，用于追踪
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