API 概述 - Shannon

简介

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）。生产环境请启用。

启用后，通过请求头传递 API 密钥：

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": "情感：积极"
}

错误（400/401/404/429/500）：

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

速率限制

默认：每个 API 密钥 60 请求/分钟
响应头：X-RateLimit-Limit、X-RateLimit-Remaining、X-RateLimit-Reset
429 响应：包含 Retry-After 头部

事件流式传输

Shannon 为实时监控发出 33 种事件类型： 核心事件：

WORKFLOW_STARTED、WORKFLOW_COMPLETED
AGENT_STARTED、AGENT_COMPLETED
STATUS_UPDATE、DATA_PROCESSING
ERROR_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

Python SDK

官方 Python 客户端，支持流式传输

REST 客户端

使用任何 HTTP 客户端库

下一步

提交任务

任务提交 API

流式传输事件

实时事件流

会话 API

会话管理

认证

API 密钥和 JWT 认证

任务控制

暂停、恢复和取消任务

Python SDK

Python 入门

​简介

​快速开始

​API 端点

​核心任务操作

​任务控制

​实时流式传输

​会话管理

​定时任务（周期性任务）

​认证

​审批

​OpenAI 兼容 API

​健康检查和可观测性

​认证

​响应格式

​速率限制

​事件流式传输

​客户端接入

​最佳实践

​1. 使用会话 ID 保持上下文

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

​3. 处理速率限制

​4. 使用幂等性键

​SDK

Python SDK

REST 客户端

​下一步

提交任务

流式传输事件

会话 API

认证

任务控制

Python SDK

简介

快速开始

API 端点

核心任务操作

任务控制

实时流式传输

会话管理

定时任务（周期性任务）

认证

审批

OpenAI 兼容 API

健康检查和可观测性

认证

响应格式

速率限制

事件流式传输

客户端接入

最佳实践

1. 使用会话 ID 保持上下文

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

3. 处理速率限制

4. 使用幂等性键

SDK

下一步