신호 제공자

추가된 항목: @mastra/core@1.39.0

:::실험적

이 기능은 베타 버전입니다. API가 안정될 때까지 주요 버전 변경 없이 주요 변경 사항이 발생할 수 있습니다.

:::

신호 Provider는 GitHub, Slack, CI(지속적 통합) 또는 자체 API 같은 외부 소스를 모니터링하고 구독 중인 Agent 스레드로 알림 신호를 푸시합니다.

신호 제공자를 사용해야 하는 경우

외부 시스템이 Agent가 반응해야 하는 이벤트를 생성하고 Mastra가 구독 기록을 관리하도록 하려는 경우 신호 제공자를 사용합니다.

• 소스는 풀 요청, 채널 또는 빌드와 같이 스레드가 관심을 갖는 리소스에 연결된 이벤트를 내보냅니다.
• 어떤 스레드가 어떤 외부 리소스를 감시하는지 추적하는 한 곳이 필요합니다.
• 폴링이나 웹후크 또는 둘 다를 통해 이벤트를 수신하려고 합니다.

일회성 이벤트만 스레드에 푸시해야 하는 경우 다음을 호출하세요.agent.sendNotificationSignal() directly instead.

신호 제공자의 작동 방식

신호 Provider는 신호 시스템에서 신호를 생성하는 측입니다. 외부 이벤트를 스레드로 가져오며, 신호 API는 스레드가 이를 소비하는 방식을 제어합니다. 신호 제공자는 세 가지 기능을 결합합니다.

• 구독 추적: SignalProvider 기본 클래스는 각 Agent 스레드를 감시 중인 외부 리소스에 매핑하는 인메모리 레지스트리를 유지합니다.
• 수집: 풀 기반 소스에는 poll()을, 푸시 기반 소스에는 handleWebhook()을 재정의합니다.
• 전달: 이벤트가 구독과 일치하면 protected notify() 헬퍼를 호출하여 연결된 Agent의 스레드로 알림 신호를 전달합니다. 공급자를 Agent에 전달하여 등록합니다.

Agent는 Provider를 연결하고 pollInterval이 설정되어 있으면 폴링을 시작합니다. 또한 Provider가 노출하는 모든 프로세서 또는 Tool을 병합합니다.

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처럼 알림을 지원하는 스토리지 어댑터가 필요합니다. notify()가 알림 레코드를 저장할 수 있도록 Mastra 인스턴스에 스토리지를 구성하세요. :::

빠른 시작

다음 예에서는 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 },
      )
    }
  }
}

Agent에 공급자를 등록하고 보고 싶은 파이프라인에 스레드를 구독하세요.

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 등록, 스레드 구독 및 테스트를 갖춘 완전한 폴링 Provider를 빌드하는 방법은 신호 Provider 빌드하기를 참조하세요. :::

폴링 및 웹훅 제공자

외부 소스가 앱으로 이벤트를 푸시하지 않는다면 폴링을 사용하세요. pollInterval을 설정하고 poll(subscriptions)를 재정의합니다. 각 구독에는 스레드 대상과 검사할 외부 리소스 ID가 포함됩니다. 외부 소스가 앱을 호출할 수 있다면 웹후크를 사용하세요. handleWebhook(request)를 재정의하고, 페이로드를 파싱하고, 일치하는 구독을 찾은 다음, 일치 항목마다 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()은 자동으로 마운트되는 HTTP 경로가 아니라 Provider 메서드입니다. 요청 본문, 헤더 및 경로 매개변수를 전달하여 자체 엔드포인트에서 호출하세요. 구독, 폴링, 수명 주기 및 notify()에 대한 자세한 내용은 SignalProvider 레퍼런스를 참조하세요. 중복 제거 및 병합 필드를 포함한 전체 알림 페이로드 구조는 Agent.sendNotificationSignal() 레퍼런스를 참조하세요.

내장 웹훅 제공자

일반적인 웹후크 소스에는 하위 클래스를 작성하는 대신 WebhookSignalProvider를 사용하세요. 페이로드에서 리소스 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')

웹후크가 도착하면 라우트에서 webhooks.handleWebhook({ body, headers })를 호출하세요. Provider는 추출된 리소스 ID를 구독과 대조하고 일치하는 각 스레드에 알립니다.

고급 공급자 기능

공급자는 이벤트 수집 이상의 기능을 지원할 수 있습니다. 소스에 필요한 기능만 추가하세요.

• 영속 구독: 기본 레지스트리는 인메모리 방식이며 프로세스별로 존재합니다. 재시작 후에도 구독을 유지해야 한다면 직접 영속화한 다음 start()에서 다시 불러오세요.
• 수명 주기 후크: 비동기 설정에는 start(), 정리에는 stop()을 재정의하세요. stop()을 재정의할 때는 기본 Provider가 폴링을 중지하고 레지스트리를 비울 수 있도록 super.stop()을 호출하세요.
• 프로세서 및 Tool: getInputProcessors() 또는 getOutputProcessors()에서 프로세서를 반환하고, getTools()에서 Agent가 호출할 수 있는 Tool을 반환하세요. @mastra/github-signals 패키지는 GitHub pull request를 감시하고 댓글, 리뷰 상태, 지속적 통합 상태 및 병합에 관해 스레드에 알리는 프로덕션용 신호 Provider입니다. 폴링, 영속 구독, Tool, 프로세서 및 수명 주기 후크의 참고 구현으로 활용하세요.

src/mastra/agents/dev-agent.ts

import { Agent } from '@mastra/core/agent'
import { GithubSignals } from '@mastra/github-signals'

export const devAgent = new Agent({
  id: 'dev-agent',
  name: 'Dev Agent',
  instructions: 'Help triage pull request activity.',
  model: 'openai/gpt-5.6-sol',
  signals: [new GithubSignals()],
})

관련된

• 가이드: 신호 제공자 구축
• 신호
• 알림 신호
• SignalProvider참조
• WebhookSignalProvider참조