ReqManagedAgents.Provider behaviour (ReqManagedAgents v0.9.1)

Copy Markdown View Source

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.

: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

conn()

@type conn() :: term()

Provider-private connection / session handle.

event()

@type event() :: %{required(String.t()) => term()}

A raw, decoded provider event (string-keyed wire map).

handle()

@type handle() :: term()

Provider-private handle to a provisioned, reusable server-side resource.

input()

@type input() :: term()

Provider-private input that drives the next turn.

spec()

@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).

terminal()

@type terminal() :: :end_turn | :requires_action | :terminated

Callbacks

consumer(conn)

@callback consumer(conn()) :: pid() | nil

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

kickoff_input(opts)

@callback kickoff_input(opts :: keyword()) :: input()

Input that kicks off the conversation (the initial user message).

mode()

@callback mode() :: :streaming | :request_response

normalize(list)

@callback normalize([event()]) :: ReqManagedAgents.TurnResult.t()

Fold a turn's accumulated events into the canonical turn outcome (carries raw events).

open(opts, subscriber)

@callback open(opts :: keyword(), subscriber :: pid()) :: {:ok, conn()} | {:error, term()}

Establish the connection/session; for :streaming, open the event stream to subscriber.

pending_tool_uses(list)

(optional)
@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.

poll_turn(conn, input)

(optional)
@callback poll_turn(conn(), input()) :: {:ok, [event()], conn()} | {:error, term()}

:request_response only — run one turn synchronously: send input, return the turn's events.

provision(spec, opts)

@callback provision(spec(), opts :: keyword()) :: {:ok, handle()} | {:error, term()}

Create (or look up) the provider-side agent resource for spec; return a durable handle.

push_input(conn, input)

(optional)
@callback push_input(conn(), input()) :: :ok | {:error, term()}

:streaming only — post input; events arrive asynchronously at the subscriber.

reconnect(conn, subscriber, seen)

(optional)
@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.

ref(conn)

@callback ref(conn()) :: reference() | nil

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

resume_input(tool_uses, results)

@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).

resumed?(conn)

@callback resumed?(conn()) :: boolean()

Whether open/2 consolidated an EXISTING session into conn (a resume) rather than creating a fresh one. Gates Session's reattach behavior (#66).

session_id(conn)

@callback session_id(conn()) :: String.t() | nil

The provider-side session id conn carries, if any (Session surfaces it in SessionResult/SessionInfo; nil when the concept doesn't apply).

supports_outcomes?()

(optional)
@callback supports_outcomes?() :: boolean()

Optional — true when the provider natively honors the :outcome kickoff (user.define_outcome).

teardown(handle, opts)

(optional)
@callback teardown(handle(), opts :: keyword()) :: :ok | {:error, term()}

Delete the provider-side resource named by handle. opts carries the client / test seam.

text_delta(event)

(optional)
@callback text_delta(event()) :: String.t() | nil

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

turn_boundary?(event)

(optional)
@callback turn_boundary?(event()) :: boolean()

:streaming only — does this event close a turn (so accumulated events form one)?

user_input(text)

@callback user_input(text :: String.t()) :: input()

Input for a follow-up user message into a running session.

Functions

result_of(id, tool_event)

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