Metadata-Version: 2.4
Name: astronomer-otto-sdk
Version: 0.0.3
Summary: Python SDK for programmatically driving the Otto agent binary
Project-URL: Homepage, https://astronomer.io/otto
Project-URL: Documentation, https://astronomer.io/otto
Author-email: Astronomer <humans@astronomer.io>
License-Expression: Apache-2.0
License-File: LICENSE
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: ruff>=0.5.0; extra == 'dev'
Requires-Dist: ty>=0.0.31; extra == 'dev'
Description-Content-Type: text/markdown

# astronomer-otto-sdk (Python)

Python SDK for programmatically driving the [otto](https://astronomer.io/otto) agent binary. Spawns otto in `--rpc` mode and exposes an `OttoClient` async context manager and a `query()` async iterator over JSON-lines on stdin/stdout.

Useful for building your own agents on top of otto, or as a drop-in shape for any code that already shells out to a Claude-style coding-agent SDK.

## Install

```bash
pip install astronomer-otto-sdk
```

The import path is `otto_sdk`. Zero runtime dependencies (stdlib only). Requires Python 3.10+.

## Prerequisites

1. `otto` binary installed. Resolution: `otto_path` option → `$OTTO_PATH` → `~/.astro/bin/otto` → `shutil.which("otto")`.
2. Astro env vars in the calling process:
   - `ASTRO_TOKEN`, `ASTRO_DOMAIN`, `ASTRO_ORGANIZATION`
   - `AIRFLOW_API_URL` + creds if the agent needs Airflow access

## One-shot: `query()`

```python
from otto_sdk import QueryOptions, query

async for event in query(QueryOptions(prompt="list dags", cwd="./project")):
    if event["type"] == "tool_execution_start":
        print(f"→ {event['toolName']}")
    elif event["type"] == "message_end":
        print(event["message"])
```

Or the batch variant:

```python
from otto_sdk import QueryOptions, run_query

result = await run_query(QueryOptions(prompt="summarize", cwd="./project"))
print(result.final_text)
```

## Multi-turn: `OttoClient`

```python
from otto_sdk import OttoClient, OttoOptions

async with OttoClient(OttoOptions(cwd="./project", no_session=True)) as client:
    await client.prompt("list the dags")
    async for event in client.events():
        ...  # stream events for this turn

    await client.prompt("now describe the first one")
    async for event in client.events():
        ...

    state = await client.get_state()
    print(state["messageCount"])
```

## Events

`AgentEvent` is a dict with a required `type` key. Narrow on `event["type"]` (or `match` in 3.10+):

- `agent_start` / `agent_end`
- `turn_start` / `turn_end`
- `message_start` / `message_update` / `message_end`
- `tool_execution_start` / `tool_execution_update` / `tool_execution_end`

See `otto_sdk.protocol` for the full TypedDict schema.

## Options

```python
@dataclass
class OttoOptions:
    otto_path: str | None = None       # override binary location
    cwd: str | None = None              # defaults to os.getcwd()
    env: dict[str, str | None] | None = None
    provider: str | None = None         # default "astronomer"
    model: str | None = None            # otto's current default if None
    no_session: bool = False            # skip ~/.astro/otto/sessions/ persistence
    session_path: str | None = None     # resume an existing .jsonl session (maps to --session)
    thinking_level: ThinkingLevel | None = None  # "off"|"minimal"|"low"|"medium"|"high"|"xhigh"
    extra_args: list[str] = []
    on_stderr: Callable[[str], None] | None = None
```

`thinking_level` is applied right after `start()`. You can also change it mid-session with `await client.set_thinking_level(level)`.

