python-api.mdx

title	Python API
slug	python-api
description	Drive the OpenSRE agent in-process from Python.

Use the Python API when your code runs on the same machine as OpenSRE and you want agent responses without invoking the CLI. For network access, use the HTTP API. To supply your own output sink, tools, prompts, or error reporting and drive one agent across many turns, see Hosting the Agent.

OpenSRE as a teammate in your daily loop

OpenSRE functions as an autonomous engineering teammate embedded directly into your daily operational workflows. Rather than treating the agent as an isolated chat interface, you can embed the Python API into scheduled cron jobs, CI/CD pipelines, and automated incident triage loops to perform recurring tasks — auditing repository health, triaging production alerts, and delivering digests to team channels like Slack, webhooks, or ticketing systems.

Prerequisites

The standalone CLI installer does not provide an importable package. Install from a source checkout:

git clone https://github.com/Tracer-Cloud/opensre.git
cd opensre && make install

Configure a provider once (opensre onboard) — the session API reuses the same config and credentials as the CLI. Run your script inside the checkout's environment with uv run python your_script.py.

Register adapters before the first turn:

from bootstrap.process import EMBEDDED_PROFILE, configure_process

configure_process(EMBEDDED_PROFILE)

Daily engineering recipes

Recipe 1: Querying repository and workflow metrics

Use session.chat() to query configured developer tools and observability sources (e.g. GitHub star velocity, PR backlog, deployment status):

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession

# 1. Register tools and adapter plugins once per process
configure_process(EMBEDDED_PROFILE)

# 2. Start an in-process session connected to configured integrations
session = AgentSession.start()

# 3. Query repository trends or metrics (requires GitHub integration configured)
result = session.chat("what's our GitHub star velocity over the last 7 days?")
action_ok = (
    result.action_result.handled
    and not result.action_result.has_unhandled_clause
    and result.action_result.accounting_status == "completed"
)
if not (result.answered or action_ok) or result.cancelled or not result.primary_response_text:
    raise RuntimeError(f"Turn failed: {result.primary_response_text or 'no response produced'}")

print(result.primary_response_text)

Recipe 2: Automated alert triage

Hand a raw alert payload from your alertmanager, webhook, or monitoring system to the agent as a chat turn — it calls your connected integrations to gather context and answers with its findings:

import json

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession

configure_process(EMBEDDED_PROFILE)
session = AgentSession.start()

alert_payload = {
    "alert_name": "HighLatency",
    "service": "checkout-api",
    "severity": "warning",
    "description": "p99 latency exceeded 1200ms in us-east-1",
}

result = session.chat(f"Triage this alert:\n{json.dumps(alert_payload)}")
print(result.primary_response_text)

One API — chat

Every surface goes through the same verb:

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession

configure_process(EMBEDDED_PROFILE)

session = AgentSession.start()
result = session.chat("why is checkout-api slow?")
action_ok = (
    result.action_result.handled
    and not result.action_result.has_unhandled_clause
    and result.action_result.accounting_status == "completed"
)
if (result.answered or action_ok) and not result.cancelled:
    print(result.primary_response_text)

Or the one-liner that wires the same boot step for you:

from bootstrap.embedded import start_embedded_session

session = start_embedded_session()

AgentSession.start() resolves the environment, opens a session, and attaches an agent with the standard ports — the same tools and prompts the interactive shell uses. It does not register adapters (core may not import bootstrap); call configure_process first or use start_embedded_session.

Always verify turn success before trusting chat text:

• For conversational questions and digests, the agent synthesizes an answer (result.answered).
• For action-only turns (tools handled the request directly without an LLM call), verify result.action_result.handled with not result.action_result.has_unhandled_clause and result.action_result.accounting_status == "completed".
• When a turn fails (for example the LLM provider is unreachable) or is cancelled (result.cancelled), the failure details land in result.primary_response_text.

Internal seams (not for hosts)

Chat hosts terminate at dispatch_chat_turn → run_turn. Do not invent parallel public entrypoints.

Unattended daily delivery and background loops

For recurring daily workflows (such as scheduled morning digests or CI health checks), start an in-process session and forward findings to team channels:

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession

configure_process(EMBEDDED_PROFILE)

session = AgentSession.start()

result = session.chat("summarize critical alerts and deployment changes from the last 24h")
action_ok = (
    result.action_result.handled
    and not result.action_result.has_unhandled_clause
    and result.action_result.accounting_status == "completed"
)
if not (result.answered or action_ok) or result.cancelled or not result.primary_response_text:
    raise RuntimeError(f"Scheduled digest failed: {result.primary_response_text or 'no response produced'}")

summary = result.primary_response_text
# Forward summary to team notification webhooks (Slack, email, or ticketing)
print("Delivering daily digest:\n", summary)

In the interactive shell, recurring unattended runs are managed with /loops — each loop sends its result to the handles OpenSRE can reach (Telegram, Slack, and the local shell inbox).

A conversation

Each chat call is one turn in the same session, so follow-ups see earlier context:

session.chat("list unresolved Sentry issues from the last 24 hours")
result = session.chat("which of those affect checkout?")

To resume an existing session, pass its ID:

from core.agent_harness import AgentSession, SessionConfig

session = AgentSession.start(SessionConfig(session_id="abc123"))

Run until a goal is complete

Use chat_until_goal when one request may need several agent turns. Pass an explicit SessionGoal so the completion condition, checklist, and turn limit do not depend on the first turn inferring them:

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession
from core.agent_harness.session_goal import SessionGoal

configure_process(EMBEDDED_PROFILE)
session = AgentSession.start()

outcome = session.chat_until_goal(
    "Audit active production alert rules and summarize risky thresholds.",
    goal=SessionGoal(
        condition=(
            "Every active production alert rule has been checked and risky "
            "thresholds have been summarized."
        ),
        checklist=(
            "List active production alert rules",
            "Check each threshold and evaluation window",
            "Summarize risky thresholds",
        ),
        max_outer_turns=5,
    ),
)

print(outcome.goal.status)
print(outcome.turn_count)
if outcome.last_result.answered:
    print(outcome.last_result.primary_response_text)

The loop stops when the goal is achieved, paused, cancelled, cleared, or its turn budget is exhausted. The result contains the final goal, the last_result from chat, and turn_count. Without goal=, the first action turn must attach a goal; otherwise chat_until_goal returns after that one turn. Use cancel_requested to stop between turns and on_progress to receive checklist updates.

Custom grounding context

By default, the agent builds prompts from the session. To supply a custom system prompt or retrieved context, pass a provider:

from core.agent_harness import AgentSession, SessionConfig

session = AgentSession.start(SessionConfig(prompts=my_provider))

If omitted, core.agent_harness.spi.defaults.DefaultPromptContextProvider is used. A custom provider must implement core.agent_harness.ports.PromptContextProvider. The same prompts= argument is accepted by DefaultHeadlessBuild.agent() on the custom-ports path below.

Custom output and ports

start() buffers all output. To capture tool progress yourself (for example to stream to a websocket), build the agent and pass your own sink:

from core.agent_harness import AgentSession, SessionConfig
from core.agent_harness.runtime import DefaultHeadlessBuild
from core.agent_harness.turns.headless_adapters import BufferOutputSink

session = AgentSession(SessionConfig())
startup = session.startup()
sink = BufferOutputSink()
agent = DefaultHeadlessBuild(session=startup.session, output=sink).agent(
    prompts=my_provider,  # optional — same port as SessionConfig.prompts
)
session.attach_agent(agent)

session.chat("summarize open incidents")
print(sink.lines)      # rendered output lines
print(sink.streamed)   # streamed answer chunks

Any object implementing the OutputSink protocol (core.agent_harness.ports.OutputSink: print, render_response_header, render_error, stream) may replace BufferOutputSink. DefaultHeadlessBuild also takes a custom logger, console, and prompt surface; its agent() takes your own tool provider (tools=, usually a configured DefaultToolProvider) and gather ports — see its docstring.

External tools

Register a package before the first tool lookup:

from tools.registry import register_external_tool_package
import my_agent_tools

register_external_tool_package(my_agent_tools)

Requirements:

1. Declare tools with surfaces=("action",) when the action loop should be able to call them. The @tool default is ("chat",).
2. Tools may be defined in the package __init__.py or in submodules; both are discovered after registration.