ExAgent.Server (ExAgent v0.5.0)

Copy Markdown View Source

A supervised, stateful wrapper around ExAgent.run/3.

ExAgent.Server keeps an agent alive across many runs: it preserves the conversation history, accumulates token usage, threads stateful models (like ExAgent.Models.Test) from one run to the next, and emits ExAgent.Events over ExAgent.PubSub so UIs (LiveView, CLI, channels) can observe every run in real time.

It does not coordinate multiple participants or own shared state — that is ExAgent.Session's job (Roadmap Phase 3). It is purely a conversation with state: model + history + usage + events.

Starting

agent = ExAgent.new(model: "openai:gpt-4o", instructions: "Be a DM.")
{:ok, pid} = ExAgent.AgentSupervisor.start_agent(agent: agent, name: :dm)

or standalone:

{:ok, pid} = ExAgent.Server.start_link(agent: agent, name: :dm)

Options: :agent (required), :agent_id, :name, :pubsub (nil/:none/:local/{module, config}), :max_pending (default 8), :metadata.

Concurrency model

Runs execute in a supervised task (ExAgent.TaskSupervisor), so the GenServer keeps answering abort/1, health/1 and backpressure during long runs.

  • chat/3 — synchronous: blocks the caller until the run finishes.
  • send_message/3 — asynchronous: returns {:ok, request_id} immediately; the result arrives as a :run_finished/:run_failed event.
  • stream/3 — asynchronous text streaming: deltas arrive as :text_delta events on the agent topic.
  • steer/2 — enqueue a high-priority follow-up for the next run. It does not mutate an HTTP request already in flight.
  • abort/1 — cancel the in-flight run and emit :server_request_cancelled.

Backpressure: while a run is in flight, chat/3 returns {:error, :busy}; send_message/3/steer/2 enqueue up to max_pending requests and otherwise return {:error, :queue_full}.

Events

Published on ExAgent.Event.agent_topic(agent_id) ("exagent:agent:<agent_id>"). Subscribers receive {:exagent_event, %ExAgent.Event{}}. The seq field is monotonic per agent.

Summary

Functions

Cancel the in-flight run, if any. Returns :ok. Idempotent.

Run one prompt synchronously and return {:ok, result} (the ExAgent.run/3 result map) or {:error, reason}. Refuses to start if a run is already in flight, returning {:error, :busy}.

Runtime health: status and pending queue depth.

The current conversation history (messages threaded across runs).

Clear the conversation history and accumulated usage — start a fresh conversation on the same supervised agent. Only allowed when idle; returns {:error, :busy} if a run is in flight. When a store is configured, the cleared (empty) state is checkpointed.

Run one prompt asynchronously. Returns {:ok, request_id} immediately; the result is delivered as a :run_finished/:run_failed event on the agent topic. Returns {:error, :queue_full} if the pending queue is full.

Replace the model of an idle agent. Returns :ok or {:error, :busy | reason}.

Start a supervised, stateful agent.

Enqueue a high-priority follow-up for the next run. It is placed at the front of the pending queue (or started immediately if idle). It does not modify an HTTP request already in flight. Returns {:ok, request_id}.

Stream a prompt's text deltas. Returns {:ok, request_id} immediately; deltas arrive as :text_delta events and the final result as a :run_finished event on the agent topic. Returns {:error, :busy} if a run is in flight.

Accumulated token usage across all runs so far.

Functions

abort(server)

@spec abort(GenServer.server()) :: :ok

Cancel the in-flight run, if any. Returns :ok. Idempotent.

chat(server, prompt, opts \\ [])

@spec chat(GenServer.server(), String.t(), keyword()) ::
  {:ok, ExAgent.result()} | {:error, term()}

Run one prompt synchronously and return {:ok, result} (the ExAgent.run/3 result map) or {:error, reason}. Refuses to start if a run is already in flight, returning {:error, :busy}.

The GenServer call uses :infinity timeout by default (LLM runs can be long); pass timeout: ms in opts to override. Run options (:deps, :model_settings, …) are forwarded to ExAgent.run/3.

health(server)

@spec health(GenServer.server()) :: %{status: atom(), pending: non_neg_integer()}

Runtime health: status and pending queue depth.

history(server)

@spec history(GenServer.server()) :: [ExAgent.Message.t()]

The current conversation history (messages threaded across runs).

reset(server)

@spec reset(GenServer.server()) :: :ok | {:error, :busy}

Clear the conversation history and accumulated usage — start a fresh conversation on the same supervised agent. Only allowed when idle; returns {:error, :busy} if a run is in flight. When a store is configured, the cleared (empty) state is checkpointed.

send_message(server, prompt, opts \\ [])

@spec send_message(GenServer.server(), String.t(), keyword()) ::
  {:ok, String.t()} | {:error, :queue_full}

Run one prompt asynchronously. Returns {:ok, request_id} immediately; the result is delivered as a :run_finished/:run_failed event on the agent topic. Returns {:error, :queue_full} if the pending queue is full.

Run options are forwarded to ExAgent.run/3.

set_model(server, model_spec)

@spec set_model(GenServer.server(), ExAgent.Model.model() | String.t()) ::
  :ok | {:error, :busy | term()}

Replace the model of an idle agent. Returns :ok or {:error, :busy | reason}.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Start a supervised, stateful agent.

Options

  • :agent — (required) an ExAgent.t() built with ExAgent.new/1.
  • :agent_id — stable id for correlation/events; auto-generated if absent.
  • :name — registered name for the GenServer.
  • :pubsubnil/:none/:local/{module, config} (default nil).
  • :storenil/:ets/{module, config} (default nil, no
                persistence). When set, history/usage are checkpointed
                after every run and rehydrated on restart.
  • :max_pending — max queued async requests before :queue_full (default 8).
  • :metadata — free-form map attached to every emitted event.

steer(server, prompt, opts \\ [])

@spec steer(GenServer.server(), String.t(), keyword()) ::
  {:ok, String.t()} | {:error, :queue_full}

Enqueue a high-priority follow-up for the next run. It is placed at the front of the pending queue (or started immediately if idle). It does not modify an HTTP request already in flight. Returns {:ok, request_id}.

stream(server, prompt, opts \\ [])

@spec stream(GenServer.server(), String.t(), keyword()) ::
  {:ok, String.t()} | {:error, :busy}

Stream a prompt's text deltas. Returns {:ok, request_id} immediately; deltas arrive as :text_delta events and the final result as a :run_finished event on the agent topic. Returns {:error, :busy} if a run is in flight.

Phase 1 scope: this surfaces text deltas via the existing streaming core. It does not run a full agentic tool loop (use chat/3/send_message/3 for that).

usage(server)

Accumulated token usage across all runs so far.