신호 제공자 구축

이 가이드에서는 일정 간격으로 외부 서비스를 폴링하고 감시된 리소스가 변경될 때마다 Agent 스레드에 알림을 푸시하는 신호 공급자를 구축합니다. 연장하는 방법을 배우게 됩니다.SignalProvider기본 클래스, 구독 추적, 폴 루프에서 알림 보내기, Agent에 공급자 등록 등이 있습니다.

예제에서는 가짜 CI 서비스에서 빌드 파이프라인을 감시하지만 패턴은 문제 추적기, 상태 API, 대기열 또는 자체 백엔드 등 모든 풀 기반 소스에 적용됩니다.

:::실험적

신호 제공자는 베타 버전입니다. API가 안정될 때까지 주요 버전 변경 없이 주요 변경 사항이 발생할 수 있습니다.

:::

전제조건

• Node.js v22.13.0 이상 설치
• 지원되는 Model Provider의 API 키
• 기존 Mastra 프로젝트. 필요한 경우 설치 가이드를 따르세요. 이 가이드는 신호를 개괄적으로 이해하고 있다고 가정합니다. 전체 API 범위는 SignalProvider 레퍼런스를 참조하세요.

알림 저장소 추가

신호 Provider는 알림 신호를 스레드로 푸시하며, 알림에는 알림 도메인을 지원하는 스토리지 어댑터가 필요합니다. Mastra 인스턴스에서 스토리지를 구성하세요.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { LibSQLStore } from '@mastra/libsql'

export const mastra = new Mastra({
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: 'file:./mastra.db',
  }),
})

LibSQL, PostgreSQL, MongoDB는 모두 알림 레코드를 지원합니다. 알림 스토리지가 없으면 Provider의 notify() 호출이 런타임에 오류를 발생시킵니다.

외부 서비스 클라이언트 생성

실제 공급자는 외부 API를 호출합니다. 이 가이드를 독립적으로 유지하려면 파이프라인의 빌드 상태를 반환하는 작은 가짜 CI 클라이언트를 생성하세요. 나중에 실제 API 클라이언트로 바꾸세요.

src/mastra/signals/ci-client.ts

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

// Returns a random status so you can see notifications fire while testing.
export async function fetchBuildStatus(pipeline: string): Promise<BuildStatus> {
  const states: BuildStatus['status'][] = ['passed', 'failed', 'running']
  const status = states[Math.floor(Math.random() * states.length)]!
  return { id: `build_${Date.now()}`, pipeline, status }
}

이 클라이언트는 호출할 때마다 무작위 상태를 반환합니다. 실제 API에 연결하면 이 파일만 변경됩니다.

신호 제공자 구축

SignalProvider를 확장하고 추상 id 필드를 구현한 다음 pollInterval을 설정하고 poll()을 재정의하세요. 기본 클래스는 활성 구독마다 지정된 간격으로 poll()을 호출합니다. 관심 있는 빌드에 대해서만 알림을 내보내세요.

src/mastra/signals/ci-signals.ts

import { SignalProvider } from '@mastra/core/signals'
import type { SignalProviderTarget, SignalSubscription } from '@mastra/core/signals'
import { fetchBuildStatus } from './ci-client'

export class CiSignals extends SignalProvider<'ci-signals'> {
  readonly id = 'ci-signals' as const
  readonly pollInterval = 10_000 // poll every 10 seconds

  // Public API so callers can subscribe a thread to a pipeline.
  watch(target: SignalProviderTarget, pipeline: string): SignalSubscription {
    return this.subscribe(target, pipeline)
  }

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

  async poll(subscriptions: SignalSubscription[]): Promise<void> {
    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 },
      )
    }
  }
}

참고할 몇 가지 사항:

• subscribe()와 unsubscribe()는 기본 클래스에서 protected입니다. 호출자가 구독을 관리할 수 있도록 자체 public 메서드(watch / unwatch)로 감싸세요.
• externalResourceId는 Provider별 문자열입니다. 파이프라인 이름일 수 있으며, GitHub Provider에서는 "github:owner/repo#123"과 같은 값을 사용할 수 있습니다.
• notify()는 연결된 Agent의 스레드에 알림 신호를 전달합니다. Provider가 Agent에 등록된 적이 없으면 오류가 발생합니다.
• dedupeKey는 동일한 오류가 두 번 저장되는 것을 방지합니다.

Agent에 공급자 등록

signals를 통해 Provider를 Agent에 전달하세요. Agent가 Provider를 연결하고 폴링 루프를 자동으로 시작합니다.

src/mastra/agents/dev-agent.ts

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

export const ciSignals = new CiSignals()

export const devAgent = new Agent({
  id: 'dev-agent',
  name: 'Dev Agent',
  instructions: 'Help the user triage CI build failures.',
  model: 'openai/gpt-5.6-sol',
  signals: [ciSignals],
})

Mastra에 Agent를 등록하고 첫 번째 단계부터 스토리지를 추가합니다.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { LibSQLStore } from '@mastra/libsql'
import { devAgent } from './agents/dev-agent'

export const mastra = new Mastra({
  agents: { devAgent },
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: 'file:./mastra.db',
  }),
})

스레드 구독

Provider는 스레드가 감시하는 리소스만 폴링합니다. poll()이 확인할 대상이 있도록 스레드를 파이프라인에 구독시키세요.

src/mastra/subscribe.ts

import { ciSignals } from './agents/dev-agent'

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

예를 들어 설정 스크립트 또는 API 경로에서 Agent가 등록된 후 이 작업을 한 번 실행합니다. 구독은 공급자의 Memory 내 레지스트리에 있으므로 다시 시작한 후에 다시 구독하세요.

신호 제공자 테스트

개발 서버를 시작합니다:

npm

npm run dev

pnpm

pnpm run dev

Yarn

yarn dev

Bun

bun run dev

스레드가 구독되었는지 확인한 다음 로그를 살펴보세요. 가짜 클라이언트는 각 폴링마다 무작위 상태를 반환하므로 몇 주기 내에 실패한 빌드가 구독 스레드에 대한 알림을 트리거하는 것을 볼 수 있습니다.

Agent가 알림에 반응하는 것을 보려면 스레드를 구독하고 스트리밍하세요.

src/mastra/watch-thread.ts

const subscription = await devAgent.subscribeToThread({
  resourceId: 'user_123',
  threadId: 'thread_456',
})

for await (const chunk of subscription.stream) {
  console.log(chunk)
}

빌드가 실패하면 Model은 알림을 컨텍스트로 받습니다.

<notification source="ci-signals" type="ci-status" priority="high" status="delivered">Build failed for acme/app:main</notification>

가짜 클라이언트는 상태를 무작위로 지정하고 Model은 응답을 자유롭게 표현하므로 출력은 비결정적이므로 정확한 표현은 다양합니다.

다음 단계

이 신호 제공자를 다음으로 확장할 수 있습니다.

• fetchBuildStatus()를 실제 API 클라이언트로 교체하세요.

• 다시 시작한 후에도 유지되도록 구독을 저장한 다음 start()에서 다시 불러오세요.

• 푸시 기반 소스를 위해 handleWebhook()을 사용하는 webhook 진입점을 추가하세요.

• Agent가 자체 구독을 관리할 수 있도록 getTools()를 통해 subscribe 및 unsubscribe Tool을 노출하세요.

• 알림에 중복 제거 또는 일괄 처리가 필요하면 dedupeKey와 coalesceKey를 사용하세요. 자세히 알아보기:

• 신호 제공자 구축

• 신호

• SignalProvider참조

• WebhookSignalProvider참조