跳转到主要内容

​
简介

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/tasksPOST提交新任务
/api/v1/tasks/streamPOST提交任务并获取流 URL（201）
/api/v1/tasksGET使用过滤器列出任务
/api/v1/tasks/{id}GET获取任务状态和结果
/api/v1/tasks/{id}/streamGET流式传输特定任务的事件
/api/v1/tasks/{id}/eventsGET获取任务事件历史
/api/v1/tasks/{id}/timelineGET获取 Temporal 执行时间线

​
任务控制

端点方法描述
/api/v1/tasks/{id}/cancelPOST取消正在运行的任务
/api/v1/tasks/{id}/pausePOST在下一个检查点暂停任务
/api/v1/tasks/{id}/resumePOST恢复已暂停的任务
/api/v1/tasks/{id}/control-stateGET获取任务控制状态（暂停/运行）

​
实时流式传输

端点方法描述
/api/v1/stream/sse?workflow_id={id}GET服务器发送事件流
/api/v1/stream/ws?workflow_id={id}GETWebSocket 流

​
会话管理

端点方法描述
/api/v1/sessionsGET列出会话（分页）
/api/v1/sessions/{sessionId}GET获取会话详情
/api/v1/sessions/{sessionId}/historyGET获取会话聊天历史
/api/v1/sessions/{sessionId}/eventsGET获取会话事件（按轮次分组）
/api/v1/sessions/{sessionId}PATCH更新会话标题（UUID 或 external_id）
/api/v1/sessions/{sessionId}DELETE删除会话（软删除；幂等 204）

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

端点方法描述
/api/v1/schedulesPOST创建新的定时任务
/api/v1/schedulesGET列出定时任务（支持过滤）
/api/v1/schedules/{id}GET获取定时任务详情
/api/v1/schedules/{id}/runsGET获取定时任务执行历史
/api/v1/schedules/{id}PUT更新定时任务配置
/api/v1/schedules/{id}/pausePOST暂停定时任务
/api/v1/schedules/{id}/resumePOST恢复定时任务
/api/v1/schedules/{id}DELETE删除定时任务

​
认证

端点方法描述需要认证
/api/v1/auth/registerPOST注册新用户否
/api/v1/auth/loginPOST登录并获取 JWT 令牌否
/api/v1/auth/refreshPOST刷新 JWT 访问令牌否
/api/v1/auth/meGET获取当前用户信息是
/api/v1/auth/refresh-keyPOST重新生成 API 密钥是
/api/v1/auth/api-keysGET列出用户的 API 密钥是
/api/v1/auth/api-keysPOST创建新的 API 密钥是
/api/v1/auth/api-keys/{id}DELETE撤销 API 密钥是

​
审批

端点方法描述
/api/v1/approvals/decisionPOST提交人工审批决策（人机协作任务）

​
OpenAI 兼容 API

Shannon 提供与 OpenAI 兼容的 API，便于与现有 OpenAI SDK 和工具集成。详细信息请参阅 OpenAI 兼容 API 参考。
端点方法描述
/v1/chat/completionsPOST聊天补全（支持流式传输）
/v1/modelsGET列出可用模型
/v1/models/{model}GET获取特定模型详情

​
健康检查和可观测性

端点方法描述需要认证
/healthGET健康检查否
/readinessGET就绪探针否
/openapi.jsonGETOpenAPI 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 入门