跳转到内容

输入关键词开始搜索文档。

流式协议

最后更新 查看 Markdown 源文件

本页面记录了线路层面的字节级细节:即 createAgentRouter(agent) 为每个智能体会话提供的 HTTP 接口。路由挂载、认证和 CORS 属于应用层决策,请参阅路由指南;Flue Agent SDK 客户端封装了此协议,因此大多数应用无需手动调用。

该协议在 Node.js 和 Cloudflare 目标上完全一致——两者分发到相同的处理器,因此下面描述的每个请求格式、响应格式、头部字段和错误信息均适用于两者。

所有路由相对于路由器的挂载路径,加上调用方选择的会话 ID:

  • POST /:id — 投递一条消息(202 录入)。
  • GET /:id — 读取会话:view=history(默认)返回一个物化快照;view=updates 返回增量块,可选通过长轮询或 SSE 进行实时推送。
  • HEAD /:id — 以头部形式返回流元数据,无响应体。
  • POST /:id/abort — 中止进行中和排队的工作。
  • GET /:id/attachments/:attachmentId — 附件字节下载。

:id 必须是非空的路径段;空值或仅含空白的 ID 将以 invalid_request(400)拒绝。路由未列出的方法将以 method_not_allowed(405)拒绝,并附带 Allow 头。对从未接收过消息的会话的任何读取路由请求均返回 stream_not_found(404)——会话及其流由首次被接受的 POST 创建。

偏移量

偏移量是一个恢复令牌,用于定位会话持久记录流中的位置:

0000000000000000_0000000000000003
  • 格式为两个 16 位零填充整数通过 _ 连接(Durable Streams 偏移量格式)。在 Flue 中,第一个组件始终为 0。
  • -1 是「在第一个批次之前」的哨兵值:从 -1 读取将重放整个会话。
  • 偏移量寻址的是持久记录_批次_,而非消息。一次消息投递通常会产生多个批次,因此偏移量的增长速度快于消息的出现。如果一个批次中的所有记录都是内部投射且不产生可见块,那么 updates 响应可能为空,而 Stream-Next-Offset 仍在推进。
  • 将偏移量视为不透明值:从响应中获取(录入或快照的 offset 字段、Stream-Next-Offset 头部、SSE control 事件),并原样传回。不要自行构造或解析它们。
  • 读取是排他的:在偏移量 X 处的读取返回在 X _之后_记录的数据。
  • 超过当前流头部的偏移量会导致读取失败并返回 conversation_stream_store_failure(500)。格式错误的偏移量(除 -1 或双组件形式外的任何内容)将以 invalid_request(400)拒绝。

协调头部

三个响应头部用于跨请求协调读取:

  • Stream-Next-Offset — 恢复读取的偏移量。出现在 202 录入响应、快照响应、非 SSE 的 updates 响应和 HEAD 中。在 SSE 模式下,相同的值通过 control 事件传递。
  • Stream-Up-To-Date — 当响应到达产出时刻的持久头部时,值为 true。当更多数据已经可用时,该头部缺失(永远不是 false);继续从 Stream-Next-Offset 读取。在快照和 HEAD 响应中始终为 true。
  • Location — 仅在 202 录入响应中出现:会话的流 URL(与响应体中的 streamUrl 镜像),遵循 Durable Streams 流创建约定。

浏览器不会跨域暴露这些头部,除非你的 CORS 中间件在 Access-Control-Expose-Headers 中列出了它们;请参阅 CORS。

POST /:id — 消息录入

向会话投递一条消息。请求体为 JSON(Content-Type: application/json),与服务端 dispatch(...) 接受的 DeliveredMessage 格式相同,另有两个可选的保留顶层字段。@flue/sdk 导出了 DeliveredMessage 和 DeliveredAttachment。

type PromptBody = DeliveredMessage & {
  initialData?: unknown;
  uid?: string | null;
};

type DeliveredMessage =
  | { kind: 'user'; body: string; attachments?: DeliveredAttachment[] }
  | {
      kind: 'signal';
      type: string;
      body: string;
      attributes?: Record<string, string>;
      tagName?: string;
    };

type DeliveredAttachment =
  | { type: 'image'; data: string; mimeType: string; filename?: string } // data: base64
  | { type: 'document'; data: string; mimeType: 'application/pdf'; filename?: string };
  • kind: 'user' — 直接的用户聊天轮次。attachments 接受图片和文档;data 为 base64 编码,限制为 14,680,064 个字符(14 × 1024 × 1024)——超出长度的数据将以 invalid_request(400)拒绝。文档的 mimeType 必须为 application/pdf,否则请求将以 invalid_request(400)拒绝。
  • kind: 'signal' — 结构化事件投递。type 必须非空。body 是纯字符串;如需结构化载荷请自行 JSON 序列化。tagName 必须是合法的 XML 标签名(^[A-Za-z_][A-Za-z0-9_.-]*$);它在模型上下文中以未转义形式渲染为信号信封,因此宽松的值将以 invalid_request(400)拒绝。
  • initialData — 实例创建数据,仅在此次发送创建会话时使用。当智能体声明了 initialData 模式时,创建发送会根据该模式进行验证;不匹配的情况将在任何持久化之前以 invalid_request(400)拒绝。
  • uid — 发送条件。字符串值仅投递给具有该 uid 的实例:实例不存在或 uid 不匹配将以 agent_instance_not_found(404)拒绝。null 仅在不存在实例时创建:已有实例将以 agent_instance_exists(409)拒绝,其 details 字段会指出已有的 uid。省略该字段则无条件投递。将字符串 uid 与 initialData 组合使用是矛盾的(条件禁止创建),将以 invalid_request(400)拒绝。条件失败不会留下任何持久记录。

dispatch(...) 接受的纯字符串简写不在线路协议中:请求体必须是对象格式,否则将以 invalid_request(400)拒绝。

该路由还接受 W3C 追踪上下文请求头部 traceparent 和 tracestate;有效的 traceparent 会将接受的提交链接到调用方的分布式追踪中。

录入响应

录入是即发即忘的:一旦消息被持久接受,服务端立即返回 202 Accepted,此时智能体尚未运行。

{
  streamUrl: string; // 会话的读取 URL(当前 URL,去除查询参数)
  offset: string; // 录入时的持久头部
  submissionId: string; // 此次投递结算的身份标识
  uid: string; // 被联络实例的 uid
}
  • streamUrl — 从请求 URL 派生,去除了查询字符串。通过 Location 头部镜像。
  • offset — 录入后会话的持久头部,即消息本身已被记录后的状态。从该偏移量开始的 updates 读取将观察到智能体响应产生的所有内容,而无需重放历史或已录入的消息。通过 Stream-Next-Offset 头部镜像。
  • submissionId — 与 submission-settled 块和快照的 settlements 条目匹配,因此客户端可以等待此特定投递的结果。
  • uid — 被联络实例的 uid:当此次发送创建了实例时铸造,继续已有实例时回传。将其作为 uid 发送条件传回以联络同一个实例。

没有同步的「等待回复」模式:任何 ?wait 查询参数将以 invalid_request(400)拒绝。从会话流中读取结果,或使用 SDK 的 wait(...)。

当 Content-Type 错误时,非 JSON 请求体将以 unsupported_media_type(415)拒绝;当 JSON 无法解析时,以 invalid_json(400)拒绝。

GET /:id?view=history — 快照

返回会话的一个物化快照:每条消息都被还原为完整的、可渲染的部分。view=history 是默认值;普通的 GET /:id 等同于该请求。除 history 或 updates 外的任何 view 值将以 invalid_request(400)拒绝,将 view=history 与 offset、tail 或 live 组合使用也是如此。

响应:200,Content-Type: application/json,Cache-Control: no-store,Stream-Next-Offset 设为快照的 offset,Stream-Up-To-Date: true。

有界读取

三个可选的查询参数将读取限制在记录的一部分。游标是从先前响应的 before 字段获取的不透明令牌。不要构造或解析游标:游标指向某一代流中的一条消息。

  • limit=N — 最新的 N 条消息(正整数)。N 计算每条消息,包括隐藏和诊断性的。窗口可能包含超过 N 条:它总是向后延伸以包含仍能接收实时更新的每条消息,例如仍在流式传输的响应位于之后投递的消息之上。响应是完整快照(相同的 offset、incarnation 和每个结算)加上 before。
  • from=<cursor> — 从游标的消息(含)到头部的每条消息,作为完整快照加上 before。不能与 limit 组合使用。
  • before=<cursor> — 更早的页面:严格在游标之前的消息,最旧的在前,可选地用 limit=N 仅取最新的 N 条。响应是 { v: 1, conversationId, messages, before },没有 offset、incarnation 或 settlements,也没有 Stream-Next-Offset 头。

before 是下一页更早页面的游标(指向返回的最旧消息),当响应到达对话开头时为 null。无界读取省略它。将 before 与 from 组合、重复参数、非游标的值或非正整数的 limit 将以 invalid_request(400)拒绝。来自早期流代的游标(流在游标发出后被重置并重新增长)或其消息已消失的游标将以 history_cursor_not_found(410)拒绝,被拒绝的游标在 meta.cursor 中。即使新一代包含相同 ID 的消息也是如此。重新读取最新窗口。早于有界读取的运行时忽略这些参数并返回整个对话。

FlueConversationSnapshot

@flue/sdk 将快照类型导出为 FlueConversationSnapshot、FlueConversationMessage、FlueConversationPart 和 FlueConversationSettlement。

interface FlueConversationSnapshot {
  v: 1;
  conversationId: string;
  offset: string;
  messages: FlueConversationMessage[];
  settlements: FlueConversationSettlement[];
  before?: string | null; // bounded reads only
}

interface FlueConversationSettlement {
  submissionId: string;
  outcome: 'completed' | 'failed' | 'aborted';
  error?: unknown;
  answeredBySubmissionId?: string;
  timestamp?: string;
}

interface FlueConversationMessage {
  id: string;
  role: 'user' | 'assistant' | 'system';
  purpose: 'user' | 'assistant' | 'dispatch' | 'advisory';
  display: 'visible' | 'hidden' | 'diagnostic';
  submissionId?: string;
  turnId?: string;
  signal?: { tagName?: string; attributes?: Record<string, string> };
  settlement?: { outcome: 'failed' | 'aborted' };
  timestamp?: string;
  parts: FlueConversationPart[];
  metadata?: Record<string, unknown>;
}

type FlueConversationPart =
  | { type: 'text'; text: string; state: 'streaming' | 'done' }
  | { type: 'reasoning'; text: string; state: 'streaming' | 'done' }
  | { type: `data-${string}`; data: unknown }
  | { type: 'file'; mediaType: string; id?: string; size?: number; url?: string; filename?: string }
  | {
      type: 'dynamic-tool';
      toolName: string;
      toolCallId: string;
      state: 'input-available';
      input: unknown;
    }
  | {
      type: 'dynamic-tool';
      toolName: string;
      toolCallId: string;
      state: 'output-available';
      input: unknown;
      output: unknown;
      durationMs?: number;
    }
  | {
      type: 'dynamic-tool';
      toolName: string;
      toolCallId: string;
      state: 'output-error';
      input: unknown;
      errorText: string;
      durationMs?: number;
    };
  • v — 快照模式版本;当前为 1。
  • offset — 此快照被还原到的持久头部,包括不投射到任何可见消息的批次。从该偏移量恢复 updates 读取将精确地产生此快照之后的变更。
  • messages — 按顺序排列的会话记录。一条 assistant 消息代表一次完整的响应:一次提交的每个模型步骤都会折叠到该提交的第一条 assistant 消息中,各部分跨步骤累积。
  • settlements — 此会话上每个已结算提交的最终结果。error 携带供 failed/aborted 结果使用的调用方安全错误值。timestamp 是结算记录的捕获时间。
  • role/purpose/display — role 是粗粒度的渲染通道;purpose 对语义进行分类(dispatch = 投递的信号,advisory = 运行时建议);display 是可见性提示(visible 主聊天,diagnostic 活动面板材料,hidden 管道层)。
  • signal — 仅出现在从信号投递投射的 system 角色消息上;携带投递的 tagName 和 attributes。
  • settlement — 仅出现在运行时在提交以 failed 或 aborted 结算时追加的最终建议消息上;该消息的 submissionId 指明了已结算的提交。已完成的提交不会获得时间线标记(assistant 回复本身就是标记);settlements 仍是程序化的结果索引。
  • timestamp — 服务端捕获时间(ISO 8601),对应消息背后的持久记录:user/system 消息的用户或信号记录(输入应用到对话的时间,而非提交被接受的时间),assistant 响应的首个步骤开始时间(后续步骤保持不变)。每个角色都有此字段,包括在该字段投射之前记录的对话。这是服务端时钟时间——在单个同步跨度内写入的记录(特别是在 Cloudflare Workers 上)可能共享时间戳,且时钟跨主机不单调——因此按数组位置排序,而非按时间戳。
  • metadata — 完全由智能体编写(response-metadata 钩子)。运行时不填充任何内容;usage 或 model 等键是应用层约定。服务端捕获时间在 timestamp 上。
  • parts — text/reasoning 在实时响应进行中时携带 state: 'streaming',完成后为 'done'。data-<name> 部分是命名的客户端数据写入,每次写入对应一个部分,按发出顺序排列。file 部分通过 id 引用附件;url 永远不会由服务端设置(运行时不知道公共挂载路径——SDK 在客户端解析它,而 GET /:id/attachments/:attachmentId 是底层路由)。dynamic-tool 部分经历 input-available → output-available/output-error 的状态变化;durationMs 是工具处理器的执行时间,在该字段出现之前记录的结果中缺失。

快照仅覆盖每个智能体实例的一个会话:默认的根会话。子会话(子智能体任务和其他内部会话)永远不会通过此接口暴露。规范的持久记录模式同样不会暴露——快照和更新块是线路上唯一的读取格式。

GET /:id?view=updates — 更新

返回偏移量之后的会话变更,以 JSON 数组形式的更新块呈现。

查询参数:

  • offset — 必需,恰好一次:-1 或先前返回的偏移量。缺失、重复或格式错误的值将以 invalid_request(400)拒绝。
  • live — 可选:long-poll 或 sse。任何其他值将以 invalid_request(400)拒绝。省略 = 立即返回可用内容。
  • tail — 此接口不支持;以 invalid_request(400)拒绝。流后缀可以省略消息开始、压缩边界和较早的增量,因此无法安全投射。
  • from、limit — 用于 conversation-reset 快照的可选窗口,由有界观察发送。有了它们,每个重置快照在服务端裁剪到窗口,带有 before 游标,而不是携带整个记录。当 from 游标在当前流代中仍可解析时,从该游标的消息开始;否则持有最新的 limit 条消息。与有界历史读取相同的规则适用,以及相同的 invalid_request 验证。没有它们时,重置携带整个记录。before 以 invalid_request(400)拒绝。

不使用 live 时,响应为 200,Content-Type: application/json,Cache-Control: no-store,附带 Stream-Next-Offset 和(当读取到达头部时)Stream-Up-To-Date: true。响应体是块数组——当 offset 之后没有记录时为空。

单个响应最多覆盖 100 个持久批次(固定的服务器页面大小;线路上没有参数可以更改它)。当 Stream-Up-To-Date 缺失时,表示更多数据已经可用:从返回的 Stream-Next-Offset 发起下一次读取。

块是相对于 offset 处会话状态的差异。仅从你持有其状态的偏移量恢复——快照的 offset,或 -1(全新读取以携带完整快照的 conversation-reset 开始)。当本地状态丢失时,请求新的快照,而不是猜测。

为 updates 读取提供服务时,需要先通过 offset 重建会话的还原状态,然后进行投射——没有持久化的重放缓存——因此每次读取或重连的初始化成本随会话持久流的总长度增长。拥有非常大对话的应用应测量重连延迟,避免不必要的重连循环。

live=long-poll

当 offset 处有可用数据时,与普通的 updates 读取完全相同——响应立即返回。当没有可用数据时,服务器保持请求直到新数据到达或 30 秒窗口期满,以先到者为准:

  • 新数据到达 → 返回 200,附带块数组,如上所述。
  • 超时 → 返回 200,附带空数组 [],Stream-Next-Offset 不变,Stream-Up-To-Date: true。重新发起请求以继续等待。
  • 客户端在等待期间断开连接 → 响应被丢弃,状态码 499,无响应体。

live=sse

保持连接无限期打开,并以服务器推送事件的形式推送块。响应:200,Content-Type: text/event-stream,Cache-Control: no-cache。

event: data
data:[{ "type": "message-delta", ... }, ...]

event: control
data:{"streamNextOffset":"0000000000000000_0000000000000007","upToDate":true}

: heartbeat
  • data 事件 — JSON 块数组,每次读取周期一个事件。仅在该周期产生块时才会发出。
  • control 事件 — 响应体中的流协调信息(因为头部无法在流中途更新):streamNextOffset(字符串)和 upToDate(仅在追上进度时以 true 出现)。在每次读取周期后发出,包括空周期,因此已追上进度的流仍会至少每 30 秒产生一个 control 事件。断开连接后从最后的 streamNextOffset 重连。
  • : heartbeat 注释行 — 每 15 秒一次,防止中间代理超时空闲连接。

流在服务端永远不会结束;它会一直运行直到客户端断开。SSE 在重连间是至少一次的传输——通过 position(见下文)对块进行去重。

ConversationStreamChunk

每个块携带一个 type、conversationId 和一个 position。@flue/sdk 将联合类型导出为 ConversationStreamChunk(用于第一方渲染器;应用代码应使用 SDK 物化的 observe() 状态)。

type ConversationStreamChunk = ChunkBody & { position: { batch: number; index: number } };

type ChunkBody =
  | { type: 'conversation-reset'; conversationId: string; snapshot: FlueConversationSnapshot }
  | { type: 'message-appended'; conversationId: string; message: FlueConversationMessage }
  | {
      type: 'message-started';
      conversationId: string;
      messageId: string;
      submissionId?: string;
      turnId?: string;
      metadata?: Record<string, unknown>;
      timestamp?: string;
    }
  | {
      type: 'message-metadata';
      conversationId: string;
      messageId: string;
      metadata: Record<string, unknown>;
    }
  | { type: 'data-part'; conversationId: string; messageId: string; name: string; data: unknown }
  | {
      type: 'message-delta';
      conversationId: string;
      messageId: string;
      kind: 'text' | 'reasoning';
      delta: string;
    }
  | {
      type: 'tool-input';
      conversationId: string;
      messageId: string;
      toolCallId: string;
      toolName: string;
      input: unknown;
      timestamp?: string;
    }
  | {
      type: 'tool-output';
      conversationId: string;
      toolCallId: string;
      output: unknown;
      durationMs?: number;
      timestamp?: string;
    }
  | {
      type: 'tool-output-error';
      conversationId: string;
      toolCallId: string;
      errorText: string;
      durationMs?: number;
      timestamp?: string;
    }
  | { type: 'message-completed'; conversationId: string; messageId: string; timestamp?: string }
  | {
      type: 'submission-settled';
      conversationId: string;
      submissionId: string;
      outcome: 'completed' | 'failed' | 'aborted';
      error?: unknown;
      timestamp?: string;
    };
  • position — a monotonic ordering token: batch is the durable batch ordinal the chunk was projected from, index its position within that batch’s projection. { batch, index } is globally unique and ordered across the conversation; compare lexicographically (batch, then index) to dedupe redelivered chunks. Otherwise opaque — do not interpret the numbers.
  • conversation-reset — replace all accumulated state with the embedded snapshot. Emitted when a batch contains a structural boundary (conversation creation, compaction); the reset subsumes every other chunk of its batch, so a fresh read from offset=-1 begins with one. The embedded snapshot may already contain settlements — check snapshot.settlements as well as submission-settled chunks when awaiting an outcome.
  • message-appended — a complete message (user turn or system signal), in the same message format as the snapshot, including its timestamp (the same value a later snapshot projects).
  • message-started — an assistant response opened. metadata carries agent-authored response metadata available at start. Assistant chunks are pre-coalesced: every model step of a submission addresses the submission’s first assistant messageId, so accumulating parts per messageId reproduces the snapshot’s one-message-per-response shape. A later message-started for an already-open messageId is a continuation, not a new message.
  • message-metadata — agent-authored metadata for an open response; merge onto the message.
  • data-part — one named client data write; append a data-<name> part.
  • message-delta — streamed text or reasoning content; append to the open part of that kind, opening one if none is open. A kind change or message-completed closes the open part.
  • tool-input / tool-output / tool-output-error — tool-call lifecycle, correlated by toolCallId. Input arrives on the assistant message; outputs update the matching dynamic-tool part.
  • message-completed — the assistant response closed; mark streaming parts done.
  • submission-settled — the terminal outcome of one submission, matching the admission response’s submissionId.
  • timestamp — capture time (ISO 8601) of the underlying durable record, present on boundary chunks (message-started, tool-input, tool-output, tool-output-error, message-completed, submission-settled). message-appended carries it on the embedded message instead. message-delta deliberately omits it for wire weight; interpolate between stamped boundaries. The SDK copies message-started’s timestamp onto the assistant message it opens (a continuation keeps the first step’s) and submission-settled’s onto the settlement.

HEAD /:id

以头部形式返回流元数据,无响应体:Content-Type: application/json,Cache-Control: no-store,Stream-Next-Offset(当前持久头部),Stream-Up-To-Date: true。流不存在时返回 404,附带错误头部和无响应体。

POST /:id/abort

中止会话的所有持久工作:正在运行的提交及其后面排队的所有内容。无需请求体。响应:200,JSON 响应体:

{
  aborted: boolean;
}
  • aborted — 当存在进行中或排队的工作且现在正在被中止时为 true;当会话处于空闲状态时为 false。中止会记录一个持久意图并立即返回——受影响的提交会异步结算为 aborted 结果。通过 submission-settled 块或快照的 settlements 观察结算状态。

非 POST 方法将以 method_not_allowed(405)拒绝,Allow: POST。

GET /:id/attachments/:attachmentId

提供单个附件的字节内容。:attachmentId 是 file 部分的 id;需要进行 URI 编码。查找范围限定在会话的默认根会话中——属于子会话的附件永远不会被提供服务,并返回 attachment_not_found(404),未知 ID 同理。尚不存在的会话返回 stream_not_found(404)。

响应:200,附带原始字节和:

  • Content-Type — 附件存储的 MIME 类型。
  • Content-Length — 字节大小。
  • Content-Disposition: inline。
  • Cache-Control: private, max-age=31536000, immutable — 附件内容以摘要为键且不可变,因此客户端可以无限期缓存。
  • Content-Security-Policy: sandbox — MIME 类型由上传方控制,因此直接导航被中和为不透明源;<img>/<a> 子资源加载不受影响。

非 GET 方法将以 method_not_allowed(405)拒绝,Allow: GET。

错误响应

此接口上的每个错误都以规范的 Flue 信封格式渲染,Content-Type: application/json:

{
  error: {
    type: string;      // 稳定的机器可读分类
    message: string;
    details: string;
    dev?: string;      // 仅限本地开发环境
    meta?: Record<string, unknown>;
  };
}

根据 type 分支;消息文本不是 API。类型代码、状态和字段语义在错误参考中有文档。此接口使用的状态:invalid_request 和 invalid_json(400)、agent_instance_not_found 和 stream_not_found 和 attachment_not_found(404)、method_not_allowed(405)、agent_instance_exists(409)、history_cursor_not_found(410,有界历史读取)、unsupported_media_type(415)、runtime_unavailable(503,本地开发重载,带 Retry-After),以及 internal_error 或 conversation_stream_store_failure(500)。未知的服务器失败永远不会泄露其原始消息——它们渲染为通用的 internal_error。

固定响应头部

每个读取和错误响应都携带两个浏览器安全头部:X-Content-Type-Options: nosniff 和 Cross-Origin-Resource-Policy: cross-origin。

协议按设计不设置其他横切头部:

  • 没有 Access-Control-* 头部——CORS 是应用层中间件;请参阅 CORS,包括应暴露哪些协调头部。
  • 没有认证挑战——保护挂载点是应用层中间件;请参阅保护你的智能体。
  • 会话读取不设置缓存验证器(ETag、Last-Modified)——偏移量是恢复机制,会话响应为 no-store。