SDK Reference¶

This guide covers the Orcheo Python SDK for programmatic workflow execution. The SDK ships two building blocks — OrcheoClient (URL/header composition) and HttpWorkflowExecutor (synchronous HTTP execution with retries) — plus the orcheo CLI for everything else (uploading, listing, credentials, publishing; see the CLI Reference).

Installation¶

pip install orcheo-sdk
# or with uv
uv tool install orcheo-sdk

Quick Start¶

import os
from orcheo_sdk import HttpWorkflowExecutor, OrcheoClient

client = OrcheoClient(base_url="http://localhost:2025")
executor = HttpWorkflowExecutor(
    client,
    auth_token=os.environ.get("ORCHEO_SERVICE_TOKEN"),
)

result = executor.trigger_run(
    "my-workflow-id",
    workflow_version_id="version-id",
    triggered_by="sdk",
    inputs={"query": "What is RAG?"},
)
print(result)

Tip: orcheo code scaffold <workflow_id> generates this snippet pre-filled with the workflow's latest version id.

`OrcheoClient`¶

A lightweight helper for composing Orcheo backend requests:

client = OrcheoClient(
    base_url="https://orcheo.example.com",
    default_headers={"X-Orcheo-Workspace": "my-workspace"},  # optional
    request_timeout=30.0,
)

client.workflow_collection_url()          # .../api/workflows
client.workflow_trigger_url("wf-id")      # .../api/workflows/wf-id/runs
client.websocket_url("wf-id")             # ws(s)://.../ws/workflow/wf-id

`HttpWorkflowExecutor`¶

A synchronous executor built on httpx with exponential-backoff retries for transient server errors (500/502/503/504 by default):

executor = HttpWorkflowExecutor(
    client,
    auth_token=os.environ.get("ORCHEO_SERVICE_TOKEN"),  # sent as Bearer token
    timeout=30.0,
    max_retries=3,
    backoff_factor=0.5,
)

Triggering runs¶

result = executor.trigger_run(
    "workflow-id",
    workflow_version_id="version-id",     # required
    triggered_by="ci",                    # actor recorded on the run
    inputs={"query": "search query"},
    runnable_config={"configurable": {"temperature": 0.7}},  # optional
)

The return value is the backend's run payload (a dict). Failures raise RuntimeError after retries are exhausted; the underlying httpx exception is attached as the cause.

Validating credentials¶

report = executor.validate_credentials("workflow-id", actor="ci")

Triggers vault credential validation for the workflow and returns the backend response.

Streaming Execution¶

Run updates stream over the backend WebSocket endpoint. Compose the URL with client.websocket_url(workflow_id) and connect with any WebSocket client, passing the token per the WebSocket authentication options. The orcheo workflow run CLI command streams these updates by default.

Environment-Based Configuration¶

The CLI (and code generated by orcheo code scaffold) respects:

Variable	Description
`ORCHEO_API_URL`	Backend API URL
`ORCHEO_SERVICE_TOKEN`	Service token for authentication

State Management¶

Orcheo workflows maintain a typed state object that flows between nodes (orcheo.graph.state.State):

from orcheo.graph.state import State

# State provides:
#   inputs               - workflow inputs
#   node_results         - node outputs (keyed by node name)
#   messages             - conversation messages (MessagesState)
#   structured_response  - final structured output
#   config               - runtime config

The node_results dictionary accumulates outputs from task nodes, enabling downstream nodes to access upstream outputs via variable interpolation (e.g., {{node_results.retriever.documents}}).

Integration Example: Batch Processing¶

HttpWorkflowExecutor is synchronous; parallelize with threads if needed:

import os
from concurrent.futures import ThreadPoolExecutor
from orcheo_sdk import HttpWorkflowExecutor, OrcheoClient

client = OrcheoClient(base_url="http://localhost:2025")
executor = HttpWorkflowExecutor(
    client, auth_token=os.environ.get("ORCHEO_SERVICE_TOKEN")
)


def run_query(query: str) -> dict:
    return executor.trigger_run(
        "query-processor",
        workflow_version_id="version-id",
        triggered_by="batch",
        inputs={"query": query},
    )


with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(run_query, ["q1", "q2", "q3"]))