Agent.stream()

.stream() メソッドは、高度な機能と柔軟な形式により、Agent からのレスポンスをリアルタイムでストリーミングできます。このメソッドはメッセージと省略可能なストリーミングオプションを受け取り、Mastra ネイティブ形式と AI SDK v5 以降の両方に対応した最新のストリーミング機能を提供します。

使用例

const stream = await agent.stream('message for agent')

情報

モデルの互換性: このメソッドは V2 モデル向けに設計されています。V1 モデルでは .streamLegacy() メソッドを使用してください。フレームワークはモデルのバージョンを自動的に検出し、不一致がある場合はエラーをスローします。

パラメーター

messages:

string | string[] | CoreMessage[] | AiMessageType[] | UIMessageWithMetadata[]

Agent に送信するメッセージ。単一の文字列、文字列の配列、または構造化されたメッセージオブジェクトを指定できます。

options?:

AgentExecutionOptions<Output, Format>

ストリーミング処理の省略可能な設定。

AgentExecutionOptions<Output, Format>

maxSteps?:

number

実行中に処理するステップの最大数。

scorers?:

MastraScorers | Record<string, { scorer: MastraScorer['name']; sampling?: ScoringSamplingConfig }>

実行結果に対して実行する評価 scorer。

scorer:

string

使用する scorer の名前。

sampling?:

ScoringSamplingConfig

scorer のサンプリング設定。

type:

'none' | 'ratio'

サンプリング戦略の種類。サンプリングを無効にするには 'none'、割合に基づくサンプリングには 'ratio' を使用します。

rate?:

number

サンプリング率（0〜1）。type が 'ratio' の場合は必須です。

onIterationComplete?:

(context: IterationCompleteContext) => { continue?: boolean; feedback?: string } | void | Promise<{ continue?: boolean; feedback?: string } | void>

各イテレーションの完了後に呼び出されるコールバック関数。進捗の監視、Agent を導くフィードバックの提供、実行の早期停止に使用します。コールバックは、現在のテキスト、Tool 呼び出し、終了理由など、イテレーションに関するコンテキストを受け取ります。

context.iteration:

number

現在のイテレーション番号（1 始まり）。

context.maxIterations:

number | undefined

許可される最大イテレーション数（設定されている場合）。

context.text:

string

このイテレーションのテキストレスポンス。

context.isFinal:

boolean

これが最後のイテレーションかどうか。

context.finishReason:

string

このイテレーションが終了した理由（例: 'stop'、'length'、'tool-calls'）。

context.toolCalls:

ToolCall[]

このイテレーションで行われた Tool 呼び出し。

context.messages:

MastraDBMessage[]

これまでに蓄積されたすべてのメッセージ。

return.continue?:

boolean

実行を早期に停止するには false に設定します。

return.feedback?:

string

Agent の次のイテレーションを導くフィードバックメッセージ。

isTaskComplete?:

IsTaskCompleteConfig

タスクが完了したかを検証する完了スコアリング設定。Mastra の評価 scorer を使用し、Agent のレスポンスが完了条件を満たすかを自動的に確認します。

scorers:

MastraScorer[]

タスクの完了を評価する scorer の配列。各 scorer は 0（失敗）または 1（成功）を返します。

strategy?:

'all' | 'any'

scorer の結果を組み合わせる戦略。'all' ではすべての scorer の成功が必要で、'any' では 1 つ以上の成功が必要です。

onComplete?:

(result: IsTaskCompleteRunResult) => void | Promise<void>

タスク完了チェックの終了時に呼び出されるコールバック。各 scorer のスコアを含む結果を受け取ります。

parallel?:

boolean

scorer を並列実行するかどうか。

timeout?:

number

すべての scorer の完了を待機する最大時間（ミリ秒）。

suppressFeedback?:

boolean

true の場合、完了チェックのフィードバックに印を付け、利用側が表示出力から非表示にできるようにします。フィードバックはチェックが失敗した場合にのみ、次のイテレーションを導くため会話へ追加されます。

delegation?:

DelegationConfig

サブ Agent への委譲設定。Agent が他の Agent にタスクを委譲するタイミングを制御および監視し、委譲の変更や拒否、supervisor を導くフィードバックの提供に使用します。

onDelegationStart?:

(context: DelegationStartContext) => DelegationStartResult | void | Promise<DelegationStartResult | void>

サブ Agent に委譲する前に呼び出されます。委譲パラメーターの変更、委譲自体の拒否、または context.requestContext の変更によるサブ Agent 実行の Request Context への項目追加に使用します。

onDelegationComplete?:

(context: DelegationCompleteContext) => { feedback?: string } | void | Promise<{ feedback?: string } | void>

サブ Agent への委譲が完了した後に呼び出されます。コンテキストには以降の実行を停止する bail() メソッドが含まれ、{ feedback } を返して supervisor の次のアクションを導くことができます。フィードバックは assistant メッセージとして supervisor の Memory に保存されます。

messageFilter?:

(context: MessageFilterContext) => MastraDBMessage[] | Promise<MastraDBMessage[]>

サブ Agent に委譲する前に呼び出されるコールバック関数。サブ Agent に渡すメッセージの絞り込みに使用します。

tracingContext?:

TracingContext

span の階層とメタデータに使用する Tracing コンテキスト。

returnScorerData?:

boolean

詳細なスコアリングデータをレスポンスに含めるかどうか。

onChunk?:

(chunk: ChunkType) => Promise<void> | void

ストリーミング中、各チャンクに対して呼び出されるコールバック関数。

onError?:

({ error }: { error: Error | string }) => Promise<void> | void

ストリーミング中にエラーが発生したときに呼び出されるコールバック関数。

onAbort?:

(event: any) => Promise<void> | void

ストリームが中止されたときに呼び出されるコールバック関数。

abortSignal?:

AbortSignal

Agent の実行を中止するための Signal オブジェクト。Signal が中止されると、Agent が委譲した実行中のサブ Agent を含む、進行中のすべての処理が終了します。

activeTools?:

Array<keyof ToolSet> | undefined

実行中に使用できる有効な Tool 名の配列。

prepareStep?:

PrepareStepFunction<any>

複数ステップの実行で各ステップの前に呼び出されるコールバック関数。

context?:

ModelMessage[]

Agent に提供する追加のコンテキストメッセージ。

structuredOutput?:

StructuredOutputOptions<S extends ZodTypeAny = ZodTypeAny>

構造化出力の生成を微調整するオプション。

schema:

StandardJSONSchemaV1

期待される出力構造を定義する標準 JSON Schema。

model?:

MastraLanguageModel

構造化出力の生成に使用する Language Model。指定すると、Agent は Tool 呼び出し、テキスト、構造化出力を含む複数ステップで応答できます。

errorStrategy?:

'strict' | 'warn' | 'fallback'

schema 検証エラーを処理する戦略。'strict' はエラーをスローし、'warn' は警告をログに記録し、'fallback' はフォールバック値を使用します。

fallbackValue?:

<S extends ZodTypeAny>

schema の検証に失敗し、errorStrategy が 'fallback' の場合に使用するフォールバック値。

instructions?:

string

構造化出力モデルへの追加指示。

jsonPromptInjection?:

boolean | 'system' | 'inline' | 'auto'

JSON schema をモデルに渡す方法を制御します。'auto' に設定すると、対応している場合はネイティブの構造化出力を使用し、それ以外の場合はインラインのプロンプト注入を使用します。

providerOptions?:

ProviderOptions

内部の構造化 Agent に渡す Provider 固有のオプション。思考モデルの reasoning effort など、モデルの動作を制御するために使用します（例: { openai: { reasoningEffort: 'low' } }）。

outputProcessors?:

Processor[]

Agent に設定された出力 Processor を上書きします。出力 Processor は、Agent からのメッセージをユーザーに返す前に変更または検証できます。processOutputResult 関数と processOutputStream 関数のいずれか、または両方を実装する必要があります。

includeRawChunks?:

boolean

ストリーム出力に未加工のチャンクを含めるかどうか（一部のモデル Provider では使用できません）。

inputProcessors?:

Processor[]

Agent に設定された入力 Processor を上書きします。入力 Processor は、メッセージが Agent によって処理される前に変更または検証できます。processInput 関数を実装する必要があります。

instructions?:

string

この生成に限り、Agent のデフォルト指示を上書きするカスタム指示。新しい Agent インスタンスを作成せず、Agent の動作を動的に変更する場合に便利です。

system?:

string | string[] | CoreSystemMessage | SystemModelMessage | CoreSystemMessage[] | SystemModelMessage[]

プロンプトに含めるカスタム system メッセージ。単一の文字列、メッセージオブジェクト、またはそのいずれかの配列を指定できます。system メッセージは、Agent の主な指示を補足するコンテキストや動作指示を提供します。

output?:

Zod schema | JsonSchema7

**非推奨。** 同じ結果を得るには、モデルを指定せずに structuredOutput を使用してください。期待される出力構造を定義します。JSON Schema オブジェクトまたは Zod schema を指定できます。

memory?:

object

Memory の設定。Memory の管理にはこの方法を推奨します。

thread:

string | { id: string; metadata?: Record<string, any>, title?: string }

会話 thread。文字列 ID、または id と省略可能な metadata を持つオブジェクトとして指定します。

resource:

string

thread に関連付けられたユーザーまたはリソースの識別子。

options?:

MemoryConfig

lastMessages、readOnly、semanticRecall、workingMemory、filterIncompleteToolCalls を含む Memory の動作設定。

onTitleGenerated?:

(title: string) => void | Promise<void>

thread のタイトルが生成され、ストレージに永続化されたときに非同期で呼び出されるコールバック。タイトル生成はバックグラウンドで実行され、ストリームの終了後に完了する場合があります。Memory オプションで generateTitle が有効で、thread に既存のタイトルがない場合にのみ呼び出されます。

onFinish?:

StreamTextOnFinishCallback<any> | StreamObjectOnFinishCallback<OUTPUT>

ストリーミングの完了時に呼び出されるコールバック関数。最終結果を受け取ります。

onStepFinish?:

StreamTextOnStepFinishCallback<any> | never

各実行ステップの後に呼び出されるコールバック関数。ステップの詳細を JSON 文字列として受け取ります。構造化出力では使用できません。

telemetry?:

TelemetrySettings

ストリーミング中の OTLP telemetry 収集の設定（Tracing ではありません）。

isEnabled?:

boolean

telemetry を有効または無効にします。実験段階ではデフォルトで無効です。

recordInputs?:

boolean

入力の記録を有効または無効にします。デフォルトでは有効です。機密情報の記録を避けるため、入力の記録を無効にすることもできます。

recordOutputs?:

boolean

出力の記録を有効または無効にします。デフォルトでは有効です。機密情報の記録を避けるため、出力の記録を無効にすることもできます。

functionId?:

string

この関数の識別子。telemetry データを関数ごとにグループ化するために使用します。

modelSettings?:

CallSettings

Model-specific settings like temperature, maxOutputTokens, topP, etc. These settings control how the language model generates responses.

temperature?:

number

Controls randomness in generation (0-2). Higher values make output more random.

maxOutputTokens?:

number

Maximum number of tokens to generate in the response. Note: Use maxOutputTokens (not maxTokens) as per AI SDK v5 convention.

maxRetries?:

number

Maximum number of retry attempts for failed requests.

topP?:

number

Nucleus sampling parameter (0-1). Controls diversity of generated text.

topK?:

number

Top-k sampling parameter. Limits vocabulary to k most likely tokens.

presencePenalty?:

number

Penalty for token presence (-2 to 2). Reduces repetition.

frequencyPenalty?:

number

Penalty for token frequency (-2 to 2). Reduces repetition of frequent tokens.

stopSequences?:

string[]

Stop sequences. If set, the model will stop generating text when one of the stop sequences is generated.

toolChoice?:

'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }

ストリーミング中に Agent が Tool を使用する方法を制御します。

'auto':

string

Tool を使用するかどうかをモデルに判断させます（デフォルト）。

'none':

string

Tool を一切使用しません。

'required':

string

モデルに 1 つ以上の Tool の使用を必須とします。

{ type: 'tool'; toolName: string }:

object

名前を指定した特定の Tool の使用をモデルに必須とします。

toolsets?:

ToolsetsInput

ストリーミング中に Agent で利用可能にする追加の Toolset。

clientTools?:

ToolsInput

リクエストの 'client' 側で実行される Tool。これらの Tool の定義には execute 関数がありません。

hooks?:

ToolHooks

Tool 呼び出しの前後に実行される、実行ごとの hook。この実行では、一致する Agent レベルの hook を上書きします。beforeToolCall は { proceed: false, output } を返して Tool 呼び出しをスキップできます。

savePerStep?:

boolean

各ストリームステップの完了後に、メッセージを段階的に保存します（デフォルト: false）。

requireToolApproval?:

boolean

true の場合、すべての Tool 呼び出しで実行前の明示的な承認が必要です。ストリームは tool-call-approval チャンクを発行し、approveToolCall() または declineToolCall() が呼び出されるまで一時停止します。

autoResumeSuspendedTools?:

boolean

true の場合、ユーザーが同じ thread に新しいメッセージを送信すると、中断された Tool を自動的に再開します。Agent は Tool の resumeSchema に基づき、ユーザーのメッセージから resumeData を抽出します。Memory の設定が必要です。

toolCallConcurrency?:

number

同時に実行する Tool 呼び出しの最大数。承認が必要になる可能性がある場合のデフォルトは 1、それ以外は 10 です。

providerOptions?:

Record<string, Record<string, JSONValue>>

基盤となる LLM Provider に渡す、Provider 固有の追加オプション。構造は { providerName: { optionKey: value } } です。例: { openai: { reasoningEffort: 'high' }, anthropic: { maxTokens: 1000 } }。

openai?:

Record<string, JSONValue>

OpenAI 固有のオプション。例: { reasoningEffort: 'high' }

anthropic?:

Record<string, JSONValue>

Anthropic 固有のオプション。例: { maxTokens: 1000 }

google?:

Record<string, JSONValue>

Google 固有のオプション。例: { safetySettings: [...] }

[providerName]?:

Record<string, JSONValue>

その他の Provider 固有のオプション。キーは Provider 名、値は Provider 固有のオプションのレコードです。

runId?:

string

この生成実行の一意な ID。追跡やデバッグに便利です。

requestContext?:

RequestContext

依存性注入とコンテキスト情報のための Request Context。

tracingContext?:

TracingContext

子 span の作成とメタデータの追加に使用する Tracing コンテキスト。Mastra の Tracing システムを使用すると自動的に注入されます。

currentSpan?:

Span

子 span の作成とメタデータの追加に使用する現在の span。カスタムの子 span の作成や、実行中の span 属性の更新に使用します。

tracingOptions?:

TracingOptions

Tracing 設定のオプション。

metadata?:

Record<string, any>

ルート Trace span に追加するメタデータ。ユーザー ID、セッション ID、feature flag などのカスタム属性を追加する場合に便利です。

requestContextKeys?:

string[]

この Trace のメタデータとして抽出する追加の RequestContext キー。ネストされた値にはドット記法（例: 'user.id'）を使用できます。

traceId?:

string

この実行で使用する Trace ID（1〜32 文字の 16 進数）。指定すると、この Trace は指定された Trace の一部になります。

parentSpanId?:

string

この実行で使用する親 span ID（1〜16 文字の 16 進数）。指定すると、ルート span はこの span の子として作成されます。

tags?:

string[]

この Trace に適用するタグ。Trace の分類とフィルタリングに使用する文字列ラベルです。

versions?:

VersionOverrides

サブ Agent への委譲で使用する、呼び出しごとのバージョン上書き。Mastra インスタンスレベルのバージョンにマージされ、requestContext を通じてサブ Agent の呼び出しへ自動的に伝播します。editor パッケージが必要です。Editor のバージョン管理を参照してください。

VersionOverrides

agents?:

Record<string, VersionSelector>

Agent ID とそのバージョンセレクターのマップ。

VersionSelector

versionId?:

string

ID で特定のバージョンを指定します。

status?:

'draft' | 'published'

この公開ステータスに該当する最新バージョンを指定します。

untilIdle?:

boolean | { maxIdleMs?: number }

設定すると、バックグラウンドタスクによる継続処理の間もストリームを開いたままにします。バックグラウンドタスクが完了すると Agent は LLM を自動的に再呼び出し、同じ fullStream を通じて継続ターンをストリーミングします。デフォルト設定（アイドルタイムアウト 5 分）を使用するには true を、設定を変更するには maxIdleMs を持つオブジェクトを渡します。Memory が必要です。独立した streamUntilIdle() メソッドを置き換えるものです。

maxIdleMs?:

number

ターン間でこの時間（ミリ秒）アイドル状態が続くと、外側のストリームを閉じます。タイマーは wrapper がターン間にある間だけ動作します。デフォルト: 5 分。

戻り値

stream:

MastraModelOutput<Output>

ストリーミング出力にアクセスできる MastraModelOutput インスタンスを返します。

traceId?:

string

Tracing が有効な場合に、この実行に関連付けられる Trace ID。ログの関連付けや実行フローのデバッグに使用します。

spanId?:

string

Tracing が有効な場合に、この実行に関連付けられるルート span ID。span 単位の検索や関連付けに使用します。

詳細な使用例

Mastra 形式（デフォルト）

index.ts

import { stepCountIs } from 'ai-v5'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// Access text stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

// or access full stream
for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get full text after streaming
const fullText = await stream.text

AI SDK v5 以降の形式

AI SDK v5 以降でストリームを使用するには、ユーティリティ関数 toAISdkStream で変換します。

index.ts

import { stepCountIs, createUIMessageStreamResponse } from 'ai'
import { toAISdkStream } from '@mastra/ai-sdk'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// In an API route for frontend integration
return createUIMessageStreamResponse({
  stream: toAISdkStream(stream, { from: 'agent' }),
})

コールバックの使用

すべてのコールバック関数をトップレベルのプロパティとして使用できるようになり、API がより扱いやすくなりました。

index.ts

const stream = await agent.stream('Tell me a story', {
  onFinish: result => {
    console.log('Streaming finished:', result)
  },
  onStepFinish: step => {
    console.log('Step completed:', step)
  },
  onChunk: chunk => {
    console.log('Received chunk:', chunk)
  },
  onError: ({ error }) => {
    console.error('Streaming error:', error)
  },
  onAbort: event => {
    console.log('Stream aborted:', event)
  },
})

// Process the stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

オプションを使用する高度な例

index.ts

import { z } from 'zod'
import { stepCountIs } from 'ai'

await agent.stream('message for agent', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
  memory: {
    thread: 'user-123',
    resource: 'test-app',
  },
  toolChoice: 'auto',
  // Structured output with better DX
  structuredOutput: {
    schema: z.object({
      sentiment: z.enum(['positive', 'negative', 'neutral']),
      confidence: z.number(),
    }),
    model: 'openai/gpt-5.6-sol',
    errorStrategy: 'warn',
  },
  // Output processors for streaming response validation
  outputProcessors: [
    new ModerationProcessor({ model: 'openrouter/openai/gpt-oss-safeguard-20b' }),
    new BatchPartsProcessor({ maxBatchSize: 3, maxWaitTime: 100 }),
  ],
})

Responses WebSocket transport

Provider オプションを指定して、Responses WebSocket ストリーミングを有効にします。これはストリーミング呼び出しにのみ適用され、OpenAI の直接モデルと Azure OpenAI Responses のデプロイでサポートされています。WebSocket ストリーミングを利用できない場合、Mastra は HTTP ストリーミングへフォールバックします。デフォルトでは、ストリームの終了時に Mastra が WebSocket を閉じます。

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket', // 'websocket' | 'fetch' | 'auto'
      websocket: {
        url: 'wss://api.openai.com/v1/responses',
        closeOnFinish: true, // default
      },
    },
  },
})

Azure OpenAI では、useResponsesAPI: true を指定して gateway を設定し、providerOptions.azure.transport を使用します。

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    azure: {
      transport: 'websocket',
      store: false,
      websocket: { closeOnFinish: true },
    },
  },
})

ストリームの終了後も接続を開いたままにするには、closeOnFinish: false を設定し、手動で閉じます。

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket',
      websocket: { closeOnFinish: false },
    },
  },
})

// Later, when you're done with the connection:
stream.transport?.close()

Responses WebSocket 接続では、一度に 1 つのレスポンスを実行します。同じ WebSocket transport 上で previous_response_id を含む継続リクエストが重複すると、Mastra はそのリクエストを拒否します。レスポンスチェーンの次のターンを送信する前に、実行中のストリームが終了するまで待ってください。

関連項目

• レスポンスの生成
• レスポンスのストリーミング
• Agent の承認