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)

Structured output

Pass output_schema (a plain JSON Schema dict; for pydantic use MyModel.model_json_schema()) and otto registers a synthetic submit_final_answer tool with that schema, instructs the agent to deliver its final result through it, and nudges the model (bounded retries) if it tries to finish without a valid call. run_query() extracts and re-validates the result:

from otto_sdk import QueryOptions, run_query

result = await run_query(QueryOptions(
    prompt="diagnose the failing DAG",
    cwd="./project",
    output_schema={
        "type": "object",
        "properties": {
            "diagnosis": {"type": "string"},
            "severity": {"type": "string", "enum": ["low", "medium", "high"]},
        },
        "required": ["diagnosis", "severity"],
        "additionalProperties": False,
    },
))

match result.structured_output_status:
    case "produced":       # final_structured_output is schema-valid — use it
        print(result.final_structured_output)
    case "invalid":        # extracted but failed validation
        print(result.structured_output_errors)
    case "missing":        # run ended without a valid submit_final_answer call
        ...
    case "not_requested":  # no output_schema was set
        ...

• On Anthropic models, the retry after a nudge is API-forced via tool_choice — the model cannot answer the recovery turn with prose. Additionally, with OTTO_STRICT_TOOL_USE=1 in the environment, eligible models (Sonnet 4.5+/5.x, Haiku 4.5+) register the tool with strict: true (grammar-constrained decoding — args schema-valid by construction). Strict is opt-in until the Astronomer gateway forwards strict schemas correctly.
• structured_output_status distinguishes "the model never delivered" ("missing") from "delivered but not schema-valid" ("invalid") — before this field, both looked like final_structured_output is None or required callers to hand-check the shape.
• Validation runs SDK-side with jsonschema against your original schema — defense in depth on top of otto's own arg validation.
• When the run gives up, otto also emits a custom message event with customType == STRUCTURED_OUTPUT_MISSING_CUSTOM_TYPE whose details carry diagnostics (reason, attempt counts, last validation error) — useful for logging; structured_output_status does not depend on it.
• CLI parity: otto --mode json --output-schema ... exits 4 when the run completes without structured output.

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 — one low-level run. agent_end carries willRetry: true when pi is about to retry a failed provider call, so it does not mean the turn is over.
• agent_settled — the turn is over: pi will not retry or continue on its own. events() ends here. Needs otto v0.1.19 or later.
• 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.

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

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

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)

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

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