Contract a streaming agent backend implements so one ReqManagedAgents.Session loop can
drive ANY provider. A provider owns its transport mode and invocation end-to-end:
:streaming(push) — the server holds a connection open and pushes events; the client posts inputs out-of-band; a turn ends on a boundary event (turn_boundary?/1).:request_response(pull) — the client calls (poll_turn/2) and the server answers the whole turn; resume re-sends the conversation delta.
The canonical vocabulary uses Anthropic's custom_tool_use / custom_tool_result terms and
names ONLY the client-side / return-of-control species; server-side tools are observe-only.
Raw preservation. Normalization is additive, never lossy: turn_outcome carries the raw,
JSON-decoded provider events it was derived from alongside the normalized fields, so a
consumer can always cross-reference the provider's own wire documentation.
Summary
Types
Provider-private connection / session handle.
A raw, decoded provider event (string-keyed wire map).
Provider-private handle to a provisioned, reusable server-side resource.
Provider-private input that drives the next turn.
A provider-agnostic agent definition — the input to provisioning and the cache key.
provision/2 coerces its input via ReqManagedAgents.Agent.Spec.new/1 at the boundary,
so any Spec-shaped map is accepted; the callback itself is typed against the validated
struct (#70, generalizes #68).
Callbacks
The linked process consuming the stream on conn's behalf, if any. nil when there is no
live consumer (e.g. :request_response, or a not-yet-consolidated resume).
Input that kicks off the conversation (the initial user message).
Fold a turn's accumulated events into the canonical turn outcome (carries raw events).
Establish the connection/session; for :streaming, open the event stream to subscriber.
Optional — recover unanswered custom tool uses from events (the session's full
accumulated raw event history, oldest first).
:request_response only — run one turn synchronously: send input, return the turn's events.
Create (or look up) the provider-side agent resource for spec; return a durable handle.
:streaming only — post input; events arrive asynchronously at the subscriber.
:streaming only — after a stream drop, re-open the stream (delivering to subscriber) and
return any unanswered tool calls to re-drive locally, plus the grown seen set.
The live stream's tag conn carries, if any — Session matches inbound
{:managed_agents, ref, _} messages against it. nil for a :request_response conn or a
:streaming conn with no stream open yet (e.g. a not-yet-consolidated resume).
Input that resumes the loop after local tools ran (the mode's resume contract).
Whether open/2 consolidated an EXISTING session into conn (a resume) rather than
creating a fresh one. Gates Session's reattach behavior (#66).
The provider-side session id conn carries, if any (Session surfaces it in
SessionResult/SessionInfo; nil when the concept doesn't apply).
Optional — true when the provider natively honors the :outcome kickoff (user.define_outcome).
Delete the provider-side resource named by handle. opts carries the client / test seam.
Optional — map ONE raw event to a normalized text chunk, or nil.
Optional — the provider's client-held conversation history in its native wire shape
(string-keyed message maps), or nil. Implemented by providers whose history lives
in the conn (e.g. Local); server-held providers (CMA, AgentCore) omit it. Session
embeds the value into SessionResult.transcript at terminal. Entries are
provider-verbatim — never interpreted by this library.
:streaming only — does this event close a turn (so accumulated events form one)?
Input for a follow-up user message into a running session.
Functions
Extract a canonical %ToolResult{} from a Tools.run/7 wire event
(user.custom_tool_result shape), given the tool-use id it answers.
Types
@type conn() :: term()
Provider-private connection / session handle.
A raw, decoded provider event (string-keyed wire map).
@type handle() :: term()
Provider-private handle to a provisioned, reusable server-side resource.
@type input() :: term()
Provider-private input that drives the next turn.
@type spec() :: ReqManagedAgents.Agent.Spec.t()
A provider-agnostic agent definition — the input to provisioning and the cache key.
provision/2 coerces its input via ReqManagedAgents.Agent.Spec.new/1 at the boundary,
so any Spec-shaped map is accepted; the callback itself is typed against the validated
struct (#70, generalizes #68).
@type terminal() :: :end_turn | :requires_action | :terminated
Callbacks
The linked process consuming the stream on conn's behalf, if any. nil when there is no
live consumer (e.g. :request_response, or a not-yet-consolidated resume).
Input that kicks off the conversation (the initial user message).
@callback mode() :: :streaming | :request_response
@callback normalize([event()]) :: ReqManagedAgents.TurnResult.t()
Fold a turn's accumulated events into the canonical turn outcome (carries raw events).
Establish the connection/session; for :streaming, open the event stream to subscriber.
@callback pending_tool_uses([event()]) :: [ReqManagedAgents.ToolUse.t()]
Optional — recover unanswered custom tool uses from events (the session's full
accumulated raw event history, oldest first).
A requires_action turn's per-batch normalize/1 can resolve to zero
custom_tool_uses when the ids its own stop condition references live in an
EARLIER, already-processed batch (the referencing idle and the tool uses it
names don't always land in the same batch). When that happens, Session calls
this callback instead of driving an empty resume — implement it the same way
reconnect/3 recovers unanswered tool calls across a stream drop (e.g. via
ReqManagedAgents.Consolidate.unanswered_tool_uses/1). Return [] when nothing
is recoverable; Session then surfaces a loud protocol-state error rather than
ever POSTing an empty events list.
Every use returned here is re-run at-least-once: Session re-invokes
ReqManagedAgents.Handler.handle_tool_call/3 (or /4) for it regardless of
whether that tool_use id was already dispatched earlier in the session —
see ReqManagedAgents.Handler's moduledoc.
:request_response only — run one turn synchronously: send input, return the turn's events.
Create (or look up) the provider-side agent resource for spec; return a durable handle.
:streaming only — post input; events arrive asynchronously at the subscriber.
@callback reconnect(conn(), subscriber :: pid(), seen :: MapSet.t()) :: {:ok, conn(), [ReqManagedAgents.ToolUse.t()], MapSet.t()} | {:error, term()}
:streaming only — after a stream drop, re-open the stream (delivering to subscriber) and
return any unanswered tool calls to re-drive locally, plus the grown seen set.
Optional: required only for providers that must recover events or unanswered tool
calls across a reconnect (streaming drops, or list-events surfaces). A
request_response provider that only reattaches may omit it — Session defaults to
{:ok, conn, [], seen}.
The live stream's tag conn carries, if any — Session matches inbound
{:managed_agents, ref, _} messages against it. nil for a :request_response conn or a
:streaming conn with no stream open yet (e.g. a not-yet-consolidated resume).
@callback resume_input( tool_uses :: [ReqManagedAgents.ToolUse.t()], results :: [ReqManagedAgents.ToolResult.t()] ) :: input()
Input that resumes the loop after local tools ran (the mode's resume contract).
Whether open/2 consolidated an EXISTING session into conn (a resume) rather than
creating a fresh one. Gates Session's reattach behavior (#66).
Returning true routes Session through the :resume → :reconnect continuation.
If the provider does not implement reconnect/3, resume proceeds with no event
recovery ({:ok, conn, [], seen}) — correct for request_response reattach. A
synchronous Session.run/2 resume with no :prompt has nothing to deliver and
parks until the run timeout — reattach callers should always pass prompt:.
The provider-side session id conn carries, if any (Session surfaces it in
SessionResult/SessionInfo; nil when the concept doesn't apply).
@callback supports_outcomes?() :: boolean()
Optional — true when the provider natively honors the :outcome kickoff (user.define_outcome).
Delete the provider-side resource named by handle. opts carries the client / test seam.
Optional — map ONE raw event to a normalized text chunk, or nil.
When implemented, the Session emits %{"type" => "rma.text_delta", "text" => chunk}
through handle_event immediately after forwarding the raw event (additive
normalization: alongside, never instead of, the raw event; never stored in
SessionResult.events). Chunk granularity is whatever the provider's wire exposes —
true streaming deltas on AgentCore (contentBlockDelta), whole message blocks on
Claude Managed Agents (agent.message).
Optional — the provider's client-held conversation history in its native wire shape
(string-keyed message maps), or nil. Implemented by providers whose history lives
in the conn (e.g. Local); server-held providers (CMA, AgentCore) omit it. Session
embeds the value into SessionResult.transcript at terminal. Entries are
provider-verbatim — never interpreted by this library.
:streaming only — does this event close a turn (so accumulated events form one)?
Input for a follow-up user message into a running session.
Functions
@spec result_of(String.t(), event()) :: ReqManagedAgents.ToolResult.t()
Extract a canonical %ToolResult{} from a Tools.run/7 wire event
(user.custom_tool_result shape), given the tool-use id it answers.