ag_ui/server/
mod.rs

1//! Host an [AG-UI] agent in Rust.
2//!
3//! AG-UI is the protocol between a user-facing application and an agent
4//! backend: a POST carrying [`RunAgentInput`],
5//! answered by a stream of typed events. This crate is the server half —
6//! implement [`Agent`], hand it to [`run()`], and you have a stream a transport
7//! can serialize. [`ag-ui-axum`] mounts it on a router; nothing here depends on
8//! a web framework, an executor or an LLM client.
9//!
10//! ```
11//! use ag_ui::{Event, EventType, RunAgentInput, RunOutcome};
12//! use ag_ui::server::{Agent, Result, RunContext, run};
13//! use futures_util::StreamExt;
14//! use serde::{Deserialize, Serialize};
15//!
16//! /// State the client mirrors and the agent updates.
17//! #[derive(Default, Serialize, Deserialize)]
18//! struct Draft {
19//!     revision: u32,
20//!     title: String,
21//! }
22//!
23//! struct Editor;
24//!
25//! impl Agent for Editor {
26//!     type State = Draft;
27//!
28//!     async fn run(&self, ctx: &mut RunContext<Draft>) -> Result<RunOutcome> {
29//!         // A step brackets a phase of the run. Its guard emits
30//!         // STEP_FINISHED on drop, so an early `?` cannot skip it.
31//!         let mut step = ctx.step("draft")?;
32//!
33//!         // Reasoning the client can render, in its own REASONING_* block.
34//!         step.think("The user wants a title.")?;
35//!
36//!         // A message streams as TEXT_MESSAGE_START / _CONTENT* / _END.
37//!         let mut message = step.assistant_message()?;
38//!         message.delta("Naming it ")?;
39//!         message.delta("\"Q3 plan\".")?;
40//!         message.end()?;
41//!
42//!         // Publishing state diffs against the last snapshot and sends
43//!         // whichever of STATE_SNAPSHOT / STATE_DELTA is smaller.
44//!         step.update_state(|draft| {
45//!             draft.revision += 1;
46//!             draft.title = "Q3 plan".into();
47//!         })?;
48//!
49//!         drop(step); // or just let it fall out of scope
50//!         Ok(RunOutcome::Success)
51//!     }
52//! }
53//!
54//! # let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
55//! # rt.block_on(async {
56//! let input = RunAgentInput::new("thread-1", "run-1");
57//! let events: Vec<Event> = run(Editor, input)
58//!     .map(|event| event.expect("the stream should not break"))
59//!     .collect()
60//!     .await;
61//!
62//! let types: Vec<EventType> = events.iter().map(Event::event_type).collect();
63//! assert_eq!(
64//!     types,
65//!     [
66//!         EventType::RunStarted,
67//!         EventType::StepStarted,
68//!         EventType::ReasoningStart,
69//!         EventType::ReasoningMessageStart,
70//!         EventType::ReasoningMessageContent,
71//!         EventType::ReasoningMessageEnd,
72//!         EventType::ReasoningEnd,
73//!         EventType::TextMessageStart,
74//!         EventType::TextMessageContent,
75//!         EventType::TextMessageContent,
76//!         EventType::TextMessageEnd,
77//!         EventType::StateSnapshot,
78//!         EventType::StepFinished,
79//!         EventType::RunFinished,
80//!     ]
81//! );
82//! # });
83//! ```
84//!
85//! # The four things that shape this API
86//!
87//! **Protocol misuse should not compile.** Event ordering is enforced by
88//! [typestate handles](emit) that borrow the run context mutably, so
89//! interleaving two messages is a borrow-check error. The handles emit their
90//! terminating event on `Drop`, so it cannot be forgotten. What the borrow
91//! checker cannot catch — raw [`emit`](RunContext::emit) calls — a runtime
92//! [ordering verifier](verify) catches, on by default.
93//!
94//! **The emit path is synchronous.** `Drop` cannot be async, so a handle cannot
95//! `await` while emitting its terminator: `msg.delta(text)?` takes no `.await`.
96//! Emitters push into an unbounded channel and the transport drains it.
97//!
98//! **Executor-agnostic.** `futures` primitives throughout, no tokio in the
99//! dependency list, no `spawn`. [`CancellationToken`] is an `AtomicBool` and a
100//! waker list rather than `tokio_util`'s. Polling the stream is what runs the
101//! agent.
102//!
103//! **One extension point.** Everything that observes or rewrites the stream is
104//! a [`StreamTransformer`]. There is no parallel builder of callbacks; the
105//! hooks other SDKs expose that way are built-in transformers here —
106//! [`FilterToolCalls`], [`ToolResultToState`].
107//!
108//! # Human in the loop
109//!
110//! Return [`RunOutcome::Interrupt`] to pause
111//! a run. The client answers, and the next request carries the answers in
112//! [`RunContext::resume`]:
113//!
114//! ```
115//! # use ag_ui::{Interrupt, ResumeStatus, RunOutcome};
116//! # use ag_ui::server::{Agent, Result, RunContext};
117//! # struct Approver;
118//! impl Agent for Approver {
119//!     type State = ();
120//!
121//!     async fn run(&self, ctx: &mut RunContext<()>) -> Result<RunOutcome> {
122//!         match ctx.resume_for("delete-everything") {
123//!             None => Ok(RunOutcome::interrupt(vec![Interrupt::new(
124//!                 "delete-everything",
125//!                 "tool_approval",
126//!             )])),
127//!             Some(answer) if answer.status == ResumeStatus::Resolved => {
128//!                 ctx.say("Done.")?;
129//!                 Ok(RunOutcome::Success)
130//!             }
131//!             Some(_) => {
132//!                 ctx.say("Cancelled.")?;
133//!                 Ok(RunOutcome::Success)
134//!             }
135//!         }
136//!     }
137//! }
138//! ```
139//!
140//! # Subagents
141//!
142//! An agent that delegates opens a [`RunContext::subagent`] scope. Everything
143//! emitted through the handle — text, tool calls, reasoning, steps, nested
144//! subagents — comes out attributed to that invocation, bracketed by
145//! `SUBAGENT_STARTED` and `SUBAGENT_FINISHED`, so a client can group the
146//! output by who produced it:
147//!
148//! ```
149//! # use ag_ui::RunOutcome;
150//! # use ag_ui::server::{Agent, Result, RunContext};
151//! # struct Supervisor;
152//! impl Agent for Supervisor {
153//!     type State = ();
154//!
155//!     async fn run(&self, ctx: &mut RunContext<()>) -> Result<RunOutcome> {
156//!         let mut planner = ctx.subagent("planner")?;
157//!         planner.say("Two tasks: scope, then risks.")?;
158//!         {
159//!             let mut estimator = planner.subagent("estimator")?;   // nested
160//!             estimator.say("About a day each.")?;
161//!         }                                                          // SUBAGENT_FINISHED
162//!         planner.finish_with(serde_json::json!({ "tasks": 2 }))?;
163//!
164//!         ctx.say("Plan ready.")?;                                   // the parent's own
165//!         Ok(RunOutcome::Success)
166//!     }
167//! }
168//! ```
169//!
170//! What a consumer sees is a producer-side choice, because a client older than
171//! subagent support fails while decoding the lifecycle events. The default
172//! sends the stream as emitted; [`SubagentVisibility::inline`] flattens it to
173//! the pre-subagent shape and [`SubagentVisibility::hidden`] keeps only the
174//! parent's own events — both are ordinary transformers.
175//!
176//! # Features
177//!
178//! - `verify` *(default)* — the ordering state machine. Off, the whole
179//!   verifier is a zero-sized type whose checks compile away.
180//!
181//! [AG-UI]: https://docs.ag-ui.com
182//! [`ag-ui-axum`]: https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/axum/index.html
183//
184// `ag-ui-core` items are spelled as absolute links to the published rustdoc
185// rather than as intra-doc paths. `cargo doc --no-deps` — which is what CI and
186// the Pages deploy both run — cannot emit a path into a crate it is not
187// documenting, and it does not warn: a cross-crate intra-doc link silently
188// becomes literal `[text]`, and the `[text](path)` form silently becomes an
189// href of `path`, which renders as a link and 404s. See the `doc-links` job.
190//! [`RunAgentInput`]: https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/input/struct.RunAgentInput.html
191//! [`RunOutcome::Interrupt`]: https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/outcome/enum.RunOutcome.html#variant.Interrupt
192
193pub mod agent;
194pub mod cancel;
195pub mod context;
196pub mod emit;
197pub mod error;
198pub mod run;
199pub mod state;
200pub mod transform;
201pub mod verify;
202
203pub use agent::{Agent, AgentState, BoxAgent, DynAgent};
204pub use cancel::{CancellationToken, Cancelled};
205pub use context::RunContext;
206pub use emit::{
207    EventReceiver, MessageHandle, ReasoningHandle, StepGuard, SubagentHandle, ToolCallHandle,
208};
209pub use error::{Error, Result, Rule, VerificationError};
210pub use run::{Runner, run};
211pub use state::{StateManager, StatePublish};
212pub use transform::{
213    FilterToolCalls, StreamTransformer, SubagentFilter, SubagentVisibility, ToolResultToState,
214    TransformerChain,
215};
ag_ui/server/mod.rs

ag_ui/server/
mod.rs
