Pilot Python SDK

cuitty-pilot provides both async (PilotClient) and synchronous (PilotClientSync) Python clients for the Pilot API. Built on httpx with dataclass types and context manager support.

Install

sfw pip install cuitty-pilot

Requires Python 3.10+ and httpx.

Quick start

Async

import asyncio
from cuitty_pilot import PilotClient, StartRunRequest

async def main():
    async with PilotClient("http://localhost:4320", auth_token="pk_...") as client:
        # List all playbooks
        playbooks = await client.list_playbooks()

        # Start a run
        resp = await client.start_run(StartRunRequest(
            playbook_id="cloudflare.dns.add-a-record",
            mode="autonomous",
            inputs={"zone": "example.com", "record_name": "app", "ip_address": "1.2.3.4"},
        ))
        print(f"Run started: {resp['id']} ({resp['status']})")

        # Get run details
        run = await client.get_run(resp["id"])
        print(f"Steps: {len(run.steps)}")

asyncio.run(main())

Synchronous

from cuitty_pilot import PilotClientSync, StartRunRequest

with PilotClientSync("http://localhost:4320", auth_token="pk_...") as client:
    playbooks = client.list_playbooks()
    resp = client.start_run(StartRunRequest(
        playbook_id="cloudflare.dns.add-a-record",
        inputs={"zone": "example.com", "record_name": "app", "ip_address": "1.2.3.4"},
    ))
    run = client.get_run(resp["id"])
    client.close()

PilotClient (async)

The primary client. Supports async with as a context manager.

client = PilotClient(
    base_url="http://localhost:4320",
    auth_token="ey...",     # Optional bearer token
    timeout=30.0,           # Per-request timeout in seconds (default 30.0)
)

PilotClientSync

Synchronous wrapper with an identical API. Supports with as a context manager.

client = PilotClientSync(
    base_url="http://localhost:4320",
    auth_token="ey...",
    timeout=30.0,
)

All methods are the same as PilotClient but without await.

Playbooks

# List all
playbooks = await client.list_playbooks()

# Filter by target
cf_playbooks = await client.list_playbooks(target="cloudflare")

# Get one
pb = await client.get_playbook("cloudflare.dns.add-a-record")
print(pb.title, pb.document_yaml)

# Create
resp = await client.create_playbook(CreatePlaybookRequest(
    target="cloudflare",
    title="Add CNAME record",
    yaml=playbook_yaml,
    authored_by="human",
))

# Update
await client.update_playbook("cloudflare.dns.add-cname",
    UpdatePlaybookRequest(yaml=updated_yaml))

# Delete
await client.delete_playbook("cloudflare.dns.add-cname")

Runs

# Start a run
resp = await client.start_run(StartRunRequest(
    playbook_id="cloudflare.dns.add-a-record",
    mode="interactive",
    inputs={"zone": "example.com", "record_name": "api", "ip_address": "1.2.3.4"},
    provider_profile_id="cf-prod",
))

# Get run with steps
run = await client.get_run(resp["id"])
for step in run.steps:
    print(f"{step.step_id}: {step.status} ({step.duration_ms}ms)")

# List runs with filters
running = await client.list_runs(status="running", limit=10)

# Abort
await client.abort_run(run_id)

# Approve a step
await client.approve_run(run_id, ApproveRequest(
    step_id="save",
    decision="approve",
))

Types

All types are Python dataclass instances.

Core types

Request types

Error handling

The SDK raises typed exceptions from a common PilotError base class:

from cuitty_pilot import PilotAPIError, PilotConnectionError

try:
    await client.start_run(StartRunRequest(playbook_id="nonexistent"))
except PilotAPIError as e:
    print(f"API error {e.status}: {e.message}")
except PilotConnectionError as e:
    print(f"Cannot reach Pilot: {e}")
    if e.cause:
        print(f"Underlying: {e.cause}")

See also

• JavaScript SDK — TypeScript client with event streaming
• Playbook reference — YAML schema for playbook documents
• Quickstart — record and replay your first workflow