SignalProvider

新增于： @mastra/core@1.39.0

用于构建 Signal Provider 的抽象基类。Signal Provider 监视外部来源（API、webhook、事件流），并通过内置订阅注册表将通知 Signal 推送至 Agent thread。

Signal Provider 默认不是处理器。需要拦截 Agent 执行的 Provider 会从 getInputProcessors() 或 getOutputProcessors() 返回处理器。暴露供 Agent 调用的 Tool 的 Provider 会从 getTools() 返回它们。

如需可直接使用的基于 webhook 的 Provider，请参阅 WebhookSignalProvider。

使用示例

每 30 秒检查一次 API 的轮询 Provider：

src/signals/slack-signals.ts

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：

src/mastra/index.ts

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

const agent = new Agent({
  id: 'agent',
  signals: [new SlackSignals()],
})

Agent 调用 connect(this) 并注册 Provider 返回的所有处理器或 Tool，然后开始轮询。

构造函数参数

SignalProvider 是抽象类。子类调用不带参数的 super()。

属性

id:

TId extends string

此 Provider 的唯一标识符。子类必须将其实现为 readonly 属性。

name?:

string

Provider 的人类可读显示名称。

pollInterval?:

number

以毫秒为单位的轮询间隔。设置后，框架会按此间隔调用 poll()。对于仅 webhook 的 Provider，请保留为 undefined 或 0。

isConnected:

boolean

此 Provider 是否已连接到 Agent。调用 connect() 后返回 true。内部用于在 Agent.__fork() 期间跳过重新连接。

方法

连接

connect(agent)

由 Agent 构造函数调用。建立双向链接，以便 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)

当 Provider 的 Agent 注册到 Mastra 实例时调用。可重写此方法以访问 storage 或其他 Mastra 服务。务必调用 super.__registerMastra(mastra)。

override __registerMastra(mastra) {
  super.__registerMastra(mastra)
  // this.mastra is now available
}

处理器和 Tool 集成

getInputProcessors()

返回此 Provider 需要注册到 Agent 的输入处理器。当 Provider 要拦截 Agent 输入步骤时重写（例如注入上下文提示或检测 Tool 调用）。

getInputProcessors() {
  return [this]
}

返回：InputProcessorOrWorkflow[]

getOutputProcessors()

返回此 Provider 需要注册到 Agent 的输出处理器。当 Provider 要拦截 Agent 输出步骤时重写。

getOutputProcessors() {
  return [this]
}

返回：OutputProcessorOrWorkflow[]

getTools()

返回此 Provider 向 Agent 暴露的 Tool。当 Provider 添加可由 Agent 调用的 Tool（如订阅或取消订阅命令）时重写。

getTools() {
  return {
    subscribe_pr: createTool({ /* ... */ }),
    unsubscribe_pr: createTool({ /* ... */ }),
  }
}

返回：Record<string, unknown>

订阅跟踪

subscribe(target, externalResourceId, metadata?)

为 thread 订阅外部资源。这是受保护的方法：请从 Provider 实现内部调用。

const sub = this.subscribe(
  { threadId: 'thread-1', resourceId: 'user-1' },
  'github:mastra-ai/mastra#123',
  { pr: 123 },
)

返回：SignalSubscription：创建的订阅，或具有已合并 metadata 的现有订阅。

target:

SignalProviderTarget

要订阅的 thread。必须包含 threadId 和 resourceId。

externalResourceId:

string

外部资源的 Provider 特定标识符（例如，"github:owner/repo#123"）。

metadata?:

Record<string, unknown>

与订阅一起存储的附加数据。在重复订阅时合并到现有 metadata。

unsubscribe(target, externalResourceId)

移除订阅。

const removed = this.unsubscribe(
  { threadId: 'thread-1', resourceId: 'user-1' },
  'github:mastra-ai/mastra#123',
)

返回：boolean：移除时为 true；不存在匹配订阅时为 false。

getSubscriptions()

返回此 Provider 的所有活跃订阅。

const allSubs = this.getSubscriptions()

返回：SignalSubscription[]

getSubscriptionsForResource(externalResourceId)

返回特定外部资源的所有订阅。

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)

返回特定 thread 的所有订阅。

const subs = this.getSubscriptionsForThread({
  threadId: 'thread-1',
  resourceId: 'user-1',
})

返回：SignalSubscription[]

hasSubscription(target, externalResourceId)

检查订阅是否存在。

if (this.hasSubscription(target, 'github:mastra-ai/mastra#123')) {
  // already subscribed
}

返回：boolean

unsubscribeAll(target)

移除一个 thread 的全部订阅。

const removed = this.unsubscribeAll({
  threadId: 'thread-1',
  resourceId: 'user-1',
})

返回：number：已移除订阅的数量。

subscriptionCount

此 Provider 的活跃订阅总数。

if (this.subscriptionCount === 0) {
  // nothing to poll
}

返回：number

轮询

poll(subscriptions)

在每个轮询周期中使用所有活跃订阅调用。重写以检查外部来源并发出通知。框架会防止轮询周期重叠：若 poll() 调用耗时超过 pollInterval，则跳过下一周期。

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()

启动轮询计时器。由 Agent 在 connect() 后调用。具有幂等性：多次调用没有影响。

provider.startPolling()

stopPolling()

停止轮询计时器。

provider.stopPolling()

Webhooks

handleWebhook(request)

处理传入的 webhook 请求。重写以解析 payload 并将其与订阅匹配，然后发出通知 Signal。有关可直接使用的实现，请参阅 WebhookSignalProvider。

验证 webhook 请求后，从应用程序定义的 HTTP 端点调用此方法。

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()

在 connect() 后调用以运行异步初始化。当设置需要 Agent 或 Mastra 实例可用时重写。

async start() {
  await this.loadInitialState()
}

stop()

关闭时调用。默认实现会停止轮询并清除所有订阅。

provider.stop()

通知

notify(notification, target)

向已连接的 Agent 发送通知 Signal。这是对 agent.sendNotificationSignal() 的受保护便捷封装。

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:

object

通知 payload。

object

source:

string

通知来源的标识符。

kind:

string

事件类型（例如，"pr-updated"、"new-message"）。

summary:

string

事件的人类可读摘要。

priority?:

"high" | "medium" | "low"

通知优先级。

payload?:

unknown

附加到通知的任意数据。

target:

SignalProviderTarget

要通知的 thread。必须包含 threadId 和 resourceId。

类型

SignalSubscription

由 subscribe() 返回的订阅对象。

id:

string

订阅的唯一标识符。

providerId:

string

拥有此订阅的 Provider。

threadId:

string

接收 Signal 的 thread。

resourceId:

string

拥有该 thread 的资源。

externalResourceId:

string

外部资源的 Provider 特定标识符（例如，"github:owner/repo#123"）。

subscribedAt:

Date

创建订阅的时间。

metadata:

Record<string, unknown>

与订阅一起存储的 Provider 特定 metadata。

SignalProviderTarget

标识特定的 Agent thread。

threadId:

string

目标 thread。

resourceId:

string

拥有该 thread 的资源。

agentId?:

string

Agent 标识符。

类型守卫

isSignalProvider(obj)

用于检查 SignalProvider 实例的运行时检查。

import { isSignalProvider } from '@mastra/core/signals'

if (isSignalProvider(obj)) {
  obj.connect(agent)
}

返回：boolean