astronomer-otto-sdk 0.0.2

Python SDK for programmatically driving the Otto agent binary

astronomer-otto-sdk (Python)

Python SDK for programmatically driving the 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

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()

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:

from otto_sdk import QueryOptions, run_query

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

Multi-turn: OttoClient

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

@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).