tracer-cloud--opensre/docs/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](/api). To supply your own output sink, tools, prompts, or error
reporting and drive one agent across many turns, see
[Hosting the Agent](/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:

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

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

```python
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:

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

# 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)
print(report.report)
```

## One API — chat and investigate

Every surface uses the same two verbs:

```python
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)

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

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

```python
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; it uses the payload runner installed by `configure_process`.

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 installed payload runner (`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:

```python
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-investigations):
- `/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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.