> ## Documentation Index
> Fetch the complete documentation index at: https://opensre.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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](/docs/guides/api). To supply your own output sink, tools, prompts, or error
reporting and drive one agent across many turns, see
[Hosting the Agent](/docs/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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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`](/docs/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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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 theme={null}
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.