Steps

Steps are preconfigured document processing pipelines. Each step defines a specific processing workflow — you provide a document and DocuTray executes the step's pipeline, returning structured results. Steps are always executed asynchronously.

You can also manage your organization's steps over the API — list, inspect, create, update, and delete them without going through support. See Managing Steps.

Quick Start

from pathlib import Path
from docutray import Client

client = Client(api_key="YOUR_API_KEY")

# Execute a step
status = client.steps.run_async(
    step_id="step_abc123",
    file=Path("document.pdf")
)

# Wait for completion
result = status.wait()

if result.is_success():
    print(result.data)

Response

# status is a StepExecutionStatus
print(status.execution_id)        # "exec_abc123"
print(status.status)              # "SUCCESS"
print(status.data)                # Extracted data
print(status.original_filename)   # "document.pdf"
print(status.request_timestamp)   # When execution started
print(status.response_timestamp)  # When execution completed

Polling and Status

Steps are always asynchronous. You can poll for status manually or use the SDK's built-in wait() method.

Using wait() (Recommended)

status = client.steps.run_async(
    step_id="step_abc123",
    file=Path("document.pdf")
)

# Wait with automatic polling
result = status.wait()

if result.is_success():
    print("Step completed successfully")
    print(result.data)
elif result.is_error():
    print(f"Step failed: {result.error}")

Manual Polling

status = client.steps.get_status("exec_abc123")

if status.is_success():
    print(status.data)
elif status.status == "ENQUEUED" or status.status == "PROCESSING":
    print("Still processing...")

Input Methods

Steps support the same input methods as Convert and Identify: file upload, URL, and base64.

# File upload
status = client.steps.run_async(
    step_id="step_abc123",
    file=Path("document.pdf")
)

# URL
status = client.steps.run_async(
    step_id="step_abc123",
    url="https://example.com/document.pdf"
)

# Base64
import base64
with open("document.pdf", "rb") as f:
    encoded = base64.b64encode(f.read()).decode()

status = client.steps.run_async(
    step_id="step_abc123",
    file_base64=encoded,
    content_type="application/pdf"
)

Parameters

Complete Code

from pathlib import Path
from docutray import Client, NotFoundError, DocuTrayError

client = Client(api_key="YOUR_API_KEY")

try:
    # Execute a step with metadata
    status = client.steps.run_async(
        step_id="step_abc123",
        file=Path("invoice.pdf"),
        document_metadata={"source": "email", "customer_id": "cust_456"}
    )

    print(f"Execution started: {status.execution_id}")

    # Wait for completion
    result = status.wait()

    if result.is_success():
        print("Step completed!")
        print(f"Result: {result.data}")
    elif result.is_error():
        print(f"Step failed: {result.error}")

except NotFoundError:
    print("Step not found — check the step ID")
except DocuTrayError as e:
    print(f"Error: {e.message}")
finally:
    client.close()

Managing Steps

Besides executing steps, your organization can manage them over the API:

Every request is scoped to the organization the API key belongs to — never to the other organizations its owner may be a member of. A step from another organization is indistinguishable from one that does not exist: both return 404.

Setup

Every management example below builds on this snippet. Run it first, or paste it above whichever example you are copying.

import requests

BASE_URL = "https://app.docutray.com"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
JSON_HEADERS = {**HEADERS, "Content-Type": "application/json"}

The Step Resource

List Steps

Steps are returned newest first, paginated.

response = requests.get(
    f"{BASE_URL}/api/steps",
    headers=HEADERS,
    params={"search": "invoice", "page": 1, "limit": 20},
)
response.raise_for_status()
body = response.json()

for step in body["data"]:
    print(f"{step['id']}: {step['name']} — {step['executions30d']} runs in 30d")

print(f"Total: {body['pagination']['total']}")

Response

{
  "data": [
    {
      "id": "step_abc123",
      "name": "Invoice intake",
      "description": "Identifies and extracts incoming invoices",
      "isActive": true,
      "identification": { "types": ["invoice", "credit_note"] },
      "executions30d": 412,
      "lastRunAt": "2024-01-15T10:30:00.000Z",
      "createdAt": "2023-11-02T08:12:00.000Z",
      "updatedAt": "2024-01-10T16:45:00.000Z",
      "createdBy": "user_xyz789"
    }
  ],
  "pagination": {
    "total": 1,
    "page": 1,
    "limit": 20
  }
}

Get a Step

response = requests.get(f"{BASE_URL}/api/steps/step_abc123", headers=HEADERS)
response.raise_for_status()

step = response.json()["data"]
print(step["name"], step["isActive"], step["identification"]["types"])

The response is a single step wrapped in data, with the same fields as a list entry.

Create a Step

Only name is required. identification.types takes document type codes — the stable codeType values you get from Document Types — and every code must be a published type your organization can access, either its own or a public one. Draft types are rejected whoever owns them — your own or a public one not yet published — so publish first. Duplicate codes are collapsed, so the stored list can be shorter than the one you sent.

response = requests.post(
    f"{BASE_URL}/api/steps",
    headers=JSON_HEADERS,
    json={
        "name": "Invoice intake",
        "description": "Identifies and extracts incoming invoices",
        "isActive": True,
        "identification": {"types": ["invoice", "credit_note"]},
    },
)

if response.status_code == 400:
    print(response.json())  # includes "inaccessibleTypes" when a code is not allowed
response.raise_for_status()

step = response.json()["data"]
print(f"Created {step['id']}")

A successful create returns 201 with the new step under data, with executions30d at 0 and lastRunAt at null.

Omit identification — or send an empty list — to let the step consider every document type your organization can access. If any code is not accessible, the request fails with 400, the response lists the offending codes under inaccessibleTypes, and nothing is created.

Fields the contract does not define are ignored rather than rejected: sending aiModelConfig, webhookUrl or any other internal setting still returns 201, and the setting is dropped. PATCH does the same — except that a body made up only of unrecognized fields is left with nothing to change and fails with 400.

Update a Step

PATCH is partial: fields you leave out keep their current value. At least one field must be present.

# Rename and restrict the step to a single document type
response = requests.patch(
    f"{BASE_URL}/api/steps/step_abc123",
    headers=JSON_HEADERS,
    json={
        "name": "Invoice intake (EU)",
        "identification": {"types": ["invoice"]},
    },
)
response.raise_for_status()

# Deactivate without touching anything else
requests.patch(
    f"{BASE_URL}/api/steps/step_abc123",
    headers=JSON_HEADERS,
    json={"isActive": False},
).raise_for_status()

A successful update returns 200 with the whole updated step under data, in the same shape GET returns — not an empty 204.

An update is all-or-nothing. identification is validated before anything is written, so a body that mixes it with scalar fields lands entirely or not at all: PATCH {"name": "...", "identification": {"types": []}} against a fixed-type step returns 409 and the rename is discarded along with it. If you want a rename to survive an identification change being rejected, send them as two requests.

identification.types replaces the whole list — it is not merged. To add a type, send the existing codes plus the new one. Sending a non-empty list also switches the step to automatic identification, replacing any fixed document type it was configured with.

Codes the step already has configured are always accepted, even if your organization can no longer access them — for example a public type that was later unpublished — so an accessibility check will never reject a code you just read back. Adding a code you cannot access is still rejected with 400. That is only about accessibility, though: re-sending identification unchanged is not safe in general, for the reasons below.

Concurrency is not uniform across fields. A write that includes identification.types reads, merges, and writes the step's configuration under a version guard, retrying transparently; only sustained contention returns 409, and then you should re-read the step and re-apply your change. A write touching only name, description or isActive has no such guard: two concurrent updates both succeed and the last one wins. Serialize your own writes if that matters to you.

Delete a Step

response = requests.delete(f"{BASE_URL}/api/steps/step_abc123", headers=HEADERS)

if response.status_code == 409:
    print(response.json())  # executions still in flight
response.raise_for_status()

print(response.json()["data"])  # {"id": "step_abc123", "deleted": True}

A successful delete returns 200 with {"id": "...", "deleted": true} under data — not an empty 204.

Deletion is rejected with 409 while the step has executions in flight — enqueued, in progress, or uploading. An upload that never completed stops blocking the deletion 15 minutes after it started, so a request that died mid-upload does not keep the step alive forever. The response normally reports how many under activeExecutions; if executions keep arriving faster than the delete can be applied, it returns 409 without that field. Either way: wait for the executions to finish, or deactivate the step first so no new ones arrive, and retry.

Errors

SDK Reference

For detailed class and method documentation:

• Python SDK Steps Reference — execution
• Node.js SDK Steps Reference — execution
• CLI Steps Commands — execution
• REST API Reference — execution and management
• Steps Webhook Events — STEP_STARTED, STEP_COMPLETED, STEP_FAILED