> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# SignalProvider

**新增于：** `@mastra/core@1.39.0`

用于构建 Signal Provider 的抽象基类。Signal Provider 监视外部来源（API、webhook、事件流），并通过内置订阅注册表将通知 Signal 推送至 Agent thread。

Signal Provider 默认不是处理器。需要拦截 Agent 执行的 Provider 会从 `getInputProcessors()` 或 `getOutputProcessors()` 返回处理器。暴露供 Agent 调用的 Tool 的 Provider 会从 `getTools()` 返回它们。

如需可直接使用的基于 webhook 的 Provider，请参阅 [`WebhookSignalProvider`](https://mastra.zisheng.pro/reference/signals/webhook-signal-provider)。

## 使用示例

每 30 秒检查一次 API 的轮询 Provider：

```typescript
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：

```typescript
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 的唯一标识符。子类必须将其实现为 readonly 属性。

**name** (`string`): Provider 的人类可读显示名称。

**pollInterval** (`number`): 以毫秒为单位的轮询间隔。设置后，框架会按此间隔调用 poll()。对于仅 webhook 的 Provider，请保留为 undefined 或 0。

**isConnected** (`boolean`): 此 Provider 是否已连接到 Agent。调用 connect() 后返回 true。内部用于在 Agent.\_\_fork() 期间跳过重新连接。

## 方法

### 连接

#### `connect(agent)`

由 Agent 构造函数调用。建立双向链接，以便 Provider 将 Signal 发送回 Agent。可重写此方法，以在链接建立后执行额外设置。务必调用 `super.connect(agent)`。

```typescript
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)`

当 Provider 的 Agent 注册到 Mastra 实例时调用。可重写此方法以访问 storage 或其他 Mastra 服务。务必调用 `super.__registerMastra(mastra)`。

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

### 处理器和 Tool 集成

#### `getInputProcessors()`

返回此 Provider 需要注册到 Agent 的输入处理器。当 Provider 要拦截 Agent 输入步骤时重写（例如注入上下文提示或检测 Tool 调用）。

```typescript
getInputProcessors() {
  return [this]
}
```

返回：`InputProcessorOrWorkflow[]`

#### `getOutputProcessors()`

返回此 Provider 需要注册到 Agent 的输出处理器。当 Provider 要拦截 Agent 输出步骤时重写。

```typescript
getOutputProcessors() {
  return [this]
}
```

返回：`OutputProcessorOrWorkflow[]`

#### `getTools()`

返回此 Provider 向 Agent 暴露的 Tool。当 Provider 添加可由 Agent 调用的 Tool（如订阅或取消订阅命令）时重写。

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

返回：`Record<string, unknown>`

### 订阅跟踪

#### `subscribe(target, externalResourceId, metadata?)`

为 thread 订阅外部资源。这是受保护的方法：请从 Provider 实现内部调用。

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

返回：`SignalSubscription`：创建的订阅，或具有已合并 metadata 的现有订阅。

**target** (`SignalProviderTarget`): 要订阅的 thread。必须包含 threadId 和 resourceId。

**externalResourceId** (`string`): 外部资源的 Provider 特定标识符（例如，"github:owner/repo#123"）。

**metadata** (`Record<string, unknown>`): 与订阅一起存储的附加数据。在重复订阅时合并到现有 metadata。

#### `unsubscribe(target, externalResourceId)`

移除订阅。

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

返回：`boolean`：移除时为 `true`；不存在匹配订阅时为 `false`。

#### `getSubscriptions()`

返回此 Provider 的所有活跃订阅。

```typescript
const allSubs = this.getSubscriptions()
```

返回：`SignalSubscription[]`

#### `getSubscriptionsForResource(externalResourceId)`

返回特定外部资源的所有订阅。

```typescript
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)`

返回特定 thread 的所有订阅。

```typescript
const subs = this.getSubscriptionsForThread({
  threadId: 'thread-1',
  resourceId: 'user-1',
})
```

返回：`SignalSubscription[]`

#### `hasSubscription(target, externalResourceId)`

检查订阅是否存在。

```typescript
if (this.hasSubscription(target, 'github:mastra-ai/mastra#123')) {
  // already subscribed
}
```

返回：`boolean`

#### `unsubscribeAll(target)`

移除一个 thread 的全部订阅。

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

返回：`number`：已移除订阅的数量。

#### `subscriptionCount`

此 Provider 的活跃订阅总数。

```typescript
if (this.subscriptionCount === 0) {
  // nothing to poll
}
```

返回：`number`

### 轮询

#### `poll(subscriptions)`

在每个轮询周期中使用所有活跃订阅调用。重写以检查外部来源并发出通知。框架会防止轮询周期重叠：若 `poll()` 调用耗时超过 `pollInterval`，则跳过下一周期。

```typescript
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()` 后调用。具有幂等性：多次调用没有影响。

```typescript
provider.startPolling()
```

#### `stopPolling()`

停止轮询计时器。

```typescript
provider.stopPolling()
```

### Webhooks

#### `handleWebhook(request)`

处理传入的 webhook 请求。重写以解析 payload 并将其与订阅匹配，然后发出通知 Signal。有关可直接使用的实现，请参阅 [`WebhookSignalProvider`](https://mastra.zisheng.pro/reference/signals/webhook-signal-provider)。

验证 webhook 请求后，从应用程序定义的 HTTP 端点调用此方法。

```typescript
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 实例可用时重写。

```typescript
async start() {
  await this.loadInitialState()
}
```

#### `stop()`

关闭时调用。默认实现会停止轮询并清除所有订阅。

```typescript
provider.stop()
```

### 通知

#### `notify(notification, target)`

向已连接的 Agent 发送通知 Signal。这是对 `agent.sendNotificationSignal()` 的受保护便捷封装。

```typescript
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`): 通知 payload。

**notification.source** (`string`): 通知来源的标识符。

**notification.kind** (`string`): 事件类型（例如，"pr-updated"、"new-message"）。

**notification.summary** (`string`): 事件的人类可读摘要。

**notification.priority** (`"high" | "medium" | "low"`): 通知优先级。

**notification.payload** (`unknown`): 附加到通知的任意数据。

**target** (`SignalProviderTarget`): 要通知的 thread。必须包含 threadId 和 resourceId。

## 类型

### `SignalSubscription`

由 `subscribe()` 返回的订阅对象。

**id** (`string`): 订阅的唯一标识符。

**providerId** (`string`): 拥有此订阅的 Provider。

**threadId** (`string`): 接收 Signal 的 thread。

**resourceId** (`string`): 拥有该 thread 的资源。

**externalResourceId** (`string`): 外部资源的 Provider 特定标识符（例如，"github:owner/repo#123"）。

**subscribedAt** (`Date`): 创建订阅的时间。

**metadata** (`Record<string, unknown>`): 与订阅一起存储的 Provider 特定 metadata。

### `SignalProviderTarget`

标识特定的 Agent thread。

**threadId** (`string`): 目标 thread。

**resourceId** (`string`): 拥有该 thread 的资源。

**agentId** (`string`): Agent 标识符。

## 类型守卫

### `isSignalProvider(obj)`

用于检查 `SignalProvider` 实例的运行时检查。

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

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

返回：`boolean`