Workflows の概要

Workflow では、単一 Agent の推論に頼らず、明確で構造化されたステップを使って複雑なタスクの流れを定義できます。タスクの分割方法、ステップ間のデータの流れ、実行のタイミングを完全に制御できます。Workflow はデフォルトで組み込みの実行エンジンを使いますが、Inngest など、マネージドインフラストラクチャを提供する Workflow runner にデプロイすることもできます。

AI Agent 向け： 一時的なスクリプトを作成せずにこの Workflow を直接開始するには、npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' を実行します。このコマンドには実行中の Mastra サーバーが必要です。npx mastra dev でローカルサーバーを起動するか、--url でアクセス可能なサーバーのベース URL を渡してください。別の入力を作成する前に、npx mastra api workflow run start --schema を実行します。API CLI の検出、対象指定、スキーマ、認証、エラー処理に関する詳細なガイダンスを利用するには、npx skills add mastra-ai/skills --skill mastra で Mastra の Skill をインストールしてください。

Workflow を使う場面

事前に明確に定義され、特定の実行順序を持つ複数のステップからなるタスクには Workflow を使います。ステップ間でのデータの流れや変換方法、各段階で呼び出すプリミティブを細かく制御できます。

基本原則

Mastra Workflow は次の原則に基づいて動作します。

• createStep で入出力スキーマとビジネスロジックを指定し、ステップを定義する。
• createWorkflow でステップを組み合わせ、実行フローを定義する。
• Workflow を実行してシーケンス全体を処理する。一時停止、再開、結果のストリーミングを標準でサポートします。

Workflow のステップを作成する

ステップは Workflow の構成要素です。createStep() に inputSchema と outputSchema を指定し、受け取るデータと返すデータを定義します。どちらのスキーマも Standard JSON Schema（Zod、Valibot、ArkType など）で定義できます。

execute 関数でステップの処理を定義します。コードベース内の関数、外部 API、Agent、Tool の呼び出しに使用できます。

Zod

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    formatted: z.string(),
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

const step1 = createStep({
  id: 'step-1',
  inputSchema: toStandardJsonSchema(
    v.object({
      message: v.string(),
    }),
  ),
  outputSchema: toStandardJsonSchema(
    v.object({
      formatted: v.string(),
    }),
  ),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

設定オプションの一覧は、Step を参照してください。

Agent と Tool を使う

Workflow のステップから、登録済みの Agent を呼び出したり、Tool を直接インポートして実行したりできます。詳しくは Tool の使用を参照してください。

Workflow を作成する

createWorkflow() に inputSchema と outputSchema を指定し、Workflow が受け取るデータと返すデータを定義します。どちらのスキーマも Standard JSON Schema（Zod、Valibot、ArkType など）で定義できます。.then() でステップを追加し、.commit() で Workflow を完成させます。

Zod

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { z } from "zod";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    output: z.string()
  })
})
  .then(step1)
  .commit();

Valibot

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import * as v from "valibot";
import { toStandardJsonSchema } from "@valibot/to-json-schema";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: toStandardJsonSchema(v.object({
    message: v.string()
  })),
  outputSchema: toStandardJsonSchema(v.object({
    output: v.string()
  }))
})
  .then(step1)
  .commit();

ArkType

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { type } from "arktype";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: type({
    message: "string"
  }),
  outputSchema: type({
    output: "string"
  })
})
  .then(step1)
  .commit();

設定オプションの一覧は、Workflow Class を参照してください。

Control Flow を理解する

Workflow はさまざまなメソッドで構成できます。選択するメソッドによって、各ステップのスキーマ構造が決まります。詳しくは Control Flow を参照してください。

Studio

Studio を開き、Workflows タブから Workflow を選択します。

• グラフビュー：中央のパネルに Workflow のステップと実行フローが表示されます。
• 入力フォーム：右側のサイドバーに Workflow の inputSchema からフォームが生成されます。入力後、実行を開始します。
• ライブステータス：実行中は各ステップのステータスがグラフ上でリアルタイムに更新されます。サイドバーには Workflow の入力、出力、状態、ログが表示されます。
• Time travel：実行完了後に個々のステップを再実行し、調査や再試行ができます。

Workflow の状態

Workflow の状態を使うと、すべてのステップの inputSchema と outputSchema を経由させずに、ステップ間で値を共有できます。進捗の追跡、結果の蓄積、Workflow 全体での設定共有に使用します。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  stateSchema: z.object({ counter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // Read from state
    console.log(state.counter)

    // Update state for subsequent steps
    setState({ ...state, counter: state.counter + 1 })

    return { formatted: inputData.message.toUpperCase() }
  },
})

状態スキーマ、初期状態、一時停止と再開をまたぐ永続化、ネストした Workflow については、Workflow の状態を参照してください。

ステップとしての Workflow

Workflow をステップとして使用すると、より大きな構成の中でそのロジックを再利用できます。入出力には、基本原則で説明したものと同じスキーマ規則が適用されます。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});
const step2 = createStep({...});

const childWorkflow = createWorkflow({
  id: "child-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(step1)
  .then(step2)
  .commit();

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(childWorkflow)
  .commit();

Workflow を複製する

ロジックを再利用しつつ、新しい ID で個別に追跡するには、cloneWorkflow() で Workflow を複製します。各複製は独立して実行され、ログや Observability Tool では別々の Workflow として表示されます。

src/mastra/workflows/test-workflow.ts

import { cloneWorkflow } from "@mastra/core/workflows";

const step1 = createStep({...});

const parentWorkflow = createWorkflow({...})
const clonedWorkflow = cloneWorkflow(parentWorkflow, { id: "cloned-workflow" });

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .then(clonedWorkflow)
  .commit();

Workflow を登録する

アプリケーション全体で利用できるように、Workflow を Mastra インスタンスへ登録します。登録すると Agent や Tool から呼び出せるようになり、ログや Observability 機能などの共有リソースにもアクセスできます。

src/mastra/index.ts

import { Mastra } from '@mastra/core/mastra'
import { testWorkflow } from './workflows/test-workflow'

export const mastra = new Mastra({
  workflows: { testWorkflow },
})

Workflow を参照する

Workflow は Agent、Tool、Mastra Client、コマンドラインから実行できます。構成に応じて mastra または mastraClient インスタンスの .getWorkflow() を呼び出し、参照を取得します。

const testWorkflow = mastra.getWorkflow('testWorkflow')

情報

直接インポートするより mastra.getWorkflow() を推奨する理由は2つあります。

1. Mastra インスタンスの設定（ロガー、テレメトリ、ストレージ、登録済み Agent、ベクトルストア）にアクセスできる
2. Workflow の入出力スキーマを TypeScript で完全に型推論できる

getWorkflow() には Workflow の登録キー（Mastra に追加したときのキー）を指定します。id プロパティで Workflow を取得する getWorkflowById() もありますが、同じレベルの型推論は得られません。

Workflow を実行する

Workflow には2つの実行モードがあります。start はすべてのステップが完了してから結果を返し、stream は実行中にイベントを発行します。最終結果だけが必要なら start、進捗の監視やステップ完了時の処理が必要なら stream を選びます。

.start()

createRun() で Workflow の実行インスタンスを作成し、Workflow の inputSchema に一致する inputData を指定して .start() を呼び出します。すべてのステップが実行され、最終結果が返されます。

const run = await testWorkflow.createRun()

const result = await run.start({
  inputData: {
    message: 'Hello world',
  },
})

if (result.status === 'success') {
  console.log(result.result)
}

.stream()

.createRun() で Workflow の実行インスタンスを作成し、Workflow の inputSchema に一致する inputData を指定して .stream() を呼び出します。fullStream を反復処理して進捗を追跡し、result を待機して Workflow の最終結果を取得します。

const run = await testWorkflow.createRun()

const stream = run.stream({
  inputData: {
    message: 'Hello world',
  },
})

for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get the final result (same type as run.start())
const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Workflow の結果型

run.start() と stream.result はどちらも、status プロパティに基づく判別共用体を返します。ステータスは success、failed、suspended、tripwire、paused のいずれかです。ステータスにかかわらず、result.status、result.input、result.steps と、任意の result.state には常に安全にアクセスできます。

また、ステータスに応じて利用できるプロパティが異なります。

ステータス	固有のプロパティ	説明
success	result	Workflow の出力データ
failed	error	失敗の原因となったエラー
tripwire	tripwire	reason、retry?、metadata?、processorId? を含む
suspended	suspendPayload, suspended	一時停止データと、一時停止したステップのパスの配列
paused	（なし）	共通プロパティのみ利用可能

ステータス固有のプロパティにアクセスするには、先に status を確認します。

const result = await run.start({ inputData: { message: 'Hello world' } })

if (result.status === 'success') {
  console.log(result.result) // Only available when status is "success"
} else if (result.status === 'failed') {
  console.log(result.error.message)
} else if (result.status === 'suspended') {
  console.log(result.suspendPayload)
}

Workflow の出力

次は、input、steps、result プロパティを含む、成功した Workflow の結果例です。

{
  "status": "success",
  "steps": {
    "step-1": {
      "status": "success",
      "payload": {
        "message": "Hello world"
      },
      "output": {
        "formatted": "HELLO WORLD"
      }
    },
    "step-2": {
      "status": "success",
      "payload": {
        "formatted": "HELLO WORLD"
      },
      "output": {
        "emphasized": "HELLO WORLD!!!"
      }
    }
  },
  "input": {
    "message": "Hello world"
  },
  "result": {
    "emphasized": "HELLO WORLD!!!"
  }
}

ストリーミング

汎用的な writer API の使用方法は、ストリーミングを参照してください。

Workflow ストリームのペイロードを調べる

ストリームに書き込まれたイベントは、発行されるチャンクに含まれます。これらのチャンクを調べると、イベント型、中間値、ステップ固有のデータなどのカスタムフィールドにアクセスできます。

const testWorkflow = mastra.getWorkflow('testWorkflow')

const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}

if (result!.status === 'suspended') {
  // if the workflow is suspended, we can resume it with the resumeStream method
  const resumedStream = await run.resumeStream({
    resumeData: { value: 'resume data' },
  })

  for await (const chunk of resumedStream) {
    console.log(chunk)
  }
}

中断された Workflow ストリームを再開する

Workflow ストリームが何らかの理由で閉じられたり中断されたりした場合は、resumeStream メソッドで再開できます。Workflow のイベントを監視するための新しい ReadableStream が返されます。

const newStream = await run.resumeStream()

for await (const chunk of newStream) {
  console.log(chunk)
}

Agent を使用する Workflow

Agent の textStream を Workflow ステップの writer にパイプします。これにより部分的な出力がストリーミングされ、Mastra は Agent の使用量を Workflow の実行へ自動的に集約します。

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'

export const testStep = createStep({
  execute: async ({ inputData, mastra, writer }) => {
    const { city } = inputData

    const testAgent = mastra?.getAgent('testAgent')
    const stream = await testAgent?.stream(`What is the weather in ${city}?`)

    await stream!.textStream.pipeTo(writer!)

    return {
      value: await stream!.text,
    }
  },
})

実行中の Workflow を再起動する

Workflow の実行がサーバーとの接続を失った場合、最後にアクティブだったステップから再起動できます。これは、実行中にサーバーとの接続が失われる可能性がある長時間実行型の Workflow に便利です。Workflow の実行を再起動すると、最後にアクティブだったステップから実行を再開し、そこから処理を続けます。

restartAllActiveWorkflowRuns() ですべての実行中 Workflow を再起動する

ある Workflow の実行中のすべての run を再起動するには、restartAllActiveWorkflowRuns() を使います。各 run を手動でループして再起動する必要はありません。

workflow.restartAllActiveWorkflowRuns()

restart() で実行中の Workflow を再起動する

実行中の Workflow を最後にアクティブだったステップから再起動するには、restart() を使います。そのステップから実行を再開し、Workflow の処理を続けます。

const run = await workflow.createRun()

const result = await run.start({ inputData: { value: 'initial data' } })

const restartedResult = await run.restart()

実行中の Workflow を特定する

Workflow が実行中の場合、status は running または waiting です。Workflow の status を確認して実行中か判断し、active を使って実行中の Workflow を特定できます。

const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
  console.log(activeRuns.runs)
}

注記

ローカルの mastra サーバーを実行すると、サーバーの起動時に実行中のすべての Workflow が自動的に再起動されます。

RequestContext を使う

リクエスト固有の値にアクセスするには、RequestContext を使います。これにより、リクエストのコンテキストに応じて動作を調整できます。

src/mastra/workflows/test-workflow.ts

export type UserTier = {
  'user-tier': 'enterprise' | 'pro'
}

const step1 = createStep({
  execute: async ({ requestContext }) => {
    const userTier = requestContext.get('user-tier') as UserTier['user-tier']

    const maxResults = userTier === 'enterprise' ? 1000 : 50

    return { maxResults }
  },
})

詳しくは Request Context を参照してください。

ヒント

型安全なリクエストコンテキストのスキーマ検証については、スキーマ検証を参照してください。

関連情報

Workflow を詳しく学ぶには、実践例を通して基本概念を説明する Workflow ガイドを参照してください。

• Workflow の状態
• Control Flow
• 一時停止と再開
• エラー処理
• Workers：専用のバックグラウンドプロセスで Workflow を実行する
• 📹 Mastra による Agentic Workflow ワークショップ