Signals

新增於： @mastra/core@1.39.0

beta

此功能目前處於 beta 階段。在 API 穩定之前，即使沒有提升主要版本，也可能出現破壞性變更。

Signals 是透過 thread 與 Agent 互動的一種方式。你毋須每次互動都呼叫 agent.stream()，而可以訂閱 thread，然後傳送訊息或 signal。Mastra 會在 Agent 閒置時喚醒它、將輸入加入正在運行的 Agent 迴圈，或把輸入排入下一個 turn 的佇列。

使用訊息 API 傳送由用戶撰寫的輸入。至於背景任務通知、政策提示或 processor 產生的上下文等較底層系統上下文，則使用 sendSignal()。

📹 觀看影片

觀看 Mastra signals 概覽，了解 signals 如何喚醒及引導長時間運行的 Agent。

何時使用 signals

當 Agent thread 需要接收原來 stream() 呼叫以外的新輸入或上下文時，便可使用 signals。以下情況特別適合使用：運行期間用戶傳送後續訊息、背景系統需要向 thread 加入上下文，或外部事件需要喚醒、更新或通知 Agent。

由用戶撰寫的輸入使用 sendMessage() 和 queueMessage()；較底層的系統上下文使用 sendSignal()；持久狀態通道使用 sendStateSignal()；若外部事件需要建立持久的通知收件箱記錄，則使用 sendNotificationSignal()。

快速開始

建立 Agent、訂閱 thread，然後向該 thread 傳送訊息。當訊息喚醒 Agent 或進入正在運行的迴圈時，訂閱便會收到活躍的 stream。

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 stream 時，sendMessage() 會成為該 Agent 迴圈內的新輸入。當 thread 閒置時，Mastra 會啟動 stream，並以該訊息作為第一個輸入。

訊息輸入

立即傳送訊息

當用戶預期活躍的 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>

沒有屬性的訊息會以純文字用戶輸入傳送。

將訊息排入下一個 turn

若用戶傳送後續訊息，但目前的模型呼叫應先完成，請使用 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 活躍時，它會在目前的運行完成後開始新的運行，以保留 turn 的順序。

Signal 上下文

控制底層 signal 行為

需要傳送由系統產生的上下文，而非由用戶撰寫的輸入時，請使用 sendSignal()。外部事件請使用 type: 'notification'。Mastra 預設會把 signals 傳送至活躍的運行，並喚醒閒置的 thread。你可以使用 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

若閒置喚醒 stream 需要模型設定、Tools 或運行時上下文等選項，請傳入 ifIdle.streamOptions。有關 ifActive、ifIdle、分支屬性及 streamOptions，請參閱 Agent.sendSignal() 參考資料。

傳送通知上下文

Signals 具有語義上的 type 和面向 LLM 的 tagName。使用 type 描述 signal 類別，並使用 tagName 控制模型所看到的 XML 標籤。

外部事件請使用 type: 'notification'。Reactive signals 預留給 processor 或運行時產生的上下文，例如政策指引、背景任務結果及自動載入的指示。

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 和屬性名稱。它們可以包含字母、數字、底線、句點及連字號，並且必須以字母或底線開頭。

儲存支援

支援較豐富 Memory 及 signal Workflow 的儲存介接器可使用通知收件箱儲存：libSQL、PostgreSQL 和 MongoDB。這些介接器透過 getStore('notifications') 提供通知記錄。

傳送 processor 上下文

Processors 可以在運行期間傳送 reactive signals。Processor 應檢查聊天記錄、回應特定觸發條件，並避免重複傳送相同的上下文。

以下範例示範一個 processor：當 Tool 呼叫讀取 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
  },
}

Reactive signals 預設使用 tagName: 'system-reminder'，因此模型會收到如下形式的上下文：

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

當訂閱的 thread 處於活躍狀態時，等待 sendSignal() 完成可保留 stream 回應的順序。

條件式屬性

使用 ifActive.attributes 和 ifIdle.attributes，可根據傳送時 Agent 處於活躍還是閒置狀態，為輸入加上相應的上下文標籤。最上層的 attributes 一律套用；當輸入獲接受時，Mastra 會把所選分支的 attributes 合併其中。有關分支專用屬性，請參閱 Agent.sendMessage() 參考資料和 Agent.sendSignal() 參考資料。

狀態及通知 signals

狀態 signals

狀態 signals 提供具名且限定於 thread 範圍的上下文通道。它們適用於會隨時間改變的持久上下文，例如瀏覽器狀態、編輯器狀態或背景監察器的結果。

當外部產生者偵測到狀態變更時，請使用 sendStateSignal()。每個狀態 signal 都會識別狀態通道、由產生者擁有的快取鍵，以及該更新屬於快照還是差異。

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 接受狀態 signal 後，會在 thread 上儲存精簡的追蹤 metadata。如果產生者在該狀態仍然有效時再次傳送相同的 cacheKey 和模式，Mastra 便會略過重複項目。

當 processor 擁有狀態通道時，請使用 computeStateSignal()。Mastra 會在 processInputStep() 之後，每個模型輸入步驟呼叫它一次。有關狀態 signal 欄位及傳回值，請參閱 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()。當目前的訊息清單不包含最新快照時，它會從訊息記錄解析兩者；合併和比較差異的邏輯仍由 processor 負責。

contextWindow.hasSnapshot 會告知 processor，活躍的訊息視窗是否已包含此狀態通道的快照。如果值為 false，請傳回新的 snapshot，以便即使較舊的狀態訊息已從上下文視窗裁走，模型仍可看到目前狀態。

內置瀏覽器上下文 processor 會以 browser id 發出狀態，並使用快照及差異模式。

通知 signals

通知 signals 代表 GitHub 活動、電郵、Slack 提及、CI 狀態、事故、錄音或直接訊息等外部事件。若事件應建立持久的收件箱記錄，請使用 agent.sendNotificationSignal()。

通知傳送分為兩個階段。在接收階段，agent.sendNotificationSignal() 會儲存通知記錄，並解析 Agent 的傳送政策。在派送階段，Mastra 會處理到期記錄，並發出完整通知或摘要 signals。

預設的傳送政策會考慮優先級。緊急通知會立即傳送；優先級較低的通知則可能彙整成摘要，或等待 thread 閒置。有關通知欄位，請參閱 Agent.sendNotificationSignal() 參考資料；有關 notifications.deliveryPolicy 設定，請參閱 Agent 建構函式參考資料；有關收件箱 Tool 操作，請參閱 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。如果收件箱 Tool 先讀取通知，它可以注入完整通知 signal，並把記錄標記為 seen，以避免重複傳送完整通知。

若部分通知應等待另一個派送時段或摘要彙整，請在 Agent 上設定傳送政策。若要自動傳送延後的通知及摘要彙整，請在 Mastra 層級啟用排程派送。有關 notifications.deliveryPolicy，請參閱 Agent 建構函式參考資料；有關運行時通知派送設定，請參閱 Mastra 類別參考資料。

通知收件箱 Tool

使用 createNotificationInboxTool()，可讓 Agent 透過一個 Tool 執行收件箱操作，毋須使用多個 CRUD Tools。當 Agent 收到 <notification-summary> signal 後，需要取得摘要背後的完整記錄時，請使用 read。通知內容會以 signals 而非一般 Tool 輸出的形式傳送。有關設定範例、輸入 schema 及操作行為，請參閱 createNotificationInboxTool() 參考資料。

sendNotificationSignal() 需要支援 notifications 的儲存 domain。只有較底層、具通知形式但應繞過收件箱儲存的上下文，才使用 sendSignal({ type: 'notification' })。

分散式及 serverless 部署

Signals 透過 pub/sub backend 協調運行。當 signal 到達實作了 LeaseProvider 的 backend 時，Mastra 會取得目標 thread 的租約，確保一次只有一個程序擁有對話，然後喚醒 Agent 或把輸入傳送到正在運行的迴圈。沒有租約功能的 backend 會退回至一律授予擁有權的 no-op；這適用於單一程序，但不適用於跨執行個體的環境。

預設的記憶體內 pub/sub 無法跨越執行個體邊界。在 Vercel 等 serverless 平台或任何多執行個體部署中，後續 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 同時實作事件傳送合約及分散式租約，因此單一 backend 即可處理跨執行個體的 signal 傳送及租約擁有權。Vercel 的受管理 Redis 整合及 Upstash Redis 均十分合適。如要進一步了解何時需要分散式 pub/sub，請參閱 PubSub 指南及 RedisStreamsPubSub 參考資料。

相容性及 API

相容性

Mastra 仍接受舊版 signal payload，例如 type: 'user-message' 和 type: 'system-reminder'。它會在內部把它們正規化為新的類別及標籤形式：

• type: 'user-message'：正規化為 type: 'user' 和 tagName: 'user'
• type: 'system-reminder'：正規化為 type: 'reactive' 和 tagName: 'system-reminder'

現有已儲存的 signal 資料列及較舊的 client 會繼續透過相容層載入。當伺服器支援時，新 client 會呼叫訊息 routes；React 的 thread signal 路徑偵測到較舊的伺服器時，則會退回舊版 /signals route。有關完整的訊息、signal 及訂閱類型，請參閱 Agent signals 參考資料。

核准 Tool 呼叫

當訂閱的運行暫停以等待 Tool 核准時，請使用訂閱原生的方法核准或拒絕 Tool 呼叫。恢復後的 chunks 會透過現有的 thread 訂閱送達。有關請求及回應格式，請參閱 client.getAgent().sendToolApproval() 參考資料和伺服器 Agent routes。

使用 HTTP routes

如果你直接透過 HTTP 呼叫 Mastra，立即傳送訊息請使用 POST /api/agents/:agentId/send-message，把訊息排入下一個 turn 則使用 POST /api/agents/:agentId/queue-message。訂閱原生的 Tool 核准請使用 POST /api/agents/:agentId/send-tool-approval。有關請求及回應 schema，請參閱伺服器 routes 參考資料。

使用 client SDK

JavaScript client 提供 thread signal API。

傳送 thread 輸入之前，請先使用 subscribeToThread()，讓 client 可以呈現接收該輸入或因應輸入而被喚醒的 stream。

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 訂閱連線

如果你自行提供 Server-Sent Events (SSE) endpoint 用於 thread 訂閱，請在 stream 閒置時定期傳送 heartbeat frames。這可避免瀏覽器、proxies 及負載平衡器在下一個 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)
})

請把 heartbeats 與 client 端的重新連線邏輯配合使用。Heartbeats 可減少閒置中斷，而當網絡或運行時仍然關閉 stream 時，重新連線則可恢復連接。

相關內容

• Agent.sendMessage()
• Agent.queueMessage()
• Agent.sendSignal()
• Agent.sendStateSignal()
• Agent.subscribeToThread()
• createNotificationInboxTool()
• client.getAgent().sendMessage()
• client.getAgent().queueMessage()
• client.getAgent().sendSignal()
• 伺服器 Agent routes
• client.getAgent().subscribeToThread()
• client.getAgent().sendToolApproval()
• RedisStreamsPubSub
• 📹 Mastra signals 工作坊