Python API

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 (tools and investigation):

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 and root-cause analysis

Use session.investigate() to pass a raw alert payload from your alertmanager, webhook, or monitoring system directly into OpenSRE’s investigation pipeline:

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession
from tools.investigation.capability import run_investigation_payload

# 1. Register tools and investigation adapters
configure_process(EMBEDDED_PROFILE)

# 2. Start an in-process session
session = AgentSession.start()

# 3. Triage and investigate a production alert payload
alert_payload = {
    "alert_name": "HighLatency",
    "service": "checkout-api",
    "severity": "warning",
    "description": "p99 latency exceeded 1200ms in us-east-1",
}

report = session.investigate(alert_payload, runner=run_investigation_payload)
print(report.report)

One API — chat and investigate

Every surface uses the same two verbs:

from bootstrap.process import EMBEDDED_PROFILE, configure_process
from core.agent_harness import AgentSession
from tools.investigation.capability import run_investigation_payload

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)

report = session.investigate(
    {"alert_name": "HighLatency", "severity": "warning"},
    runner=run_investigation_payload,
)
print(report.report)

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. investigate does not require an attached chat agent; pass it the run_investigation_payload runner as shown above (core may not import tools, so the caller supplies it).

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. Investigation terminates at the payload runner the caller passes to investigate (run_investigation_payload). 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 with live gathering enabled to query active integrations and forward findings to team channels:

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

configure_process(EMBEDDED_PROFILE)

# Start session with live evidence gathering enabled for configured integrations
session = AgentSession.start(gather=GatherPhase(enabled=True))

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, asynchronous investigations and automated notifications are managed via /background:

• /background on enables async background investigation launches.
• /background notify set telegram (or email, rocketchat, buzz) routes completed RCA reports directly to your team’s notification channels upon task completion.
• /background show <task_id> inspects delivery status and findings.

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",) (or include "action"). The @tool default is ("investigation",) only; the chat and Python API action loop does not load investigation-only tools.
2. Tools may be defined in the package __init__.py or in submodules; both are discovered after registration.