> ## Documentation Index
> Fetch the complete documentation index at: https://docs.retab.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Workflows

> Build visual document processing pipelines with drag-and-drop blocks

## What are Workflows?

Workflows are visual, block-based pipelines that let you chain together multiple document processing operations. Instead of writing code for each step, you can drag and drop blocks onto a canvas, connect them, and create powerful document automation flows.

A workflow typically consists of:

* **Input blocks** - Entry points for data:
  * **Document** - Upload files (PDF, images, Word, Excel)
  * **JSON Input** - Pass structured JSON data
* **Processing blocks** - Operations like Extract, Parse, Split, Classifier
* **Logic blocks** - Conditional flows like review gates, Function, If/Else routing, and API Call

### Workflow Evals

Workflow evals validate individual block outputs with saved inputs and
assertions. Use them to check that an Extract, Function, Split, or Classifier
block still behaves as expected after you change schemas, prompts, code, or
block configuration.

Learn more in [Workflow Evals](/workflows/evals).

### Experiments

Experiments measure a block's consistency by replaying the same Extract, Split,
Classifier, or split-by-key For Each block over a fixed document set with
multiple consensus passes. Use them to compare block configurations, find
low-agreement documents or fields, and decide what needs stricter evals.

Learn more in [Experiments](/workflows/Experiments).

## Creating a Workflow

1. Navigate to the **Workflows** section in your dashboard
2. Click **Create Workflow** to open a new canvas
3. Drag blocks from the sidebar onto the canvas
4. Connect blocks by dragging from output handles to input handles
5. Configure each block by clicking on it
6. Your workflow auto-saves as you build

## Connecting Blocks

Blocks communicate through **handles** that define the type of data they accept or produce:

| Handle Type | Icon  | Description                               |
| ----------- | ----- | ----------------------------------------- |
| **File**    | 📎    | Document files (PDF, images, Word, Excel) |
| **JSON**    | `{ }` | Structured data extracted from documents  |

### Connection Rules

* **File → File**: Pass documents between processing blocks
* **JSON → JSON**: Pass extracted data between logic blocks
* Each input handle accepts only **one connection**
* Connections validate automatically to prevent incompatible links

## Declarative Workflow Spec

You can also define and manage workflows from YAML. A declarative spec uses
`apiVersion: workflows.retab.com/v1alpha2` and keeps topology in `spec.edges`.
Every edge endpoint is explicit: it names the block and the raw runtime handle.

```yaml theme={null}
apiVersion: workflows.retab.com/v1alpha2
kind: Workflow
metadata:
  id: wf_abc123
  name: Invoice Workflow

spec:
  blocks:
    start_document-node:
      type: start_document
      label: Input Document

    extract-node:
      type: extract
      label: Extract Fields
      config:
        inputs:
          - name: source_doc
            type: file
            is_primary: true
        json_schema:
          type: object
          properties: {}

  edges:
    - source:
        block: start_document-node
        handle: output-file-0
      target:
        block: extract-node
        handle: input-file-source_doc
```

The SDK exposes `validate()` and `get()` under `client.workflows.spec`, while
`plan()` and `apply()` live on the top-level `client.workflows` resource:

<CodeGroup>
  ```python Python theme={null}
  validation = client.workflows.spec.validate(yaml_definition)
  plan = client.workflows.plan(yaml_definition)
  created = client.workflows.apply(yaml_definition)
  updated = client.workflows.apply(yaml_definition, workflow_id="wf_abc123")
  exported = client.workflows.spec.get(created.workflow_id)
  ```

  ```typescript TypeScript theme={null}
  const validation = await client.workflows.spec.validate(yamlDefinition);
  const plan = await client.workflows.plan(yamlDefinition);
  const created = await client.workflows.apply(yamlDefinition);
  const updated = await client.workflows.apply(yamlDefinition, undefined, "wf_abc123");
  const exported = await client.workflows.spec.get(created.workflowId);
  ```

  ```php PHP theme={null}
  $validation = $client->workflows()->spec()->validate($yamlDefinition);
  $plan = $client->workflows()->plan(yamlDefinition: $yamlDefinition);
  $created = $client->workflows()->apply(yamlDefinition: $yamlDefinition);
  $updated = $client->workflows()->apply(yamlDefinition: $yamlDefinition, workflowId: 'wf_abc123');
  $exported = $client->workflows()->spec()->get($created->workflowId);
  ```

  ```ruby Ruby theme={null}
  require 'retab'

  client = Retab::Client.new(api_key: ENV['RETAB_API_KEY'])

  validation = client.workflows.spec.validate(yaml_definition: yaml_definition)
  plan = client.workflows.plan(yaml_definition: yaml_definition)
  created = client.workflows.apply(yaml_definition: yaml_definition)
  updated = client.workflows.apply(yaml_definition: yaml_definition, workflow_id: 'wf_abc123')
  exported = client.workflows.spec.get(workflow_id: created.workflow_id)
  ```

  ```csharp C# theme={null}
  using System;
  using System.Threading.Tasks;
  using Retab;
  using RetabClient = Retab.Retab;

  var client = new RetabClient("YOUR_API_KEY");

  var yamlDefinition = """
  apiVersion: workflows.retab.com/v1alpha2
  kind: Workflow
  metadata:
    id: wf_invoice_demo
    name: Invoice Workflow
  spec:
    blocks:
      start:
        type: start_json
        label: Input JSON
        config:
          json_schema:
            type: object
            required:
              - value
            properties:
              value:
                type: string
    edges: []
  """;

  var validation = await client.Workflows.Spec.ValidateAsync(
      new WorkflowSpecValidateOptions { YamlDefinition = yamlDefinition }
  );
  var plan = await client.Workflows.PlanAsync(
      new WorkflowsPlanOptions { YamlDefinition = yamlDefinition }
  );
  var created = await client.Workflows.ApplyAsync(
      new WorkflowsApplyOptions { YamlDefinition = yamlDefinition }
  );
  var updated = await client.Workflows.ApplyAsync(
      new WorkflowsApplyOptions { YamlDefinition = yamlDefinition, WorkflowId = "wf_abc123" }
  );
  var exported = await client.Workflows.Spec.GetAsync(created.WorkflowId);
  ```

  ```go Go theme={null}
  package main

  import (
  	"context"
  	"fmt"
  	"log"

  	retab "github.com/retab-dev/retab/clients/go"
  )

  func main() {
  	client, err := retab.NewClient("")
  	if err != nil {
  		log.Fatal(err)
  	}

  	yamlDefinition := `apiVersion: workflows.retab.com/v1alpha2
  kind: Workflow
  metadata:
    id: wf_invoice_demo
    name: Invoice Workflow
  spec:
    blocks:
      start:
        type: start_json
        label: Input JSON
        config:
          json_schema:
            type: object
            required:
              - value
            properties:
              value:
                type: string
    edges: []
  `
  	validation, err := client.Workflows.Spec.Validate(context.Background(), &retab.WorkflowSpecValidateParams{YamlDefinition: yamlDefinition})
  	if err != nil {
  		log.Fatal(err)
  	}
  	plan, err := client.Workflows.Plan(context.Background(), &retab.WorkflowsPlanParams{YamlDefinition: yamlDefinition})
  	if err != nil {
  		log.Fatal(err)
  	}
  	created, err := client.Workflows.Apply(context.Background(), &retab.WorkflowsApplyParams{YamlDefinition: yamlDefinition})
  	if err != nil {
  		log.Fatal(err)
  	}
  	updated, err := client.Workflows.Apply(context.Background(), &retab.WorkflowsApplyParams{YamlDefinition: yamlDefinition, WorkflowID: retab.Ptr("wf_abc123")})
  	if err != nil {
  		log.Fatal(err)
  	}
  	exported, err := client.Workflows.Spec.Get(context.Background(), created.WorkflowID)
  	if err != nil {
  		log.Fatal(err)
  	}

  	fmt.Println(validation, plan, created, updated, exported)
  }
  ```

  ```rust Rust theme={null}
  use retab::{models::DeclarativeWorkflowRequest, resources::{workflow_spec, workflows}, Retab};

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let yaml_definition = r#"apiVersion: workflows.retab.com/v1alpha2
  kind: Workflow
  metadata:
    id: wf_invoice_demo
    name: Invoice Workflow
  spec:
    blocks:
      start:
        type: start_json
        label: Input JSON
        config:
          json_schema:
            type: object
            required:
              - value
            properties:
              value:
                type: string
    edges: []
  "#.to_string();
  let validation = client
      .workflows().spec()
      .validate(workflow_spec::ValidateParams::new(DeclarativeWorkflowRequest::new(yaml_definition.clone())))
      .await?;
  let plan = client
      .workflows()
      .plan(workflows::PlanParams::new(DeclarativeWorkflowRequest::new(yaml_definition.clone())))
      .await?;
  let created = client
      .workflows()
      .apply(workflows::ApplyParams::new(DeclarativeWorkflowRequest::new(yaml_definition.clone())))
      .await?;
  let updated = client
      .workflows()
      .apply(workflows::ApplyParams {
          body: DeclarativeWorkflowRequest::new(yaml_definition.clone()),
          workflow_id: Some("wf_abc123".to_string()),
      })
      .await?;
  let exported = client.workflows().spec().get(&created.workflow_id).await?;

  println!("{validation:#?} {plan:#?} {updated:#?} {exported:#?}");
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().spec().validate("metadata:\n  id: invoice-workflow\n", null);
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

Use `validate()` for parse and handle checks, `plan()` to preview changes,
`apply()` to create a new workflow from YAML, `apply(..., workflow_id=...)` to
modify an existing workflow draft, and `get()` to get canonical YAML from an
existing workflow.
`plan()` and `apply()` return Terraform-style `summary`, `resource_changes`,
and `rendered_plan` fields so clients can inspect exactly what changed.
Call `client.workflows.publish(workflow_id)` separately when the draft should
become the live published workflow. Publishing happens **inside the current
environment** — the API key decides whether you are publishing into test or
production.

For endpoint details, see:

* [Validate Workflow Spec](/api-reference/workflows/spec/validate)
* [Plan Workflow Spec](/api-reference/workflows/spec/plan)
* [Apply Workflow Spec](/api-reference/workflows/spec/apply)
* [Apply Workflow Spec to Existing Workflow](/api-reference/workflows/apply-to)
* [Export Workflow Spec](/api-reference/workflows/spec)

## Edit Mode vs Run Mode

Workflows have two operational modes:

### Edit Mode

* Add, remove, and configure blocks
* Create and delete connections
* Rename the workflow
* View generated Python code

### Run Mode

* Upload documents to input blocks
* Execute the workflow step-by-step
* View results at each stage
* Download processed files and extracted data

Toggle between modes using the switch at the top of the canvas.

## Running a Workflow

A workflow is fundamentally an asynchronous job. When you start it, Retab creates a workflow run, executes each step on the server, and stores the results on that run. You can then poll the run until it finishes and inspect the stored step outputs.

For the SDK and HTTP endpoint details, see the workflow API reference:

* [List Workflows](/api-reference/workflows/list)
* [Get Workflow](/api-reference/workflows/get)
* [Run Workflow](/api-reference/workflows/runs/create)
* [Get Run](/api-reference/workflows/runs/get)
* [List Steps](/api-reference/workflows/steps/list)
* [Get Step](/api-reference/workflows/steps/get)

### From the Dashboard

1. Switch to **Run Mode**
2. Upload a document to each **Document** input block
3. Click **Run Workflow**
4. Watch as each block processes (status indicators show progress)
5. Click on output handles to view results

### Using the SDK

The Python, Node, and Go SDKs expose workflow metadata, graph authoring, run execution, and typed step inspection:

* `client.workflows.*` / `client.Workflows.*` for `list()`, `get()`, `create()`, `update()`, `delete()`, and `publish()`
* `client.workflows.blocks.*` / `client.Workflows.Blocks.*` and `client.workflows.edges.*` / `client.Workflows.Edges.*` for programmatic graph changes
* `client.workflows.runs.*` / `client.Workflows.Runs.*` and `client.workflows.steps.*` / `client.Workflows.Steps.*` for running flows and reading results

#### Discover input block IDs

Workflow run inputs are keyed by the IDs of your `start_document` and `start_json` blocks. List the workflow's blocks to discover them.

<CodeGroup>
  ```python Python theme={null}
  from retab import Retab

  client = Retab()

  blocks = client.workflows.blocks.list("wf_abc123")

  document_start_id = next(block.id for block in blocks.data if block.type == "start_document")
  json_start_id = next(block.id for block in blocks.data if block.type == "start_json")

  ```

  ```typescript TypeScript theme={null}
  import { Retab } from "@retab/node";

  const client = new Retab({ apiKey: process.env.RETAB_API_KEY });

  const blocks = await client.workflows.blocks.list({ workflowId: "wf_abc123" });

  const documentStartId = blocks.data.find((block) => block.type === "start_document")?.id;
  const jsonStartId = blocks.data.find((block) => block.type === "start_json")?.id;

  if (!documentStartId || !jsonStartId) {
    throw new Error("Workflow is missing a required input block");
  }
  ```

  ```go Go theme={null}
  package main

  import (
  	"context"
  	"fmt"
  	"log"

  	retab "github.com/retab-dev/retab/clients/go"
  )

  func main() {
  	ctx := context.Background()

  	client, err := retab.NewClient("")
  	if err != nil {
  		log.Fatal(err)
  	}

  	blocks, err := client.Workflows.Blocks.List(ctx, &retab.WorkflowBlocksListParams{WorkflowID: "wf_abc123"})
  	if err != nil {
  		log.Fatal(err)
  	}

  	var documentStartID string
  	var jsonStartID string
  	for _, block := range blocks.Data {
  		switch block.Type {
  		case "start_document":
  			documentStartID = block.ID
  		case "start_json":
  			jsonStartID = block.ID
  		}
  	}

  	fmt.Println(documentStartID)
  	fmt.Println(jsonStartID)
  }
  ```

  ```ruby Ruby theme={null}
  require 'retab'

  client = Retab::Client.new(api_key: ENV['RETAB_API_KEY'])

  blocks = client.workflows.blocks.list(workflow_id: 'wf_abc123')

  document_start_id = blocks.data.find { |block| block.type == 'start_document' }&.id
  json_start_id = blocks.data.find { |block| block.type == 'start_json' }&.id

  if document_start_id.nil? || json_start_id.nil?
    raise 'Workflow is missing a required input block'
  end

  puts document_start_id
  puts json_start_id
  ```

  ```php PHP theme={null}
  <?php
  require 'vendor/autoload.php';

  use Retab\Client;
  use Retab\Resource\WorkflowBlockType;

  $client = new Client();

  $blocks = $client->workflows()->blocks()->list('wf_abc123');

  $documentStartId = null;
  $jsonStartId = null;
  foreach ($blocks->data as $block) {
      if ($block->type === WorkflowBlockType::StartDocument) {
          $documentStartId = $block->id;
      } elseif ($block->type === WorkflowBlockType::StartJson) {
          $jsonStartId = $block->id;
      }
  }

  if ($documentStartId === null || $jsonStartId === null) {
      throw new RuntimeException('Workflow is missing a required input block');
  }
  ```

  ```csharp C# theme={null}
  using System;
  using System.Linq;
  using System.Threading.Tasks;
  using Retab;
  using RetabClient = Retab.Retab;

  var client = new RetabClient("YOUR_API_KEY");

  var blocks = await client.Workflows.Blocks.ListAsync(
      new WorkflowBlocksListOptions { WorkflowId = "wf_abc123" }
  );

  string? documentStartId = null;
  string? jsonStartId = null;
  await foreach (var block in blocks)
  {
      if (block.Type == WorkflowBlockType.StartDocument)
      {
          documentStartId = block.Id;
      }
      else if (block.Type == WorkflowBlockType.StartJson)
      {
          jsonStartId = block.Id;
      }
  }

  if (documentStartId is null || jsonStartId is null)
  {
      throw new InvalidOperationException("Workflow is missing a required input block");
  }
  ```

  ```rust Rust theme={null}
  use retab::{enums::WorkflowBlockType, resources::workflow_blocks::ListParams, Retab};

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let blocks = client
      .workflows().blocks()
      .list(ListParams::new("wf_abc123"))
      .await?;

  let document_start_id = blocks
      .data
      .iter()
      .find(|block| block.type_ == WorkflowBlockType::StartDocument)
      .map(|block| block.id.clone());
  let json_start_id = blocks
      .data
      .iter()
      .find(|block| block.type_ == WorkflowBlockType::StartJson)
      .map(|block| block.id.clone());

  println!("{document_start_id:?} {json_start_id:?}");
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().blocks().list("wf_abc123", null, null, 10L);
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

#### Run and wait for completion

Workflows support two input maps:

* `documents` for Document (`start_document`) blocks
* `json_inputs` for JSON Input (`start_json`) blocks

<CodeGroup>
  ```python Python theme={null}
  import time
  from pathlib import Path

  from retab import Retab

  client = Retab()

  workflow = client.workflows.get("wf_abc123")
  blocks = client.workflows.blocks.list(workflow.id)
  document_start_id = next(block.id for block in blocks.data if block.type == "start_document")
  json_start_id = next(block.id for block in blocks.data if block.type == "start_json")

  run = client.workflows.runs.create(
      workflow_id=workflow.id,
      documents={
          document_start_id: Path("path/to/invoice.pdf"),
      },
      json_inputs={
          json_start_id: {"customer_id": "cust_123", "priority": "high"},
      },
  )

  terminal_statuses = {"completed", "error", "cancelled"}
  while run.lifecycle.status not in terminal_statuses and run.lifecycle.status != "awaiting_review":
      time.sleep(1)
      run = client.workflows.runs.get(run.id)

  print(run.lifecycle.status)
  if run.lifecycle.status == "awaiting_review":
      print(run.lifecycle.waiting_for_block_ids)
  elif run.lifecycle.status == "error":
      raise RuntimeError(run.lifecycle.message)
  elif run.lifecycle.status == "cancelled":
      raise RuntimeError(run.lifecycle.reason or "Workflow run was cancelled")
  else:
      for step_summary in client.workflows.steps.list(run.id):
          step = client.workflows.steps.get(step_summary.step_id)
          if step.handle_outputs:
              print(step.block_id, step.handle_outputs)

  ```

  ```typescript TypeScript theme={null}
  import { Retab } from "@retab/node";

  const client = new Retab({ apiKey: process.env.RETAB_API_KEY });

  const workflow = await client.workflows.get("wf_abc123");
  const blocks = await client.workflows.blocks.list({ workflowId: workflow.id });
  const documentStartId = blocks.data.find((block) => block.type === "start_document")?.id;
  const jsonStartId = blocks.data.find((block) => block.type === "start_json")?.id;

  if (!documentStartId || !jsonStartId) {
    throw new Error("Workflow is missing a required input block");
  }

  let run = await client.workflows.runs.create(
    workflow.id,
    {
      [documentStartId]: "path/to/invoice.pdf",
    },
    {
      [jsonStartId]: { customer_id: "cust_123", priority: "high" },
    },
  );

  const terminalStatuses = new Set(["completed", "error", "cancelled"]);
  while (!terminalStatuses.has(run.lifecycle.status) && run.lifecycle.status !== "awaiting_review") {
    await new Promise((resolve) => setTimeout(resolve, 1000));
    run = await client.workflows.runs.get(run.id);
  }

  console.log(run.lifecycle.status);
  if (run.lifecycle.status === "awaiting_review") {
    console.log(run.lifecycle.waitingForBlockIds);
  } else if (run.lifecycle.status === "error") {
    throw new Error(run.lifecycle.message);
  } else if (run.lifecycle.status === "cancelled") {
    throw new Error(run.lifecycle.reason ?? "Workflow run was cancelled");
  } else {
    const steps = await client.workflows.steps.list({ runId: run.id });
    for (const stepSummary of steps.data) {
      const step = await client.workflows.steps.get(stepSummary.stepId);
      if (step.handleOutputs && Object.keys(step.handleOutputs).length > 0) {
        console.log(step.blockId, step.handleOutputs);
      }
    }
  }
  ```

  ```go Go theme={null}
  package main

  import (
  	"context"
  	"fmt"
  	"log"
  	"time"

  	retab "github.com/retab-dev/retab/clients/go"
  )

  func main() {
  	ctx := context.Background()

  	client, err := retab.NewClient("")
  	if err != nil {
  		log.Fatal(err)
  	}

  	workflow, err := client.Workflows.Get(ctx, "wf_abc123")
  	if err != nil {
  		log.Fatal(err)
  	}

  	blocks, err := client.Workflows.Blocks.List(ctx, &retab.WorkflowBlocksListParams{WorkflowID: workflow.ID})
  	if err != nil {
  		log.Fatal(err)
  	}

  	var documentStartID string
  	var jsonStartID string
  	for _, block := range blocks.Data {
  		switch block.Type {
  		case "start_document":
  			documentStartID = block.ID
  		case "start_json":
  			jsonStartID = block.ID
  		}
  	}

  	run, err := client.Workflows.Runs.Create(ctx, &retab.WorkflowRunsCreateParams{
  		WorkflowID: workflow.ID,
  		Documents: &map[string]any{
  			documentStartID: "path/to/invoice.pdf",
  		},
  		JSONInputs: &map[string]any{
  			jsonStartID: map[string]any{
  				"customer_id": "cust_123",
  				"priority":    "high",
  			},
  		},
  	})
  	if err != nil {
  		log.Fatal(err)
  	}

  	deadline := time.Now().Add(10 * time.Minute)
  	for {
  		status := run.Lifecycle.Status()
  		if status == "completed" || status == "error" || status == "failed" || status == "cancelled" || status == "awaiting_review" {
  			break
  		}
  		if time.Now().After(deadline) {
  			log.Fatalf("timed out waiting for workflow run %s", run.ID)
  		}
  		time.Sleep(time.Second)
  		run, err = client.Workflows.Runs.Get(ctx, run.ID)
  		if err != nil {
  			log.Fatal(err)
  		}
  	}

  	status := run.Lifecycle.Status()
  	fmt.Println(status)
  	if status == "awaiting_review" {
  		return
  	}
  	if status != "completed" {
  		log.Fatalf("workflow finished with lifecycle %s", status)
  	}

  	steps, err := client.Workflows.Steps.List(ctx, &retab.WorkflowStepsListParams{RunID: &run.ID})
  	if err != nil {
  		log.Fatal(err)
  	}
  	for _, stepSummary := range steps.Data {
  		step, err := client.Workflows.Steps.Get(ctx, stepSummary.StepID, &retab.WorkflowStepsGetParams{RunID: &run.ID})
  		if err != nil {
  			log.Fatal(err)
  		}
  		if len(step.HandleOutputs) > 0 {
  			fmt.Println(step.BlockID, step.HandleOutputs)
  		}
  	}
  }
  ```

  ```ruby Ruby theme={null}
  require 'retab'

  client = Retab::Client.new(api_key: ENV['RETAB_API_KEY'])

  workflow = client.workflows.get(workflow_id: 'wf_abc123')
  blocks = client.workflows.blocks.list(workflow_id: workflow.id)
  document_start_id = blocks.data.find { |block| block.type == 'start_document' }&.id
  json_start_id = blocks.data.find { |block| block.type == 'start_json' }&.id

  run = client.workflows.runs.create(
    workflow_id: workflow.id,
    documents: {
      document_start_id => 'path/to/invoice.pdf',
    },
    json_inputs: {
      json_start_id => { customer_id: 'cust_123', priority: 'high' },
    },
  )

  terminal_statuses = ['completed', 'error', 'cancelled']
  until terminal_statuses.include?(run.lifecycle.status) || run.lifecycle.status == 'awaiting_review'
    sleep 1
    run = client.workflows.runs.get(run_id: run.id)
  end

  puts run.lifecycle.status
  case run.lifecycle.status
  when 'awaiting_review'
    puts run.lifecycle.waiting_for_block_ids
  when 'error'
    raise run.lifecycle.message
  when 'cancelled'
    raise(run.lifecycle.reason || 'Workflow run was cancelled')
  else
    client.workflows.steps.list(run_id: run.id).data.each do |step_summary|
      step = client.workflows.steps.get(step_id: step_summary.step_id, run_id: run.id)
      next if step.handle_outputs.nil? || step.handle_outputs.empty?
      puts "#{step.block_id} #{step.handle_outputs}"
    end
  end
  ```

  ```php PHP theme={null}
  <?php
  require 'vendor/autoload.php';

  use Retab\Client;
  use Retab\Resource\WorkflowBlockType;

  $client = new Client();

  $workflow = $client->workflows()->get('wf_abc123');
  $blocks = $client->workflows()->blocks()->list($workflow->id);

  $documentStartId = null;
  $jsonStartId = null;
  foreach ($blocks->data as $block) {
      if ($block->type === WorkflowBlockType::StartDocument) {
          $documentStartId = $block->id;
      } elseif ($block->type === WorkflowBlockType::StartJson) {
          $jsonStartId = $block->id;
      }
  }

  $run = $client->workflows()->runs()->create(
      workflowId: $workflow->id,
      documents: [
          $documentStartId => 'path/to/invoice.pdf',
      ],
      jsonInputs: [
          $jsonStartId => ['customer_id' => 'cust_123', 'priority' => 'high'],
      ],
  );

  $terminalStatuses = ['completed', 'error', 'cancelled'];
  $deadline = time() + 600;
  while (!in_array($run->lifecycle->status, $terminalStatuses, true)
      && $run->lifecycle->status !== 'awaiting_review') {
      if (time() > $deadline) {
          throw new RuntimeException("Timed out waiting for workflow run {$run->id}");
      }
      sleep(1);
      $run = $client->workflows()->runs()->get($run->id);
  }

  echo $run->lifecycle->status . PHP_EOL;
  if ($run->lifecycle->status === 'awaiting_review') {
      print_r($run->lifecycle->waitingForBlockIds ?? []);
  } elseif ($run->lifecycle->status === 'error') {
      throw new RuntimeException($run->lifecycle->message ?? 'error');
  } elseif ($run->lifecycle->status === 'cancelled') {
      throw new RuntimeException($run->lifecycle->reason ?? 'Workflow run was cancelled');
  } else {
      foreach ($client->workflows()->steps()->list(runId: $run->id)->data as $stepSummary) {
          $step = $client->workflows()->steps()->get($stepSummary->stepId, runId: $run->id);
          if (!empty($step->handleOutputs)) {
              echo $step->blockId . PHP_EOL;
              print_r($step->handleOutputs);
          }
      }
  }
  ```

  ```rust Rust theme={null}
  use retab::{models::CreateWorkflowRunRequest, resources::workflow_runs, Retab};
  use std::path::PathBuf;
  use std::time::{Duration, Instant};

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let request = CreateWorkflowRunRequest::new("wf_abc123")
      .with_document("start_document-block-id", PathBuf::from("path/to/invoice.pdf"));
  let mut run = client
      .workflows().runs()
      .create(workflow_runs::CreateParams::new(request))
      .await?;

  let deadline = Instant::now() + Duration::from_secs(600);
  while !matches!(
      &run.lifecycle,
      retab::models::WorkflowRunLifecycleOneOf::CompletedTerminal(_)
          | retab::models::WorkflowRunLifecycleOneOf::ErrorTerminal(_)
          | retab::models::WorkflowRunLifecycleOneOf::CancelledTerminal(_)
          | retab::models::WorkflowRunLifecycleOneOf::AwaitingReviewRun(_)
  ) {
      if Instant::now() > deadline {
          return Err("timed out waiting for workflow run".into());
      }
      tokio::time::sleep(Duration::from_secs(1)).await;
      run = client.workflows().runs().get(&run.id).await?;
  }

  println!("{}", run.id);
  ```

  ```csharp C# theme={null}
  using System;
  using System.Collections.Generic;
  using Retab;
  using RetabClient = Retab.Retab;

  static string LifecycleStatus(object? lifecycle)
  {
      return lifecycle?.GetType().GetProperty("Status")?.GetValue(lifecycle)?.ToString() ?? "unknown";
  }

  var client = new RetabClient("YOUR_API_KEY");
  var run = await client.Workflows.Runs.CreateAsync(
      new WorkflowRunsCreateOptions
      {
          WorkflowId = "wf_abc123",
          Documents = new Dictionary<string, WorkflowRunDocumentInput>
          {
              ["start_document-block-id"] = MimeData.FromFile("path/to/invoice.pdf"),
          },
      }
  );

  var deadline = DateTimeOffset.UtcNow.AddMinutes(10);
  while (LifecycleStatus(run.Lifecycle) is not ("completed" or "error" or "cancelled" or "awaiting_review"))
  {
      if (DateTimeOffset.UtcNow > deadline)
      {
          throw new TimeoutException($"Timed out waiting for workflow run {run.Id}");
      }
      await Task.Delay(TimeSpan.FromSeconds(1));
      run = await client.Workflows.Runs.GetAsync(run.Id);
  }

  Console.WriteLine(LifecycleStatus(run.Lifecycle));
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().get("wf_abc123");
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

`steps.list(run.id)` returns the step roster for a run. For the full execution record for one block, including typed inputs and outputs, use `steps.get(run.id, block_id)`.

#### Inspect step outputs

Start with `steps.list(run.id)` when you need the blocks that ran. Then call `steps.get(run.id, block_id)` for the specific execution record you want to inspect.

Step payloads are normalized into `HandlePayload` objects. For JSON-producing blocks, `extracted_data` is shorthand for the default `output-json-0` handle.

<CodeGroup>
  ```python Python theme={null}
  # Step roster:
  for step in client.workflows.steps.list(run.id):
      print(step.block_id, step.lifecycle.status)
      if step.artifact:
          print(step.artifact.operation, step.artifact.id)

  # Full execution record for one step:

  step = client.workflows.steps.get(run.id, "extract-block-id")
  print(step.lifecycle.status)
  if step.extracted_data:
  print(step.extracted_data)

  ```

  ```typescript TypeScript theme={null}
  // Step roster:
  const steps = await client.workflows.steps.list({ runId: run.id });
  for (const step of steps) {
    console.log(step.blockId, step.lifecycle.status);
    if (step.artifact) {
      console.log(step.artifact.operation, step.artifact.id);
    }
  }

  // Full execution record for one step:
  const step = await client.workflows.steps.get(run.id, "extract-block-id");
  console.log(step.lifecycle.status);
  const extractedData = step.handleOutputs["output-json-0"]?.data;
  if (extractedData) {
    console.log(extractedData);
  }
  ```

  ```go Go theme={null}
  // Step roster:
  steps, err := client.Workflows.Steps.List(ctx, run.ID)
  if err != nil {
  	log.Fatal(err)
  }
  for _, step := range steps {
  	fmt.Println(step.BlockID, step.Lifecycle["status"])
  	if step.Artifact != nil {
  		fmt.Println(step.Artifact.Operation, step.Artifact.ID)
  	}
  }

  // Full execution record for one step:
  step, err := client.Workflows.Steps.Get(ctx, run.ID, "extract-block-id")
  if err != nil {
  	log.Fatal(err)
  }
  fmt.Println(step.Lifecycle["status"])
  if extractedData := step.ExtractedData(); extractedData != nil {
  	fmt.Println(extractedData)
  }
  ```

  ```ruby Ruby theme={null}
  # Step roster:
  client.workflows.steps.list(run_id: run.id).data.each do |step|
    puts "#{step.block_id} #{step.lifecycle.status}"
    if step.artifact
      puts "#{step.artifact.operation} #{step.artifact.id}"
    end
  end

  # Full execution record for one step:
  step = client.workflows.steps.get(step_id: 'extract-step-id', run_id: run.id)
  puts step.lifecycle.status
  extracted_data = step.handle_outputs['output-json-0']&.data
  puts extracted_data if extracted_data
  ```

  ```php PHP theme={null}
  <?php
  // Step roster
  foreach ($client->workflows()->steps()->list(runId: $run->id)->data as $step) {
      echo $step->blockId . ' ' . ($step->lifecycle->status ?? '') . PHP_EOL;
      if ($step->artifact !== null) {
          echo $step->artifact->operation . ' ' . $step->artifact->id . PHP_EOL;
      }
  }

  // Full execution record for one step:
  $step = $client->workflows()->steps()->get('extract-block-id', runId: $run->id);
  echo ($step->lifecycle->status ?? '') . PHP_EOL;
  $extractedData = $step->handleOutputs['output-json-0']->data ?? null;
  if ($extractedData !== null) {
      print_r($extractedData);
  }
  ```

  ```rust Rust theme={null}
  use retab::{resources::workflow_steps, Retab};

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let run_id = "wfrun_abc123".to_string();
  let steps = client
      .workflows().steps()
      .list(workflow_steps::ListParams {
          run_id: Some(run_id.clone()),
          ..Default::default()
      })
      .await?;

  for summary in steps.data {
      println!("{} {:?}", summary.block_id, summary.lifecycle);
      if let Some(artifact) = summary.artifact {
          println!("{} {}", artifact.operation, artifact.id);
      }
  }

  let step = client
      .workflows().steps()
      .get("extract-step-id", workflow_steps::GetParams::default())
      .await?;
  println!("{:?}", step.handle_outputs);
  ```

  ```csharp C# theme={null}
  using Retab;
  using RetabClient = Retab.Retab;

  static string LifecycleStatus(object? lifecycle)
  {
      return lifecycle?.GetType().GetProperty("Status")?.GetValue(lifecycle)?.ToString() ?? "unknown";
  }

  var client = new RetabClient("YOUR_API_KEY");
  var steps = await client.Workflows.Steps.ListAsync(
      new WorkflowStepsListOptions { RunId = run.Id }
  );

  await foreach (var summary in steps)
  {
      Console.WriteLine($"{summary.BlockId} {LifecycleStatus(summary.Lifecycle)}");
      if (summary.Artifact is not null)
      {
          Console.WriteLine($"{summary.Artifact.Operation} {summary.Artifact.Id}");
      }
  }

  var step = await client.Workflows.Steps.GetAsync("extract-step-id");
  Console.WriteLine(step.HandleOutputs);
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().steps().list("run_abc123", null, "block_abc123", "step_abc123", null, null, null, null, 10L);
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

Use `steps.list(run.id, block_ids=[...])` when you only need a subset of step summaries. Use `steps.get(run.id, block_id)` when you need the normalized execution record for a single block.

#### Fetch the artifact record

Some blocks persist a durable artifact record. `step.artifact` is only the stable
pointer:

```json theme={null}
{ "operation": "conditional_evaluation", "id": "ceval_abc123" }
```

Use `client.workflows.artifacts.get(step.artifact)` to dereference that pointer.
The response is the backing record flattened with `operation` at the top level,
so consumers can dispatch on one object without juggling an extra `record`
wrapper.

<CodeGroup>
  ```python Python theme={null}
  step = client.workflows.steps.get(run.id, "conditional-block-id")
  if step.artifact:
      artifact = client.workflows.artifacts.get(step.artifact)
      print(artifact.operation)
      print(artifact.matched_condition_ids)
      print(artifact.evaluations)
  ```

  ```typescript TypeScript theme={null}
  const step = await client.workflows.steps.get(run.id, "function-block-id");
  if (step.artifact) {
    const artifact = await client.workflows.artifacts.get(step.artifact);
    console.log(artifact.operation);
    console.log(artifact.output);
    console.log(artifact.error);
  }
  ```

  ```ruby Ruby theme={null}
  step = client.workflows.steps.get(step_id: 'conditional-step-id', run_id: run.id)
  if step.artifact
    artifact = client.workflows.artifacts.get(artifact_id: step.artifact.id)
    puts artifact['operation']
    puts artifact['matched_condition_ids']
    puts artifact['evaluations']
  end
  ```

  ```php PHP theme={null}
  $step = $client->workflows()->steps()->get('conditional-block-id', runId: $run->id);
  if ($step->artifact !== null) {
      $artifact = $client->workflows()->artifacts()->get($step->artifact->id);
      echo $artifact->operation->value . PHP_EOL;
      print_r($artifact->matchedConditionIds ?? null);
      print_r($artifact->evaluations ?? null);
  }
  ```

  ```go Go theme={null}
  package main

  import (
  	"context"
  	"fmt"
  	"log"

  	retab "github.com/retab-dev/retab/clients/go"
  )

  func main() {
  	client, err := retab.NewClient("")
  	if err != nil {
  		log.Fatal(err)
  	}

  	step, err := client.Workflows.Steps.Get(context.Background(), "conditional-step-id", nil)
  	if err != nil {
  		log.Fatal(err)
  	}
  	if step.Artifact != nil {
  		artifact, err := client.Workflows.Artifacts.Get(context.Background(), step.Artifact.ID)
  		if err != nil {
  			log.Fatal(err)
  		}

  		fmt.Println(artifact)
  	}
  }
  ```

  ```rust Rust theme={null}
  use retab::Retab;
  use retab::resources::workflow_steps;

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let step = client
      .workflows().steps()
      .get("conditional-step-id", workflow_steps::GetParams::default())
      .await?;
  if let Some(artifact_ref) = step.artifact {
      let artifact = client.workflows().artifacts().get(&artifact_ref.id).await?;
      println!("{artifact:#?}");
  }
  ```

  ```csharp C# theme={null}
  using Retab;
  using RetabClient = Retab.Retab;

  var client = new RetabClient("YOUR_API_KEY");
  var step = await client.Workflows.Steps.GetAsync("conditional-step-id");
  if (step.Artifact is not null)
  {
      var artifact = await client.Workflows.Artifacts.GetAsync(step.Artifact.Id);
      Console.WriteLine(artifact);
  }
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().steps().get("step_abc123", "run_abc123");
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

`workflows.artifacts.list(run.id)` dereferences every artifact produced by a
run. Pass `operation=` or `block_id=` when you only need a subset.

<CodeGroup>
  ```python Python theme={null}
  condition_records = client.workflows.artifacts.list(
      run.id,
      operation="conditional_evaluation",
  )
  ```

  ```typescript TypeScript theme={null}
  const conditionRecords = await client.workflows.artifacts.list({
    runId: run.id,
    operation: "conditional_evaluation",
  });
  ```

  ```ruby Ruby theme={null}
  condition_records = client.workflows.artifacts.list(
    run_id: run.id,
    operation: 'conditional_evaluation',
  )
  ```

  ```php PHP theme={null}
  use Retab\Resource\StepArtifactRefOperation;

  $conditionRecords = $client->workflows()->artifacts()->list(
      runId: $run->id,
      operation: StepArtifactRefOperation::ConditionalEvaluation,
  );
  ```

  ```go Go theme={null}
  package main

  import (
  	"context"
  	"fmt"
  	"log"

  	retab "github.com/retab-dev/retab/clients/go"
  )

  func main() {
  	client, err := retab.NewClient("")
  	if err != nil {
  		log.Fatal(err)
  	}

  	runID := "wfrun_abc123"
  	operation := retab.WorkflowArtifactsOperation(retab.StepArtifactRefOperationConditionalEvaluation)
  	records, err := client.Workflows.Artifacts.List(context.Background(), &retab.WorkflowArtifactsListParams{
  		RunID:     &runID,
  		Operation: &operation,
  	})
  	if err != nil {
  		log.Fatal(err)
  	}

  	fmt.Println(records.Data)
  }
  ```

  ```rust Rust theme={null}
  use retab::{enums::WorkflowArtifactsOperation, resources::workflow_artifacts::ListParams, Retab};

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let run_id = "wfrun_abc123";
  let condition_records = client
      .workflows().artifacts()
      .list(ListParams {
          run_id: Some(run_id.into()),
          operation: Some(WorkflowArtifactsOperation::ConditionalEvaluation),
          ..Default::default()
      })
      .await?;

  println!("{:#?}", condition_records.data);
  ```

  ```csharp C# theme={null}
  using Retab;
  using RetabClient = Retab.Retab;

  var client = new RetabClient("YOUR_API_KEY");
  var conditionRecords = await client.Workflows.Artifacts.ListAsync(
      new WorkflowArtifactsListOptions
      {
          RunId = run.Id,
          Operation = WorkflowArtifactsOperation.ConditionalEvaluation,
      }
  );

  Console.WriteLine(conditionRecords);
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().artifacts().list("run_abc123", null, "block_abc123", "step_abc123", null, null, 10L);
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

| operation                | produced by               | record includes                                            |
| ------------------------ | ------------------------- | ---------------------------------------------------------- |
| `extraction`             | `extract`                 | extraction result, choices, likelihoods, schema details    |
| `split`                  | `split`                   | split result and output document grouping                  |
| `classification`         | `classifier`              | selected class and consensus details                       |
| `parse`                  | `parse`                   | parsed document content                                    |
| `edit`                   | `edit`                    | edited document result                                     |
| `partition`              | `for_each_sentinel_start` | partitioned items for the loop                             |
| `conditional_evaluation` | `conditional`             | `evaluations`, `selected_handles`, `matched_condition_ids` |
| `while_loop_termination` | `while_loop`              | termination reason and final condition evaluations         |
| `api_call_invocation`    | `api_call`                | request/response attempts, retry trace, and final error    |
| `function_invocation`    | `function`                | function inputs, output, duration, and final error         |

#### Build workflows from code

The same SDK can create and publish workflow graphs:

<CodeGroup>
  ```python Python theme={null}
  workflow = client.workflows.create(name="Invoice Pipeline")
  blocks = client.workflows.blocks.list(workflow.id)
  start_document_block = next(block for block in blocks.data if block.type == "start_document")

  extract_block = client.workflows.blocks.create(
  workflow.id,
  id="extract-invoice",
  type="extract",
  label="Extract Invoice",
  position_x=320,
  position_y=0,
  config={
  "json_schema": {
  "type": "object",
  "properties": {
  "invoice_number": {"type": "string"},
  "total_amount": {"type": "number"},
  },
  },
  },
  )

  client.workflows.edges.create(
  workflow.id,
  id="edge-start-to-extract",
  source_block=start_document_block.id,
  target_block=extract_block.id,
  source_handle="output-file-0",
  target_handle="input-file-0",
  )

  client.workflows.publish(workflow.id, description="Initial version")

  ```

  ```typescript TypeScript theme={null}
  const workflow = await client.workflows.create("Invoice Pipeline");
  const blocks = await client.workflows.blocks.list({ workflowId: workflow.id });
  const startDocumentBlock = blocks.data.find((block) => block.type === "start_document");

  if (!startDocumentBlock) {
    throw new Error("Workflow is missing a start_document block");
  }

  const extractBlock = await client.workflows.blocks.create(workflow.id, "extract", "extract-invoice", "Extract Invoice", 320, 0, undefined, undefined, {
      json_schema: {
        type: "object",
        properties: {
          invoice_number: { type: "string" },
          total_amount: { type: "number" },
        },
      },
    });

  await client.workflows.edges.create(workflow.id, startDocumentBlock.id, extractBlock.id, "edge-start-to-extract", "output-file-0", "input-file-0");

  await client.workflows.publish(workflow.id, "Initial version");
  ```

  ```go Go theme={null}
  workflow, err := client.Workflows.Create(ctx, retab.CreateWorkflowRequest{
  	Name: "Invoice Pipeline",
  })
  if err != nil {
  	log.Fatal(err)
  }

  blocks, err := client.Workflows.Blocks.List(ctx, workflow.ID)
  if err != nil {
  	log.Fatal(err)
  }

  var startDocumentBlock retab.WorkflowBlock
  for _, block := range blocks.Data {
  	if block.Type == "start_document" {
  		startDocumentBlock = block
  		break
  	}
  }

  extractBlock, err := client.Workflows.Blocks.Create(ctx, workflow.ID, retab.WorkflowBlockCreateRequest{
  	ID:        "extract-invoice",
  	Type:      "extract",
  	Label:     "Extract Invoice",
  	PositionX: 320,
  	PositionY: 0,
  	Config: map[string]any{
  		"json_schema": map[string]any{
  			"type": "object",
  			"properties": map[string]any{
  				"invoice_number": map[string]any{"type": "string"},
  				"total_amount":   map[string]any{"type": "number"},
  			},
  		},
  	},
  })
  if err != nil {
  	log.Fatal(err)
  }

  _, err = client.Workflows.Edges.Create(ctx, workflow.ID, retab.WorkflowEdgeCreateRequest{
  	ID:           "edge-start-to-extract",
  	SourceBlock:  startDocumentBlock.ID,
  	TargetBlock:  extractBlock.ID,
  	SourceHandle: "output-file-0",
  	TargetHandle: "input-file-0",
  })
  if err != nil {
  	log.Fatal(err)
  }

  _, err = client.Workflows.Publish(ctx, workflow.ID, retab.PublishWorkflowRequest{
  	Description: "Initial version",
  })
  if err != nil {
  	log.Fatal(err)
  }
  ```

  ```ruby Ruby theme={null}
  require 'retab'

  client = Retab::Client.new(api_key: ENV['RETAB_API_KEY'])

  workflow = client.workflows.create(name: 'Invoice Pipeline')
  blocks = client.workflows.blocks.list(workflow_id: workflow.id)
  start_document_block = blocks.data.find { |block| block.type == 'start_document' }

  raise 'Workflow is missing a start_document block' if start_document_block.nil?

  extract_block = client.workflows.blocks.create(
    workflow_id: workflow.id,
    id: 'extract-invoice',
    type: 'extract',
    label: 'Extract Invoice',
    position_x: 320,
    position_y: 0,
    config: {
      json_schema: {
        type: 'object',
        properties: {
          invoice_number: { type: 'string' },
          total_amount: { type: 'number' },
        },
      },
    },
  )

  client.workflows.edges.create(
    workflow_id: workflow.id,
    id: 'edge-start-to-extract',
    source_block: start_document_block.id,
    target_block: extract_block.id,
    source_handle: 'output-file-0',
    target_handle: 'input-file-0',
  )

  client.workflows.publish(workflow_id: workflow.id, description: 'Initial version')
  ```

  ```php PHP theme={null}
  <?php
  require 'vendor/autoload.php';

  use Retab\Client;
  use Retab\Resource\WorkflowBlockCreateRequestType;
  use Retab\Resource\WorkflowBlockType;

  $client = new Client();

  $workflow = $client->workflows()->create(name: 'Invoice Pipeline');
  $blocks = $client->workflows()->blocks()->list($workflow->id);

  $startDocumentBlock = null;
  foreach ($blocks->data as $block) {
      if ($block->type === WorkflowBlockType::StartDocument) {
          $startDocumentBlock = $block;
          break;
      }
  }

  if ($startDocumentBlock === null) {
      throw new RuntimeException('Workflow is missing a start_document block');
  }

  $extractBlock = $client->workflows()->blocks()->create(
      workflowId: $workflow->id,
      type: WorkflowBlockCreateRequestType::Extract,
      id: 'extract-invoice',
      label: 'Extract Invoice',
      positionX: 320,
      positionY: 0,
      config: [
          'json_schema' => [
              'type' => 'object',
              'properties' => [
                  'invoice_number' => ['type' => 'string'],
                  'total_amount' => ['type' => 'number'],
              ],
          ],
      ],
  );

  $client->workflows()->edges()->create(
      workflowId: $workflow->id,
      sourceBlock: $startDocumentBlock->id,
      targetBlock: $extractBlock->id,
      sourceHandle: 'output-file-0',
      targetHandle: 'input-file-0',
      id: 'edge-start-to-extract',
  );

  $client->workflows()->publish($workflow->id, description: 'Initial version');
  ```

  ```rust Rust theme={null}
  use retab::{
      enums::{WorkflowBlockCreateRequestType, WorkflowBlockType},
      models::{CreateWorkflowRequest, WorkflowBlockCreateRequest, WorkflowEdgeCreateRequest},
      resources::{workflow_blocks, workflow_edges, workflows},
      Retab,
  };

  let client = Retab::new(std::env::var("RETAB_API_KEY")?);
  let workflow = client
      .workflows()
      .create(workflows::CreateParams::new(CreateWorkflowRequest {
          name: Some("Invoice Pipeline".to_string()),
          description: None,
          project_id: "project_abc".to_string(),
      }))
      .await?;
  let blocks = client
      .workflows().blocks()
      .list(workflow_blocks::ListParams::new(&workflow.id))
      .await?;
  let start = blocks
      .data
      .iter()
      .find(|block| block.type_ == WorkflowBlockType::StartDocument)
      .expect("workflow is missing a start_document block");

  let mut request = WorkflowBlockCreateRequest::new(&workflow.id, WorkflowBlockCreateRequestType::Extract);
  request.id = Some("extract-invoice".to_string());
  request.label = Some("Extract Invoice".to_string());
  request.position_x = Some(320.0);
  request.config = Some([(
      "json_schema".to_string(),
      serde_json::json!({
          "type": "object",
          "properties": {
              "invoice_number": { "type": "string" },
              "total_amount": { "type": "number" }
          }
      }),
  )]
  .into_iter()
  .collect());
  let extract = client
      .workflows().blocks()
      .create(workflow_blocks::CreateParams::new(request))
      .await?;

  let mut edge = WorkflowEdgeCreateRequest::new(&workflow.id, &start.id, &extract.id);
  edge.id = Some("edge-start-to-extract".to_string());
  edge.source_handle = Some("output-file-0".to_string());
  edge.target_handle = Some("input-file-0".to_string());
  client
      .workflows().edges()
      .create(workflow_edges::CreateParams::new(edge))
      .await?;
  client
      .workflows()
      .publish(&workflow.id, workflows::PublishParams::default())
      .await?;
  ```

  ```csharp C# theme={null}
  using System.Collections.Generic;
  using System.Linq;
  using Retab;
  using RetabClient = Retab.Retab;

  var client = new RetabClient("YOUR_API_KEY");
  var workflow = await client.Workflows.CreateAsync(
      new WorkflowsCreateOptions { Name = "Invoice Pipeline" }
  );
  var blocks = await client.Workflows.Blocks.ListAsync(
      new WorkflowBlocksListOptions { WorkflowId = workflow.Id }
  );
  var start = blocks.Data.First(block => block.Type == WorkflowBlockType.StartDocument);
  var extract = await client.Workflows.Blocks.CreateAsync(
      new WorkflowBlocksCreateOptions
      {
          WorkflowId = workflow.Id,
          Id = "extract-invoice",
          Type = WorkflowBlockCreateRequestType.Extract,
          Label = "Extract Invoice",
          PositionX = 320,
          Config = new Dictionary<string, object>
          {
              ["json_schema"] = new Dictionary<string, object>
              {
                  ["type"] = "object",
                  ["properties"] = new Dictionary<string, object>
                  {
                      ["invoice_number"] = new Dictionary<string, object> { ["type"] = "string" },
                      ["total_amount"] = new Dictionary<string, object> { ["type"] = "number" },
                  },
              },
          },
      }
  );
  await client.Workflows.Edges.CreateAsync(
      new WorkflowEdgesCreateOptions
      {
          WorkflowId = workflow.Id,
          SourceBlock = start.Id,
          TargetBlock = extract.Id,
          SourceHandle = "output-file-0",
          TargetHandle = "input-file-0",
          Id = "edge-start-to-extract",
      }
  );
  await client.Workflows.PublishAsync(workflow.Id, new WorkflowsPublishOptions());
  ```

  ```java Java theme={null}
  import com.retab.RetabClient;

  public final class Example {
    public static void main(String[] args) throws Exception {
      RetabClient client = new RetabClient(System.getenv("RETAB_API_KEY"));

      var result = client.workflows().create("Invoice Processing", "Extract invoice fields", null);
      System.out.println(result);
    }
  }
  ```
</CodeGroup>

Use `client.workflows.list()` or `client.workflows.get(workflow_id)` when you need to browse existing workflows before launching a run.

### Reading Workflow Results

The standard production pattern is to run the workflow, keep the returned `run.id`,
and poll the run until `lifecycle.status` reaches `completed`, `error`, `cancelled`,
or `awaiting_review`.

1. Start the workflow from the SDK or API
2. Receive a `run.id` and an initial lifecycle immediately
3. Poll the workflow run until it finishes or waits for review
4. Read the step results from the completed run

The workflow run is the source of truth for execution state and outputs. This is
enough for many scripts, internal tools, and backend services.

## Workflow Execution Order

Workflows execute in **topological order** based on the block connections:

1. Start from Document input blocks
2. Process each block once all its inputs are ready
3. Continue until all blocks are processed or an error occurs
4. Read outputs from the completed run and its step results

If a block fails, execution stops and the error is displayed on that block.

### Conditional Routing

When using **Classifier** or **If/Else** blocks, only the branches that receive data are executed. Blocks on skipped branches are marked as "skipped" rather than failed.

## Viewing Generated Code

Every workflow can be exported as Python code. Click **View Code** in the sidebar to see the equivalent SDK calls for your workflow. This is useful for:

* Integrating workflows into your existing codebase
* Running workflows in production environments
* Understanding how the visual blocks translate to API calls

## Best Practices

<AccordionGroup>
  <Accordion title="Start simple">
    Begin with a single Extract or Parse block, then gradually add complexity.
    Test each addition before moving on.
  </Accordion>

  <Accordion title="Use descriptive labels">
    Rename blocks to describe their purpose (e.g., "Invoice Data" instead of
    "Extract 1"). This makes complex workflows easier to understand.
  </Accordion>

  <Accordion title="Add notes for documentation">
    Use Note blocks to document sections of your workflow. They don't affect
    execution but help explain the logic.
  </Accordion>

  <Accordion title="Validate with review gates">
    For critical data, add a review gate to the extraction block. This ensures a
    reviewer checks low-likelihood results before they proceed.
  </Accordion>

  <Accordion title="Use Classifier for document routing">
    When processing different document types, use a Classifier block to route each
    document to the appropriate extraction schema.
  </Accordion>

  <Accordion title="Test with sample documents">
    Before deploying, run your workflow with representative sample documents to
    catch edge cases.
  </Accordion>
</AccordionGroup>

## Example: Invoice Processing Workflow

Here's a common workflow pattern for processing invoices:

```mermaid theme={null}
flowchart LR
    A[Start\nDocument] --> B[Extract\nInvoice Fields]
    B --> C[Review Overlay\nHuman Review]

    style A fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
    style B fill:#e3f2fd,stroke:#2196f3,color:#0d47a1
    style C fill:#fff3e0,stroke:#ff9800,color:#e65100
```

1. **Start** block accepts the invoice PDF
2. **Extract** block pulls out vendor, amount, date, line items
3. The extract block's **review gate** flags low-likelihood extractions for review
4. Read the verified data from the completed workflow run

## Example: Multi-Document Classification Workflow

For workflows that process mixed document bundles:

```mermaid theme={null}
flowchart TD
    A[Start\nDocument Bundle] --> B{Classifier}

    B -->|Invoice| C[Extract\nInvoice Schema]
    B -->|Contract| D[Extract\nContract Schema]
    B -->|Receipt| E[Extract\nReceipt Schema]

    C --> F[Function\nCalculate Totals]
    D --> G[Function\nExtract Dates]
    E --> H[Function\nCategorize Items]

    F --> I[Merge JSON]
    G --> I
    H --> I

    style A fill:#e8f5e9,stroke:#4caf50,color:#1b5e20
    style B fill:#f3e5f5,stroke:#9c27b0,color:#4a148c
    style C fill:#e3f2fd,stroke:#2196f3,color:#0d47a1
    style D fill:#e3f2fd,stroke:#2196f3,color:#0d47a1
    style E fill:#e3f2fd,stroke:#2196f3,color:#0d47a1
    style F fill:#fff8e1,stroke:#ffc107,color:#ff6f00
    style G fill:#fff8e1,stroke:#ffc107,color:#ff6f00
    style H fill:#fff8e1,stroke:#ffc107,color:#ff6f00
    style I fill:#e0f2f1,stroke:#009688,color:#004d40
```

1. **Classifier** routes documents by category (Invoice, Contract, Receipt)
2. Each **Extract** block uses a document-specific schema
3. **Function** blocks compute derived fields for each document type
4. **Merge JSON** combines results from all branches into a single output