strait-python 0.1.1

Python SDK for the Strait API

The official Python SDK for Strait -- an open-source job execution and workflow orchestration platform for humans and AI agents. Define jobs, build DAG workflows, run durable AI agents with cost tracking, and manage the full run lifecycle from Python.

Sync and async clients, 19 domain services (~186 operations), authoring DSL, composition helpers, FSM state machines, and PEP 561 type hints.

Website | Platform Repo | Documentation | PyPI

Installation

pip install strait-python

Requires Python 3.11+.

Quick Start

from strait import Client

client = Client(base_url="https://api.strait.dev", api_key="your-key")

# List jobs
jobs = client.jobs.list()

# Trigger a job
run = client.jobs.trigger(job["id"], {"payload": {"key": "value"}})

# Get a run
run = client.runs.get(run["id"])

Async

from strait import AsyncClient

async with AsyncClient.from_env() as client:
    jobs = await client.jobs.list()
    run = await client.jobs.trigger("job-id", {"payload": {}})

Configuration

Three ways to create a client:

# From environment variables (STRAIT_BASE_URL, STRAIT_API_KEY)
client = Client.from_env()

# From strait.json file (token always from STRAIT_API_KEY env var)
client = Client.from_file()

# Inline
client = Client(base_url="https://api.strait.dev", api_key="your-key")

All methods work with AsyncClient too. See Configuration Guide for details.

Features

Domain Operations (19 Services)

All 186 API operations organized into typed service classes:

Authoring DSL

Define jobs, workflows, and AI agents with a type-safe DSL:

from strait.authoring import define_job, define_workflow, JobOptions, WorkflowOptions
from strait.authoring import job_step, approval_step

wf = define_workflow(WorkflowOptions(
    name="Order Pipeline", slug="order-pipeline", project_id="proj-1",
    steps=[
        job_step("validate", "validate-job"),
        job_step("charge", "charge-job", depends_on=["validate"]),
        approval_step("approve", depends_on=["charge"]),
        job_step("ship", "ship-job", depends_on=["approve"]),
    ],
))

See Authoring Guide for AI agents, event definitions, RunContext, and test harness.

Composition Helpers

from strait.composition import with_retry, wait_for_run, paginate, RetryOptions

# Retry with exponential backoff
result = with_retry(lambda: client.jobs.trigger("j1", payload), RetryOptions(attempts=5))

# Wait for a run to complete
run = wait_for_run(
    get_run=lambda rid: client.runs.get(rid),
    get_status=lambda r: r["status"],
    run_id="run-1",
)

See Composition Guide for cost budgets, checkpoint resume, pagination, and more.

Error Handling

All errors inherit from StraitError with typed exceptions for each HTTP status:

from strait import NotFoundError, RateLimitedError

try:
    job = client.jobs.get("nonexistent")
except NotFoundError as e:
    print(f"Not found: {e} (status={e.status})")
except RateLimitedError as e:
    print(f"Rate limited, retry after backoff")

See Error Handling Guide for the full exception hierarchy.

Middleware

from strait import Client, Middleware

mw = Middleware(
    on_request=lambda ctx: print(f"-> {ctx.method} {ctx.url}"),
    on_response=lambda ctx: print(f"<- {ctx.status} ({ctx.duration_ms}ms)"),
)
client = Client.from_env(middleware=[mw])

FSM State Machines

from strait.fsm import transition_run, RunStatus, RunEvent

next_status = transition_run(RunStatus.EXECUTING, RunEvent.COMPLETE)
assert next_status == RunStatus.COMPLETED

Documentation

Development

make bootstrap   # Create venv + install deps
make lint        # Run ruff (lint + format check)
make typecheck   # Run mypy (strict mode)
make test        # Run pytest with coverage
make run-all     # lint + typecheck + test

See CONTRIBUTING.md for the full contributor guide.

License

MIT