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

# Workflows の概要

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

**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`](https://mastra.zisheng.pro/ja/reference/workflows/step) で入出力スキーマとビジネスロジックを指定し、**ステップ**を定義する。
- [`createWorkflow`](https://mastra.zisheng.pro/ja/reference/workflows/workflow) で**ステップ**を組み合わせ、実行フローを定義する。
- **Workflow** を実行してシーケンス全体を処理する。一時停止、再開、結果のストリーミングを標準でサポートします。

## Workflow のステップを作成する

ステップは Workflow の構成要素です。`createStep()` に `inputSchema` と `outputSchema` を指定し、受け取るデータと返すデータを定義します。どちらのスキーマも [Standard JSON Schema](https://standardschema.dev/json-schema)（[Zod](https://zod.dev/)、[Valibot](https://valibot.dev/)、[ArkType](https://arktype.io/) など）で定義できます。

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

**Zod**:

```typescript
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**:

```typescript
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**:

```typescript
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`](https://mastra.zisheng.pro/ja/reference/workflows/step) を参照してください。

### Agent と Tool を使う

Workflow のステップから、登録済みの Agent を呼び出したり、Tool を直接インポートして実行したりできます。詳しくは [Tool の使用](https://mastra.zisheng.pro/ja/docs/agents/using-tools)を参照してください。

## Workflow を作成する

`createWorkflow()` に `inputSchema` と `outputSchema` を指定し、Workflow が受け取るデータと返すデータを定義します。どちらのスキーマも [Standard JSON Schema](https://standardschema.dev/json-schema)（[Zod](https://zod.dev/)、[Valibot](https://valibot.dev/)、[ArkType](https://arktype.io/) など）で定義できます。`.then()` でステップを追加し、`.commit()` で Workflow を完成させます。

**Zod**:

```typescript
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**:

```typescript
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**:

```typescript
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](https://mastra.zisheng.pro/ja/reference/workflows/workflow) を参照してください。

### Control Flow を理解する

Workflow はさまざまなメソッドで構成できます。選択するメソッドによって、各ステップのスキーマ構造が決まります。詳しくは [Control Flow](https://mastra.zisheng.pro/ja/docs/workflows/control-flow) を参照してください。

## Studio

[Studio](https://mastra.zisheng.pro/ja/docs/studio/overview) を開き、**Workflows** タブから Workflow を選択します。

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

## Workflow の状態

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

```typescript
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 の状態](https://mastra.zisheng.pro/ja/docs/workflows/workflow-state)を参照してください。

## ステップとしての Workflow

Workflow をステップとして使用すると、より大きな構成の中でそのロジックを再利用できます。入出力には、[基本原則](https://mastra.zisheng.pro/ja/docs/workflows/control-flow)で説明したものと同じスキーマ規則が適用されます。

```typescript
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 として表示されます。

```typescript
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 機能などの共有リソースにもアクセスできます。

```typescript
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()` を呼び出し、参照を取得します。

```typescript
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()` を呼び出します。すべてのステップが実行され、最終結果が返されます。

```typescript
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 の最終結果を取得します。

```typescript
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` を確認します。

```typescript
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 の結果例です。

```json
{
  "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 の使用方法は、[ストリーミング](https://mastra.zisheng.pro/ja/guides/concepts/streaming)を参照してください。

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

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

```typescript
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` が返されます。

```typescript
const newStream = await run.resumeStream()

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

### Agent を使用する Workflow

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

```typescript
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 を手動でループして再起動する必要はありません。

```typescript
workflow.restartAllActiveWorkflowRuns()
```

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

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

```typescript
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 を特定できます。

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

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

## `RequestContext` を使う

リクエスト固有の値にアクセスするには、[RequestContext](https://mastra.zisheng.pro/ja/docs/server/request-context) を使います。これにより、リクエストのコンテキストに応じて動作を調整できます。

```typescript
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](https://mastra.zisheng.pro/ja/docs/server/request-context) を参照してください。

> **ヒント:** 型安全なリクエストコンテキストのスキーマ検証については、[スキーマ検証](https://mastra.zisheng.pro/ja/docs/server/request-context)を参照してください。

## 関連情報

Workflow を詳しく学ぶには、実践例を通して基本概念を説明する [Workflow ガイド](https://mastra.zisheng.pro/ja/guides/guide/ai-recruiter)を参照してください。

- [Workflow の状態](https://mastra.zisheng.pro/ja/docs/workflows/workflow-state)
- [Control Flow](https://mastra.zisheng.pro/ja/docs/workflows/control-flow)
- [一時停止と再開](https://mastra.zisheng.pro/ja/docs/workflows/suspend-and-resume)
- [エラー処理](https://mastra.zisheng.pro/ja/docs/workflows/error-handling)
- [Workers](https://mastra.zisheng.pro/ja/docs/deployment/workers)：専用のバックグラウンドプロセスで Workflow を実行する
- 📹 [Mastra による Agentic Workflow ワークショップ](https://www.youtube.com/watch?v=HGt8pVPpX9g)