---
title: Pilot Python SDK
description: "Async and sync Python clients for the Cuitty Pilot API — manage playbooks, runs, and providers with httpx."
section: Pilot
order: 16
updatedAt: 2026-06-01
slug: pilot/sdk/python
---
# 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

```bash
sfw pip install cuitty-pilot
```

Requires Python 3.10+ and `httpx`.

## Quick start

### Async

```python
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

```python
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.

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

| Method | Returns | Description |
|--------|---------|-------------|
| `health()` | `HealthStatus` | Check server health |
| `list_playbooks(target?)` | `list[PlaybookSummary]` | List playbooks, optionally filtered |
| `get_playbook(id)` | `Playbook` | Get playbook by ID (latest version) |
| `create_playbook(req)` | `dict` (`{id, version, created}`) | Create a new playbook |
| `update_playbook(id, req)` | `dict` (`{id, version}`) | Update (version-bump) a playbook |
| `delete_playbook(id)` | `dict` (`{deleted}`) | Delete all versions |
| `start_run(req)` | `dict` (`{id, status}`) | Start a new run |
| `list_runs(status?, playbook_id?, limit?)` | `list[Run]` | List runs with optional filters |
| `get_run(id)` | `Run` | Get run details including steps |
| `abort_run(id)` | `dict` (`{aborted, runId}`) | Abort a running or queued run |
| `approve_run(id, req)` | `dict` (`{approved, runId, stepId, decision}`) | Submit approval decision |
| `list_providers()` | `list[Provider]` | List all provider profiles |
| `close()` | `None` | Close the underlying HTTP client |

## PilotClientSync

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

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

All methods are the same as `PilotClient` but without `await`.

## Playbooks

```python
# 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

```python
# 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

| Type | Fields |
|------|--------|
| `Playbook` | `id`, `version`, `target`, `title`, `document_yaml`, `authored_by`, `created_at` |
| `PlaybookSummary` | `id`, `version`, `target`, `title`, `authored_by` |
| `Run` | `id`, `project_id`, `playbook_id`, `playbook_version`, `mode`, `status`, `inputs_json`, `outputs_json`, `provider_profile_id?`, `started_at?`, `completed_at?`, `failed_step_id?`, `error_message?`, `steps: list[RunStep]` |
| `RunStep` | `id`, `run_id`, `step_id`, `action`, `status`, `selector_tried_json`, `ai_used`, `started_at?`, `completed_at?`, `duration_ms?`, `error_message?` |
| `Provider` | `id`, `project_id`, `provider`, `label`, `base_url`, `last_authenticated?`, `expires_at?` |
| `HealthStatus` | `status`, `database`, `driver_pool: dict`, `provider_sessions: dict`, `last_checked` |

### Request types

| Type | Fields |
|------|--------|
| `StartRunRequest` | `playbook_id`, `mode="autonomous"`, `version?`, `inputs?: dict`, `provider_profile_id?`, `project_id?` |
| `CreatePlaybookRequest` | `target`, `title`, `yaml`, `authored_by?` |
| `UpdatePlaybookRequest` | `yaml`, `target?`, `title?`, `authored_by?` |
| `ApproveRequest` | `step_id`, `decision="approve"` |

## Error handling

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

```python
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}")
```

| Exception | Properties | When |
|-----------|-----------|------|
| `PilotError` | base class | All SDK errors |
| `PilotAPIError` | `status: int`, `message: str` | Non-2xx HTTP response |
| `PilotConnectionError` | `message: str`, `cause: Exception \| None` | Network or timeout failure |

## See also

- [JavaScript SDK](/docs/pilot/sdk/javascript) — TypeScript client with event streaming
- [Playbook reference](/docs/pilot/playbook-reference) — YAML schema for playbook documents
- [Quickstart](/docs/pilot/quickstart) — record and replay your first workflow