신호 제공자

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

신호 제공자를 구축하기 위한 추상 기본 클래스입니다. 신호 제공자는 외부 소스(API, 웹후크, 이벤트 스트림)를 모니터링하고 내장된 구독 레지스트리를 통해 알림 신호를 Agent 스레드에 푸시합니다.

신호 Provider는 기본적으로 프로세서가 아닙니다. Agent 실행을 가로채야 하는 Provider는 getInputProcessors() 또는 getOutputProcessors()에서 프로세서를 반환합니다. Agent가 호출할 수 있는 Tool을 노출하는 Provider는 getTools()에서 해당 Tool을 반환합니다. 즉시 사용 가능한 웹훅 기반 공급자에 대해서는 다음을 참조하세요.WebhookSignalProvider.

사용예

30초마다 API를 확인하는 폴링 제공자:

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의 고유 식별자입니다. 서브클래스는 이를 읽기 전용 속성으로 구현해야 합니다.

name?:

string

사람이 읽을 수 있는 Provider 표시 이름입니다.

pollInterval?:

number

밀리초 단위의 폴링 간격입니다. 설정하면 프레임워크가 이 간격으로 poll()을 호출합니다. 웹훅 전용 Provider의 경우 undefined 또는 0으로 두세요.

isConnected:

boolean

이 Provider가 Agent에 연결되어 있는지를 나타냅니다. connect()가 호출된 후 true를 반환합니다. Agent.__fork() 중 재연결을 건너뛰기 위해 내부적으로 사용됩니다.

행동 양식

연결

connect(agent)

Agent 생성자에 의해 호출됩니다. 공급자가 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)

공급자의 Agent가 Mastra 인스턴스에 등록될 때 호출됩니다. 스토리지 또는 기타 Mastra 서비스에 액세스하려면 재정의하세요. 항상 전화해super.__registerMastra(mastra).

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

프로세서 및 Tool 통합

getInputProcessors()

이 공급자를 Agent에 등록해야 하는 반환 입력 프로세서입니다. 공급자가 Agent 입력 단계(예: 컨텍스트 힌트 삽입 또는 Tool 호출 감지)를 가로채는 경우 재정의합니다.

getInputProcessors() {
  return [this]
}

보고:InputProcessorOrWorkflow[]

getOutputProcessors()

이 공급자를 Agent에 등록해야 하는 반환 출력 프로세서입니다. 공급자가 Agent 출력 단계를 가로채는 경우 재정의합니다.

getOutputProcessors() {
  return [this]
}

보고:OutputProcessorOrWorkflow[]

getTools()

이 공급자가 Agent에 노출하는 반환 Tool입니다. 공급자가 구독 또는 구독 취소 명령과 같은 Agent 호출 가능 Tool을 추가하면 재정의됩니다.

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

보고:Record<string, unknown>

구독 추적

subscribe(target, externalResourceId, metadata?)

외부 리소스에 대한 스레드를 구독합니다. 이는 보호된 메서드입니다. 공급자 구현 내에서 호출하세요.

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

반환: SignalSubscription: 생성된 구독 또는 메타데이터가 병합된 기존 구독입니다.

target:

SignalProviderTarget

구독할 스레드입니다. threadId와 resourceId를 포함해야 합니다.

externalResourceId:

string

외부 리소스의 Provider별 식별자입니다(예: "github:owner/repo#123").

metadata?:

Record<string, unknown>

구독과 함께 저장할 추가 데이터입니다. 중복 구독 시 기존 메타데이터에 병합됩니다.

unsubscribe(target, externalResourceId)

구독을 제거합니다.

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

반환: boolean: 제거되었으면 true, 일치하는 구독이 없었으면 false입니다.

getSubscriptions()

이 공급자에 대한 모든 활성 구독을 반환합니다.

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)

특정 스레드에 대한 모든 구독을 반환합니다.

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)

스레드에 대한 모든 구독을 제거합니다.

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

반환: number: 제거된 구독 수입니다.

subscriptionCount

이 공급자에 대한 총 활성 구독 수입니다.

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

웹훅

handleWebhook(request)

들어오는 웹훅 요청을 처리합니다. 페이로드를 구문 분석하고 구독과 일치시킨 다음 알림 신호를 보내도록 재정의하세요. 바로 사용할 수 있는 구현은 WebhookSignalProvider를 참조하세요. 웹훅 요청을 확인한 후 애플리케이션 정의 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에게 알림 신호를 보냅니다. 이것은 보호된 편의 포장지입니다.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

알림 페이로드입니다.

object

source:

string

알림 소스의 식별자입니다.

kind:

string

이벤트 유형입니다(예: "pr-updated", "new-message").

summary:

string

사람이 읽을 수 있는 이벤트 요약입니다.

priority?:

"high" | "medium" | "low"

알림 우선순위입니다.

payload?:

unknown

알림에 첨부된 임의의 데이터입니다.

target:

SignalProviderTarget

알림을 보낼 스레드입니다. threadId와 resourceId를 포함해야 합니다.

유형

SignalSubscription

에서 반환한 구독 객체subscribe().

id:

string

구독의 고유 식별자입니다.

providerId:

string

이 구독을 소유한 Provider입니다.

threadId:

string

신호를 수신하는 스레드입니다.

resourceId:

string

스레드를 소유한 리소스입니다.

externalResourceId:

string

외부 리소스의 Provider별 식별자입니다(예: "github:owner/repo#123").

subscribedAt:

Date

구독이 생성된 시점입니다.

metadata:

Record<string, unknown>

구독과 함께 저장된 Provider별 메타데이터입니다.

SignalProviderTarget

특정 Agent 스레드를 식별합니다.

threadId:

string

대상으로 지정할 스레드입니다.

resourceId:

string

스레드를 소유한 리소스입니다.

agentId?:

string

Agent 식별자입니다.

타입 가드

isSignalProvider(obj)

런타임 확인SignalProvider instances.

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

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

보고:boolean