Workflows SDK — Python

ltinteg-workflows-sdk is the generated Python client for the workflow engine. It wraps the same https://workflow-api.orkestia.dev REST surface as the Node SDK: one start_… helper per workflow, plus a small runtime client.

Requires Python ≥ 3.10. Depends on pydantic>=2 and requests>=2.28.

Install

Versioned wheels ship with each GitHub release of the SDK. Install the wheel that matches the catalog you are targeting:

pip install ltinteg-workflows-sdk

If your environment cannot see the published wheel, download it from the SDK's GitHub Releases page and pip install the file. The release tag matches the version in the package metadata.

Start a run

import os
from ltinteg_workflows_sdk import LtIntegWorkflowsClient
from ltinteg_workflows_sdk.github import auth

client = LtIntegWorkflowsClient(
    "https://workflow-api.orkestia.dev",
    token=os.environ["ORKESTIA_TOKEN"],  # org-member or end-user Bearer
)

run = auth.start_validate_token(client, token="ghp_…")
print(run.workflow_id, run.terminal_status)

Do not pass organization_uuid. The server resolves it from the token.

run.terminal_status is the contract you should branch on: "success", "failed", or None while the run is still open. run.state is a progress / debug label, not the outcome. Failure details land in run.errors.

The same loop as Node and REST

import os
import time
from ltinteg_workflows_sdk import LtIntegWorkflowsClient

client = LtIntegWorkflowsClient(
    "https://workflow-api.orkestia.dev",
    token=os.environ["ORKESTIA_TOKEN"],
)

run = client.start(
    "aws.s3.create_bucket",
    initial_data={
        "bucket": "my-app-assets",
        "connection_uuid": "…",
        "region": "us-east-1",
    },
)

# Point-in-time status until terminal
while run.terminal_status is None:
    time.sleep(2)
    run = client.get(run.workflow_id)

if run.terminal_status != "success":
    raise RuntimeError(run.errors)

print(run.state_data)

You can also call the REST surface directly — the SDK is a typed convenience, not a second API:

curl -sS https://workflow-api.orkestia.dev/api/workflows/$WORKFLOW_ID \
  -H "Authorization: Bearer $ORKESTIA_TOKEN"

See API & tooling for the full endpoint table (start, schema, stream, history, retry).

Auth

Same two-token model as Node. For org automation, pass an org-member token or an API token from Settings → API tokens. For an app acting as a signed-in end-user, pass the JWT minted by @orkestia/auth (typically your backend received it from the browser, or the browser called the workflow API itself).

client = LtIntegWorkflowsClient(
    "https://workflow-api.orkestia.dev",
    token=end_user_jwt,
)

Node vs Python

If you need live transition streaming in Python today, open GET /api/workflows/{id}/stream with requests (SSE) using the same Bearer token — or use the Node SDK.

Related