Metadata-Version: 2.4
Name: astronomer-otto-sdk
Version: 0.0.4
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: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; 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.

## Pre-tool-use hooks

Async callbacks invoked over RPC before each tool call, exactly like the Claude
Agent SDK's `PreToolUse` hooks. Each hook receives the tool name + args and can:

- **allow** the call (short-circuits Otto's permission rule engine for that call)
- **deny** it with a reason (the agent surfaces the reason and replans)
- **pass** (no opinion — falls through to other hooks and Otto's permissions)
- optionally **mutate the args** before execution

Hooks are dispatched in registration order. First explicit `deny` wins; a later
hook's `deny` overrides an earlier `allow`. Hook exceptions and per-hook
timeouts are treated as `deny` — security-style hooks fail closed.

```python
from otto_sdk import (
    HookResult,
    OttoClient,
    OttoOptions,
    PreToolUseHookEntry,
    PreToolUseInput,
)

async def echo_only_bash(payload: PreToolUseInput) -> HookResult | None:
    command = payload["tool_input"].get("command", "")
    if str(command).strip().startswith("echo "):
        return {"decision": "allow"}
    return {"decision": "deny", "reason": "only `echo` commands are allowed"}

options = OttoOptions(
    pre_tool_use_hooks=[
        PreToolUseHookEntry(matcher="bash", hook=echo_only_bash, name="echo-only"),
    ],
)
async with OttoClient(options) as client:
    await client.prompt("Run bash: echo hello")
    ...
```

### Matchers

`PreToolUseHookEntry.matcher` decides which tools a hook fires on. The SDK
filters by matcher **before** invoking, so hooks never need defensive
`if tool_name != ...` checks.

- `"*"` — match every tool.
- `"bash"` — exact name, case-insensitive.
- `"bash|webfetch"` — pipe-separated alternatives, case-insensitive.
- `re.compile(r"^mcp__")` — regex against the verbatim tool name.
- `frozenset({"a", "b"}).__contains__` — callable predicate (best for registry-driven gating).

Tool names arrive **verbatim** in the input. Pi built-ins are lowercase
(`"bash"`, `"read"`); MCP tools keep their MCP-server-given casing
(`"mcp__astro_tools__Astro_RunDAGOnTestDeployment"`).

### Long-blocking hooks (UI approval gates)

Hooks can legitimately block for tens of minutes — e.g. polaris's approval
workflow blocks on Redis pub/sub waiting for the user to click Approve. Set
`timeout_ms` per hook (the global default is 30s):

```python
PreToolUseHookEntry(
    matcher=APPROVAL_REQUIRED.__contains__,
    hook=approval_hook,
    timeout_ms=1_800_000,  # 30 minutes
)
```

A timeout produces `deny` with reason `"hook <name> timed out"`. Otto itself
imposes no ceiling by default; set `OTTO_PRE_TOOL_USE_TIMEOUT_MS` env var for
an ops kill-switch.

### Patching tool args

Return `tool_input` to fully replace the args (it's a full replacement, not a
merge):

```python
async def add_timeout(payload: PreToolUseInput) -> HookResult:
    patched = dict(payload["tool_input"])
    patched.setdefault("timeout", 30)
    return {"decision": "allow", "tool_input": patched}
```

> **Known Pi quirk:** the `tool_execution_start` event Pi emits between the
> hook and tool execution carries the **original** args, not the patched
> ones. The tool itself runs with the patched args (verified) and
> `tool_execution_end` carries the patched output. Embedders displaying tool
> calls in a UI should read `pre_tool_use_response.tool_input` (their own
> copy) or wait for `tool_execution_end`.

### Composition (first-deny-wins)

```python
options = OttoOptions(
    pre_tool_use_hooks=[
        PreToolUseHookEntry(matcher="webfetch", hook=url_allowlist),
        PreToolUseHookEntry(matcher="bash", hook=command_filter),
        PreToolUseHookEntry(matcher=is_gated, hook=approval_required),
    ],
)
```

See `examples/pre_tool_use.py` for the basic shape and
`examples/approval_workflow.py` for the polaris-style approve/reject/timeout
pattern.

## 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"
    pre_tool_use_hooks: list[PreToolUseHookEntry] = []
    hook_timeout_ms: int = 30_000       # default per-hook timeout
    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)`.

