Sagents.AgentServer (Sagents v0.10.0)
Copy MarkdownGenServer that wraps a DeepAgent and its State, managing execution lifecycle and broadcasting events via PubSub.
The AgentServer provides:
- Asynchronous agent execution
- State management and tracking
- Event broadcasting for UI updates
- Human-in-the-loop interrupt handling
Understanding agent_id
The agent_id is a runtime identifier used for process management and
inter-process communication. It serves several critical purposes:
Process Registration
The agent_id is used to construct a Registry key via get_name(agent_id),
which returns a :via tuple for GenServer registration:
- Format:
{:via, Registry, {Sagents.Registry, {:agent_server, agent_id}}} - Ensures only one AgentServer process exists per agent_id
- Enables process lookup without maintaining PIDs
PubSub Topics
The agent_id forms the basis for PubSub topic construction:
- Topic format:
"agent_server:#{agent_id}" - External clients subscribe using:
AgentServer.subscribe(agent_id) - Events broadcast include: status changes, LLM deltas, todos updates
Middleware Context
The agent_id is passed to middleware during initialization, enabling:
- Coordination with agent-specific services (FileSystemServer, SubAgentsDynamicSupervisor)
- Parent-child relationship establishment in SubAgent hierarchies
- Per-agent resource isolation (virtual filesystems, etc.)
Supervision Tree Coordination
The agent_id flows through the entire supervision tree via AgentSupervisor,
ensuring all child processes (FileSystemServer, AgentServer,
SubAgentsDynamicSupervisor) are coordinated under the same agent context.
What agent_id IS NOT
Not Part of Conversation State: The agent_id is NOT included in
serialized state (via export_state/1). It's a runtime identifier, not
conversation data. This separation provides important benefits:
- Flexibility: Restore the same conversation state under a different
agent_id - State Cloning: Clone conversations for testing or forking scenarios
- Clean Architecture: Clear separation between runtime identity and data
When restoring state via start_link_from_state/2, you must provide the
agent_id as a parameter. This enables use cases like:
# Restore with same agent_id
AgentServer.start_link_from_state(saved_state, agent_id: "conversation-123")
# Clone conversation with different agent_id
AgentServer.start_link_from_state(saved_state, agent_id: "conversation-123-fork")The agent_id can be any value that makes sense for your application:
- Database conversation IDs:
"conv_a1b2c3d4" - User-scoped identifiers:
"user-#{user_id}-session-#{session_id}" - Randomly generated GUIDs:
UUID.uuid4() - Application-defined values:
"demo-agent-001"
Events
The server broadcasts events on the topic "agent_server:#{agent_id}".
All events are wrapped in an {:agent, event} tuple to help consumers
identify and route events from AgentServer. This is similar to how
FileSystemServer wraps its events in {:file_system, event}.
Event Format
Events are received in the format {:agent, event} where event is one of:
Todo Events
{:agent, {:todos_updated, todos}}- Complete snapshot of current TODO list
Status Events
{:agent, {:status_changed, :idle, nil}}- Server ready for work (also broadcast after successful execution completion){:agent, {:status_changed, :running, nil}}- Agent executing{:agent, {:status_changed, :interrupted, interrupt_data}}- Awaiting human decision{:agent, {:status_changed, :paused, pause_reason}}- Infrastructure pause; the payload is the cause the mode attached (nil when it attached none){:agent, {:status_changed, :cancelled, nil}}- Execution was cancelled by user{:agent, {:status_changed, :error, reason}}- Execution failed
Node Transfer Events
{:agent, {:node_transferring, data}}- Agent is about to leave this node (broadcast during terminate/2)data.from_node- The node the agent is leaving
{:agent, {:node_transferred, data}}- Agent has been restored on a new node (broadcast on startup after restore)data.to_node- The node the agent has been restored to
Shutdown Events
{:agent, {:agent_shutdown, shutdown_data}}- Agent is shutting downshutdown_data.agent_id- The agent identifiershutdown_data.reason- Shutdown reason (:inactivity|:no_viewers)shutdown_data.last_activity_at- DateTime of last activityshutdown_data.shutdown_at- DateTime when shutdown was initiated
Tool Execution Events (consolidated)
{:agent, {:tool_execution_update, status, tool_info}}- Tool execution lifecycle updatestatusis one of::executing,:completed,:failed:executing→tool_infocontains:call_id,:name,:display_text,:arguments:completed→tool_infocontains:call_id,:name,:result:failed→tool_infocontains:call_id,:name,:error
{:agent, {:display_message_updated, display_msg}}- Tool status updated in DB (only whendisplay_message_persistenceis configured)
LLM Streaming Events
{:agent, {:llm_deltas, [%MessageDelta{}]}}- Streaming tokens/deltas received (list of deltas){:agent, {:llm_message, %Message{}}}- Complete message received and processed{:agent, {:llm_token_usage, %TokenUsage{}}}- Token usage information
Message Persistence Events
{:agent, {:display_message_saved, display_message}}- Broadcast after message is persisted viadisplay_message_persistencebehaviour. The{:llm_message, ...}event is also broadcast alongside this event
Queued Message Events
{:agent, {:message_queued, %Message{}}}- A message arrived while a run was in flight and is being held for delivery at the next run boundary. Seeadd_message/3andqueue_message_from_tool/3. Subscribe to this rather than reading a "queued" signal offadd_message/2's return value, which stays:okdeliberately.
Note: File events are NOT broadcast by AgentServer. Files are managed by
FileSystemServer which provides its own event handling mechanism.
Debug Events
When debug PubSub is configured, additional debug events are broadcast on the
topic "agent_server:debug:#{agent_id}". These events provide deeper insight
into agent execution for debugging and monitoring purposes.
Debug events are also wrapped in {:agent, {:debug, event}} for consistent
routing with regular events.
Middleware Debug Events
{:agent, {:debug, {:agent_state_update, state}}}- Middleware state update with full state snapshot
Queued Message Debug Events
{:agent, {:debug, {:messages_drained, count}}}- A queued message was delivered and a follow-up run started. Note that no:idlestatus is broadcast between the two runs; this is the event for observers who need to know the extra run happened.{:agent, {:debug, {:pending_message_held, :error}}}- A run failed while a message was queued. The message is kept, not delivered.{:agent, {:debug, {:auto_execution_limit_reached, limit}}}- Too many consecutive runs were started by queued messages with no human input in between. The message was added to the conversation but no further run was started.
Usage
# Start a server
{:ok, agent} = Agent.new(
agent_id: "my-agent-1",
model: model,
base_system_prompt: "You are a helpful assistant."
)
initial_state = State.new!(%{
messages: [Message.new_user!("Write a hello world program")]
})
{:ok, _pid} = AgentServer.start_link(
agent: agent,
initial_state: initial_state,
name: AgentServer.get_name("my-agent-1")
)
# Subscribe to events
AgentServer.subscribe("my-agent-1")
# Execute the agent
:ok = AgentServer.execute("my-agent-1")
# Cancel execution if needed
:ok = AgentServer.cancel("my-agent-1")
# Listen for events
receive do
{:todos_updated, todos} -> IO.inspect(todos, label: "Current TODOs")
{:status_changed, :idle, nil} -> IO.puts("Done!")
endHuman-in-the-Loop Example
# Configure agent with interrupts
{:ok, agent} = Agent.new(
agent_id: "my-agent-1",
model: model,
interrupt_on: %{"write_file" => true}
)
{:ok, _pid} = AgentServer.start_link(
agent: agent,
initial_state: state,
name: AgentServer.get_name("my-agent-1")
)
AgentServer.subscribe("my-agent-1")
# Execute
AgentServer.execute("my-agent-1")
# Wait for interrupt
receive do
{:status_changed, :interrupted, interrupt_data} ->
# Display interrupt_data.action_requests to user
decisions = get_user_decisions(interrupt_data)
AgentServer.resume("my-agent-1", decisions)
end
# Wait for completion
receive do
{:status_changed, :idle, nil} -> :ok
end
Summary
Functions
Add a message to the agent's state and transition to idle if completed.
Gets count of currently running agents.
Gets detailed information about a running agent.
Cancel a running LLM task.
Dismiss a terminal :halt interrupt and transition the server back to :idle.
Execute the agent.
Export the current conversation state to a serializable format.
Gets the agent configuration for the given agent.
Get the current inactivity status of an agent.
Get server info including status, state, and any error or interrupt data.
Gets metadata about the agent server including status and last activity.
Get the name of the AgentServer process for a specific agent.
Get the pid of the AgentServer process for a specific agent.
Get the current status of the server.
Gets all running agents matching a glob pattern.
Lists all currently running agent processes.
Send a targeted message to a specific middleware in a running AgentServer.
Hook fired after a subscriber is registered. Default no-op. Override to send a snapshot of the current state to the new subscriber so it can sync without polling.
Request the AgentServer to publish a specific debug PubSub message or event.
Request the AgentServer to publish an specific PubSub message or event.
Queue a message into the conversation from inside a running tool or middleware.
Reset the agent's state and filesystem to start fresh.
Restore agent state from a previously exported state.
Resume agent execution after an interrupt.
Check if an agent is running.
Persist a synthetic display message and broadcast it to subscribers.
Start an AgentServer.
Start a new AgentServer with restored state.
Stop the AgentServer.
Subscribe a process to events from this AgentServer.
Touch the agent to indicate activity and reset the inactivity timer.
Unsubscribe a process from events on the given channel.
Updates both the agent configuration and state.
Types
Functions
@spec add_message(String.t(), LangChain.Message.t(), keyword()) :: :ok | {:error, term()}
Add a message to the agent's state and transition to idle if completed.
This is useful for conversational interfaces where you want to add a new user message after the agent has completed a previous execution.
Returns :ok on success.
Adding a message while the agent is running
A :user message that arrives while the agent is :running is queued,
not rejected. It is delivered at the next run boundary: appended to the
conversation once the in-flight run completes successfully, which then starts
a follow-up run. add_message/2 still returns :ok in that case.
This is deliberate and load-bearing: previously the message was written into
the rolling state and then silently destroyed when the canonical state from
Sagents.Agent.execute/3 replaced it wholesale, while the caller got a misleading
{:error, "Cannot execute, server is in state: running"} despite the message
having been accepted. Returning :ok is both more truthful and non-breaking
for consumers that match only on :ok and {:error, reason}.
A consumer that wants to render a "queued" affordance should subscribe to the
{:message_queued, message} event rather than read it off the return value.
Messages that arrive with a non-:user role while running are still
rejected. Injecting an assistant turn mid-run is not a supported operation.
Options
:display- controls the transcript half of the message, independently of what the model sees. Three values:- absent (default) - existing behavior. The configured
Sagents.MessagePreprocessorruns if one is set; with none, the display and LLM halves are the same message. - a
%LangChain.Message{}- the caller supplies both halves directly. The configured preprocessor is not run; the caller has already made the decision the preprocessor exists to make. :none- model-visible, display-invisible. Nothing reaches the transcript.
- absent (default) - existing behavior. The configured
The two halves are independent messages, including their role. Because
Sagents.DisplayMessagePersistence.save_message/3 derives the transcript
entry's attribution from message.role, a message that is :user to the
model can be :assistant in the transcript. That is the shape a
tool-initiated injection needs: nobody typed it, so attributing it to the
author would put words in their mouth.
Examples
# After agent completes
:ok = AgentServer.add_message("my-agent-1", Message.new_user!("What's next?"))
:ok = AgentServer.execute("my-agent-1")
# The author typed a shorthand; the model receives the expansion.
:ok = AgentServer.add_message("my-agent-1", Message.new_user!(expanded),
display: Message.new_user!("/changelog for the latest release"))
# Model-visible, transcript-invisible.
:ok = AgentServer.add_message("my-agent-1", Message.new_user!(playbook),
display: :none)
@spec agent_count() :: non_neg_integer()
Gets count of currently running agents.
Returns the total number of AgentServer processes registered in the Sagents.Registry.
Examples
AgentServer.agent_count()
# => 5
Gets detailed information about a running agent.
Returns a map with agent status and state information, or nil if the agent
is not running.
Return Value
If the agent is running, returns a map containing:
:agent_id- The agent identifier:pid- The process ID:status- Current execution status (:idle,:running,:interrupted, etc.):state- Exported state snapshot:message_count- Number of messages in the state:has_interrupt- Boolean indicating if there's pending interrupt data
Examples
AgentServer.agent_info("conversation-1")
# => %{
# agent_id: "conversation-1",
# pid: #PID<0.1234.0>,
# status: :idle,
# state: %State{...},
# message_count: 5,
# has_interrupt: false
# }
AgentServer.agent_info("nonexistent")
# => nil
Cancel a running LLM task.
Stops the currently executing agent task and transitions the server to completed status.
Returns {:error, reason} if the server is not running (no task to cancel).
Examples
:ok = AgentServer.cancel("my-agent-1")
Dismiss a terminal :halt interrupt and transition the server back to :idle.
Used to acknowledge a halt (e.g., user clicked "Dismiss" on the halt panel)
without sending a new message. Clears interrupt_data, calls
State.cancel_pending_interrupts/1, transitions status to :idle, and
broadcasts the status change.
Only valid when status is :interrupted and the pending interrupt is a
halt (or a :multiple_interrupts batch containing a halt). Other interrupt
types — HITL approvals, ask_user_question — require an explicit response
via resume/2 (which the relevant middleware interprets), so this function
returns {:error, "interrupt requires explicit response (use resume)"}
for those.
Examples
:ok = AgentServer.dismiss_interrupt("my-agent-1")
Execute the agent.
Starts agent execution asynchronously. The server will broadcast events as the
agent runs. Returns :ok immediately.
Returns {:error, reason} if the server is not idle (already running, interrupted, etc.).
Examples
:ok = AgentServer.execute("my-agent-1")
Export the current conversation state to a serializable format.
This can be persisted to a database and later used to restore the conversation state. The exported state uses string keys (not atoms) for compatibility with JSON/JSONB storage.
Returns a map with string keys containing:
"version"- Serialization format version"state"- The conversation state (messages, todos, metadata)"serialized_at"- ISO8601 timestamp
What is NOT included:
- Agent configuration (middleware, tools, model) - must come from application code
agent_id- runtime identifier provided when restoring
This design allows you to restore the same conversation state under a different agent_id, enabling use cases like state cloning and conversation forking.
Examples
state = AgentServer.export_state("my-agent-1")
# Save to database
MyApp.Conversations.save_agent_state(conversation_id, state)
@spec get_agent(String.t()) :: {:ok, Sagents.Agent.t()} | {:error, :not_found}
Gets the agent configuration for the given agent.
Returns
{:ok, Agent.t()}- The agent struct{:error, :not_found}- Agent not found
Examples
{:ok, agent} = AgentServer.get_agent("my-agent-1")
# => {:ok, %Agent{agent_id: "my-agent-1", model: %ChatModel{...}, ...}}
Get the current inactivity status of an agent.
Returns a map with:
:inactivity_timeout- Configured timeout in milliseconds (or nil/:infinity):last_activity_at- DateTime of last activity:timer_active- Boolean indicating if timer is currently running:time_since_activity- Milliseconds since last activity (or nil if no activity yet)
Examples
status = AgentServer.get_inactivity_status("my-agent-1")
# => %{
# inactivity_timeout: 300_000,
# last_activity_at: ~U[2025-11-06 10:15:30.123Z],
# timer_active: true,
# time_since_activity: 45_000
# }
Get server info including status, state, and any error or interrupt data.
Returns a map with:
:status- Current status:state- Current State:interrupt_data- Interrupt data if status is:interrupted:error- Error reason if status is:error
Examples
info = AgentServer.get_info("my-agent-1")
Gets metadata about the agent server including status and last activity.
Returns
{:ok, metadata}- Map with agent metadata{:error, :not_found}- Agent not found
Metadata Fields
:status- Current status atom (:idle, :running, :interrupted, :error, :cancelled):last_activity_at- DateTime of last activity (may be nil):conversation_id- Conversation ID (may be nil):node- The Erlang node where this agent is running
Examples
{:ok, metadata} = AgentServer.get_metadata("my-agent-1")
# => {:ok, %{status: :idle, last_activity_at: ~U[2024-01-01 12:00:00Z], conversation_id: "conv-123"}}
@spec get_name(String.t()) :: GenServer.name()
Get the name of the AgentServer process for a specific agent.
Examples
name = AgentServer.get_name("my-agent-1")
GenServer.call(name, :get_status)
Get the pid of the AgentServer process for a specific agent.
Examples
pid = AgentServer.get_pit("my-agent-1")
send(pid, message)
Get the current status of the server.
Returns one of:
:idle- Server ready for work:running- Agent executing:interrupted- Awaiting human decision:cancelled- Execution was cancelled:error- Execution failed:not_running- Agent process does not exist
Examples
:idle = AgentServer.get_status("my-agent-1")
:not_running = AgentServer.get_status("non-existent-agent")
Gets all running agents matching a glob pattern.
Supports wildcard patterns using * which matches any sequence of characters.
Examples
# Get all conversation agents
AgentServer.list_agents_matching("conversation-*")
# => ["conversation-1", "conversation-2", "conversation-123"]
# Get all user agents
AgentServer.list_agents_matching("user-*")
# => ["user-42", "user-99"]
# Get specific prefix
AgentServer.list_agents_matching("demo-*")
# => ["demo-agent-001"]
@spec list_running_agents() :: [String.t()]
Lists all currently running agent processes.
Returns a list of agent_ids for all running AgentServer processes registered in the Sagents.Registry.
Examples
AgentServer.list_running_agents()
# => ["conversation-1", "conversation-2", "user-123"]
Send a targeted message to a specific middleware in a running AgentServer.
The message is routed through the middleware registry and delivered to the
middleware's handle_message/3 callback. The middleware is identified by its
ID (module name by default, or a custom string if configured via :id option).
This is a fire-and-forget operation — the caller does not wait for a response. If the AgentServer is not running, the message is silently dropped.
Use Cases
There are two primary use cases for this function:
1. External notifications (LiveViews, controllers, other processes)
Send context updates or configuration changes to a middleware from outside the
agent system. The middleware decides how to handle the message — typically by
updating state metadata that before_model/2 reads on the next LLM call.
# LiveView: user switched to editing a different blog post
AgentServer.notify_middleware(agent_id, MyApp.UserContext, {:post_changed, %{
slug: "/blog/getting-started-with-elixir",
title: "Getting Started with Elixir"
}})
# Controller: user changed a preference
AgentServer.notify_middleware(agent_id, MyApp.Preferences, {:preference_changed, :verbose, true})2. Async task results (middleware sending messages to itself)
Middleware that spawns background tasks (e.g., title generation, embedding computation) uses this to send results back to the AgentServer for state updates.
# Inside an async task spawned by the middleware
AgentServer.notify_middleware(agent_id, middleware_id, {:title_generated, title})Parameters
agent_id- The agent identifier (used to locate the AgentServer process)middleware_id- The middleware ID to route the message to (module name or custom string)message- Any term to be delivered to the middleware'shandle_message/3callback
Returns
:ok— always returns:ok, even if the AgentServer is not running
Examples
# Notify by module name (default middleware ID)
AgentServer.notify_middleware("conv-123", MyApp.Middleware.UserContext, {:post_changed, post})
# Notify by custom ID (when middleware was configured with `id: "english_title"`)
AgentServer.notify_middleware("conv-123", "english_title", {:regenerate, %{}})
Hook fired after a subscriber is registered. Default no-op. Override to send a snapshot of the current state to the new subscriber so it can sync without polling.
Request the AgentServer to publish a specific debug PubSub message or event.
Designed to make it easier for middleware to publish debug messages to the Agent's debug PubSub. Debug events are useful for development and debugging but separate from user-facing events.
Standardized Middleware Action Pattern
Middleware should use the :middleware_action tuple pattern to avoid event
proliferation:
{:middleware_action, middleware_module, action_data}Where:
middleware_module- The middleware module (atom) that generated the eventaction_data- Middleware-specific action tuple or data
This pattern allows the debug UI to handle all middleware events generically without needing to know about every possible middleware-specific event type.
Examples
# From ConversationTitle middleware
AgentServer.publish_debug_event_from(
agent_id,
{:middleware_action, Sagents.Middleware.ConversationTitle, {:title_generation_started, user_text}}
)
AgentServer.publish_debug_event_from(
agent_id,
{:middleware_action, Sagents.Middleware.ConversationTitle, {:title_generation_completed, title}}
)
# From custom middleware
AgentServer.publish_debug_event_from(
agent_id,
{:middleware_action, MyApp.CustomMiddleware, {:validation_started, params}}
)
Request the AgentServer to publish an specific PubSub message or event.
Designed to make it easier for a middleware desiring to publish messages to the Agent's PubSub.
A PubSub message is only broadcast if the AgentServer is configured with PubSub.
@spec queue_message_from_tool(String.t(), LangChain.Message.t(), keyword()) :: :ok | {:error, :no_server}
Queue a message into the conversation from inside a running tool or middleware.
This is the tool-facing door to the same queue add_message/3 uses. The
message is held for the duration of the current run and delivered at the run
boundary, arriving as an ordinary :user message at the head of a follow-up
run. Nothing downstream needs to distinguish it from a message the user typed,
because there is nothing to distinguish: it is an ordinary user message.
Use this when a tool needs to hand the model instructions rather than data. Models are trained to treat a tool result as data and a user turn as instruction, so returning a playbook as tool-result text is a functional downgrade, not a stylistic one.
Why this is a cast
handle_call(:cancel, ...) blocks the AgentServer in Task.shutdown(task, 2_000). Tool code runs inside that Task. A tool blocked in a GenServer.call
back to its own server during a cancel is a mutual wait that only resolves
when the 2s grace expires and the task is brutal-killed. Both existing
tool-to-server helpers (publish_event_from/2, save_synthetic_message_from/2)
are casts for the same reason.
The cost is that the caller cannot learn whether the queue accepted the
message. The GenServer.whereis guard below covers the dominant failure mode:
no server at all, which is unit tests and bare Sagents.Agent.execute/3. It returns
{:error, :no_server} so a tool can fall back to inline content rather than
silently doing nothing.
Options
Same :display option as add_message/3, with one difference: a configured
Sagents.MessagePreprocessor is never run for a queued message. The
behaviour is scoped to messages a human submitted; a tool-generated playbook
is machine-generated and there is nothing coherent for a preprocessor to do
with it.
Prefer display: :none. The model emits a narration turn after the tool
result anyway, and that turn becomes its own display message. Supplying a
display: acknowledgement as well produces two acknowledgements back to
back. Reach for an explicit display: only when the wording, role, or
guaranteed presence of the acknowledgement has to be deterministic.
Returns
:ok- the cast was sent{:error, :no_server}- no AgentServer is running for thisagent_id
Limitations
Sub-agents register under {:sub_agent, id}, not {:agent_server, id}, and
block their own process for the whole of SubAgentServer.execute/1. A tool
running inside a sub-agent therefore gets {:error, :no_server} and should
take its fallback path. Queueing into the parent conversation is
deliberately not offered.
Examples
def activate_command(%{"command" => name} = args, context) do
playbook = render_playbook(name, args)
case AgentServer.queue_message_from_tool(context.state.agent_id,
Message.new_user!(playbook),
display: :none
) do
:ok ->
{:ok, "The #{name} command has been initiated. Its instructions " <>
"will arrive as the next message. Acknowledge briefly and stop."}
{:error, :no_server} ->
# Degraded path: no AgentServer (unit test, one-shot Agent.execute).
# Return the playbook inline rather than silently doing nothing.
{:ok, playbook}
end
end
Reset the agent's state and filesystem to start fresh.
This clears:
- All messages
- All TODOs
- Middleware state
- Memory-only files (completely removed)
- In-memory modifications to persisted files (discarded)
This preserves:
- Metadata (configuration)
- Persisted files (reverted to pristine state from storage)
Status transitions:
:completed,:error, or:cancelled→:idle(ready for new execution)- Other statuses remain unchanged
Returns :ok on success.
Examples
# After agent completes or encounters error
:ok = AgentServer.reset("my-agent-1")
# Now you can execute again with clean state
:ok = AgentServer.execute("my-agent-1")
Restore agent state from a previously exported state.
This updates an already-running agent to restore its state from a
previously serialized format. The state should be a map with string
keys (as returned by export_state/1).
Returns :ok on success or {:error, reason} on failure.
Examples
# Load from database
{:ok, persisted_state} = MyApp.Conversations.load_agent_state(conversation_id)
# Restore into existing agent
:ok = AgentServer.restore_state("my-agent-1", persisted_state)
Resume agent execution after an interrupt.
Parameters
agent_id- The agent identifierresume_data- Data to resume with (polymorphic per middleware). For HITL: list of decision maps. For AskUserQuestion: response map.
Examples
# HITL resume
decisions = [
%{type: :approve},
%{type: :edit, arguments: %{"path" => "safe.txt"}},
%{type: :reject}
]
:ok = AgentServer.resume("my-agent-1", decisions)
# AskUserQuestion resume
:ok = AgentServer.resume("my-agent-1", %{type: :answer, selected: ["PostgreSQL"]})
Check if an agent is running.
Persist a synthetic display message and broadcast it to subscribers.
Designed for middleware that needs to record user-facing transcript entries
that do not correspond to an LLM message (for example, a user's answer to an
ask_user question, or a "user cancelled" notification).
The attrs map should contain at minimum :message_type, :content_type, and
:content. AgentServer routes the request to the configured
Sagents.DisplayMessagePersistence implementation and broadcasts the saved
record via {:display_message_saved, msg}.
No-op if the AgentServer was not configured with both
display_message_persistence and conversation_id, or if the configured
persistence module does not implement save_synthetic_message/3.
Start an AgentServer.
Options
:agent- The Agent struct (required):initial_state- Initial State (default: empty state):initial_subscribers- List of{channel, pid}tuples to enroll as subscribers beforeinit/1returns. Use this to atomically start the server and subscribe — every event broadcast (including the initial{:status_changed, :idle, nil}and any{:node_transferred, _}after a Horde restore) is delivered to listed pids. Channels are:mainand:debug. Default:[].:pubsub- PubSub configuration as{module(), atom()}tuple ornil(default: nil). Used only for presence wiring (subscribing toPhoenix.Presencediff broadcasts). Per-agent events are delivered directly to subscribers viaSagents.Publisher, no PubSub required.:name- Server name registration (optional, defaults toget_name(agent.agent_id)):inactivity_timeout- Timeout in milliseconds for automatic shutdown due to inactivity (default: 300_000 - 5 minutes) Set tonilor:infinityto disable automatic shutdown:shutdown_delay- Delay in milliseconds to allow the supervisor to gracefully stop all children (default: 5000):conversation_id- Optional conversation identifier for message persistence (default: nil):agent_persistence- Module implementingSagents.AgentPersistencefor state snapshots (default: nil):display_message_persistence- Module implementingSagents.DisplayMessagePersistencefor display messages (default: nil)
Examples
# Start with automatic name (recommended)
{:ok, pid} = AgentServer.start_link(
agent: agent,
initial_state: state
)
# With PubSub enabled
{:ok, pid} = AgentServer.start_link(
agent: agent,
initial_state: state,
pubsub: {Phoenix.PubSub, :my_app_pubsub}
)
# Start with explicit name (advanced use cases)
{:ok, pid} = AgentServer.start_link(
agent: agent,
initial_state: state,
name: :my_custom_name
)
# With custom inactivity timeout
{:ok, pid} = AgentServer.start_link(
agent: agent,
inactivity_timeout: 600_000 # 10 minutes
)
# Disable automatic shutdown
{:ok, pid} = AgentServer.start_link(
agent: agent,
inactivity_timeout: nil
)
@spec start_link_from_state( map(), keyword() ) :: GenServer.on_start()
Start a new AgentServer with restored state.
This is the preferred way to resume a conversation from persisted state.
The persisted_state should be a map with string keys (as returned by
export_state/1).
The agent_id will be used as the runtime identifier for this agent,
enabling process registration and PubSub topic setup. You can restore
the same conversation state under a different agent_id, which is useful
for state cloning or conversation forking.
Agent Configuration from Code
REQUIRED: You MUST provide the agent from your application code using the
:agent option. The persisted state contains ONLY conversation state (messages,
todos, metadata). Agent configuration (middleware, tools, model) comes from your
application code.
This design ensures:
- Library upgrades automatically benefit all conversations
- Code changes automatically apply to all conversations
- Per-user capabilities can be controlled via application logic
Options
:agent_id- The runtime identifier for this agent (required):agent- Agent struct from code (REQUIRED):pubsub- PubSub configuration as{module(), atom()}tuple ornil(default: nil):name- Server name registration (optional, defaults toget_name(agent_id)):inactivity_timeout- Timeout in milliseconds (default: 300_000):shutdown_delay- Delay in milliseconds (default: 5000)
Examples
# Standard restoration pattern
{:ok, persisted_state} = MyApp.Conversations.load_agent_state(conversation_id)
# Agent from code (ALWAYS required)
{:ok, agent} = MyApp.Agents.create_agent(
agent_id: "my-agent-1",
model: model,
middleware: [TodoList, FileSystem, SubAgent]
)
# Start with restored state
{:ok, pid} = AgentServer.start_link_from_state(
persisted_state,
agent: agent,
agent_id: "my-agent-1",
pubsub: {Phoenix.PubSub, :my_app_pubsub}
)
# Clone conversation with a different agent_id
{:ok, pid} = AgentServer.start_link_from_state(
persisted_state,
agent: agent,
agent_id: "my-agent-clone",
pubsub: {Phoenix.PubSub, :my_app_pubsub}
)
@spec stop(String.t()) :: :ok
Stop the AgentServer.
Examples
:ok = AgentServer.stop("my-agent-1")
@spec subscribe(String.t(), :main | :debug, pid() | nil) :: {:ok, pid(), reference()} | {:error, :process_not_found}
Subscribe a process to events from this AgentServer.
Events on the :main channel are delivered as {:agent, event} messages;
events on the :debug channel are delivered as {:agent, {:debug, event}}
and provide additional insight into middleware state, sub-agent activity,
and similar diagnostic data not surfaced on the main channel.
Delivery is via direct send/2. The producer monitors the subscriber so
departure is cleaned up automatically — but subscribers should also
Process.monitor/1 the returned server_pid to detect server death.
Arguments
agent_id— the agent's id.channel—:main(default) or:debug.subscriber_pid— the pid to receive events. Defaults toself()whennil.
Returns {:ok, server_pid, monitor_ref} on success, or
{:error, :process_not_found} if no AgentServer is running for agent_id.
Examples
# Most common: subscribe self() to the main channel.
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1")
# Subscribe to debug events (used by sagents_live_debugger and similar).
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1", :debug)
# Subscribe a foreign pid (e.g. a bridge GenServer that proxies events).
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1", :main, bridge_pid)
@spec touch(String.t()) :: :ok
Touch the agent to indicate activity and reset the inactivity timer.
This is useful when external events indicate activity (e.g., user viewing the agent in the UI, clicking tabs, etc.) to keep the agent alive and prevent automatic shutdown due to inactivity.
Returns :ok immediately (non-blocking).
Examples
:ok = AgentServer.touch("my-agent-1")
Unsubscribe a process from events on the given channel.
Mirrors subscribe/3. Defaults channel to :main and subscriber_pid
to self() (when nil). Always returns :ok.
Updates both the agent configuration and state.
This is the recommended way to restore a conversation:
- Create agent from current code using your agent factory
- Load state from database
- Call this function to update the running AgentServer
Parameters
agent_id- The agent server's identifieragent- The new agent configuration (from current code)state- The restored state (from database)
Examples
# Restore conversation
{:ok, agent} = MyApp.Agents.create_demo_agent(agent_id: "demo-123")
{:ok, state_data} = MyApp.Conversations.load_state(conversation_id)
{:ok, state} = Sagents.State.from_serialized(state_data["state"])
:ok = AgentServer.update_agent_and_state("demo-123", agent, state)Returns
:ok- Agent and state updated successfully{:error, reason}- If agent server is not running or update fails