astronomer-otto-sdk 0.0.5

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.

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