Signal providers

新增于： @mastra/core@1.42.0

signal provider 监控外部来源（例如 GitHub、Slack、持续集成（CI）或你自己的 API），并将通知 signal推送到已订阅的 agent threads。

何时使用 signal provider

当外部系统产生 agent 应作出反应的事件，并且你希望 Mastra 替你管理订阅簿记时，请使用 signal provider。

• 来源发出的事件与 thread 关心的资源相关联，例如 pull request、channel 或 build。
• 你希望有一个地方跟踪哪些 thread 在监视哪些外部资源。
• 你希望通过轮询或 webhook，或两者兼用，接收事件。

如果只需向 thread 推送一个一次性事件，请直接调用 agent.sendNotificationSignal()。

Signal providers 如何工作

signal provider 是 signals系统的生产端。它将外部事件带入 thread，而 signal API 控制 thread 如何消费它们。

signal provider 组合了三项能力：

• 订阅跟踪： SignalProvider 基类维护一个内存注册表，将每个 agent thread 映射到它所监视的外部资源。
• 摄取： 对于基于拉取的来源，重写 poll()；对于基于推送的来源，重写 handleWebhook()。
• 投递： 当事件匹配订阅时，调用受保护的 notify() 辅助方法，将通知 signal 转发到所连接 agent 的 thread。

通过将 provider 传给 agent 来注册它。

agent 会连接 provider，并在设置了 pollInterval 时开始轮询。它还会合并 provider 暴露的任何 processors 或工具。

src/mastra/agents/support-agent.ts

import { Agent } from '@mastra/core/agent'
import { CiSignals } from '../signals/ci-signals'

export const supportAgent = new Agent({
  id: 'support-agent',
  name: 'Support Agent',
  instructions: 'Help the user triage updates.',
  model: 'openai/gpt-5.6-sol',
  signals: [new CiSignals()],
})

备注

通知投递需要支持通知的存储适配器，例如 libSQL、PostgreSQL 或 MongoDB。请在 Mastra 实例上配置存储，以便 notify() 能存储通知记录。

快速开始

以下示例演示一个轮询 provider，它监视 CI 流水线，并在已订阅的流水线失败时发出通知。

src/mastra/signals/ci-signals.ts

import { SignalProvider } from '@mastra/core/signals'
import type { SignalProviderTarget, SignalSubscription } from '@mastra/core/signals'

type BuildStatus = {
  id: string
  status: 'passed' | 'failed'
}

const builds = new Map<string, BuildStatus>([
  ['acme-app-main', { id: 'build_123', status: 'failed' }],
])

async function fetchBuildStatus(pipeline: string): Promise<BuildStatus> {
  return builds.get(pipeline) ?? { id: 'build_unknown', status: 'passed' }
}

export class CiSignals extends SignalProvider<'ci-signals'> {
  readonly id = 'ci-signals' as const
  readonly pollInterval = 30_000

  watch(target: SignalProviderTarget, pipeline: string) {
    return this.subscribe(target, pipeline)
  }

  unwatch(target: SignalProviderTarget, pipeline: string) {
    return this.unsubscribe(target, pipeline)
  }

  async poll(subscriptions: SignalSubscription[]) {
    for (const sub of subscriptions) {
      const build = await fetchBuildStatus(sub.externalResourceId)
      if (build.status !== 'failed') continue

      await this.notify(
        {
          source: this.id,
          kind: 'ci-status',
          priority: 'high',
          summary: `Build failed for ${sub.externalResourceId}`,
          payload: build,
          dedupeKey: `${this.id}:${sub.externalResourceId}:${build.id}`,
        },
        { resourceId: sub.resourceId, threadId: sub.threadId },
      )
    }
  }
}

将 provider 注册到 agent，并让某个 thread 订阅你想监视的流水线。

src/mastra/agents/support-agent.ts

import { Agent } from '@mastra/core/agent'
import { CiSignals } from '../signals/ci-signals'

export const ciSignals = new CiSignals()

export const supportAgent = new Agent({
  id: 'support-agent',
  name: 'Support Agent',
  instructions: 'Help the user triage CI updates.',
  model: 'openai/gpt-5.6-sol',
  signals: [ciSignals],
})

ciSignals.watch({ resourceId: 'user_123', threadId: 'thread_456' }, 'acme-app-main')

Mastra 会按 pollInterval 使用所有活跃订阅调用 poll()。没有订阅时它会跳过一个周期，并且不会重叠周期，因此缓慢的 poll() 不会与自身并发运行。

备注

如需一个包含通知存储、agent 注册、thread 订阅和测试的完整轮询 provider 构建示例，请参阅构建 signal provider。

轮询与 webhook providers

当外部来源不向你的应用推送事件时，请使用轮询。设置 pollInterval 并重写 poll(subscriptions)。每个订阅都包含 thread 目标和要检查的外部资源 id。

当外部来源可以调用你的应用时，请使用 webhooks。重写 handleWebhook(request) 解析 payload，然后查找匹配的订阅，并为每个匹配项调用 notify()。

src/mastra/signals/ci-signals.ts

import { SignalProvider } from '@mastra/core/signals'
import type { SignalProviderWebhookRequest } from '@mastra/core/signals'

export class CiSignals extends SignalProvider<'ci-signals'> {
  readonly id = 'ci-signals' as const

  async handleWebhook(request: SignalProviderWebhookRequest) {
    const payload = request.body as { pipeline: string; status: string }
    const subscriptions = this.getSubscriptionsForResource(payload.pipeline)

    for (const sub of subscriptions) {
      await this.notify(
        {
          source: this.id,
          kind: 'ci-status',
          priority: 'high',
          summary: `Build ${payload.status} for ${payload.pipeline}`,
          payload,
        },
        { resourceId: sub.resourceId, threadId: sub.threadId },
      )
    }

    return { status: 200, body: { matched: subscriptions.length } }
  }
}

handleWebhook() 是 provider 方法，不是自动挂载的 HTTP 路由。请从你自己的端点调用它，并传入请求体、请求头和任何路由参数。订阅、轮询、生命周期和 notify() 详情请参阅 SignalProvider 参考。完整的通知 payload 结构（包括去重和合并字段）请参阅 Agent.sendNotificationSignal() 参考。

内置 webhook provider

对于通用 webhook 来源，请使用 WebhookSignalProvider，而不是编写子类。用一个从 payload 中提取资源 id 的函数，以及一个可选的构建通知函数来配置它。

src/mastra/index.ts

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

const webhooks = new WebhookSignalProvider({
  extractResourceId: payload => (payload as { repository: string }).repository,
  buildNotification: (payload, sub) => ({
    source: 'ci',
    kind: 'build-status',
    priority: 'medium',
    summary: `Build ${(payload as { status: string }).status} for ${sub.externalResourceId}`,
  }),
})

export const supportAgent = new Agent({
  id: 'support-agent',
  name: 'Support Agent',
  instructions: 'Help the user triage updates.',
  model: 'openai/gpt-5.6-sol',
  signals: [webhooks],
})

webhooks.subscribeThread({ resourceId: 'user_123', threadId: 'thread_456' }, 'acme/app')

当 webhook 到达时，请从你的路由调用 webhooks.handleWebhook({ body, headers })。provider 会将提取出的资源 id 与其订阅匹配，并通知每个匹配的 thread。

高级 provider 能力

provider 能做的不仅是事件摄取。只添加你的来源所需的能力。

• 持久化订阅： 基础注册表是内存中且按进程隔离的。当订阅必须在重启后保留时，请自行持久化它们，然后在 start() 中重新补水（rehydrate）。
• 生命周期钩子： 重写 start() 进行异步设置，重写 stop() 进行清理。重写 stop() 时请调用 super.stop()，以便基础 provider 停止轮询并清除其注册表。
• Processors 与工具： 从 getInputProcessors() 或 getOutputProcessors() 返回 processors，并从 getTools() 返回 agent 可调用的工具。

有关监视 GitHub pull requests 的生产 provider，请参阅 GitHub Channels 集成。

相关内容

• 指南：构建 signal provider
• Signals
• 通知 signals
• SignalProvider 参考
• WebhookSignalProvider 参考
• Mastra Factory 使用 signal providers 集成 GitHub 和 Linear 等外部系统，将 issue 和 pull request 带入协作工作板。