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

# `createInngestAgent()`

`createInngestAgent()` は、既存の [`Agent`](https://mastra.zisheng.pro/ja/reference/agents/agent) を [Inngest](https://www.inngest.com/docs) による耐久実行でラップします。[`createDurableAgent()`](https://mastra.zisheng.pro/ja/reference/agents/durable-agent) と同様に、[PubSub](https://mastra.zisheng.pro/ja/docs/server/pubsub) 経由でイベントをストリーミングし、再開可能な Stream をサポートします。ただし、Agentic Loop はプロセス内ではなく Inngest の実行 Engine 上で動作します。プロセスの再起動後も Run を継続する必要がある場合や、分散環境で実行する場合に使用します。

プロセス内で耐久実行するには、[`createDurableAgent()`](https://mastra.zisheng.pro/ja/reference/agents/durable-agent) を使用します。組み込み Workflow Engine で Fire-and-forget 実行するには、[`createEventedAgent()`](https://mastra.zisheng.pro/ja/reference/agents/durable-agent) を使用します。

## 使用例

Inngest Client を設定し、Agent をラップして Mastra に登録したうえで、Inngest の Serve Endpoint を公開します。

```typescript
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
import { createInngestAgent, serve as inngestServe } from '@mastra/inngest'
import { Inngest } from 'inngest'

const inngest = new Inngest({ id: 'my-app' })

const agent = new Agent({
  id: 'my-agent',
  name: 'My Agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
})

const durableAgent = createInngestAgent({ agent, inngest })

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  server: {
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => inngestServe({ mastra, inngest }),
      },
    ],
  },
})
```

レスポンスをストリーミングし、結果を読み取ります。

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!')

const text = await output.text

cleanup()
```

## `createInngestAgent(options)`

`Agent` を Inngest による耐久実行と再開可能な Stream でラップします。

```typescript
import { createInngestAgent } from '@mastra/inngest'

const durableAgent = createInngestAgent({ agent, inngest })
```

戻り値：[`InngestAgent`](#inngestagent-interface)

### パラメーター

**agent** (`Agent`): Inngest の耐久実行でラップする Agent。InngestAgent に実装されていないメソッド（listTools() や getMemory() など）は、Proxy を介してこの Agent に委譲されます。

**inngest** (`Inngest`): Inngest Client インスタンス。Workflow イベントの送信と、SDK v4 ではリアルタイム Stream イベントの発行に使用します。

**id** (`string`): ID の上書き値。 (Default: `agent.id`)

**name** (`string`): 名前の上書き値。 (Default: `agent.name`)

**pubsub** (`PubSub`): イベントをストリーミングする PubSub インスタンス。デフォルトの InngestPubSub は Inngest Realtime を使用するため、プロセスをまたいで動作します。 (Default: `InngestPubSub`)

**cache** (`MastraServerCache`): 保存済み Stream イベントの Cache。Stream の再開を可能にします。指定すると、PubSub は自動的に CachingPubSub でラップされます。省略すると、Agent は Mastra インスタンスの Cache を継承します。

**mastra** (`Mastra`): Observability に使用する Mastra インスタンス。Agent を Mastra に登録すると自動的に設定されます。

## `InngestAgent` インターフェース

`createInngestAgent()` が返すオブジェクトです。以下の耐久実行メソッドを提供します。明示的に定義されていないプロパティやメソッド（`listTools()` や `getMemory()` など）は、Proxy を介して基になる Agent に転送されます。

### プロパティ

**id** (`string`): Agent の ID。

**name** (`string`): Agent の名前。

**agent** (`Agent`): 基になる Mastra Agent。

**inngest** (`Inngest`): Inngest Client。

**cache** (`MastraServerCache | undefined`): 再開可能な Stream が有効な場合に、解決された Cache インスタンス。

**pubsub** (`PubSub`): イベントのストリーミングに使用する PubSub インスタンス。

## メソッド

### 実行

#### `stream(messages, options?)`

Inngest の耐久実行 Engine を使用してレスポンスをストリーミングします。PubSub の購読が確立された後、Inngest イベントによって Workflow がトリガーされます。

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!', {
  onChunk: chunk => console.log(chunk),
  onFinish: result => console.log('done', result),
})

const text = await output.text
cleanup()
```

戻り値：[`Promise<InngestAgentStreamResult>`](#inngestagentstreamresult)

#### `resume(runId, resumeData, options?)`

Tool の承認後などに、中断中の Inngest Run を再開します。Storage から Workflow Snapshot を読み込み、中断中の Step を見つけて Inngest に再開イベントを送信します。

```typescript
const { output, cleanup } = await durableAgent.resume(
  runId,
  {
    approved: true,
  },
  { threadId: 'thread-1', resourceId: 'user-1' },
)

await output.text
cleanup()
```

第3引数では、Lifecycle Callback に加えて `threadId` と `resourceId` を指定できます。

**threadId** (`string`): 再開した Run に関連付ける Thread ID。

**resourceId** (`string`): 再開した Run に関連付ける Resource ID。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): ストリーミングされた Chunk ごとに呼び出されます。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): Agentic Loop の Step が完了したときに呼び出されます。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): Run が完了したときに呼び出されます。

**onError** (`(error: Error) => void | Promise<void>`): Run でエラーが発生したときに呼び出されます。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): Run が中断したときに呼び出されます。

戻り値：[`Promise<InngestAgentStreamResult>`](#inngestagentstreamresult)

#### `generate(messages, options?)`

Inngest の耐久実行 Engine 上でレスポンスを実行し、単一の `FullOutput` に解決します。Run が中断した場合、`generate()` は `finishReason: 'suspended'` を返します。`runId` オプションは省略可能です。省略すると、`generate()` が Run ID を作成して `result.runId` で返します。Run を再開するには、[`resumeGenerate()`](#resumegeneraterunid-resumedata-options) を使用します。呼び出し元で中断時の Callback が必要な場合は、`onSuspended` を指定して [`stream()`](#streammessages-options) を使用します。

```typescript
const result = await durableAgent.generate('Delete the old records', {
  requireToolApproval: true,
})

result.runId // Generated automatically
result.finishReason // 'suspended' when approval is required
```

戻り値：`Promise<FullOutput<TOutput>>`

#### `resumeGenerate(runId, resumeData, options?)`

中断中の `generate()` Run を再開し、単一の `FullOutput` に解決します。

```typescript
if (!result.runId) {
  throw new Error('Run ID is missing')
}

const resumedResult = await durableAgent.resumeGenerate(result.runId, { approved: true })
```

戻り値：`Promise<FullOutput<TOutput>>`

#### `observe(runId, options?)`

既存の Run に再接続し、Cache 済みイベントを再生してから Live イベントを配信します。Network が切断された後に使用します。既知の位置から再生を開始するには `offset` を渡します。

```typescript
const { output, cleanup } = await durableAgent.observe(runId, {
  offset: 0,
  onChunk: chunk => console.log(chunk),
})

await output.text
```

`observe()` の結果には `threadId` と `resourceId` は含まれません。

戻り値：`Promise<Omit<InngestAgentStreamResult, 'threadId' | 'resourceId'>>`

> **警告:** `observe()` が返す `cleanup()` は、Run の Registry Entry と Cache 済みイベントを削除します。Run の使用を終えた場合にのみ呼び出してください。Run が中断中で、後から再開する場合は `cleanup()` を呼び出さないでください。

#### `prepare(messages, options?)`

トリガーせずに、耐久実行用の Run を準備します。Inngest Workflow イベントを手動でトリガーするために使用できる、Serialize 済みの Workflow Input を返します。

```typescript
const { runId, messageId, workflowInput, threadId, resourceId } = await durableAgent.prepare(
  'Summarize the document',
  {
    memory: { threadId: 'thread-1', resourceId: 'user-1' },
  },
)
```

戻り値：

```typescript
interface PrepareResult {
  runId: string
  messageId: string
  workflowInput: any
  threadId?: string
  resourceId?: string
}
```

### イントロスペクション

#### `isInngestAgent(obj)`

オブジェクトが `InngestAgent` かどうかを確認する Type Guard です。

```typescript
import { isInngestAgent } from '@mastra/inngest'

if (isInngestAgent(agent)) {
  // agent is InngestAgent
}
```

戻り値：`boolean`

## Stream オプション

`stream()` は `InngestAgentStreamOptions` オブジェクトを受け取ります。[`DurableAgent.stream()`](https://mastra.zisheng.pro/ja/reference/agents/durable-agent) と同じ Agent 実行オプションに加えて、Lifecycle Callback をサポートします。

**runId** (`string`): この Run の一意の識別子。後から resume() または observe() で使用します。

**instructions** (`AgentExecutionOptions['instructions']`): この Run で使用する Agent のデフォルト instructions を上書きします。

**context** (`ModelMessage[]`): Agent に渡す追加の Context Message。

**memory** (`object`): Conversation の永続化と取得に使用する Memory 設定。

**requestContext** (`RequestContext`): この Run の動的な設定と状態を保持する Request Context。

**maxSteps** (`number`): 実行する最大 Step 数。

**toolsets** (`object`): この Run で利用できる追加の Tool Set。

**clientTools** (`object`): 実行中に利用できる Client 側の Tool。

**toolChoice** (`'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }`): Tool の選択方法。

**modelSettings** (`object`): Temperature など、Model 固有の設定。

**requireToolApproval** (`boolean`): すべての Tool Call で承認を必須にし、再開されるまで Run を中断します。

**autoResumeSuspendedTools** (`boolean`): 外部からの resume() 呼び出しを待たず、中断した Tool を自動的に再開します。

**toolCallConcurrency** (`number`): 同時に実行する Tool Call の最大数。

**includeRawChunks** (`boolean`): Provider の Raw Chunk を Stream 出力に含めます。

**maxProcessorRetries** (`number`): Generation ごとの Processor の最大 Retry 回数。

**untilIdle** (`boolean | { maxIdleMs?: number }`): 設定すると、Agent がアイドル状態になるまで、バックグラウンドタスクの継続処理をまたいで Stream を開いたままにします。デフォルトの5分間のアイドルタイムアウトを使用するには true を渡し、カスタマイズするには { maxIdleMs } を渡します。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): ストリーミングされた Chunk ごとに呼び出されます。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): Agentic Loop の Step が完了したときに呼び出されます。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): Run が完了したときに呼び出されます。

**onError** (`(error: Error) => void | Promise<void>`): Run でエラーが発生したときに呼び出されます。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): Tool の承認時など、Run が中断したときに呼び出されます。

`observe()` は Lifecycle Callback（`onChunk`、`onStepFinish`、`onFinish`、`onError`、`onSuspended`）に加えて、再生開始位置を制御する `offset` を受け取ります。

## `InngestAgentStreamResult`

`stream()` と `resume()` が返すオブジェクトです。`observe()` メソッドも同じ形式を返しますが、`threadId` と `resourceId` は含まれません。

```typescript
interface InngestAgentStreamResult<OUTPUT = undefined> {
  output: MastraModelOutput<OUTPUT>
  readonly fullStream: ReadableStream<any>
  runId: string
  threadId?: string
  resourceId?: string
  cleanup: () => void
}
```

**output** (`MastraModelOutput`): Stream 出力。全文を取得するには output.text を await し、Stream を処理するには output.fullStream を使用します。

**fullStream** (`ReadableStream`): output.fullStream に委譲される、完全なイベント Stream。

**runId** (`string`): 一意の Run ID。再接続するには resume() または observe() に渡します。

**threadId** (`string`): Memory を使用する場合の Thread ID。

**resourceId** (`string`): Memory を使用する場合の Resource ID。

**cleanup** (`() => void`): PubSub の購読を解除し、Run の Registry Entry を削除します。Run の使用を終えたら呼び出してください。

## Inngest Function の提供

`@mastra/inngest` Package は、HTTP Framework に Inngest Workflow Function を登録するための `serve()` と `createServe()` を提供します。

### `serve(options)`

Mastra Workflow を Hono（デフォルトの Framework）で提供します。Mastra から Inngest ベースの Workflow をすべて収集し、Inngest Function として登録します。

```typescript
import { serve } from '@mastra/inngest'

app.use('/inngest/api', async c => {
  return serve({ mastra, inngest })(c)
})
```

### `createServe(adapter)`

任意の Inngest Serve Adapter（`inngest/express`、`inngest/fastify`、`inngest/next` など）を受け取り、その Framework 用の Serve Function を返す Factory です。

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/express'

const serveExpress = createServe(serve)
app.use('/inngest/api', serveExpress({ mastra, inngest }))
```

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/next'

const serveNext = createServe(serve)
export const { GET, POST, PUT } = serveNext({ mastra, inngest })
```

### Serve オプション

**mastra** (`Mastra`): 登録済みの Agent と Workflow を含む Mastra インスタンス。

**inngest** (`Inngest`): Inngest Client インスタンス。

**functions** (`InngestFunction.Like[]`): Mastra Workflow とともに提供する追加の Inngest Function。

**registerOptions** (`RegisterOptions`): Inngest の登録 Handler に渡すオプション。

## 関連項目

- [DurableAgent リファレンス](https://mastra.zisheng.pro/ja/reference/agents/durable-agent)
- [Agent クラス](https://mastra.zisheng.pro/ja/reference/agents/agent)
- [Inngest デプロイガイド](https://mastra.zisheng.pro/ja/guides/deployment/inngest)
- [PubSub](https://mastra.zisheng.pro/ja/docs/server/pubsub)