Workflows - Retab Docs

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.

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.

Creating a Workflow

Navigate to the Workflows section in your dashboard
Click Create Workflow to open a new canvas
Drag blocks from the sidebar onto the canvas
Connect blocks by dragging from output handles to input handles
Configure each block by clicking on it
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.

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:

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)

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);

$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);

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)

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);

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)
}

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:#?}");

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);
  }
}

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:

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:

From the Dashboard

Switch to Run Mode
Upload a document to each Document input block
Click Run Workflow
Watch as each block processes (status indicators show progress)
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.

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")

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");
}

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)
}

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
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');
}

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");
}

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:?}");

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);
  }
}

Run and wait for completion

Workflows support two input maps:

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

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)

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);
    }
  }
}

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)
		}
	}
}

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
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);
        }
    }
}

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);

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));

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);
  }
}

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.

# 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)

// 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);
}

// 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)
}

# 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
// 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);
}

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);

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);

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);
  }
}

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:

{ "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.

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)

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);
}

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

$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);
}

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)
	}
}

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:#?}");
}

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);
}

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);
  }
}

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

condition_records = client.workflows.artifacts.list(
    run.id,
    operation="conditional_evaluation",
)

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

condition_records = client.workflows.artifacts.list(
  run_id: run.id,
  operation: 'conditional_evaluation',
)

use Retab\Resource\StepArtifactRefOperation;

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

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)
}

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);

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);

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);
  }
}

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:

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")

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");

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)
}

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
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');

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?;

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());

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);
  }
}

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.

Start the workflow from the SDK or API
Receive a run.id and an initial lifecycle immediately
Poll the workflow run until it finishes or waits for review
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:

Start from Document input blocks
Process each block once all its inputs are ready
Continue until all blocks are processed or an error occurs
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

Start simple

Begin with a single Extract or Parse block, then gradually add complexity. Test each addition before moving on.

Use descriptive labels

Rename blocks to describe their purpose (e.g., “Invoice Data” instead of “Extract 1”). This makes complex workflows easier to understand.

Add notes for documentation

Use Note blocks to document sections of your workflow. They don’t affect execution but help explain the logic.

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.

Use Classifier for document routing

When processing different document types, use a Classifier block to route each document to the appropriate extraction schema.

Test with sample documents

Before deploying, run your workflow with representative sample documents to catch edge cases.

Example: Invoice Processing Workflow

Here’s a common workflow pattern for processing invoices:

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

Example: Multi-Document Classification Workflow

For workflows that process mixed document bundles:

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

​What are Workflows?

​Workflow Evals

​Experiments

​Creating a Workflow

​Connecting Blocks

​Connection Rules

​Declarative Workflow Spec

​Edit Mode vs Run Mode

​Edit Mode

​Run Mode

​Running a Workflow

​From the Dashboard

​Using the SDK

​Discover input block IDs

​Run and wait for completion

​Inspect step outputs

​Fetch the artifact record

​Build workflows from code

​Reading Workflow Results

​Workflow Execution Order

​Conditional Routing

​Viewing Generated Code

​Best Practices

​Example: Invoice Processing Workflow

​Example: Multi-Document Classification Workflow