訊號

新增於： @mastra/core@1.39.0

beta

此功能目前為 beta。在 API 穩定前，即使未提高主要版本，也可能發生破壞性變更。

訊號是透過對話串與 Agent 互動的方式。你不必每次都以 agent.stream() 開始互動，而是可以訂閱對話串，再傳送訊息或訊號。Mastra 會在對話串閒置時喚醒 Agent、將輸入放入正在執行的 Agent 迴圈，或將輸入排入下一輪佇列。

使用訊息 API 傳送使用者撰寫的輸入。較底層的系統情境（例如背景任務通知、政策提醒或 Processor 產生的情境）則使用 sendSignal()。

📹 觀看影片

觀看 Mastra 訊號概觀，瞭解訊號如何喚醒及引導長時間執行的 Agent。

何時使用訊號

Agent 對話串需要原始 stream() 呼叫以外的新輸入或情境時，請使用訊號。當使用者在執行作用中傳送後續訊息、背景系統需要將情境加入對話串，或外部事件應喚醒、更新或通知 Agent 時，訊號很有用。

使用 sendMessage() 和 queueMessage() 傳送使用者撰寫的輸入；使用 sendSignal() 傳送較底層的系統情境；持久型狀態通道使用 sendStateSignal()；外部事件應建立持久型通知收件匣記錄時，則使用 sendNotificationSignal()。

快速開始

建立 Agent、訂閱對話串，再將訊息傳送至該對話串。訊息喚醒 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)
}

對話串有正在執行的 Agent 串流時，sendMessage() 會成為該 Agent 迴圈中的新輸入。對話串閒置時，Mastra 則會以此訊息作為第一筆輸入啟動串流。

訊息輸入

立即傳送訊息

當使用者希望作用中的 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 會等待作用中執行完成，再於同一對話串啟動新的執行。

src/mastra/signals.ts

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

對話串閒置時，queueMessage() 會立即啟動執行；對話串作用中時，則會在作用中執行完成後啟動新執行，以保留輪次順序。

訊號情境

控制底層訊號行為

需要傳送系統產生的情境，而不是使用者撰寫的輸入時，請使用 sendSignal()。外部事件使用 type: 'notification'。Mastra 預設會將訊號傳送至作用中執行，並喚醒閒置對話串。使用 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

若閒置喚醒串流需要模型設定、Tool 或執行階段情境等選項，請傳入 ifIdle.streamOptions。ifActive、ifIdle、分支屬性及 streamOptions 請參閱 Agent.sendSignal() 參考。

傳送通知情境

訊號具有語意上的 type，以及提供給 LLM 的 tagName。使用 type 描述訊號類別，並用 tagName 控制模型看到的 XML 標籤。

外部事件使用 type: 'notification'。Reactive 訊號保留給 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',
  },
)

模型會以下列情境形式接收訊號：

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

請使用符合 XML 安全規則的 tagName 與屬性名稱。它們可以包含字母、數字、底線、句點及連字號，且必須以字母或底線開頭。

儲存空間支援

支援較豐富記憶體及訊號 Workflow 的儲存空間配接器可使用通知收件匣儲存功能：libSQL、PostgreSQL 及 MongoDB。這些配接器透過 getStore('notifications') 公開通知記錄。

傳送 Processor 情境

Processor 可在執行期間傳送 Reactive 訊號。Processor 應檢查聊天記錄、回應特定觸發條件，並避免重複傳送相同情境。

下列範例示範 Tool 呼叫讀取 AGENTS.md 檔案後，注入 AGENTS.md 指示的 Processor。

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 訊號預設為 tagName: 'system-reminder'，因此模型會收到下列情境：

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

訂閱的對話串作用中時，await sendSignal() 可保留串流回顯順序。

條件式屬性

使用 ifActive.attributes 和 ifIdle.attributes，可依 Agent 在傳送時為作用中或閒置狀態，為輸入加上情境標籤。頂層 attributes 一律適用；接受輸入時，Mastra 會將所選分支的 attributes 合併其中。各分支專屬屬性請參閱 Agent.sendMessage() 參考與 Agent.sendSignal() 參考。

狀態與通知訊號

狀態訊號

狀態訊號會公開具名稱、範圍限於對話串的情境通道。它適合瀏覽器狀態、編輯器狀態或背景監看器結果等會隨時間變化的持久情境。

外部產生端偵測到狀態變更時，請使用 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 接受狀態訊號時，會將精簡的追蹤中繼資料儲存於對話串。若該狀態仍為目前狀態，而產生端再次傳送相同 cacheKey 及模式，Mastra 會略過重複項目。

Processor 擁有狀態通道時，請使用 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()。當目前訊息清單不含最新快照時，系統會從訊息記錄解析它們。Processor 仍自行負責合併及差異邏輯。

contextWindow.hasSnapshot 會告知 Processor，作用中訊息視窗是否已包含此狀態通道的快照。若為 false，請傳回新的 snapshot，讓較舊的狀態訊息從情境視窗裁剪後，模型仍能看到目前狀態。

內建瀏覽器情境 Processor 會在 browser ID 下，以快照與差異模式發出狀態。

通知訊號

通知訊號代表 GitHub 活動、電子郵件、Slack 提及、CI 狀態、事件、錄製內容或即時訊息等外部事件。事件應建立持久型收件匣記錄時，請使用 agent.sendNotificationSignal()。

通知傳送分成兩個階段。擷取時，agent.sendNotificationSignal() 會儲存通知記錄，並解析 Agent 的傳送政策。分派時，Mastra 會取用到期記錄，並發出完整通知或摘要訊號。

預設傳送政策會考量優先順序。緊急通知會立即傳送；較低優先順序的通知則可能彙整為摘要，或等待對話串閒置。通知欄位請參閱 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 先讀取通知，便可注入完整通知訊號並將記錄標記為 seen，以避免重複傳送完整通知。

若部分通知應等待不同的分派時段或摘要彙總，請在 Agent 上設定傳送政策。若要自動傳送延後的通知及摘要彙總，請在 Mastra 層級啟用排程分派。notifications.deliveryPolicy 請參閱 Agent 建構函式參考，執行階段通知分派設定請參閱 Mastra 類別參考。

通知收件匣 Tool

使用 createNotificationInboxTool()，可提供 Agent 一個處理收件匣動作的 Tool，而不必提供多個 CRUD Tool。Agent 在收到 <notification-summary> 訊號後需要摘要背後的完整記錄時，請使用 read。通知內容會以訊號而非一般 Tool 輸出傳送。設定範例、輸入結構描述及動作行為請參閱 createNotificationInboxTool() 參考。

sendNotificationSignal() 需要支援 notifications 的儲存空間領域。只有較底層、應略過收件匣儲存的通知形式情境，才使用 sendSignal({ type: 'notification' })。

分散式與 Serverless 部署

訊號會透過 pub/sub 後端協調執行。訊號抵達實作 LeaseProvider 的後端時，Mastra 會取得目標對話串的 Lease，使同一時間只有一個處理程序擁有該對話，再喚醒 Agent 或將輸入導入執行中迴圈。不具 Lease 功能的後端會退回至一律授予擁有權的空操作；這適合單一處理程序，但不適合跨執行個體使用。

預設的記憶體內 pub/sub 無法跨越執行個體邊界。在 Vercel 等 Serverless 平台或任何多執行個體部署中，後續訊號可能被路由至與執行 Agent 不同的執行個體。

若沒有共用 pub/sub，該執行個體無法連線作用中執行，而會啟動自己的執行，使原始執行不受影響，並導致對話串被處理兩次。

請在 Mastra 執行個體上設定由 Redis Streams 支援的共用 pub/sub，讓 Lease 與訊號能跨執行個體協調：

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 同時實作事件傳送合約及分散式 Lease，因此單一後端即可處理跨執行個體訊號傳送與 Lease 擁有權。Vercel 的代管 Redis 整合與 Upstash Redis 都很適合。何時需要分散式 pub/sub 的詳細資訊，請參閱 PubSub 指南與 RedisStreamsPubSub 參考。

相容性與 API

相容性

Mastra 仍接受 type: 'user-message' 和 type: 'system-reminder' 等舊版訊號 payload，並在內部將其正規化為新的類別及標籤結構：

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

現有已儲存的訊號資料列及較舊用戶端仍會透過相容層載入。伺服器支援訊息路由時，新用戶端會呼叫這些路由；React 的對話串訊號路徑偵測到較舊伺服器時，則會退回舊版 /signals 路由。完整訊息、訊號及訂閱型別請參閱 Agent 訊號參考。

核准 Tool 呼叫

已訂閱的執行因等待 Tool 核准而暫停時，請使用訂閱專用方法核准或拒絕 Tool 呼叫。恢復後的區塊會透過現有對話串訂閱抵達。請參閱 client.getAgent().sendToolApproval() 參考及伺服器 Agent 路由，瞭解請求與回應結構。

使用 HTTP 路由

若直接透過 HTTP 呼叫 Mastra，立即訊息請使用 POST /api/agents/:agentId/send-message，下一輪訊息則使用 POST /api/agents/:agentId/queue-message。訂閱專用 Tool 核准請使用 POST /api/agents/:agentId/send-tool-approval。請求與回應結構描述請參閱伺服器路由參考。

使用用戶端 SDK

JavaScript 用戶端提供對話串訊號 API。

傳送對話串輸入前，請先使用 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 訂閱保持作用

若你為對話串訂閱公開自己的 Server-Sent Events（SSE）端點，請在串流閒置時定期傳送心跳訊框。這能避免瀏覽器、Proxy 及 Load Balancer 在下一個訊號或模型區塊抵達前關閉連線。

下列範例每 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)
})

請將心跳與用戶端重新連線邏輯搭配使用。心跳可減少閒置中斷，而當網路或執行階段仍關閉串流時，重新連線可恢復作業。

相關內容

• Agent.sendMessage()
• Agent.queueMessage()
• Agent.sendSignal()
• Agent.sendStateSignal()
• Agent.subscribeToThread()
• createNotificationInboxTool()
• client.getAgent().sendMessage()
• client.getAgent().queueMessage()
• client.getAgent().sendSignal()
• 伺服器 Agent 路由
• client.getAgent().subscribeToThread()
• client.getAgent().sendToolApproval()
• RedisStreamsPubSub
• 📹 Mastra 訊號工作坊