CAI API Backend 实战指南用cai --api构建有状态的 AI 安全 Agent HTTP 服务【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/caicai --api是 CAICybersecurity AI框架内置的 HTTP 后端模式它以 FastAPI 为基础为每个会话维护独立的 Agent 实例与记忆通过 REST 路由执行 REPL 命令或向模型发送提示词。读完本文你将掌握如何启动并配置 API 服务、理解全部 18 个端点的请求/响应契约、正确设置认证与调试日志并用 Python / Node / iOS Swift / curl 从零构建流式与非流式客户端。本文以仓库中的 docs/api.md 为骨架结合 src/cai/cli.py 的环境变量约定、src/cai/agents/factory.py 的 Agent 注册机制与 pricing.json 的模型目录展开纵深讲解。一、总体架构有状态的 FastAPI 后端cai --api暴露的是一个有状态stateful的 HTTP 服务与一次性调用模型的普通 REST API 有本质区别会话即 Agent每个 session 对应一个独立的 Agent 实例携带自己的对话历史与记忆stateful: true时上下文跨请求持续累积REST 路由复用 REPL 能力/commands系列端点直接把 REPL 的斜杠命令如/memory show暴露为 HTTP 接口复用 src/cai/repl/commands 的命令处理逻辑流式分层提供两种流式粒度——/messages/stream输出高层推理步骤reasoning steps/messages/stream_tokens额外输出 token 级增量。启动后交互式文档位于/api/docsSwagger UIOpenAPI 规范位于/api/openapi.json可作为任何语言客户端的契约来源。二、启动服务命令行参数与环境变量最简单的启动方式cai --api --api-host 0.0.0.0 --api-port 8080 # 如果 8080或你指定的端口被占用服务会自动选取下一个空闲端口并在控制台打印实际端口所有 CLI 标志均可通过环境变量等价设置这也是容器化部署与docker-compose环境注入的基础Flag环境变量说明--apiCAI_API_MODE启用 HTTP 后端--api-hostCAI_API_HOST绑定主机/网卡默认127.0.0.1--api-portCAI_API_PORT绑定端口默认8000--api-reloadCAI_API_RELOAD开发模式自动重载--api-workersCAI_API_WORKERSWorker 进程数与 reload 互斥reload 时忽略说明与 src/cai/cli.py 中CAI_AGENT_TYPE默认one_tool_agent、CAI_MODEL默认alias1等既有约定一致API 模式的默认 Agent 与模型同样由环境变量驱动创建会话时未显式指定则回落到这些默认值。认证与安全API 使用客户端的ALIAS_API_KEY作为共享密钥设置ALIAS_API_KEY后所有除/health外的请求必须在请求头携带X-CAI-API-Key: $ALIAS_API_KEY请求头名称可通过CAI_API_KEY_HEADER自定义若ALIAS_API_KEY未设置API 处于无保护状态仅限本地开发为兼容旧配置CAI_API_KEY会被接受作为回退密钥。日志与调试服务端日志通过以下环境变量控制需在cai --api之前设置CAI_API_LOG_LEVELdebug或trace调高服务端日志级别CAI_API_LOG_REQUESTStrue打印请求日志方法/路径/请求头/请求体预览CAI_API_LOG_AUTHtrue打印认证决策日志便于排查为什么返回 401CAI_API_RELOADtrue开发时自动重载。一个完整的认证 全量调试启动示例ALIAS_API_KEYyour_key \ CAI_API_LOG_LEVELdebug \ CAI_API_LOG_REQUESTStrue \ CAI_API_LOG_AUTHtrue \ CAI_API_RELOADtrue \ cai --api --api-host 0.0.0.0 --api-port 8080内容类型请求/响应负载使用 JSONapplication/json流式端点使用 Server-Sent EventsSSEtext/event-stream。三、端点总览所有已认证调用都需要携带请求头X-CAI-API-Key: $ALIAS_API_KEY方法路径说明GET/api/v1/health存活检查无需认证GET/api/v1/commands列出全部 REPL 命令POST/api/v1/commands/{command}执行 REPL 命令POST/api/v1/sessions创建有状态会话GET/api/v1/sessions列出会话摘要GET/api/v1/sessions/{id}会话详情含完整历史DELETE/api/v1/sessions/{id}删除会话POST/api/v1/sessions/{id}/reset重置会话 Agent 与历史POST/api/v1/sessions/{id}/messages非流式推理POST/api/v1/sessions/{id}/messages/stream推理步骤流SSEPOST/api/v1/sessions/{id}/messages/stream_tokensToken 级流SSEGET/api/v1/sessions/{id}/history获取会话历史POST/api/v1/sessions/{id}/interrupt中断正在运行的任务POST/api/v1/sessions/{id}/reload重建会话 AgentGET/api/v1/agents列出可用 Agent 与模式GET/api/v1/models列出已知模型与定价POST/api/v1/sessions/{id}/ux/final_message/stream_tokens任务收尾说明的 token 流POST/api/v1/ux/title生成标题无会话POST/api/v1/ux/summarize生成一句话摘要无会话四、核心端点详解4.1 GET /api/v1/health存活检查无需认证。返回 200{status:ok,version:semver or dev}4.2 GET /api/v1/commands列出所有 REPL 命令名称、别名、子命令。返回 200{ commands: [ {name:/memory,description:memory ops,aliases:[],subcommands:[show]}, {name:/help,description:display help,aliases:[/h],subcommands:[]} ] }4.3 POST /api/v1/commands/{command}执行一条 REPL 命令对应 src/cai/repl/commands 中的命令处理器。Body{args: [show], auto_correct: true}响应 200{handled: true, suggested_command: null, stdout: ..., stderr: , exit_code: null}auto_correct默认true允许对命令名做模糊纠正当命令未被识别时suggested_command会给出纠错建议。4.4 会话生命周期创建 / 列出 / 详情 / 删除 / 重置创建会话POST/api/v1/sessions——每个会话拥有独立的 Agent 实例与记忆{agent: redteam_agent, model: alias1, stateful: true, metadata: {}}响应 201SessionDetailModel包含摘要与初始为空的 history。agent默认来自CAI_AGENT_TYPEmodel默认来自CAI_MODELstateful默认true。列出会话GET/api/v1/sessions返回 200{sessions: [{id:uuid,agent:redteam_agent,model:alias1,stateful:true,history_length:0, created_at:...,updated_at:...,metadata:{}}]}会话详情GET/api/v1/sessions/{id}返回摘要 完整历史OpenAI Responses 输入项列表含 user/system/assistant/tool 各类条目。删除会话DELETE/api/v1/sessions/{id}返回 204 No Content。重置会话POST/api/v1/sessions/{id}/reset重建会话 Agent 并清空历史返回 200SessionDetailModel。4.5 非流式推理POST /api/v1/sessions/{id}/messages运行 Agent 并返回最终结果。BodyInferenceRequest{input: List current risks, context: {org: acme}, max_turns: 8}响应 200InferenceResponse{ session: {id: uuid, ...}, result: { messages: [/* semantic items: messages, tool calls, outputs, ... */], history: [/* updated message list */], final_output: {/* typed final output if agent uses an output schema, else string */}, text_output: assistant final text, if any, input_guardrails: [], output_guardrails: [] } }其中input既可以是字符串也可以是ResponseInputItem[]context用于注入上下文变量max_turns限制最大轮次可对应 src/cai/cli.py 中CAI_MAX_TURNS的语义。4.6 推理步骤流POST /api/v1/sessions/{id}/messages/streamSSE实时推送高层推理步骤无 token 增量并在结束时推送最终摘要。底层机制是API 使用非流式模型调用通过服务端 RunHooks工具开始/结束、handoffs、Agent 切换把步骤实时广播出去并在每个 assistant 回合后追加一条完整文本的消息步骤——从而保证模型层流式永远关闭的同时客户端仍能获得实时的步骤更新。HeadersX-CAI-API-Key、Content-Type: application/json、Accept: text/event-stream。Body 与非流式相同。流格式为两种 SSE 事件event: reasoning_step—— 每个步骤一条data为 JSON见下event: final—— 最终事件携带{ steps, final_message, final_output }。推理步骤的 payload 类型无 token 增量// 助手生成的消息 {type:message,agent:Red Team,text:...full assistant message...} // 工具调用 {type:tool_call,agent:Red Team,tool:nmap_scan,arguments:{target:10.0.0.5}} // 工具输出 {type:tool_output,agent:Red Team,output:open ports: 22,80} // Agent 交接handoff {type:handoff,from_agent:Coordinator,to_agent:Exploiter} // 显式 Agent 切换信号 {type:agent_switched,agent:Exploiter}最终事件 payload{ steps: [ /* 流式期间发出的全部 reasoning steps */ ], final_message: ...last assistant message (if any)..., final_output: {/* 结构化输出若存在否则为 string/null */} }curl 示例SSEcurl -N \ -H Accept: text/event-stream \ -H Content-Type: application/json \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {input: List current risks} \ http://localhost:8080/api/v1/sessions/SESSION_ID/messages/stream4.7 Token 级流POST /api/v1/sessions/{id}/messages/stream_tokensSSE在推理步骤之外额外开启提供商级 token 流实时下发 token 增量。仅当你需要字符/token 级粒度时才使用。Headers 同上Body 同InferenceRequest。流事件event: tokendata 为{ type: token_delta, text: ... }每个文本增量一条event: tokendata 为{ type: message_start }与{ type: message_end }标记边界event: reasoning_step—— 高层步骤schema 与/messages/stream相同event: final—— 与/messages/stream相同的汇总。注意事项token 流消息非常密集客户端必须做好背压backpressure处理并使用流式友好的 APIiOS 端推荐使用 URLSession 流见下文示例Safari 的EventSource无法设置自定义请求头。curl 示例tokenscurl -N \ -H Accept: text/event-stream \ -H Content-Type: application/json \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {input: Write a haiku about ports} \ http://localhost:8080/api/v1/sessions/SESSION_ID/messages/stream_tokens4.8 中断与重载中断POST/api/v1/sessions/{id}/interrupt取消该会话当前正在服务端运行的 run 任务。返回 200{interrupted: true}重载POST/api/v1/sessions/{id}/reload重建会话的 Agent可选择保留消息历史。Body{preserve_history: true}返回 200SessionDetailModel。4.9 环境查询/agents 与 /modelsGET /api/v1/agents列出运行时可用 Agent 与模式源自cai.agents即 src/cai/agents 目录下注册的redteam_agent、blue_teamer、web_pentester等以及 src/cai/agents/patterns 中的 swarm 等模式。返回 200AgentsResponse{ agents: [ { name: redteam_agent, description: ..., type: agent, pattern_type: null, tools: [ {name: nmap_scan, description: Scan a host or subnet}, {name: http_get, description: Fetch a URL} ] }, { name: swarm_pattern, description: Swarm agentic pattern, type: pattern, pattern_type: swarm, tools: [] } ] }GET /api/v1/models合并预置模型目录与 pricing.json若存在返回 200ModelsResponse{ models: [ { name: alias1, provider: OpenAI, category: Alias, description: Best model for Cybersecurity AI tasks, input_cost: 0.50, output_cost: 0.50, pricing: { input_cost_per_token: 0.000005, output_cost_per_token: 0.000005, max_tokens: 128000, max_input_tokens: 200000, max_output_tokens: 128000, supports_function_calling: true, supports_vision: true, supports_response_schema: true, supports_tool_choice: true } } ] }4.10 UX 辅助端点POST /api/v1/sessions/{id}/ux/final_message/stream_tokensSSE任务完成后流式生成一段向用户解释刚刚发生了什么的收尾助手消息。应用在任务结束调用此端点发送提示词语气/指示可选传入客户端侧收集的 steps——若不传后端使用session.last_steps。BodyFinalMessageRequest{ prompt: Explain to the user what we found and next steps., steps: [ /* 可选客户端收集的步骤否则服务端使用 session.last_steps */ ], include_history: true, max_turns: 8 }流事件依次为event: token的message_start→ 若干token_delta→message_end若 UX Agent 产出步骤则穿插reasoning_step最后event: final{ steps: [...], final_message: ..., final_output: ... }。客户端把token_delta增量渲染进聊天气泡在message_end/final时收尾。POST /api/v1/ux/title通过alias1模型经 LiteLLM的一次强制 tool call 生成简洁标题不使用会话。Body{ messages: [ {role: user, content: Analiza CVE-2024-...} ], title_hint: (opcional) }返回 200{title: Analizando CVE-2024-...}。POST /api/v1/ux/summarize用alias1的一次强制 tool call 返回一句话摘要不使用会话。Body{ messages: [ {role: user, content: Escanea 10.0.0.5} ], steps: [ {type: tool_call, agent: Red Team, tool: nmap_scan, arguments: {target: 10.0.0.5}}, {type: tool_output, agent: Red Team} ], max_len: 100 }返回 200{summary_text: Tool output procesado por Red Team}。实现细节两个 UX 端点都强制tool_choice: required使用单一函数produce_title_and_summary且始终使用model: alias1 Alias 的api_base与ALIAS_API_KEY服务端不存储也不读取会话状态。五、请求/响应 Schema 字段参考公共结构HealthResponsestatus: string、version: stringCommandMetadataname如/memory、description、aliases如[/h]、subcommands如[show]CommandsResponsecommands: CommandMetadata[]CommandRequestargs: string[]可选、auto_correct: boolean默认trueCommandResponsehandled、suggested_command: string | null、stdout、stderr、exit_code: number | nullCreateSessionRequestagent可选默认CAI_AGENT_TYPE、model可选默认CAI_MODEL、stateful默认true、metadata可选SessionSummaryidUUID、agent、model、stateful、created_at/updated_atISO8601、history_length、metadataSessionDetailSessionSummary 全部字段 history: ResponseInputItem[]OpenAI Responses 输入项列表SessionsResponsesessions: SessionSummary[]InferenceRequestinput: string | ResponseInputItem[]、context: object可选、max_turns: number可选RunResultPayloadmessages: Item[]—— 运行期间生成的语义项列表history: ResponseInputItem[]—— 原始输入加生成项可直接用于续接对话final_output: any—— Agent 定义了输出 schema 时为结构化结果否则为文本或 nulltext_output: string | null—— 最后一条助手文本消息若有input_guardrails: object[]/output_guardrails: object[]—— 输入/输出的护栏判定输出。messages[] 条目的通用信封每个条目形如type: string—— 如message_output_item、tool_call_item、tool_call_output_item、handoff_output_itemagent: string | null—— 产生该条目的 Agent 名payload: object—— 底层输出/输入项的 Pydantic 模型 dumpoutput: any—— 仅tool_call_output_item携带结构化工具返回值。各类型要点message_output_itempayload为ResponseOutputMessageOpenAI Responses 消息含 content 数组text_output汇总最后一个文本块tool_call_itempayload为ResponseFunctionToolCall | ResponseComputerToolCall | ResponseFileSearchToolCall函数调用典型字段为name、argumentstool_call_output_itemoutput为解码后的工具结果handoff_output_itempayload为 handoff 输入项信封的agent与 payload 内容隐含源/目标 Agent 名。reasoning_step 流事件字段messageagent: string | null、text完整助手消息无 token 增量tool_callagent、tool工具名、arguments: object | stringtool_outputagent、output结构化工具输出handofffrom_agent: string | null、to_agent: string | nullagent_switchedagent新活跃 Agentfinal事件steps已发出的 reasoning_step payload 数组、final_message: string | null、final_output: any。六、错误与状态码状态码含义示例响应401 Unauthorized启用认证时密钥缺失/无效{detail:Invalid or missing API key}404 Not Found如未知会话 ID{detail:Session not found}422 Unprocessable Entity请求体格式错误标准 FastAPI 校验错误500 Internal Server ErrorAgent 执行意外失败{detail:Agent execution failed: ...}排查建议若出现意外 401先开启CAI_API_LOG_AUTHtrue查看认证决策日志若 500用CAI_API_LOG_LEVELdebug拉高日志级别定位 Agent 执行栈。七、构建客户端快速配方Pythonrequests iter_lines 处理 SSEimport json import os import requests BASE http://127.0.0.1:8080/api/v1 HEADERS {X-CAI-API-Key: os.environ.get(ALIAS_API_KEY, ), Content-Type: application/json} # 1) 创建会话 sess requests.post(f{BASE}/sessions, headersHEADERS, json{agent:redteam_agent,model:alias1,stateful:True}).json() sid sess[id] # 2) 非流式 res requests.post(f{BASE}/sessions/{sid}/messages, headersHEADERS, json{input:List current risks}).json() print(res[result][text_output]) # final message # 3) 流式SSE stream_headers HEADERS | {Accept: text/event-stream} with requests.post(f{BASE}/sessions/{sid}/messages/stream, headersstream_headers, json{input:List current risks}, streamTrue) as r: for line in r.iter_lines(decode_unicodeTrue): if not line: continue if line.startswith(event:): evt line.split(:, 1)[1].strip() elif line.startswith(data:): data json.loads(line.split(:, 1)[1].strip()) if evt reasoning_step: print(step:, data) elif evt final: print(final:, data)Node浏览器 EventSourceconst key process.env.ALIAS_API_KEY; const sid SESSION_ID; // create via POST /sessions const es new EventSource(http://localhost:8080/api/v1/sessions/${sid}/messages/stream, { withCredentials: false }); // Note: 浏览器中向 SSE 发送请求头需要代理或改用 fetch ReadableStream。 es.addEventListener(reasoning_step, ev console.log(step, JSON.parse(ev.data))); es.addEventListener(final, ev console.log(final, JSON.parse(ev.data)));Nodefetch ReadableStream可带认证头import fetch from node-fetch; const key process.env.ALIAS_API_KEY; const sid process.env.SID; const resp await fetch(http://localhost:8080/api/v1/sessions/${sid}/messages/stream, { method: POST, headers: { Content-Type:application/json, Accept:text/event-stream, X-CAI-API-Key: key }, body: JSON.stringify({ input: List current risks }) }); for await (const chunk of resp.body) { const s chunk.toString(); // 解析 SSE 行event: name / data: json process.stdout.write(s); }iOSSwiftURLSession 流式token 级let sid SESSION_ID var req URLRequest(url: URL(string: http://127.0.0.1:8080/api/v1/sessions/\(sid)/messages/stream_tokens)!) req.httpMethod POST req.addValue(text/event-stream, forHTTPHeaderField: Accept) req.addValue(application/json, forHTTPHeaderField: Content-Type) req.addValue(ProcessInfo.processInfo.environment[ALIAS_API_KEY] ?? , forHTTPHeaderField: X-CAI-API-Key) req.httpBody try! JSONSerialization.data(withJSONObject: [input: Hi], options: []) let task URLSession.shared.streamTask(with: req) task.resume() task.readData(ofMinLength: 1, maxLength: 8192, timeout: 0) { data, atEOF, error in if let data data, let s String(data: data, encoding: .utf8) { // 解析 SSE 行event: name / data: json print(s) } }最佳实践流式请求始终携带Accept: text/event-stream预期收到多个reasoning_step事件随后恰好一个final事件不下发 token 增量每条 message 步骤都包含完整的助手消息文本工具调用可能非常频繁客户端务必做好背压长任务运行期间保持宽松的连接超时。八、curl 快速复制粘贴# 健康检查 curl -s http://localhost:8080/api/v1/health # 列出 Agent curl -s -H X-CAI-API-Key: $ALIAS_API_KEY http://localhost:8080/api/v1/agents | jq . # 列出模型 curl -s -H X-CAI-API-Key: $ALIAS_API_KEY http://localhost:8080/api/v1/models | jq . # 列出命令 curl -s -H X-CAI-API-Key: $ALIAS_API_KEY http://localhost:8080/api/v1/commands # 执行命令 curl -s -X POST http://localhost:8080/api/v1/commands/memory \ -H Content-Type: application/json \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {args: [show]} # 创建会话 curl -s -X POST http://localhost:8080/api/v1/sessions \ -H Content-Type: application/json \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {agent: redteam_agent, model: alias1, stateful: true} # 中断与重载 curl -s -X POST -H X-CAI-API-Key: $ALIAS_API_KEY \ http://localhost:8080/api/v1/sessions/SESSION_ID/interrupt curl -s -X POST -H Content-Type: application/json -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {preserve_history: true} \ http://localhost:8080/api/v1/sessions/SESSION_ID/reload # 非流式推理 curl -s -X POST http://localhost:8080/api/v1/sessions/SESSION_ID/messages \ -H Content-Type: application/json \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {input: List current risks} # 流式推理步骤SSE curl -N -X POST http://localhost:8080/api/v1/sessions/SESSION_ID/messages/stream \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -H X-CAI-API-Key: $ALIAS_API_KEY \ -d {input: List current risks} # 重置与删除会话 curl -s -X POST -H X-CAI-API-Key: $ALIAS_API_KEY http://localhost:8080/api/v1/sessions/SESSION_ID/reset curl -s -X DELETE -H X-CAI-API-Key: $ALIAS_API_KEY http://localhost:8080/api/v1/sessions/SESSION_ID九、深入流式实现原理与底层机制对好奇的开发者理解/messages/stream的实现可以避免很多困惑API 流式永远不会开启 OpenAI chat completions 的 token 流。具体做法是用非流式模型调用运行 Agent通过 RunHooks工具 start/end、handoffs、Agent 切换发出实时事件每个 assistant 回合后追加一条完整文本的消息步骤无 token 增量。这保证了模型层流式始终关闭同时客户端仍能获得实时步骤更新——即步骤实时、文本整段的取舍token 级粒度仅由/messages/stream_tokens端点按需提供。这套机制与 CAI 的 SDK 运行器紧密关联Agent 的执行、工具调用、handoff 与事件流分别由 src/cai/sdk/agents/run.py、src/cai/sdk/agents/tool.py 与 src/cai/sdk/agents/stream_events.py 支撑/api/v1/agents返回的 Agent 清单来自 src/cai/agents 的注册表模型定价则由根目录 pricing.json 提供。若需要将 API 服务容器化部署可参考 dockerized/docker-compose.yaml 的环境变量注入方式CAI_API_HOST、CAI_API_PORT、ALIAS_API_KEY等。综上cai --api提供了一条从终端 REPL到多端 HTTP 服务的平滑路径同一套 Agent、命令与记忆体系既可通过/commands远程驱动 REPL又可通过/sessions管理有状态多轮对话还能按需选择步骤级或 token 级流式输出是构建 CAI 移动端、Web 端或自动化集成层的最佳起点。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考