Hosting the Agent

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.

Three roles, four import doors

Everything a host touches comes through four modules. Nothing else under core.agent_harness is public.

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` — `DefaultPorts`, `HeadlessPorts`, `HeadlessAgent`, `TurnBinding`, `GatherPorts`, `DefaultToolProvider`, `ActionTurnRunner`
Host	call helpers around a turn	`core.agent_harness.spi.<role>` — `session_goal`, `session_flags`, `cancel`, `accounting`, `prompt_chrome`, `integrations`, `grounding`, `defaults`
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 DefaultPorts, GatherPorts, TurnBinding

# 1. Build — the default port family for this session, plus your ports.
agent = DefaultPorts(
    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=GatherPorts(),       # 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,
)

DefaultPorts 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 HeadlessPorts — a script or test says HeadlessPorts().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 = DefaultPorts(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=GatherPorts(),
)
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 builds one agent at startup and calls handle per prompt:

agent = DefaultPorts(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.

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`	`DefaultPorts(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=GatherPorts(on_progress=…, persist=…))`
Route swallowed exceptions (Sentry, logs)	`ErrorReporter` — `report(exc, context=…, expected=…)`	`DefaultPorts(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 flags	—	`spi.session_goal`, `spi.session_flags`

Rules a host must keep

One agent per logical session; one turn at a time on it. bind_turn mutates per-turn state; the gateway holds a per-session lock across dispatch. Do not share one agent between concurrent turns.
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 reach past the doors. Modules under core.agent_harness.turns, .tools, .session and .prompts are internal; the shipped hosts pin their remaining deep imports in shrink-only tests.

Deploy and APIs

Community

Three roles, four import doors

The host loop

The same loop in the two shipped hosts

What you implement vs what you call

Rules a host must keep

​Three roles, four import doors

​The host loop

​The same loop in the two shipped hosts

​What you implement vs what you call

​Rules a host must keep

Three roles, four import doors

The host loop

The same loop in the two shipped hosts

What you implement vs what you call

Rules a host must keep