Shannon 提供 HTTP REST 和 gRPC API，用于提交任务、流式传输结果和管理 AI 智能体工作流。
基础 URL：http://localhost:8080（开发环境）
快速开始
# 提交任务
curl -X POST http://localhost:8080/api/v1/tasks \
-H "Content-Type: application/json" \
-d '{"query": "分析情感：Shannon 让 AI 变得简单"}'
# 响应
{
"task_id": "task-dev-1234567890",
"status": "submitted",
"created_at": "2025-01-20T10:00:00Z"
}
# 流式传输实时事件
curl -N "http://localhost:8080/api/v1/stream/sse?workflow_id=task-dev-1234567890"
API 端点
核心任务操作
| 端点 | 方法 | 描述 |
|---|
/api/v1/tasks | POST | 提交新任务 |
/api/v1/tasks/stream | POST | 提交任务并获取流 URL（201） |
/api/v1/tasks | GET | 使用过滤器列出任务 |
/api/v1/tasks/{id} | GET | 获取任务状态和结果 |
/api/v1/tasks/{id}/stream | GET | 流式传输特定任务的事件 |
/api/v1/tasks/{id}/events | GET | 获取任务事件历史 |
/api/v1/tasks/{id}/timeline | GET | 获取 Temporal 执行时间线 |
任务控制
| 端点 | 方法 | 描述 |
|---|
/api/v1/tasks/{id}/cancel | POST | 取消正在运行的任务 |
/api/v1/tasks/{id}/pause | POST | 在下一个检查点暂停任务 |
/api/v1/tasks/{id}/resume | POST | 恢复已暂停的任务 |
/api/v1/tasks/{id}/control-state | GET | 获取任务控制状态（暂停/运行） |
实时流式传输
| 端点 | 方法 | 描述 |
|---|
/api/v1/stream/sse?workflow_id={id} | GET | 服务器发送事件流 |
/api/v1/stream/ws?workflow_id={id} | GET | WebSocket 流 |
会话管理
| 端点 | 方法 | 描述 |
|---|
/api/v1/sessions | GET | 列出会话（分页） |
/api/v1/sessions/{sessionId} | GET | 获取会话详情 |
/api/v1/sessions/{sessionId}/history | GET | 获取会话聊天历史 |
/api/v1/sessions/{sessionId}/events | GET | 获取会话事件（按轮次分组） |
/api/v1/sessions/{sessionId} | PATCH | 更新会话标题（UUID 或 external_id） |
/api/v1/sessions/{sessionId} | DELETE | 删除会话（软删除；幂等 204） |
定时任务（周期性任务）
| 端点 | 方法 | 描述 |
|---|
/api/v1/schedules | POST | 创建新的定时任务 |
/api/v1/schedules | GET | 列出定时任务（支持过滤） |
/api/v1/schedules/{id} | GET | 获取定时任务详情 |
/api/v1/schedules/{id}/runs | GET | 获取定时任务执行历史 |
/api/v1/schedules/{id} | PUT | 更新定时任务配置 |
/api/v1/schedules/{id}/pause | POST | 暂停定时任务 |
/api/v1/schedules/{id}/resume | POST | 恢复定时任务 |
/api/v1/schedules/{id} | DELETE | 删除定时任务 |
| 端点 | 方法 | 描述 | 需要认证 |
|---|
/api/v1/auth/register | POST | 注册新用户 | 否 |
/api/v1/auth/login | POST | 登录并获取 JWT 令牌 | 否 |
/api/v1/auth/refresh | POST | 刷新 JWT 访问令牌 | 否 |
/api/v1/auth/me | GET | 获取当前用户信息 | 是 |
/api/v1/auth/refresh-key | POST | 重新生成 API 密钥 | 是 |
/api/v1/auth/api-keys | GET | 列出用户的 API 密钥 | 是 |
/api/v1/auth/api-keys | POST | 创建新的 API 密钥 | 是 |
/api/v1/auth/api-keys/{id} | DELETE | 撤销 API 密钥 | 是 |
| 端点 | 方法 | 描述 |
|---|
/api/v1/approvals/decision | POST | 提交人工审批决策（人机协作任务） |
OpenAI 兼容 API
Shannon 提供与 OpenAI 兼容的 API，便于与现有 OpenAI SDK 和工具集成。详细信息请参阅 OpenAI 兼容 API 参考。
| 端点 | 方法 | 描述 |
|---|
/v1/chat/completions | POST | 聊天补全（支持流式传输） |
/v1/models | GET | 列出可用模型 |
/v1/models/{model} | GET | 获取特定模型详情 |
健康检查和可观测性
| 端点 | 方法 | 描述 | 需要认证 |
|---|
/health | GET | 健康检查 | 否 |
/readiness | GET | 就绪探针 | 否 |
/openapi.json | GET | OpenAPI 3.0 规范 | 否 |
开发环境默认：认证已禁用（GATEWAY_SKIP_AUTH=1）。生产环境请启用。
curl -H "X-API-Key: sk_test_123456" \
http://localhost:8080/api/v1/tasks
SSE 流式传输：浏览器 EventSource 无法发送自定义请求头。
- 开发环境：设置
GATEWAY_SKIP_AUTH=1 可跳过认证。
- 生产环境：由后端（或边缘代理）发起 SSE 并注入
X-API-Key 或 Authorization: Bearer 请求头。
- SSE 端点支持
api_key 查询参数作为回退（当无法发送请求头时）。
响应格式
所有端点返回 JSON，错误格式一致：
成功（200）：
{
"task_id": "task-dev-1234567890",
"status": "COMPLETED",
"result": "情感：积极"
}
速率限制
- 默认：每个 API 密钥 60 请求/分钟
- 响应头：
X-RateLimit-Limit、X-RateLimit-Remaining、X-RateLimit-Reset
- 429 响应：包含
Retry-After 头部
事件流式传输
Shannon 为实时监控发出 33 种事件类型：
核心事件：
WORKFLOW_STARTED、WORKFLOW_COMPLETEDAGENT_STARTED、AGENT_COMPLETEDSTATUS_UPDATE、DATA_PROCESSINGERROR_OCCURRED
LLM 事件：
LLM_PROMPT、LLM_OUTPUT、LLM_PARTIAL
工具事件：
TOOL_INVOKED、TOOL_OBSERVATION
查看事件类型参考获取完整列表。
Python SDK：通过 EventType 枚举进行过滤与判断，参见SDK 流式传输。
客户端接入
推荐使用网关 REST API 或 Python SDK（仅 HTTP）：
- REST 端点：
http://localhost:8080/api/v1/*
- Python SDK：
ShannonClient(base_url="http://localhost:8080")
说明：gRPC 服务为内部接口，不属于公共 SDK 的一部分。OpenAI 兼容端点 /v1/chat/completions 仅用于兼容性。如需完整的 Shannon 功能（技能、会话工作区、研究策略），请使用 /api/v1/tasks 及相关端点。
最佳实践
1. 使用会话 ID 保持上下文
{
"query": "Q2 怎么样？",
"session_id": "user-123-analytics"
}
2. 对长任务使用流式传输事件
import requests
r = requests.get(
f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}",
stream=True
)
for line in r.iter_lines():
if line:
event = parse_sse(line)
print(event['type'])
3. 处理速率限制
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
4. 使用幂等性键
curl -X POST http://localhost:8080/api/v1/tasks \
-H "Idempotency-Key: $(uuidgen)" \
-d '{"query": "只处理一次"}'
SDK
下一步
