Introducing Custom Signal Providers for Mastra Agents

Your agents can now react to real-time events using a custom signal provider. Connect to any external source — GitHub, Slack, CI pipelines, or your own APIs and databases — and wake idle agents so they can take action.

Third-party systems emit events in different ways. With a custom signal provider you can build a notification system suitable for the source. Listen for events, poll for updates, or handle webhook requests to capture event payloads.

Before custom signal providers, integrating with external event systems meant manually managing incoming events and thread subscriptions. With a custom signal provider, Mastra tracks the subscriptions and sends matched events as notification signals to subscribed agent threads.

A custom signal provider could expose .start() and .stop() for setup and teardown, .poll() or .handleWebhook() for ingestion, and .watch() and .unwatch() for subscription management.

Get started

Install the latest @mastra/core:

Terminal

npm install @mastra/core@latest

note

Requires @mastra/core@1.42.0 or later, added in PR #17577.

Postgres signal provider

The example below receives Postgres LISTEN events and forwards matched events to a subscribed thread using .notify():

src/mastra/signals/pg-notify.ts

import { SignalProvider } from "@mastra/core/signals";
import type { SignalProviderTarget, SignalSubscription, SignalProviderWebhookRequest } from "@mastra/core/signals";
import { Client } from "pg";
 
export class PgNotifySignalProvider extends SignalProvider<"pg-notify"> {
  readonly id = "pg-notify" as const;
  readonly channel = "mastra_events";
  readonly externalResourceId = "postgres";
 
  pgClient?: Client;
 
  async start() {
    this.pgClient = new Client({ connectionString: process.env.DATABASE_URL });
    await this.pgClient.connect();
    await this.pgClient.query(`LISTEN "${this.channel}"`);
 
    this.pgClient.on("notification", async (msg) => {
      if (!msg.payload) return;
      const payload = JSON.parse(msg.payload);
 
      // Route the event to every thread subscribed to this resource
      const subscriptions = this.getSubscriptionsForResource(this.externalResourceId);
 
      for (const subscription of subscriptions) {
        await this.notify(
          {
            source: this.id,
            kind: payload.event,
            priority: payload.priority,
            summary: `${payload.event}: ${payload.schema}.${payload.name}`,
            payload
          },
          {
            resourceId: subscription.resourceId,
            threadId: subscription.threadId,
            ifIdle: { behavior: "wake" }
          }
        );
      }
    });
  }
 
  stop() {
    void this.pgClient?.end();
  }
 
  async poll(subscriptions: SignalSubscription[]) {
    // Runs on an interval — for polling-based sources
  }
 
  async handleWebhook(request: SignalProviderWebhookRequest) {
    // Called from a custom api route — for webhook-based sources
  }
 
  async watch(target: SignalProviderTarget) {
    return this.subscribe(target, this.externalResourceId);
  }
 
  async unwatch(target: SignalProviderTarget) {
    return this.unsubscribe(target, this.externalResourceId);
  }
}
 
export const pgNotify = new PgNotifySignalProvider();

You can extend the SignalProvider class and configure the methods your source needs, for example:

Setup and teardown:

.start(): Opens external connections
.stop(): Closes external connections

Ingestion:

.poll(): Runs on an interval for pull sources
.handleWebhook(): Handles a webhook request for push sources

Delivery:

.notify(): Sends a matched event as a notification signal to a subscribed thread

Subscription:

.watch(): Subscribes a thread to receive signals
.unwatch(): Removes a thread's subscription

Postgres events

The trigger below runs pg_notify on the mastra_events channel whenever a table is dropped, sending the event name and priority in the payload:

CREATE OR REPLACE FUNCTION mastra_notify_table_dropped() RETURNS event_trigger AS $$
DECLARE obj record;
BEGIN
  FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects() LOOP
    IF obj.object_type = 'table' THEN
      PERFORM pg_notify('mastra_events', json_build_object(
        'event', 'table_dropped',
        'priority', 'urgent',
        'schema', obj.schema_name,
        'name', obj.object_name
      )::text);
    END IF;
  END LOOP;
END;
$$ LANGUAGE plpgsql;
 
CREATE EVENT TRIGGER mastra_on_drop_table ON sql_drop
EXECUTE FUNCTION mastra_notify_table_dropped();

Agent config

Pass the signal provider to your agent's signals:

src/mastra/agents/db-sentinel.ts

import { Agent } from "@mastra/core/agent";
import { pgNotify } from "../signals/pg-notify";
 
export const dbSentinel = new Agent({
  id: "db-sentinel",
  name: "DB Sentinel",
  instructions: /* ... */,
  model: "anthropic/claude-opus-4-8",
  signals: [pgNotify]
});

Application code

Call .watch() to subscribe a thread to the provider:

import { pgNotify } from "../signals/pg-notify";
 
pgNotify.watch({
  resourceId: "user_123", // the resource that owns the thread — or the agent's id when testing locally in Studio
  threadId: "db-sentinel-thread"
});

For more information and full configuration options, see:

Signal providers
Signals
SignalProvider reference
WebhookSignalProvider reference
.sendNotificationSignal() reference — priority delivery behavior
Custom API routes