SignalProvider
新增於： @mastra/core@1.39.0
用於建立 Signal Provider 的抽象基底類別。Signal Provider 會監控外部來源（API、webhook、event stream），並透過內建的 subscription registry，將 notification signal push 至 Agent thread。
Signal Provider 預設不是 processor。需要攔截 Agent 執行的 Provider 會從 getInputProcessors() 或 getOutputProcessors() 傳回 processor。公開 Agent 可呼叫 Tool 的 Provider 則會從 getTools() 傳回 Tool。
可直接使用的 webhook 型 Provider 請參閱 WebhookSignalProvider。
使用範例「使用範例」的直接連結
每 30 秒檢查 API 的 polling Provider：
import { SignalProvider } from '@mastra/core/signals'
import type { SignalSubscription } from '@mastra/core/signals'
class SlackSignals extends SignalProvider<'slack-signals'> {
readonly id = 'slack-signals'
readonly pollInterval = 30_000
async poll(subscriptions: SignalSubscription[]) {
for (const sub of subscriptions) {
const messages = await fetchSlackMessages(sub.externalResourceId)
if (messages.length > 0) {
await this.notify(
{
source: 'slack',
kind: 'new-messages',
summary: `${messages.length} new messages in ${sub.externalResourceId}`,
},
{ threadId: sub.threadId, resourceId: sub.resourceId },
)
}
}
}
}
向 Agent 註冊：
import { Agent } from '@mastra/core/agent'
const agent = new Agent({
id: 'agent',
signals: [new SlackSignals()],
})
Agent 會呼叫 connect(this)，並註冊 Provider 傳回的所有 processor 或 Tool，接著開始 polling。
Constructor 參數「Constructor 參數」的直接連結
SignalProvider 是抽象類別。子類別呼叫不含引數的 super()。
屬性「屬性」的直接連結
id:
name?:
pollInterval?:
poll()。只使用 webhook 的 Provider 請保留為 undefined 或 0。isConnected:
connect() 後傳回 true。內部用於在 Agent.__fork() 期間略過重複接線。方法「方法」的直接連結
連線「連線」的直接連結
connect(agent)「connectagent」的直接連結
由 Agent constructor 呼叫。建立雙向連結，讓 Provider 能將 signal 傳回 Agent。若要在建立連結後進行其他設定，請覆寫此方法。務必呼叫 super.connect(agent)。
class MySignals extends SignalProvider<'my-signals'> {
readonly id = 'my-signals'
override connect(agent) {
super.connect(agent)
// additional setup after agent link is established
}
}
__registerMastra(mastra)「__registermastramastra」的直接連結
Provider 的 Agent 註冊至 Mastra instance 時呼叫。若要存取 Storage 或其他 Mastra 服務，請覆寫此方法。務必呼叫 super.__registerMastra(mastra)。
override __registerMastra(mastra) {
super.__registerMastra(mastra)
// this.mastra is now available
}
Processor 與 Tool 整合「Processor 與 Tool 整合」的直接連結
getInputProcessors()「getinputprocessors」的直接連結
傳回此 Provider 需要向 Agent 註冊的輸入 processor。Provider 會攔截 Agent 輸入 step 時，請覆寫此方法，例如插入 context 提示或偵測 Tool 呼叫。
getInputProcessors() {
return [this]
}
傳回：InputProcessorOrWorkflow[]
getOutputProcessors()「getoutputprocessors」的直接連結
傳回此 Provider 需要向 Agent 註冊的輸出 processor。Provider 會攔截 Agent 輸出 step 時，請覆寫此方法。
getOutputProcessors() {
return [this]
}
傳回：OutputProcessorOrWorkflow[]
getTools()「gettools」的直接連結
傳回此 Provider 公開給 Agent 的 Tool。Provider 會新增 Agent 可呼叫的 Tool（例如訂閱或取消訂閱指令）時，請覆寫此方法。
getTools() {
return {
subscribe_pr: createTool({ /* ... */ }),
unsubscribe_pr: createTool({ /* ... */ }),
}
}
傳回：Record<string, unknown>
Subscription 追蹤「Subscription 追蹤」的直接連結
subscribe(target, externalResourceId, metadata?)「subscribetarget-externalresourceid-metadata」的直接連結
讓 thread 訂閱外部資源。這是 protected 方法，請從 Provider 實作內部呼叫。
const sub = this.subscribe(
{ threadId: 'thread-1', resourceId: 'user-1' },
'github:mastra-ai/mastra#123',
{ pr: 123 },
)
傳回：SignalSubscription：新建的 subscription，或 metadata 合併後的現有 subscription。
target:
threadId 與 resourceId。externalResourceId:
"github:owner/repo#123"。metadata?:
unsubscribe(target, externalResourceId)「unsubscribetarget-externalresourceid」的直接連結
移除 subscription。
const removed = this.unsubscribe(
{ threadId: 'thread-1', resourceId: 'user-1' },
'github:mastra-ai/mastra#123',
)
傳回：boolean：已移除時為 true，沒有符合的 subscription 時為 false。
getSubscriptions()「getsubscriptions」的直接連結
傳回此 Provider 的所有有效 subscription。
const allSubs = this.getSubscriptions()
傳回：SignalSubscription[]
getSubscriptionsForResource(externalResourceId)「getsubscriptionsforresourceexternalresourceid」的直接連結
傳回特定外部資源的所有 subscription。
const subs = this.getSubscriptionsForResource('github:mastra-ai/mastra#123')
for (const sub of subs) {
await this.notify(
{ source: 'my-provider', kind: 'update', summary: 'Resource updated' },
{ threadId: sub.threadId, resourceId: sub.resourceId },
)
}
傳回：SignalSubscription[]
getSubscriptionsForThread(target)「getsubscriptionsforthreadtarget」的直接連結
傳回特定 thread 的所有 subscription。
const subs = this.getSubscriptionsForThread({
threadId: 'thread-1',
resourceId: 'user-1',
})
傳回：SignalSubscription[]
hasSubscription(target, externalResourceId)「hassubscriptiontarget-externalresourceid」的直接連結
檢查 subscription 是否存在。
if (this.hasSubscription(target, 'github:mastra-ai/mastra#123')) {
// already subscribed
}
傳回：boolean
unsubscribeAll(target)「unsubscribealltarget」的直接連結
移除 thread 的所有 subscription。
const removed = this.unsubscribeAll({
threadId: 'thread-1',
resourceId: 'user-1',
})
傳回：number：已移除的 subscription 數量。
subscriptionCount「subscriptioncount」的直接連結
此 Provider 的有效 subscription 總數。
if (this.subscriptionCount === 0) {
// nothing to poll
}
傳回：number
Polling「Polling」的直接連結
poll(subscriptions)「pollsubscriptions」的直接連結
每次 poll cycle 都會呼叫，並傳入所有有效 subscription。請覆寫此方法以檢查外部來源並發出通知。framework 會避免 poll cycle 重疊：如果一次 poll() 呼叫耗時超過 pollInterval，便會略過下一個 cycle。
async poll(subscriptions: SignalSubscription[]) {
for (const sub of subscriptions) {
const events = await checkExternalSource(sub.externalResourceId)
for (const event of events) {
await this.notify(
{ source: 'my-provider', kind: event.type, summary: event.message },
{ threadId: sub.threadId, resourceId: sub.resourceId },
)
}
}
}
startPolling()「startpolling」的直接連結
啟動 polling timer。Agent 會在 connect() 後呼叫。此方法具冪等性，多次呼叫不會產生影響。
provider.startPolling()
stopPolling()「stoppolling」的直接連結
停止 polling timer。
provider.stopPolling()
Webhook「Webhook」的直接連結
handleWebhook(request)「handlewebhookrequest」的直接連結
處理傳入的 webhook request。請覆寫此方法以解析 payload 並比對 subscription，接著發出 notification signal。可直接使用的實作請參閱 WebhookSignalProvider。
驗證 webhook request 後，請從應用程式定義的 HTTP endpoint 呼叫此方法。
async handleWebhook(request) {
const payload = request.body as { repo: string, event: string }
const subs = this.getSubscriptionsForResource(payload.repo)
for (const sub of subs) {
await this.notify(
{ source: 'github', kind: payload.event, summary: `Event on ${payload.repo}` },
{ threadId: sub.threadId, resourceId: sub.resourceId },
)
}
return { status: 200, body: { matched: subs.length } }
}
傳回：Promise<{ status?: number; body?: unknown }>
生命週期「生命週期」的直接連結
start()「start」的直接連結
在 connect() 後呼叫，以執行 async 初始化。設定需要 Agent 或 Mastra instance 才能進行時，請覆寫此方法。
async start() {
await this.loadInitialState()
}
stop()「stop」的直接連結
在關閉時呼叫。預設實作會停止 polling，並清除所有 subscription。
provider.stop()
通知「通知」的直接連結
notify(notification, target)「notifynotification-target」的直接連結
向已連線 Agent 傳送 notification signal。這是 agent.sendNotificationSignal() 的 protected 便利 wrapper。
await this.notify(
{
source: 'my-provider',
kind: 'pr-updated',
summary: 'PR #123 was updated',
priority: 'high',
payload: { prNumber: 123 },
},
{ threadId: 'thread-1', resourceId: 'user-1' },
)
notification:
source:
kind:
"pr-updated"、"new-message"。summary:
priority?:
payload?:
target:
threadId 與 resourceId。型別「型別」的直接連結
SignalSubscription「signalsubscription」的直接連結
subscribe() 傳回的 subscription 物件。
id:
providerId:
threadId:
resourceId:
externalResourceId:
"github:owner/repo#123"。subscribedAt:
metadata:
SignalProviderTarget「signalprovidertarget」的直接連結
識別特定 Agent thread。
threadId:
resourceId:
agentId?:
Type guard「Type guard」的直接連結
isSignalProvider(obj)「issignalproviderobj」的直接連結
對 SignalProvider instance 執行 runtime 檢查。
import { isSignalProvider } from '@mastra/core/signals'
if (isSignalProvider(obj)) {
obj.connect(agent)
}
傳回：boolean