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

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

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:

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

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

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

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:

```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("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`](/platform/cron) — 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:

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

## Prompt surface

The agent assembles its prompt from the session — the system prompt, connected
integrations, memory, the task plan, and earlier turns — so there is no prompt
text to pass in. What you choose is the surface the turns run on:
`interactive_shell` (the default), `headless_cli`, or `gateway`. The surface
decides the surface-specific parts of the prompt, such as setup facts and
whether the agent may open a menu.

```python
from core.agent_harness import AgentSession, SessionConfig

session = AgentSession.start(SessionConfig(), surface="headless_cli")
```

A `prompts=` provider — `SessionConfig(prompts=…)`, or
`DefaultHeadlessBuild.agent(prompts=…)` on the custom-ports path below — names
the surface instead. It implements `core.agent_harness.ports.PromptContextProvider`,
whose one method is `surface()`. If omitted,
`core.agent_harness.spi.defaults.DefaultPromptContextProvider` is used.

## 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",)` 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.