Hosting the Agent

Build the OpenSRE agent on your own ports and drive it per turn — the same way the gateway and the interactive shell do

Python API covers embedding: AgentSession.start() and chat(), with the standard ports. This page is for a host — a program that supplies its own output sink, tools, prompts, error reporting or gather progress, and drives one agent across many turns. The gateway (Slack, Discord, Telegram) and the interactive shell are both hosts, and they are the same program with different ports.

Four roles, five API modules

Everything a host or tool author touches comes through five modules — the harness API. Nothing else under core.agent_harness is part of it.

Role	You want to…	Import
Embed	run turns with the standard ports	core.agent_harness — AgentSession, SessionConfig, SessionCore, SessionManager, TurnResult, ToolCallingTurnResult, OutputSink
Host	build one agent from your ports, handle it per message	core.agent_harness.runtime — DefaultHeadlessBuild, InMemoryHeadlessBuild, HeadlessAgent, TurnBinding, GatherPhase, DefaultToolProvider, ActionTurnRunner; also the ReAct engine the investigation pipeline shares (AgentConfig, build_agent, the default LLM factories)
Host	call helpers around a turn	core.agent_harness.spi.<role> — session_goal, session_state, cancel, accounting, prompt_chrome, integrations, grounding, defaults
Write a tool	implement an action tool	core.agent_harness.tools — ActionToolScope, execute_with_action_context, capability_available_from_sources, HandoffField, EVIDENCE_KIND_VALUES
Implement	satisfy a protocol the harness calls	core.agent_harness.ports — OutputSink, ToolProvider, PromptContextProvider, ReasoningClientProvider, RunRecordFactory, ErrorReporter, TurnAccounting, SessionState, and the rebind mix-ins ConsoleBindable / OutputBindable / SessionBindable

The host loop

Build once, then one call per inbound message:

from core.agent_harness.runtime import DefaultHeadlessBuild, GatherPhase, TurnBinding

# 1. Build — the default port family for this session, plus your ports.
agent = DefaultHeadlessBuild(
    session=session,            # a SessionCore from AgentSession / SessionManager
    output=my_sink,             # your OutputSink
    error_reporter=my_reporter, # optional; default logs at debug
).agent(
    tools=my_tool_provider,     # optional; a configured DefaultToolProvider or your own
    prompts=my_prompts,         # optional; default grounding provider
    gather=GatherPhase(),       # opt in to the evidence pass; default is disabled
)

# 2. Handle — state the whole turn; the agent binds it, dispatches, and keeps
#    going while a session goal is attached. Nothing carries over from the
#    previous message.
result = agent.handle(
    text,
    TurnBinding(
        session=session,           # None keeps the current session
        tool_hooks=approval_hooks, # None means "no hooks this turn"
        console=turn_console,      # for cooperative cancel and progress lines
        confirm_fn=ask_user,       # None means "no confirmation callback"
        is_tty=False,
    ),
    accounting_factory=None,       # message -> TurnAccounting; None = default per message
    cancel_requested=lambda: cancel.is_set(),
    on_progress=render_goal_progress,
)

DefaultHeadlessBuild holds what does not vary per host — reasoning client, run records, the default tool provider and prompt provider — and takes only the inputs those defaults need (console, logger, surface). Every port you pass to agent() replaces that default. TurnBinding is a value, not a set of keyword arguments: a field you omit is this turn’s value (no hooks, no callback), never “leave alone”, so a pooled agent cannot inherit another conversation’s hooks by omission. handle is the only per-message call a host makes; dispatch (one engine turn) sits underneath it.

The in-memory family is InMemoryHeadlessBuild — a script or test says InMemoryHeadlessBuild().agent(tools=NullToolProvider()) and has an agent with zero configuration. There is one construction verb, <family>.agent(...); hosts do not call HeadlessAgent(...) directly.

The same loop in the two shipped hosts

The gateway keeps one agent per logical session and calls handle per inbound message:

agent = DefaultHeadlessBuild(session=session, output=live_sink, console=console, logger=log, surface="gateway").agent(
    tools=DefaultToolProvider(session, console, tool_action_logger=log,
                              observer_factory=…, subprocess_presenter_factory=…, slash_ports_factory=…),
    gather=GatherPhase(),
)
result = agent.handle(
    text,
    TurnBinding(session=session, tool_hooks=sink.tool_hooks, console=turn_console, is_tty=False),
    cancel_requested=cancel.is_set,
    on_progress=post_status_line,
)

The interactive shell shares the same TurnRunner → SessionAgentPool → handle path as the gateway (TTY accounting / is_tty=True). Direct construction for tests or a custom host looks like:

agent = DefaultHeadlessBuild(session=session, output=ShellOutputSink(console), console=console,
                     error_reporter=ShellErrorReporter()).agent(
    tools=shell_tool_provider(session, console, request_exit=…),
    prompts=shell_prompt_context_provider(session),
    gather=shell_gather_ports(session, console),   # console progress lines + session persistence
)
result = agent.handle(
    text,
    TurnBinding(session=session, output=sink, tool_hooks=hooks, console=console, confirm_fn=ask, is_tty=tty),
    accounting_factory=lambda message: ShellTurnAccounting(session, message, recorder),
    cancel_requested=lambda: host_cancel_requested(sink),
    on_progress=print_checklist,
)

Neither host replaces a stage of the agent (action selection, evidence gathering, answering) and neither writes its own turn loop. They differ only in ports and in what they do with the result. Production gateway/shell submissions go through the pool; expand AgentBuildConfig with resolve_agent_ports.

What you implement vs what you call

You want to…	Implement	Call
Send output somewhere new (a websocket, a chat thread)	OutputSink — print, stream, render_error, render_response_header	DefaultHeadlessBuild(output=…)
Add tools or change how they are rendered/confirmed	a ToolProvider — usually DefaultToolProvider(...) with your port factories	.agent(tools=…)
Change what the model is told about the environment	subclass spi.defaults.DefaultPromptContextProvider, override the block you need	.agent(prompts=…)
Stream progress / persist tool calls of the evidence pass	two callables	.agent(gather=GatherPhase(on_progress=…, persist=…))
Route swallowed exceptions (Sentry, logs)	ErrorReporter — report(exc, context=…, expected=…)	DefaultHeadlessBuild(error_reporter=…)
Account a turn (telemetry, token totals)	TurnAccounting — record_action_result, finalize	TurnBinding(accounting=…)
Approve or deny tool calls per conversation	—	TurnBinding(tool_hooks=…)
Follow the turn’s console or sink from your own port	ConsoleBindable.bind_console / OutputBindable.bind_output	happens on bind_turn
Drive a multi-turn goal, read or set session state	—	spi.session_goal, spi.session_state

Rules a host must keep

• One agent per logical session; one turn at a time on it. The agent enforces this: a second handle while one is running raises AgentBusyError rather than interleaving. Serialize turns before calling handle (the gateway holds a per-session lock across it); do not catch the error to retry.
• State the whole turn. Build a fresh TurnBinding per message and pass it to handle; do not call bind_turn/dispatch yourself.
• Bind on the worker thread you dispatch from. Storage scope is a ContextVar; if you hand the turn to a thread pool, wrap it with config.scope_context.in_current_scope on the calling thread.
• Do not import past the API. Modules under core.agent_harness.turns, .tools, .session and .prompts are internal; the shipped hosts pin their remaining deep imports in shrink-only tests.