Signals 信号

添加于： @mastra/core@1.39.0

Signals 让你通过 thread 与 agent 交互。无需以 agent.stream() 开始每次交互，你可以订阅一个 thread 并发送消息或 signals。Mastra 会唤醒空闲的 agent，或将输入投递到其运行中的循环；也可以将输入排队到下一轮。

用户撰写的输入请使用消息 API。sendSignal() 用于更低层的系统上下文，例如后台任务通知、策略提醒或处理器生成的上下文。

📹 观看

观看 Mastra signals 概览，了解 signals 如何唤醒并引导长时间运行的 agents。

何时使用 signals

当 agent thread 需要在原始 stream() 调用之外获得新输入或上下文时，使用 signals。signals 适用于以下场景：用户在运行进行中发送后续消息；后台系统需要向 thread 添加上下文；外部事件应唤醒、更新或通知 agent。

用户撰写的输入使用 sendMessage() 和 queueMessage()。更低层的系统上下文使用 sendSignal()。持久化的状态通道使用 sendStateSignal()；当外部事件应创建持久化的通知收件箱记录时，使用 sendNotificationSignal()。

快速开始

创建一个 agent，订阅一个 thread，然后向该 thread 发送消息。当消息唤醒 agent 或进入运行中的循环时，订阅会接收活动流。

src/mastra/signals.ts

import { Agent } from '@mastra/core/agent'

const agent = new Agent({
  id: 'support-agent',
  name: 'Support Agent',
  instructions: 'Help the user compare options.',
  model: 'openai/gpt-5.6-sol',
})

const thread = {
  resourceId: 'user_123',
  threadId: 'thread_456',
}

const subscription = await agent.subscribeToThread(thread)

await agent.sendMessage('Compare that with the previous option.', thread)

for await (const chunk of subscription.stream) {
  console.log(chunk)
}

当 thread 中存在运行中的 agent 流时，sendMessage() 会成为该 agent 循环内的新输入。当 thread 空闲时，Mastra 会以该消息作为首个输入启动一个流。

推理期间的输入

在默认循环以及实验性的 durable 和 evented agents 中，已接受的待处理消息和 signals 会中断一个仅产生了推理或元数据的模型请求。这包括等待首个 token 的阶段。Mastra 会取消该请求，从模型历史和记忆中丢弃其响应，并在同一次运行的下一请求中包含排队输入。此前已完成的响应仍保留在历史中。

一旦文本、工具调用或其他响应内容开始产生，signals 会等待下一个循环边界。正在运行的工具不会被取消或重复执行。如果运行达到 maxSteps，排队输入可能被保存但没有响应。queueMessage() 仍会等待活动运行结束后再处理。处理器直接写入消息历史中的 signals 不会中断它们自己的请求。

被中断的请求不计入 maxSteps，也不会推进 stepNumber。下一请求使用相同的逻辑步骤，因此 maxSteps: 1 仍允许一个替代响应。Mastra 不会设置单独的中断上限。重复的 signals 可以持续重启推理，直到某个请求完成、运行被取消或其总超时到期。

已流式传输的推理内容可能仍然可见，直到客户端重新加载历史。当输入到达时，如果每步输入处理器（processInputStep）正在运行，处理器会完成执行，Mastra 随后取消它们所准备的模型请求。自定义输出处理器在取消后也会继续运行，下一请求会等待它们进行中的工作结束。Mastra 会丢弃框架拥有的响应缓冲区，但不会撤销处理器状态、处理器在被取消请求收尾期间写入的数据，或外部副作用。用量总计不包含被中断的请求，即使提供商报告了它们的用量。提供商仍可能对这些请求计费。

推理中断不适用于 Inngest 引擎，也不适用于在与接受输入不同的进程中运行的 durable 和 evented 步骤。

消息输入

立即发送消息

当用户期望活动 agent 立即看到消息时，使用 sendMessage()。

src/mastra/signals.ts

agent.sendMessage(
  {
    contents: 'Use the latest customer note too.',
    attributes: { name: 'Jane', sentFrom: 'slack' },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会将带属性的消息作为 XML 包裹的用户输入接收：

<user name="Jane" sentFrom="slack">Use the latest customer note too.</user>

不带属性的消息会作为纯用户输入发送。

将消息排队到下一轮

当用户发送后续消息，但当前模型调用应先完成时，使用 queueMessage()。Mastra 会等待活动运行完成，然后在同一 thread 上启动新的运行。

src/mastra/signals.ts

agent.queueMessage('Also check whether the tests need updates.', {
  resourceId: 'user_123',
  threadId: 'thread_456',
})

当 thread 空闲时，queueMessage() 会立即启动一次运行。当 thread 处于活动状态时，它会在活动运行完成后启动新运行，以保持轮次顺序。

将带类型的 signal 排队到下一轮

当系统生成的输入需要与 queueMessage() 相同的轮次顺序，但应保留自己的 signal 类型、标签和属性时，使用 queueSignal()。例如，skill 调用可以渲染为 <skill> 元素，而不是普通用户消息：

src/mastra/signals.ts

agent.queueSignal(
  {
    type: 'reactive',
    tagName: 'skill',
    attributes: { name: 'triage' },
    contents: 'Triage the new issue.',
  },
  { resourceId: 'user_123', threadId: 'thread_456' },
)

模型会收到 <skill name="triage">Triage the new issue.</skill>。排队的 signals 与 queueMessage() 共享同一队列，因此会计入队列数量事件，并且可以用 cancelQueuedMessages() 取消。

当 signal 启动一次运行时，该运行的 AGENT_RUN span 会在 attributes.signal 下记录 signal 的 id、type、tagName 和 attributes。

取消待处理输入

使用 cancelQueuedMessages() 移除选定的待处理输入，而不停止活动运行。取消范围限定在 memory thread 内，包括共享同一 runtime 和 PubSub 实例的其他 Agents 提交的输入。

thread 必须仍有待处理消息可供取消。此示例假设在你排队消息时运行处于活动状态。空闲 thread 会立即启动该消息。

const queued = agent.queueMessage('Also check the tests.', thread)
await queued.accepted

const { cancelledSignalIds } = agent.cancelQueuedMessages({
  ...thread,
  signalIds: [queued.signal.id],
})
console.log(cancelledSignalIds)

只有本次调用在本地移除的 ID 会出现在 cancelledSignalIds 中。Mastra 仍会通过 PubSub 发布每个请求的 ID，即使本地没有任何待处理项，这样订阅该 thread 的其他进程可以取消它们匹配的待处理输入。由于该传播是异步的，结果并不确认远程取消，已经交给执行的输入也无法再触及。取消不会删除已保存的消息、通知记录或状态更新，也绝不会撤销先前的接受确认。

停止运行并清除待处理输入

默认情况下，中止 thread 会保留其排队输入。要在停止活动运行前清除待处理的 signals，请传入 clearPendingSignals: true：

subscription.abort({ clearPendingSignals: true })

没有订阅时，使用 agent.abortThreadStream({ ...thread, clearPendingSignals: true })。取消后现有监听器仍保持订阅，后续消息仍可启动新的运行。

使用共享 PubSub 后端时，runtime 接收远程 signals、取消和中止请求的过程独立于 subscribeToThread() 观察者。Clear-on-abort 会通过 PubSub 向远程活动所有者发送清除标志，但不会清除每个进程的本地队列。这两种操作都不会取消由 continueWithMessages() 创建的延续。

Signal 上下文

控制低层 signal 行为

当需要发送系统生成的上下文而非用户撰写的输入时，使用 sendSignal()。对于外部事件，使用 type: 'notification'。默认情况下，Mastra 会将 signals 投递到活动运行并唤醒空闲 threads。使用 ifActive.behavior 和 ifIdle.behavior 更改该行为。

src/mastra/signals.ts

const result = agent.sendSignal(
  {
    type: 'notification',
    contents: 'GitHub CI failed on PR #123: 3 tests failed.',
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
    ifIdle: {
      behavior: 'persist',
    },
  },
)

await result.persisted

当空闲唤醒流需要模型设置、工具或 runtime context 等选项时，传入 ifIdle.streamOptions。关于 ifActive、ifIdle、分支属性和 streamOptions，请访问 Agent.sendSignal() 参考。

发送通知上下文

Signals 具有语义化的 type 和面向 LLM 的 tagName。使用 type 描述 signal 类别。使用 tagName 控制模型看到的 XML 标签。

对于外部事件，使用 type: 'notification'。响应式 signals 保留给处理器或 runtime 生成的上下文，例如策略指引、后台任务结果和自动加载的指令。

src/mastra/signals.ts

agent.sendSignal(
  {
    type: 'notification',
    contents: 'PR #123 has a new review comment from User X about the API surface.',
    attributes: {
      source: 'github',
      pr: '123',
    },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会像这样将 signal 作为上下文接收：

<notification source="github" pr="123">PR #123 has a new review comment from User X about the API surface.</notification>

请使用 XML 安全的 tagName 和属性名。它们可以包含字母、数字、句点、连字符和下划线字符（_）。它们必须以字母或下划线开头。

存储支持

通知收件箱存储可在支持更丰富记忆和 signal 工作流的存储适配器中使用：libSQL、PostgreSQL 和 MongoDB。这些适配器通过 getStore('notifications') 暴露通知记录。

发送处理器上下文

处理器可以在运行期间发送响应式 signals。处理器应检查聊天历史，并针对特定触发条件做出反应，同时避免多次发送相同上下文。

以下示例展示了一个处理器：在工具调用读取 AGENTS.md 文件后，注入 AGENTS.md 指令。

src/mastra/processors/agents-md-reminder.ts

import type { Processor, ProcessInputStepArgs } from '@mastra/core/processors'

export const agentsMdReminderProcessor: Processor = {
  id: 'agents-md-reminder',
  async processInputStep({ messageList, sendSignal }: ProcessInputStepArgs) {
    const messages = messageList.get.all.db()
    const agentsMdPath = findAgentsMdPathFromToolCalls(messages)

    if (!agentsMdPath || hasAlreadySentAgentsMdReminder(messages, agentsMdPath)) {
      return messageList
    }

    await sendSignal?.({
      type: 'reactive',
      contents: readAgentsMdInstructions(agentsMdPath),
      attributes: {
        type: 'dynamic-agents-md',
        path: agentsMdPath,
      },
      metadata: {
        path: agentsMdPath,
      },
    })

    return messageList
  },
}

响应式 signals 默认使用 tagName: 'system-reminder'。模型会像这样接收该上下文

<system-reminder type="dynamic-agents-md" path="packages/ui/AGENTS.md">
  $agentsMdFileContents
</system-reminder>

当订阅的 thread 处于活动状态时，await sendSignal() 可保持流回显顺序。

条件属性

使用 ifActive.attributes 和 ifIdle.attributes 为输入附加取决于投递时 agent 处于活动还是空闲状态的上下文。顶层 attributes 始终生效，Mastra 在接受输入时会将所选分支的 attributes 合并到其中。关于分支专用属性，请访问 Agent.sendMessage() 参考和 Agent.sendSignal() 参考。

状态与通知 signals

状态信号

状态信号暴露具名的、thread 作用域的上下文通道。将它们用于随时间变化的持久上下文，例如浏览器状态、编辑器状态或后台监视器的结果。

当外部生产者检测到状态变化时，使用 sendStateSignal()。每个状态信号都会标识一个状态通道、一个生产者拥有的缓存键，以及该更新是快照还是增量。

src/mastra/browser-watcher.ts

await agent.sendStateSignal(
  {
    id: 'browser',
    mode: 'snapshot',
    cacheKey: 'browser:https://example.com:3-tabs',
    contents: 'Browser is open. Active tab URL: https://example.com. 3 open tabs.',
    value: {
      activeUrl: 'https://example.com',
      tabCount: 3,
      open: true,
    },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

当 Mastra 接受状态信号时，它会在 thread 上存储紧凑的跟踪元数据。如果生产者在状态仍然有效时再次发送相同的 cacheKey 和模式，Mastra 会跳过该重复项。

当处理器拥有某个状态通道时，使用 computeStateSignal()。Mastra 会在 processInputStep() 之后，每个模型输入步骤调用它一次。关于状态信号字段和返回值，请访问 Agent.sendStateSignal() 参考。

src/mastra/processors/browser-state.ts

import type { ComputeStateSignalArgs, Processor } from '@mastra/core/processors'

export const browserStateProcessor: Processor = {
  id: 'browser-state',
  stateId: 'browser',
  computeStateSignal(args: ComputeStateSignalArgs) {
    const browser = readCurrentBrowserState()
    const previous = readMostRecentBrowserState(args.activeStateSignals)
    const changed = previous ? diffBrowserState(previous, browser) : browser
    const shouldRefreshSnapshot = Boolean(args.lastSnapshot && !args.contextWindow.hasSnapshot)

    if (previous && Object.keys(changed).length === 0 && !shouldRefreshSnapshot) {
      return
    }

    const isDelta = Boolean(previous && !shouldRefreshSnapshot)

    return {
      mode: isDelta ? 'delta' : 'snapshot',
      cacheKey: stableBrowserStateCacheKey(browser),
      contents: isDelta ? describeBrowserDelta(changed) : describeBrowserSnapshot(browser),
      value: browser,
      ...(isDelta ? { delta: changed } : {}),
    }
  },
}

Mastra 会将 lastSnapshot 和 deltasSinceSnapshot 传入 computeStateSignal()。当当前消息列表不包含最新快照时，它会从消息历史中解析这些值。处理器仍负责合并和差异逻辑。

contextWindow.hasSnapshot 告知处理器当前活动消息窗口是否已包含该状态通道的快照。如果为 false，请返回新的 snapshot，这样即使较旧的状态消息被从上下文窗口中裁剪，模型也能看到当前状态。

内置的浏览器上下文处理器会以 browser id 发出状态，支持快照和增量模式。

通知信号

通知信号表示外部事件，例如 GitHub 活动、电子邮件、Slack 提及、CI 状态、事故、录音或直接消息。当事件应创建持久化收件箱记录时，使用 agent.sendNotificationSignal()。

通知投递分为两个阶段。在入口阶段，agent.sendNotificationSignal() 存储通知记录并解析 agent 的投递策略。在分发阶段，Mastra 消费到期记录并发出完整通知或摘要 signals。

默认投递策略会感知优先级。紧急通知立即投递，而较低优先级的通知可能被批处理为摘要，或等待 thread 空闲。关于通知字段，请访问 Agent.sendNotificationSignal() 参考；关于 notifications.deliveryPolicy 配置，请访问 Agent 构造函数参考；关于收件箱工具操作，请访问 createNotificationInboxTool() 参考。

src/mastra/notifications.ts

await agent.sendNotificationSignal(
  {
    source: 'github',
    kind: 'ci-status',
    priority: 'high',
    summary: 'CI failed on main: 3 tests failed.',
    payload: {
      repository: 'acme/app',
      branch: 'main',
    },
    dedupeKey: 'github:acme/app:main:ci',
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会将完整通知作为上下文接收：

<notification source="github" type="ci-status" priority="high" status="delivered">CI failed on main: 3 tests failed.</notification>

通知摘要会告诉模型收件箱记录正在等待：

<notification-summary pending="10">github: 3, email: 5, slack: 2</notification-summary>

当 Mastra 发出摘要时，它会清除 summaryAt，并在每条被摘要的记录上设置 summarySignalId。这些记录保持待处理且可读取。当 Mastra 发出完整通知时，它会设置 deliveredSignalId 并将记录标记为 delivered。如果收件箱工具先读取了某条通知，它可以注入完整通知 signal 并将记录标记为 seen，从而防止重复进行完整投递。

当某些通知应等待不同的分发窗口或摘要汇总时，在 agent 上配置投递策略。当延迟通知和摘要汇总应自动投递时，在 Mastra 层级启用计划分发。关于 notifications.deliveryPolicy，请访问 Agent 构造函数参考；关于 runtime 通知分发配置，请访问 Mastra 类参考。

通知收件箱工具

使用 createNotificationInboxTool() 为 agents 提供一个处理收件箱操作的工具，而不是许多 CRUD 工具。当 agent 需要 <notification-summary> signal 背后的完整记录时，使用 read。通知内容会作为 signals 投递，而不是普通工具输出。关于设置示例、输入 schema 和操作行为，请访问 createNotificationInboxTool() 参考。

sendNotificationSignal() 需要支持 notifications 的存储域。仅在需要绕过收件箱存储的低层通知形态上下文时，才使用 sendSignal({ type: 'notification' })。

Mastra Code 中的跨 agent 连接

跨 agent 通信是实验性的，默认关闭。在 Mastra Code 中通过 /settings 开关 "Experimental cross-agent communication" 启用并重启。嵌入式客户端可以在调用 createMastraCode() 时设置 crossAgentSignals: true。该设置会启用 thread 所有权公告和对等节点发现。它还会使 agent 连接工具可用，并让对等节点发现能够触达同一机器上其他项目中的实例（参见在其他项目中发现 agents）。

import { createMastraCode } from 'mastracode'

const mastraCode = await createMastraCode({
  crossAgentSignals: true,
})

Mastra Code 可以使用通知信号在 agent threads 之间通信。对等节点是一个精确的 { agentId, resourceId, threadId } 端点。发现是某个时间点的公告，表示该精确端点当前可用。它并不是持久连接，也不保证投递一定成功。

agent_connections_list 返回一个 peers 集合。每个对等节点包含：

• relationship：none 或 saved。这是当前发送者 thread 的权威持久状态。
• presence：advertised 或 absent。这反映当前发现过程的结果。
• displayStatus：派生的展示值：discovered、connected 或 saved。
• canAttemptSend：仅当对等节点既已保存又当前被公告时为 true。

展示状态的含义如下：

状态	含义	可以尝试发送吗？
[discovered]	当前已公告，但尚未被当前发送者 thread 保存。	否
[connected]	已被当前发送者 thread 保存，且当前已公告。	是
[saved]	已被当前发送者 thread 保存，但当前未公告。	否

使用连接工具执行不同的操作：

• agent_connect 在当前发送者 thread 中保存一个新发现的确切对等节点。
• agent_disconnect 从当前发送者 thread 移除已保存的对等节点。该对等节点无需当前被公告，重复断开也是安全的。
• agent_signal_send 要求确切的对等节点在发送时仍处于已保存且刚被公告的状态。先前发现或最近见过、但在当前发现过程中缺席的对等节点不可发送。

已保存的连接会保留在发送者 thread 中，直到被显式断开。每次发送都会独立刷新发现，并在开始路由前验证确切的端点。因此，已保存的连接记录的是协作意图，并不表示当前在线。

发送尝试到达 Core 路由后，路由结果具有权威性。根据目标 thread 和通知策略，结果可以是 wake、deliver、persist、blocked 或 discard。如果发送因没有 thread 所有者确认而直接投递失败，它不会丢失：signal 会保持待处理在目标的通知收件箱中，并且发送会报告 persist 而不是声称已投递。通知分发器会自行重试该投递，因此发送者无需重新发送。messageId 不会记录到已发送历史中，也不会被拒绝为重复项，尽管待处理通知仍会与之匹配。使用相同 messageId 重新发送时，会在通知待处理期间合并到该通知中，并在重试成功后投递第二份副本。使用 expectsReply: true 的发送会在此处返回错误，因为在 signal 被投递之前，回复义务尚未建立。

当投递策略将 signal 路由到通知摘要而不是直接投递时，发送会记录 messageId，因此使用相同 id 重试会报告重复，而不是再次路由。低优先级发送始终走这条路径并报告 persist。发往活动 thread 的高优先级或中优先级发送也会走这条路径，但在这种情况下摘要 signal 会立即发出，因此发送会报告该 signal 的路由结果（wake、deliver 或 persist），并以相同方式记录 messageId，包括 expectsReply: true 时。只有当策略安排了摘要但在发送时未发出 signal 时，expectsReply: true 的发送才无法建立其回复义务：它会返回错误，且不会将发送记录到已发送历史中，因此要建立回复义务需要使用一个会直接路由的优先级重新发送 signal。

在其他项目中发现 agents

启用跨 agent 通信后，agent_connections_list 会列出同一机器上每个项目中的 Mastra Code 实例，包括当前项目。一个实例可以连接并发送消息到另一个项目中的实例，并接收其回复。这需要 Unix socket pub/sub。Mastra Code CLI 默认启用它。嵌入式客户端在调用 createMastraCode() 时，将 unixSocketPubSub: true 与 crossAgentSignals: true 一起传入。以下情况不适用：

• Windows 上。
• 当你传入自己的 pubsub 时。
• 当通过 MASTRA_RESOURCE_ID 或 .mastracode/database.json 中的 resourceId 设置的 resource ID 无法用作目录名时。这包括空 ID、超过 128 个字符的 ID、包含 /、\ 或控制字符的 ID，以及等于 .、.. 或 _shared 的 ID。Mastra Code 在启动时记录一次警告，该实例只会列出共享其 resource ID 的实例。

Mastra Code 会将 thread 的消息和运行租约路由到该 thread 所属项目的 socket 目录。Thread 所有者查找和对等节点发现（列出你可以连接的 threads）都在共享的 /tmp/mc/_shared 目录中运行。在此更改之前版本运行的实例无法看到更新后的实例，也无法被其看到，因此更新后请重启每个实例。

任何以你的用户身份运行的本地进程都可以列出你的实例所公告的 threads 并向它们发送消息，因此只应在你信任的机器和账户上启用跨 agent 通信。

要将某个实例或测试与共享的 /tmp/mc 目录隔离，请将 MASTRACODE_SIGNALS_SOCKET_ROOT 设置为另一个绝对路径。相对路径会被忽略。请保持路径简短，因为 macOS 限制 Unix socket 路径为 104 字节。该根目录还保存每个 thread 的运行和所有权租约。使用不同根目录的实例不会与使用默认根目录的实例协调 threads。

分布式和无服务器部署

Signals 通过 pub/sub 后端协调运行。当 signal 到达一个实现了 LeaseProvider 的后端时，Mastra 会在目标 thread 上获取租约，以便同一时间只有一个进程拥有该对话，然后唤醒 agent 或将输入路由到运行中的循环。没有租约功能的后端会退回到始终授予所有权的 no-op 实现，这在单进程中没问题，但跨实例则不行。

默认的内存 pub/sub 无法跨越实例边界。在 Vercel 等无服务器平台或任何多实例部署中，后续 signal 可能被路由到与运行 agent 不同的实例。

没有共享 pub/sub 时，该实例无法触达活动运行并会启动自己的运行，导致原始运行不受影响且该 thread 被处理两次。

在 Mastra 实例上配置由 Redis Streams 支持的共享 pub/sub，以便租约和 signals 跨实例协调：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

export const mastra = new Mastra({
  agents: { agent },
  pubsub: new RedisStreamsPubSub({
    url: process.env.REDIS_URL,
    keyPrefix: 'mastra:my-app',
  }),
})

RedisStreamsPubSub 同时实现了事件投递契约和分布式租约。单个后端即可处理跨实例 signal 投递和租约所有权。Vercel 的托管 Redis 集成和 Upstash Redis 都可以很好地工作。关于何时需要分布式 pub/sub 的更多信息，请参阅 PubSub 指南和 RedisStreamsPubSub 参考。

兼容性与 API

兼容性

Mastra 仍接受旧版 signal 载荷，例如 type: 'user-message' 和 type: 'system-reminder'。它会在内部将它们规范化到新的类别和标签结构：

• type: 'user-message'：规范化为 type: 'user' 和 tagName: 'user'
• type: 'system-reminder'：规范化为 type: 'reactive' 和 tagName: 'system-reminder'

现有已存储的 signal 行和较旧客户端会继续通过兼容层加载。当服务器支持时，新客户端会调用消息路由。React 的 thread signal 路径在检测到较旧服务器时会回退到旧版 /signals 路由。关于完整的消息、signal 和订阅类型，请访问 Agent signals 参考。

批准工具调用

当订阅的运行因工具审批而暂停时，使用订阅原生方法批准或拒绝该工具调用。恢复后的 chunks 会通过现有的 thread 订阅到达。关于请求和响应结构，请访问 client.getAgent().sendToolApproval() 参考和服务器 agent 路由。

使用 HTTP 路由

如果直接通过 HTTP 调用 Mastra，立即消息使用 POST /api/agents/:agentId/send-message，下一轮消息使用 POST /api/agents/:agentId/queue-message。对于订阅原生的工具审批，使用 POST /api/agents/:agentId/send-tool-approval。关于请求和响应 schema，请参阅服务器路由参考。

使用客户端 SDK

JavaScript 客户端暴露了 thread signal API。

在发送 thread 输入之前使用 subscribeToThread()，这样客户端可以渲染接收该输入或响应唤醒的流。

src/app/chat.ts

const agent = client.getAgent('supportAgent')

const subscription = await agent.subscribeToThread({
  resourceId: 'user_123',
  threadId: 'thread_456',
})

await agent.sendMessage({
  message: 'Show the shorter version.',
  resourceId: 'user_123',
  threadId: 'thread_456',
})

await subscription.processDataStream({
  onChunk: chunk => {
    console.log(chunk)
  },
  reconnect: true,
})

长期订阅请使用 reconnect: true。关于重连选项，请访问 client.getAgent().subscribeToThread() 参考。

保持自定义 SSE 订阅存活

如果你为 thread 订阅暴露自己的 Server-Sent Events (SSE) 端点，请在流空闲时发送周期性心跳帧。这可以防止浏览器、代理和负载均衡器在下一个 signal 或模型 chunk 到达之前关闭连接。

以下示例每 25 秒发送一条 SSE 注释：

src/api/subscribe.ts

const heartbeat = setInterval(() => {
  controller.enqueue(encoder.encode(': keep-alive\n\n'))
}, 25_000)

request.signal.addEventListener('abort', () => {
  clearInterval(heartbeat)
})

将心跳与客户端重连逻辑配合使用。心跳可减少空闲断开，而重连可在网络或 runtime 仍然关闭流时恢复。

相关

• Agent.sendMessage()
• Agent.queueMessage()
• Agent.queueSignal()
• Agent.sendSignal()
• Agent.sendStateSignal()
• Agent.subscribeToThread()
• createNotificationInboxTool()
• client.getAgent().sendMessage()
• client.getAgent().queueMessage()
• client.getAgent().queueSignal()
• client.getAgent().sendSignal()
• 服务器 agent 路由
• client.getAgent().subscribeToThread()
• client.getAgent().sendToolApproval()
• RedisStreamsPubSub
• 📹 Mastra signals 工作坊