SDK Quickstart

Technical reference. This page is for integrations, advanced configuration, or platform operations. For portal tasks, start with the user guide.

This guide walks through the current Python SDK surface that is actually implemented today: build a flow with the SDK, validate it locally, and register it through the control plane client.

Prerequisites

Python 3.13+
A running Orqestra control plane
The local development stack from Local Dev Setup

1. Install the SDK

For local repository development:


uv pip install -e packages/orqestra-sdk

If you are consuming a published package instead:


pip install orqestra-sdk

Verify the import:


python -c "from orqestra_sdk import Flow, Step, OrqestraClient; print('ok')"

2. Define a flow in Python

Create my_summarizer.py:


from orqestra_sdk import Flow, LLMCall, Step
 
 
flow = (
    Flow("my-summarizer", description="Simple summarization flow")
    .trigger_api(inputs={"text": "string"})
    .step(
        LLMCall("summarize")
        .inputs(
            messages=[
                {
                    "role": "user",
                    "content": "Summarize the following text:\n\n{{trigger.text}}",
                }
            ]
        )
        .next("return_result")
    )
    .step(
        Step("return_result")
        .returns(summary="{{steps.summarize.output.content}}")
    )
    .entry("summarize")
)

3. Validate locally

The SDK can validate internal step references before you hit the control plane:


python - <<'PY'
from my_summarizer import flow
 
errors, warnings = flow.validate()
print({"errors": errors, "warnings": warnings})
PY

A healthy flow returns empty errors.

4. Register the flow with the control plane


python - <<'PY'
import asyncio
 
from my_summarizer import flow
from orqestra_sdk import OrqestraClient
 
 
async def main():
    async with OrqestraClient("http://localhost:8000") as client:
        created = await client.flows.create(
            flow=flow,
            slug="my-summarizer",
            owner_ref="team:dev",
            version_label="v1",
        )
        print(created)
 
 
asyncio.run(main())
PY

Expected response shape:


{
  "asset_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "slug": "my-summarizer",
  "asset_kind": "orchestration_flow",
  "owner_ref": "team:dev",
  "lifecycle_state": "active",
  "created_at": "2026-04-06T10:30:00Z",
  "created_by": "sdk:default"
}

5. Inspect the registered flow


python - <<'PY'
import asyncio
 
from orqestra_sdk import OrqestraClient
 
 
async def main():
    async with OrqestraClient("http://localhost:8000") as client:
        flows = await client.flows.list()
        print(flows)
 
 
asyncio.run(main())
PY

The flow client returns a list of flow assets, not a paginated envelope.

6. Optional: ask the control plane to generate a flow

The SDK also exposes the flow generation endpoint:


python - <<'PY'
import asyncio
 
from orqestra_sdk import OrqestraClient
 
 
async def main():
    async with OrqestraClient("http://localhost:8000") as client:
        generated = await client.flows.generate(
            description="Create a support triage flow that classifies a ticket and returns a priority."
        )
        print(generated["generation_id"])
        print(generated["flow_content"])
 
 
asyncio.run(main())
PY

WorkloadRuntimeClient is available as client.workload_runtime:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    # List workloads
    workloads = await client.workload_runtime.list()
    print(workloads)
 
    # Execute a workload
    result = await client.workload_runtime.execute(
        slug="my-summarizer",
        inputs={"text": "Quarterly earnings were up 12% year-over-year."},
    )
    print(result)
 
    # Create a share link for a workload execution
    share = await client.workload_runtime.create_share_link(workload_id=result["workload_id"])
    print(share["url"])

8. Plugins client — install package assets

PluginsClient is available as client.plugins:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    packages = await client.plugins.list_packages()
    print(packages)
 
    install = await client.plugins.install_package(
        package_name="orqestra-official-plugins",
        version="0.1.0",
        workspace_id="00000000-0000-0000-0000-000000000123",
        selected_pack_names=["core-connectors"],
    )
    print(install["install_id"])
 
    await client.plugins.disable_install(install["install_id"])

See the Plugins guide for package catalog and install lifecycle details.

9. Releases client — evaluate, roll back, and compare

ReleasesClient is available as client.releases:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    release_id = "r-abc123"
 
    # Trigger an evaluation run
    eval_run = await client.releases.evaluate(release_id)
    print(eval_run)
 
    # List evaluation results
    results = await client.releases.get_evaluations(release_id)
    print(results)
 
    # Create an advisory asset-candidate evaluation session
    session = await client.evaluation_sessions.create(
        subject_kind="asset_candidate",
        subject_ref="refund-policy-prompt",
        candidates=[
            {
                "source": "manual",
                "asset_kind": "prompt_template",
                "content": {"template": "Answer using {{policy}}"},
            }
        ],
    )
    print(session)
 
    # Roll back to the previous active release
    await client.releases.rollback(release_id)
 
    # Advance rollout to the next lane
    await client.releases.advance_rollout(release_id)
 
    # Compare two releases
    diff = await client.releases.compare("r-abc123", "r-def456")
    print(diff)

10. Assets client — blast radius

AssetsClient is available as client.assets:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    asset_id = "a-xyz789"
 
    # Show all releases affected by this asset
    affected = await client.assets.blast_radius(asset_id)
    print(affected)

11. Executions client — evidence graph

ExecutionsClient is available as client.executions:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    exec_id = "e-qrs456"
 
    # Trace the full governance evidence chain
    graph = await client.executions.evidence_graph(exec_id, format="graph")
    print(graph)

12. Policy simulation client

PolicySimulationClient is available as client.policy_simulation:


from orqestra_sdk import OrqestraClient
 
async with OrqestraClient("http://localhost:8000") as client:
    # Dry-run a policy evaluation
    outcome = await client.policy_simulation.simulate_policy(bundle_id="pb-111")
    print(outcome)
 
    # Compare two policy bundles
    diff = await client.policy_simulation.diff_policies(
        a_id="pb-111",
        b_id="pb-222",
    )
    print(diff)

What this quickstart covers

This guide intentionally reflects the current SDK shape:

Local flow authoring and validation
Flow registration through client.flows.create(...)
Flow generation through client.flows.generate(...)
Workload execution and share links through client.workload_runtime
Plugin package catalog and install lifecycle through client.plugins
Release evaluation, rollback, and comparison through client.releases
Asset blast-radius queries through client.assets
Execution evidence graphs through client.executions
Policy simulation and diffing through client.policy_simulation

It does not assume unpublished helpers such as local from_dict() hydration or direct “create and validate in one call” abstractions.

Next steps

API Basics — Create workloads, assets, releases, and bindings through the REST API
Guides: CLI — Use the CLI for the same core resource workflows