DurableAgent
DurableAgent は既存の Agent を耐久実行と再開可能な Stream でラップします。クライアントがイベントを失わずに切断・再接続できるよう Agentic Loop を実行し、イベントを PubSub 経由でストリーミングします。Run を単一リクエストより長く継続させる場合や、接続切断後も維持する必要がある場合に使用します。
createDurableAgent ファクトリで作成します。組み込み Workflow Engine で Fire-and-forget 実行するには createEventedAgent を使用します。Inngest を利用する実行には、@mastra/inngest の createInngestAgent を使用します。
使用例使用例への直接リンク
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'
const agent = new Agent({
id: 'my-agent',
name: 'My Agent',
instructions: 'You are a helpful assistant',
model: 'openai/gpt-5.6-sol',
})
const durableAgent = createDurableAgent({ agent })
export const mastra = new Mastra({
agents: { myAgent: durableAgent },
})
レスポンスをストリーミングして結果を読み取ります。Run の使用を終えたら、cleanup 関数で PubSub の購読を解除します。
const { output, runId, cleanup } = await durableAgent.stream('Hello!')
const text = await output.text
cleanup()
durable 設定フラグの使用using-the-durable-config-flagへの直接リンク
AgentConfig に durable: true を設定すると、Agent を Mastra インスタンスへ登録したときに createDurableAgent で自動的にラップされます。cache、pubsub、maxSteps、cleanupTimeoutMs などの高度なオプションを渡すにはオブジェクトを使います。
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
const myAgent = new Agent({
id: 'my-agent',
name: 'My Agent',
instructions: 'You are a helpful assistant',
model: 'openai/gpt-5.6-sol',
durable: true, // or: { maxSteps: 10, cleanupTimeoutMs: 60_000 }
})
export const mastra = new Mastra({
agents: { myAgent },
})
mastra.getAgent('myAgent') はラップされた DurableAgent を返します。スタンドアロン Agent（作成後に Mastra インスタンスへ登録されていない Agent）は Durable になりません。ラップは登録時に適用されます。
createDurableAgent(options)createdurableagentoptionsへの直接リンク
Agent を耐久実行と再開可能な Stream でラップします。DurableAgent の推奨作成方法です。
import { createDurableAgent } from '@mastra/core/agent/durable'
const durableAgent = createDurableAgent({ agent })
戻り値： DurableAgent
パラメータパラメータへの直接リンク
agent:
id?:
name?:
cache?:
false を設定します。pubsub?:
maxSteps?:
createEventedAgent(options)createeventedagentoptionsへの直接リンク
組み込み Workflow Engine 上で Agent を Fire-and-forget の耐久実行でラップします。createDurableAgent と同様にストリーミング可能な結果を返しますが、基盤の Workflow は Stream の接続前に完了まで実行されず、startAsync によりノンブロッキングで動作します。呼び出し元とは独立して Run を進める場合に使用します。id と name の上書きは受け付けません。
import { createEventedAgent } from '@mastra/core/agent/durable'
const eventedAgent = createEventedAgent({ agent })
戻り値： EventedAgent (a subclass of DurableAgent)
パラメータパラメータへの直接リンク
agent:
cache?:
false を設定します。pubsub?:
maxSteps?:
コンストラクターのパラメータコンストラクターのパラメータへの直接リンク
DurableAgent クラスは、createDurableAgent と同じオプションに加えて cleanupTimeoutMs を受け取ります。サブクラス化が必要な場合を除き、ファクトリを推奨します。
agent:
id?:
name?:
cache?:
false を設定します。pubsub?:
maxSteps?:
cleanupTimeoutMs?:
cleanup() の手動呼び出しを必須にするには 0 を設定します。Suspend イベントでは自動 Cleanup は実行されません。メソッドメソッドへの直接リンク
実行実行への直接リンク
stream(messages, options?)streammessages-optionsへの直接リンク
耐久実行でレスポンスをストリーミングします。Run の進行に応じて output がイベントを生成する結果をすぐに返します。
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<DurableAgentStreamResult>
resume(runId, resumeData, options?)resumerunid-resumedata-optionsへの直接リンク
Tool の承認後などに Suspend された Run を再開します。元の Stream の runId と、Run が待機していたデータを渡します。Run の Registry エントリが存在しない場合はエラーをスローします。
const { output, cleanup } = await durableAgent.resume(runId, {
approved: true,
})
await output.text
cleanup()
戻り値： Promise<DurableAgentStreamResult>
observe(runId, options?)observerunid-optionsへの直接リンク
既存の Run へ再接続し、ライブイベントを配信する前に Cache 済みイベントを再生します。ネットワーク切断後に使用します。既知の位置から再生を始めるには offset を渡します。
const { output, cleanup } = await durableAgent.observe(runId, {
offset: 0,
onChunk: chunk => console.log(chunk),
})
await output.text
デフォルトでは、observe() はイベントを無期限に待機します。Run を実行するプロセスが予期せず停止すると、Run はイベント生成を止めても完了イベントを送出しないため、監視中の Stream は永遠に待機します。待機時間を制限するには idleTimeoutMs を渡します。指定したミリ秒の間イベントがなければ Stream が終了します。最初に任意の isAlive 検査が実行されます。長時間の Tool 呼び出し中や、人の入力を待って一時停止している場合など、Run の処理が続いている間は true を返すと待機を継続します。false を返すか isAlive を省略すると、Stream はエラーで終了します。isAlive が一時的に例外をスローしても「稼働中」と見なされるため、一時的な検査失敗でライブ Stream が終了することはありません。
const { output } = await durableAgent.observe(runId, {
idleTimeoutMs: 30_000,
isAlive: () => runHeartbeat.isFresh(runId),
})
アイドルタイムアウトで Run を終了すると、エラー終了時と同じ Cleanup（以下の警告を参照）が実行され、Cache 状態は保持されず解放されます。どちらのオプションも明示的に有効化します。従来どおり無期限に待機するには省略してください。
戻り値： Promise<DurableAgentStreamResult>
observe() が返す cleanup() は、Run の Registry エントリと Cache 済みイベントを破棄します。Run の使用を終えたときだけ呼び出してください。Run が Suspend 中で後から再開する場合は、cleanup() を呼び出さないでください。Run の完了またはエラー後に自動 Cleanup Timer へ処理を任せます。Suspend イベントでは自動 Cleanup は実行されません。
prepare(messages, options?)preparemessages-optionsへの直接リンク
Run を開始せずに耐久実行の準備をします。Run を内部 Registry へ登録し、シリアライズ済み Workflow 入力を返します。Workflow を起動するタイミングと方法を制御する場合に使用します。
const { runId, messageId, workflowInput, threadId, resourceId } = await durableAgent.prepare(
'Summarize the document',
{
memory: { threadId: 'thread-1', resourceId: 'user-1' },
},
)
戻り値：
interface PrepareResult {
runId: string
messageId: string
workflowInput: any
registryEntry: object
threadId?: string
resourceId?: string
}
復旧復旧への直接リンク
recoverActiveRuns(options?)recoveractiverunsoptionsへの直接リンク
この Agent で running ステータスのまま停止している Run を検出し、最後に永続化された Snapshot から再実行します。最大 options.limit 件（デフォルト：100）の Run を復旧し、復旧結果の概要を返します。
const result = await durableAgent.recoverActiveRuns()
// { recovered: [{ runId, status }], succeeded: 2, failed: 0 }
既知の Run を1件復旧するには runId を渡します。
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })
戻り値：
interface DurableAgentRecoverActiveRunsResult {
recovered: Array<{ runId: string; status: 'success' | 'failed'; error?: Error }>
succeeded: number
failed: number
}
options.runId?:
options.limit?:
options.createdBefore?:
recover(runId, options?)recoverrunid-optionsへの直接リンク
ID を指定して1件の Run を復旧します。stream() と同じ形式のストリーミング可能な結果を返します。復旧 Stream をリアルタイムで監視する場合に使用します。
const { output, cleanup } = await durableAgent.recover('run-abc-123', {
onChunk: chunk => console.log(chunk),
onError: ({ error }) => console.error(error),
})
await output.text
cleanup()
戻り値： Promise<DurableAgentStreamResult>
Stream オプションStream オプションへの直接リンク
stream() は DurableAgentStreamOptions オブジェクトを受け取ります。以下の Agent 実行オプションとライフサイクル Callback に対応します。
runId?:
resume() または observe() で使用します。instructions?:
context?:
memory?:
requestContext?:
maxSteps?:
toolsets?:
clientTools?:
toolChoice?:
activeTools?:
modelSettings?:
Authorization、X-Api-Key など）は、シリアライズ済み Snapshot がプロセス境界を越える前に削除されます。stopWhen?:
maxSteps のみへ縮退します。system?:
requireToolApproval?:
true または false、呼び出しごとの Policy には関数を渡します。関数形式の Policy はプロセス内 Run Registry に保持されます。プロセス間の再開時は true の Shadow へフォールバックします。autoResumeSuspendedTools?:
resume() 呼び出しを待たず、Suspend した Tool を自動的に再開します。toolCallConcurrency?:
includeRawChunks?:
maxProcessorRetries?:
structuredOutput?:
untilIdle?:
true、カスタマイズするには { maxIdleMs } を渡します。非推奨の streamUntilIdle() メソッドと同等です。resume() でも使用できます。disableBackgroundTasks?:
tracingOptions?:
requestContextKeys。完全に JSON シリアライズ可能です。actor?:
transform?:
transformToolPayload Closure はプロセス内 Run Registry に保持され、JSON-safe な targets Shadow だけがシリアライズされます。prepareStep?:
PrepareStepProcessor として呼び出される Step 単位の準備 Hook。Closure のみで、プロセス内 Run Registry に保存されます。プロセス間の再開では Hook が失われます。isTaskComplete?:
onComplete はプロセス内 Run Registry に保持されます。JSON-safe な Primitive（strategy、timeout、parallel、suppressFeedback、scorerNames）はプロセス間の可観測性のためにシリアライズされます。delegation?:
onDelegationStart、onDelegationComplete、messageFilter）。Callback は準備時に Sub-agent Tool Wrapper へ組み込まれます。プロセス間の再開では Callback が失われます。versions?:
abortSignal?:
AbortController へ転送されるため、どちらからでも Run をキャンセルできます。プロセス間の再開では Signal を復元できません。再開後にも中止可能にするには、新しい Signal を resume() へ渡します。onChunk?:
onStepFinish?:
onFinish?:
onError?:
onSuspended?:
onAbort?:
abortSignal または result.abort() で Run が中止されたときに呼び出されます。onIterationComplete?:
messageList、finishReason、isFinal フラグとともに呼び出されます。Durable Agent では監視専用で、continue: false や Feedback を返しても Loop に影響しません。resume() と observe() は同じライフサイクル Callback（onChunk、onStepFinish、onFinish、onError、onSuspended）を受け取ります。observe() は再生開始位置を制御する offset も受け取ります。
DurableAgentStreamResultDurableAgentStreamResultへの直接リンク
stream()、resume()、observe()、recover() が返すオブジェクトです。
interface DurableAgentStreamResult<OUTPUT = undefined> {
output: MastraModelOutput<OUTPUT>
readonly fullStream: ReadableStream<any>
runId: string
threadId?: string
resourceId?: string
cleanup: () => void
abort: () => void
}
output:
output.text を await し、または output.fullStream を消費します。fullStream:
output.fullStream へ委譲する完全なイベント Stream。runId:
resume() または observe() へ渡します。threadId?:
resourceId?:
cleanup:
abort:
AbortController を切り替えて Run を中止します。Durable LLM 実行 Step 内では AbortError として表面化し、onAbort Callback が発火します。Run 完了後に呼び出しても安全で、その場合は何も行いません。