一次消息的一生:HTTP 传输 + 客户端工具循环

30 秒导读: 你在前端调一次 sendMessage,Tambo 会把「你注册的组件和工具」连同消息一起 POST 给后端,后端用 SSE(Server-Sent Events,服务器单向持续推事件的长连接) 把 AG-UI 事件一条条吐回来。如果模型想调用一个只有浏览器/你的 app 才能执行的工具(查本地状态、点某个 按钮、调你自己的 API),后端不会自己跑,而是发一条 tambo.run.awaiting_input 把这次 run 暂停; 前端就地把工具跑了,把结果当作下一条消息、带上 previousRunId 再发一次 run,后端接着往下想。 这样「AI → 工具 → AI → 工具 → …」就被缝成一条连续对话。本章端到端串这一趟。

本章聚焦传输层与客户端工具循环这条主线。三件事刻意不在这里展开,请看兄弟章:

• 事件累加器 / reducer 内部怎么把碎片拼成消息 —— 见 第 3 章:流式协议。
• 后端「大脑」怎么挑组件、怎么流式吐 props —— 见 第 2 章:决策循环。
• 前端拿到流式 props 后怎么渲染成活组件 —— 见 第 5 章:前端渲染。
• 组件/工具是怎么变成「LLM 能调用的东西」的 —— 见 第 1 章:注册模型。

---

1. 这是什么(零基础也能懂)

一句话定义: 「一次消息的一生」= 从前端 client.run() 发出,到后端流式回一堆事件、 中途可能停下来让前端执行本地工具、再续跑,直到 RUN_FINISHED 的完整往返过程。

它要解决的核心矛盾: 模型跑在后端,但很多工具的「手脚」长在前端。

• 后端能直接跑的:MCP 服务器工具、后端自己的系统工具 —— 它就地调完,继续往下。
• 后端跑不了的:读浏览器 localStorage、弹一个确认框、调你 app 里带用户登录态的私有 API —— 这些只有前端有执行环境。

Tambo 的答案是暂停 + 续跑:后端把这类工具调用「挂起」,通过 SSE 告诉前端「我需要你去执行这些」, 前端执行完把结果回传,后端接着想。对使用者来说,这一切藏在一个 sendMessage 调用后面。

用起来什么样(最小示意):

// 示意,非源码:注册一个只有前端能跑的工具,然后发一条消息
client.registerTool({
  name: "getSelectedRow",
  description: "返回用户当前在表格里选中的那一行",
  tool: async () => window.__grid.getSelection(), // 只有浏览器有这个
});

// 一次 run:内部可能经历「模型想调 getSelectedRow → 前端执行 → 模型接着答」
const stream = client.run("把我选中的那行导出成 CSV");
for await (const { snapshot } of stream) {
  render(snapshot.messages); // 边流边渲染
}

一句话直觉: 把一次 run 想成一通没挂断的电话。后端一直在说(SSE 长连接),偶尔说 「你那边查一下 X」然后等你回话(awaiting_input);你查完报数字(tool result),它接着说。 电话从头到尾是同一通——靠 previousRunId 把前后两段接上。

---

2. 顶层全景(它大概怎么转)

怎么读这张图: 从上往下是时间。左边是前端(浏览器 / 你的 app),右边是后端(Tambo Cloud 或你自托管的同一套 NestJS API)。中间的双箭头是 HTTP。虚线框 = 一次「工具循环」,可以转 0 到多圈。

   前端 @tambo-ai/client                         后端 apps/api (NestJS)
   ─────────────────────                         ─────────────────────
   client.run(msg)
     │  把注册表转成
     │  AvailableComponents / Tools
     ▼
   createRunStream() ──POST /v1/threads[/:id]/runs──▶  V1Controller
                                                          │ startRun(抢并发锁)
                                                          │ executeRun(开 SSE)
     ◀═══════════ SSE: RUN_STARTED, TEXT_*, TOOL_CALL_* ══╡ advanceThread() 流式
     │  handleEventStream                                 │ 顺流吐 AG-UI 事件
     │  → reducer 累积(第3章)                            │
     │                                                    │ 冒出「客户端工具」调用?
     ◀═══════════ SSE: tambo.run.awaiting_input ══════════╡ 是 → 发暂停事件,收尾本段流
     │                                                    │
  ┌──┴─ 本地执行工具 executeAllPendingTools ──────────────────────────────┐ 工具循环
  │  │  把结果打包成 tool_result 内容                                       │(0..N 圈)
  │  ▼                                                                     │
  │ executeToolsAndContinue() ──POST /v1/threads/:id/runs(带 previousRunId)▶ 又一次 run │
  └──◀════════════ SSE: 接着 TEXT_* / 再一次 awaiting_input… ══════════════┘
     │
     ◀═══════════ SSE: RUN_FINISHED ══════════════════════
     ▼
   stream.thread 兑现最终 thread 快照

部件一句话职责:

部件	干什么	在哪个文件
client.run()	对外入口,建一个 TamboStream 立即返回,后台起处理循环	packages/client/src/tambo-client.ts:222 (run)
TamboStream.processLoop	端到端主循环:收流→分发→检测暂停→执行工具→续跑	packages/client/src/tambo-stream.ts:243 (processLoop)
createRunStream	把注册表转成 API 格式,选对 HTTP 方法发出去	packages/client/src/utils/send-message.ts:229 (createRunStream)
V1Controller	NestJS 路由,开 SSE 响应头、挂连接关闭钩子	apps/api/src/v1/v1.controller.ts:342 (createThreadWithRun)、:471 (createRun)
V1Service.executeRun	驱动 advanceThread,把内部事件顺流写成 SSE	apps/api/src/v1/v1.service.ts:672 (executeRun)
ClientToolCallTracker	后端侧盯流,判断有没有「挂起的客户端工具」	apps/api/src/v1/v1-client-tools.ts:54 (ClientToolCallTracker)
executeAllPendingTools	前端侧本地把工具跑了,产出 tool_result	packages/client/src/utils/tool-executor.ts:161
ToolCallTracker	前端侧累积工具参数、按 id 取回待执行的调用	packages/client/src/utils/tool-call-tracker.ts:53

主线走一遍(高层): 发消息 → 后端开 SSE、流式生成 → 遇客户端工具就发 awaiting_input 暂停 → 前端执行、回传结果、带 previousRunId 续跑 → 无更多工具则 RUN_FINISHED → stream.thread 兑现。

---

3. 前端发起:从注册表到一条 HTTP 请求

这节讲什么: client.run() 之后、HTTP 真正发出之前,前端做了哪些事。

3.1 把「注册的东西」翻译成「模型能看到的东西」

前端本地维护两张注册表:componentList(组件)和 toolRegistry(工具)。发请求前, 必须把它们转成 API 认识的 availableComponents / tools 一起带上——模型只有在这一次请求里 看到某工具的名字和 schema,才可能调用它。

真实代码里,createRunStream 先转换、再按「有没有 threadId」二选一发请求:

// send-message.ts:257-288(节选,已简化注释)
const availableComponents = toAvailableComponents(componentList);
const availableTools = toAvailableTools(toolRegistry);

if (threadId) {
  // 已有线程 → 追加一条 run
  const stream = await client.threads.runs.run(threadId, {
    message: messageWithContext, availableComponents, tools: availableTools,
    userKey, previousRunId, toolChoice,
  });
  return { stream, initialThreadId: threadId };
} else {
  // 新线程 → 建线程并起首个 run
  const stream = await client.threads.runs.create({ /* … */ });
  return { stream, initialThreadId: undefined };
}

• toAvailableComponents / toAvailableTools 见 packages/client/src/utils/registry-conversion.ts (toAvailableComponent 把已注册组件的 props JSON Schema 打包成 API 的 AvailableComponent)。
• 两个 SDK 方法正好对应后端两个端点(见 §4):runs.run(threadId,…) → POST /v1/threads/:id/runs; runs.create(…) → POST /v1/threads/runs。

3.2 消费 SSE:交给 handleEventStream,循环里逐事件分发

createRunStream 返回的 stream 是一个异步可迭代的 AG-UI 事件流。processLoop 用 for await 逐条取,交给 reducer 累积状态(第 3 章)、并顺手喂给工具追踪器:

// tambo-stream.ts:312-353(节选)
for await (const event of handleEventStream(currentStream, { debug })) {
  if (event.type === EventType.RUN_STARTED) {
    runId = event.runId;              // 记住 runId,续跑要用
    actualThreadId ??= event.threadId; // 新线程的真实 id 从首个事件里拿
  }
  toolTracker.handleEvent(event);      // 累积工具参数(见 §5)
  dispatch({ type: "EVENT", event, threadId: actualThreadId, /* … */ });
}

runId 和 actualThreadId 是缝合下一段流的关键:前者当 previousRunId,后者定位线程。

3.3 beforeRun 钩子与 ContextHelperFn:发车前塞「随手上下文」

TamboClient 留了两个 seam,让每次 run 自动带上动态上下文(当前页面、当前时间、选中的交互组件等), 不用调用方每次手写:

seam	用途	定义处
addContextHelper(name, fn)	注册一个返回上下文对象的函数,合进 additionalContext	tambo-client.ts:698 (addContextHelper)
ContextHelperFn	上述函数的类型(可同步或返回 Promise)	tambo-client.ts:37
beforeRun	每次 run 前的回调,用于异步收集上下文	tambo-client.ts:72(TamboClientOptions.beforeRun)

诚实说明:在 packages/client 这个提交里,mergeContextForRun 只做同步合并,并在注释里明说 「异步 helper 应在流处理循环里通过 beforeRun await」;但我在本包内未找到 beforeRun 的实际调用点 (仅有声明与注释,tambo-client.ts:783-786)(inferred)。真正把 helper 结果拍进 additionalContext 的实现落在 React SDK 的镜像逻辑里——它在发车前 await getAdditionalContext()、逐个塞进 additionalContext 再调 createRunStream(react-sdk/src/v1/hooks/use-tambo-v1-send-message.ts:555-577)。 所以对 React 用户,「当前页面/时间」这类上下文是自动跟着每条消息走的。

已知坑(源码 TODO):这份上下文快照只在 run 开始时抓一次,整个多轮工具循环复用它; 若流式期间交互组件变了,续跑会带旧上下文(use-tambo-v1-send-message.ts:551-554)。

---

4. HTTP / SSE 面:后端边界怎么接这条流

这节讲什么: 请求打到 NestJS 后,控制器和服务如何把一次 run 变成一条 SSE 长流。这里是 API 契约的所在,但不深入「大脑」怎么想(第 2 章)。

4.1 两个创建端点,一套 SSE 约定

方法 + 路径	何时用	控制器方法
POST /v1/threads/runs	新建线程并起首个 run	createThreadWithRun (v1.controller.ts:342)
POST /v1/threads/:threadId/runs	在已有线程上追加 run(含续跑)	createRun (v1.controller.ts:471)
POST /v1/threads/:threadId/components/:componentId/state	组件状态回写(见 §4.3)	updateComponentState (v1.controller.ts:641)

两个创建端点都返回 text/event-stream。控制器手动设 SSE 头并把线程/运行 id 放进响应头,便于前端早拿到:

// v1.controller.ts:366-371(createThreadWithRun 节选)
response.setHeader("Content-Type", "text/event-stream");
response.setHeader("Cache-Control", "no-cache");
response.setHeader("Connection", "keep-alive");
response.setHeader("X-Thread-Id", thread.id);
response.setHeader("X-Run-Id", startResult.runId);
response.flushHeaders();

每个事件按 SSE 规范写成一行 data: <json>\n\n,写前先 sanitizeEvent 脱敏:

// v1.service.ts:978-981(emitEvent)
private emitEvent(response: Response, event: BaseEvent): void {
  const sanitized = sanitizeEvent(event);
  response.write(`data: ${JSON.stringify(sanitized)}\n\n`);
}

连接关闭 = 取消 run。 控制器挂了 response.on("close"):若不是正常 finish,就尽力 cancelRun(…, "connection_closed")(v1.controller.ts:378-393)。这样用户关标签页,后端不会白烧 token。

4.2 executeRun:把内部流顺成 SSE

executeRun 是后端的驱动核心。它先发 RUN_STARTED,把 V1 的组件/工具转成内部格式, 起 advanceThread(第 2 章的「大脑」)往一个 AsyncQueue 推事件,自己则消费队列、逐条 emitEvent:

// v1.service.ts:730-760(节选):V1 契约 → 内部格式,再启动 advanceThread
const availableComponents = convertV1ComponentsToInternal(dto.availableComponents);
const clientTools       = convertV1ToolsToInternal(dto.tools);
const streamingPromise  = this.threadsService.advanceThread(
  { projectId, contextKey, sdkVersion },
  { messageToAppend, availableComponents, clientTools, forceToolChoice },
  threadId, toolCallCounts, undefined, queue, abortController.signal,
);

消费队列时它做两件与本章相关的事:把 LLM 生成的临时 messageId 映射成真实 DB id (transformEventMessageIds,v1.service.ts:1532),以及用 ClientToolCallTracker 盯流判断 有没有挂起的客户端工具(下一节)。

4.3 组件状态回写端点(一句话)

第 5 章讲的「可交互组件」会把用户改动回写到后端:前端 client.threads.state.updateState(componentId, …) (react-sdk/src/v1/hooks/use-tambo-v1-component-state.ts:146)命中 POST …/components/:id/state, 后端 updateComponentState 支持整体替换或 JSON Patch,且线程有活跃 run 时拒绝(返回 409, v1.service.ts:1025-1033)。渲染与状态细节属第 5 章,这里只标出这条 API 边界的存在。

---

5. 客户端工具循环(本章核心)

这节讲什么: 「暂停—执行—续跑」这套机制,前后端各自怎么实现,又怎么缝成一条对话。

5.1 后端:凭什么判断「该停下来等前端」

后端在流式过程中用 ClientToolCallTracker 追踪工具调用。它只认允许清单里的客户端工具名 (dto.tools 里那些),把「已 START 但还没收到 TOOL_CALL_RESULT」的调用视为挂起:

// v1-client-tools.ts:82-96(processEvent 节选)
case EventType.TOOL_CALL_START: {
  const e = event as unknown as ToolCallStartEvent;
  if (!this.clientToolNames.has(e.toolCallName)) break; // 不是客户端工具 → 忽略
  this.pendingClientToolCalls.set(e.toolCallId, { toolName: e.toolCallName, arguments: "" });
  // …初始化参数累积缓冲…
}

关键区分:MCP / 系统工具不在这份清单里,后端会就地执行并把结果继续往流里写,前端根本感知不到暂停; 只有客户端工具会走「挂起」这条路。流跑完后,若还有挂起的调用,就发一条 awaiting_input 再 RUN_FINISHED:

// v1.service.ts:856-866(节选)
const pendingToolCalls = toolCallTracker.getPendingToolCalls();
if (pendingToolCalls.length > 0) {
  const awaitingEvent = createAwaitingInputEvent(pendingToolCalls); // 带上 id/name/arguments
  this.emitEvent(response, awaitingEvent);
}

createAwaitingInputEvent(v1-client-tools.ts:196)把每个挂起调用的 { toolCallId, toolName, arguments } 塞进事件的 value.pendingToolCalls。同时后端把这些 pendingToolCallIds 记进线程状态(释放锁时,v1.service.ts:868-881),作为并发钥匙(见 §5.4)。

5.2 前端:收到 awaiting_input 就地执行

前端主循环在分发事件时盯住这个 custom 事件,一旦命中就跳出内层收流循环,进入工具执行:

// tambo-stream.ts:367-373(节选)
if (event.type === EventType.CUSTOM) {
  const customEvent = asTamboCustomEvent(event);
  if (customEvent?.name === "tambo.run.awaiting_input") {
    pendingAwaitingInput = customEvent;
    break; // 本段流到此为止,去执行工具
  }
}

执行由 executeToolsAndContinue 编排:按事件里的 id 从 ToolCallTracker 取回已累积好参数的调用, 本地顺序执行,再把结果打包成一条新消息发出去。注意它一并把 previousRunId: runId 带上——这就是缝合点:

// send-message.ts:319-345(executeToolsAndContinue 节选)
const pendingToolCallIds = event.value.pendingToolCalls.map((tc) => tc.toolCallId);
const toolCallsToExecute = toolTracker.getToolCallsById(pendingToolCallIds);
const toolResults = await executeAllPendingTools(toolCallsToExecute, toolRegistry);
toolTracker.clearToolCalls(pendingToolCallIds);          // 清掉已执行的
const stream = await client.threads.runs.run(threadId, { // 续跑:同一线程、接上一段
  message: { role: "user", content: toolResults, additionalContext },
  previousRunId: runId,
  availableComponents: toAvailableComponents(componentList),
  tools: toAvailableTools(toolRegistry),
  userKey, toolChoice,
});
return { stream, toolResults };

• 参数从哪来:ToolCallTracker 在流式期间累积 TOOL_CALL_ARGS 碎片,TOOL_CALL_END 时 JSON.parse 成完整入参(解析失败直接抛错、不静默兜底,tool-call-tracker.ts:96-108)。
• 谁真正执行:executeAllPendingTools 顺序(非并发,避免有副作用的工具相互依赖出错)遍历, 逐个 executeClientTool;工具抛错也不崩,而是回一条 isError: true 的 tool_result 让模型自己应对(tool-executor.ts:137-150)。找不到工具则回「not found」结果(:177-190)。

5.3 「扁平循环」:AI→工具→AI→工具… 缝成一条

processLoop 用一个 while (true) 外层把「一段流」和「一次工具执行」串起来。每段流结束, 若有 awaiting_input 且允许自动执行,就换上续跑流 currentStream = continuationStream,回到循环顶重跑:

// tambo-stream.ts:379-419(骨架)
if (!pendingAwaitingInput) break;      // 没有待执行工具 → 整个 run 结束
if (!autoExecuteTools) break;          // 关掉自动执行 → 停在这
stepCount++;
if (stepCount >= maxSteps) break;      // 兜底:默认 10 步防死循环(RunOptions.maxSteps)
const { stream: continuationStream, toolResults } = await executeToolsAndContinue({ … });
dispatchToolResults(dispatch, actualThreadId, toolResults); // 本地乐观回显工具结果
currentStream = continuationStream;    // 换流,继续转

这套扁平写法(而非递归)让多轮工具链不会把调用栈越叠越深,每一圈的状态(runId、 toolTracker)都在同一作用域里显式流转。教学版全貌:

// 示意,非源码:一次 run 的骨架
let stream = await openRunStream(msg);        // 第一段
while (true) {
  const pause = await drain(stream);          // 收流直到结束或 awaiting_input
  if (!pause) break;                          // 模型不再要工具 → 收工
  const results = await runToolsLocally(pause.pendingToolCalls);
  stream = await continueRun(results, lastRunId); // 带 previousRunId 续一段
}
// 重点看:previousRunId 把每一段接成同一通「电话」

流正常跑完(无暂停),循环 break,最终快照兑现 stream.thread(tambo-stream.ts:422-434)。

5.4 缝合的正确性:并发锁 + pendingToolCallIds

续跑本质是「在同一线程上再发一次 run」,必须防止并发和错配:

• 并发锁: startRun 用一次事务里的 acquireRunLock 原子地把线程从 IDLE 抢成 WAITING; 抢不到就返回 409 CONCURRENT_RUN(v1.service.ts:497-582)。前端侧也在 client.run() 处拦 同线程重入(tambo-client.ts:239-243)。
• 工具结果校验: 续跑请求里的 tool_result 必须恰好对上线程记着的 pendingToolCallIds。 startRun 先 dedupeToolResults 去重、validateToolResults 校验(缺/多都报 INVALID_TOOL_RESULT),再在事务里落库并用快照做并发钥匙清除挂起标记;若期间被别的请求改过, 抛 PendingToolCallStateMismatchError → 409 让客户端重试(v1.service.ts:446-559)。

一句话:previousRunId 缝对话顺序,pendingToolCallIds + 锁缝并发正确性。

---

6. MCP 工具:并入同一张工具表(一句话)

外部 MCP(Model Context Protocol)服务器的工具不是另一条通道——它们被降级成普通 TamboTool 混进同一张 toolRegistry,从此和你手写的工具走完全相同的路径。React SDK 在连上 MCP 后 listTools(),给每个工具 registerTool(...),其 tool 实现就是转手调 server.client.callTool (react-sdk/src/mcp/tambo-mcp-provider.tsx:305-320);底层连接与 listTools/callTool 由 packages/client/src/mcp/mcp-client.ts:223,277(MCPClient)负责。因此 MCP 工具照样被 toAvailableTools 一起发给模型;至于它在前端执行(经 awaiting_input)还是被后端就地执行, 取决于它是否进了后端那份客户端工具允许清单(§5.1)。

---

7. 同一个后端:Cloud 与自托管

值得点破的一点:Tambo Cloud 和你自托管的,是同一套后端。 apps/api 这个 NestJS 服务既是 托管版跑的,也是 docker-compose.yml 里 api 服务(端口 8261)构建运行的那个 (SELF-HOSTING.md 的服务表:Web 8260 / API 8261 / PostgreSQL)。前端只是把 TamboClient 的 baseURL 指向 https://api.tambo.co(默认,tambo-client.ts:761-768) 或你自己的地址,本章描述的 SSE 契约与工具循环一字不变。

---

8. 边界与局限(诚实)

• 上下文快照会过期。 helper 上下文只在 run 起点抓一次并贯穿整个工具循环;流式期间交互变化不会反映到 续跑里(源码 TODO,use-tambo-v1-send-message.ts:551-554)。
• awaiting_input 判定较「宽」。 后端把任何「已 START 未 RESULT」的允许清单工具都当挂起, 不区分「有意的客户端暂停」和「配置错误卡住的调用」(v1-client-tools.ts:47-53)。
• 步数硬上限。 工具循环默认 maxSteps = 10,到顶就带着未决工具收尾并打 warn (tambo-stream.ts:389-395);真需要更长的工具链要显式调大。
• beforeRun 在 framework-agnostic 层未落地。 本包只做同步上下文合并,异步 beforeRun 仅有声明(tambo-client.ts:72,783-786)(inferred);依赖它的能力目前落在 React SDK 侧。
• 流式工具执行是「尽力而为」。 标了 tamboStreamableHint 的工具会在参数还在流的时候被节流 预跑,报错只 console.warn 不影响主流程(tool-executor.ts:39-61)——最终以 awaiting_input 那次执行为准。

---

9. 代码地图(导航索引)

主题	文件路径	符号名
对外入口 / 并发拦截 / baseURL	packages/client/src/tambo-client.ts	TamboClient.run、resolveBaseUrl
上下文 seam	packages/client/src/tambo-client.ts	addContextHelper、ContextHelperFn、TamboClientOptions.beforeRun、mergeContextForRun
端到端主循环 / 扁平工具循环	packages/client/src/tambo-stream.ts	TamboStream.processLoop
注册表→API、选端点、续跑缝合	packages/client/src/utils/send-message.ts	createRunStream、executeToolsAndContinue、dispatchToolResults
注册表转换	packages/client/src/utils/registry-conversion.ts	toAvailableComponents、toAvailableTools
前端本地执行工具	packages/client/src/utils/tool-executor.ts	executeAllPendingTools、executeClientTool、executeStreamableToolCall
前端工具参数累积	packages/client/src/utils/tool-call-tracker.ts	ToolCallTracker、getToolCallsById、clearToolCalls
awaiting_input 事件类型 / 类型守卫	packages/client/src/types/event.ts	RunAwaitingInputEvent、asTamboCustomEvent
SSE 路由 / 头 / 连接关闭取消	apps/api/src/v1/v1.controller.ts	createThreadWithRun、createRun、updateComponentState
并发锁 / 工具结果校验 / 驱动 advanceThread / 发 awaiting_input	apps/api/src/v1/v1.service.ts	startRun、executeRun、emitEvent、transformEventMessageIds
后端侧工具追踪 / 暂停事件构造	apps/api/src/v1/v1-client-tools.ts	ClientToolCallTracker、createAwaitingInputEvent
MCP 连接 / listTools / callTool	packages/client/src/mcp/mcp-client.ts	MCPClient.create、listTools、callTool
MCP 工具并入注册表	react-sdk/src/mcp/tambo-mcp-provider.tsx	registerTool(调用 callTool 的 tool 实现)
组件状态回写(第5章细讲)	react-sdk/src/v1/hooks/use-tambo-v1-component-state.ts	client.threads.state.updateState
自托管服务拓扑	docker-compose.yml / SELF-HOSTING.md	api(8261)、web(8260)、postgres