Struct Builder 

pub struct Builder<Host: Role, Handler = NullHandler, Runner = NullRun, Close = NullClose>
where
    Handler: HandleDispatchFrom<Host::Counterpart>,
    Runner: RunWithConnectionTo<Host::Counterpart>,
    Close: HandleConnectionClose<Host::Counterpart>,
{ /* private fields */ }

A JSON-RPC connection that can act as either a server, client, or both.

Builder provides a builder-style API for creating JSON-RPC servers and clients. You start by calling Role.builder() (e.g., Client.builder()), then add message handlers, and finally drive the connection with either connect_to or connect_with, providing a component implementation (e.g., ByteStreams for byte streams).

JSON-RPC Primer

JSON-RPC 2.0 has two fundamental message types:

• Requests - Messages that expect a response. They have an id field that gets echoed back in the response so the sender can correlate them.
• Notifications - Fire-and-forget messages with no id field. The sender doesn’t expect or receive a response.

Type-Driven Message Dispatch

The handler registration methods use Rust’s type system to determine which messages to handle. The type parameter you provide controls what gets dispatched to your handler:

Single Message Types

The simplest case - handle one specific message type:

connection
    .on_receive_request(async |req: InitializeRequest, responder, cx| {
        // Handle only InitializeRequest messages
        responder.respond(InitializeResponse::make())
    }, agent_client_protocol::on_receive_request!())
    .on_receive_notification(async |notif: SessionNotification, cx| {
        // Handle only SessionUpdate notifications
        Ok(())
    }, agent_client_protocol::on_receive_notification!())

Enum Message Types

You can also handle multiple related messages with a single handler by defining an enum that implements the appropriate trait (JsonRpcRequest or JsonRpcNotification):

// Define an enum for multiple request types
#[derive(Debug, Clone)]
enum MyRequests {
    Initialize(InitializeRequest),
    Prompt(PromptRequest),
}

// Implement JsonRpcRequest for your enum
impl JsonRpcRequest for MyRequests { type Response = serde_json::Value; }

// Handle all variants in one place
connection.on_receive_request(async |req: MyRequests, responder, cx| {
    match req {
        MyRequests::Initialize(init) => { responder.respond(serde_json::json!({})) }
        MyRequests::Prompt(prompt) => { responder.respond(serde_json::json!({})) }
    }
}, agent_client_protocol::on_receive_request!())

Mixed Message Types

To handle requests, notifications, and responses in one callback, use on_receive_dispatch:

// on_receive_dispatch receives requests, notifications, and responses
connection.on_receive_dispatch(async |msg: Dispatch<InitializeRequest, SessionNotification>, _cx| {
    match msg {
        Dispatch::Request(req, responder) => {
            responder.respond(InitializeResponse::make())
        }
        Dispatch::Notification(notif) => {
            Ok(())
        }
        Dispatch::Response(result, router) => {
            // Forward response to its destination
            router.route_with_result(result)
        }
    }
}, agent_client_protocol::on_receive_dispatch!())

Handler Registration

Register handlers using these methods (listed from most common to most flexible):

• on_receive_request - Handle JSON-RPC requests (messages expecting responses)
• on_receive_notification - Handle JSON-RPC notifications (fire-and-forget)
• on_receive_dispatch - Handle requests, notifications, and responses in one callback
• with_handler - Low-level primitive for maximum flexibility

Handler Ordering

Handlers are tried in the order you register them. The first handler that claims a message (by matching its type) will process it. Subsequent handlers won’t see that message:

connection
    .on_receive_request(async |req: InitializeRequest, responder, cx| {
        // This runs first for InitializeRequest
        responder.respond(InitializeResponse::make())
    }, agent_client_protocol::on_receive_request!())
    .on_receive_request(async |req: PromptRequest, responder, cx| {
        // This runs first for PromptRequest
        responder.respond(PromptResponse::make())
    }, agent_client_protocol::on_receive_request!())
    // Unknown requests receive Method not found automatically; unhandled
    // notifications are ignored.

Event Loop and Concurrency

Understanding the event loop is critical for writing correct handlers.

The Event Loop

Builder runs all handler callbacks on a single async task - the event loop. While a handler is running, the server cannot receive new messages. This means any blocking or expensive work in your handlers will stall the entire connection.

To avoid blocking the event loop, use ConnectionTo::spawn to offload serious work to concurrent tasks:

connection.on_receive_request(async |req: AnalyzeRequest, responder, cx| {
    // Clone cx for the spawned task
    cx.spawn({
        let connection = cx.clone();
        async move {
            let result = expensive_analysis(&req.data).await?;
            connection.send_notification(AnalysisComplete { result })?;
            Ok(())
        }
    })?;

    // Respond immediately without blocking
    responder.respond(AnalysisStarted { job_id: 42 })
}, agent_client_protocol::on_receive_request!())

Note that the entire connection runs within one async task, so parallelism must be managed explicitly using spawn.

The Connection Context

Handler callbacks receive a context object (cx) for interacting with the connection:

• For request handlers - Responder<R> provides respond to send the response, plus methods to send other messages
• For notification handlers - ConnectionTo provides methods to send messages and spawn tasks

Both context types support:

• send_request - Send requests to the other side
• send_notification - Send notifications
• spawn - Run tasks concurrently without blocking the event loop

The SentRequest returned by send_request provides methods like on_receiving_result that help you avoid accidentally blocking the event loop while waiting for responses.

Driving the Connection

After adding handlers, you must drive the connection using one of two modes:

Server Mode: connect_to()

Use connect_to when you only need to respond to incoming messages:

connection
    .on_receive_request(async |req: MyRequest, responder, cx| {
        responder.respond(MyResponse { status: "ok".into() })
    }, agent_client_protocol::on_receive_request!())
    .connect_to(MockTransport)  // Runs until connection closes or error occurs
    .await?;

The connection will process incoming messages and invoke your handlers until the connection is closed or an error occurs.

Client Mode: connect_with()

Use connect_with when you need to both handle incoming messages AND send your own requests/notifications:

connection
    .on_receive_request(async |req: MyRequest, responder, cx| {
        responder.respond(MyResponse { status: "ok".into() })
    }, agent_client_protocol::on_receive_request!())
    .connect_with(MockTransport, async |cx| {
        // You can send requests to the other side
        let response = cx.send_request(InitializeRequest::make())
            .block_task()
            .await?;

        // And send notifications
        cx.send_notification(StatusUpdate { message: "ready".into() })?;

        Ok(())
    })
    .await?;

The connection will serve incoming messages in the background while your client closure runs. When the closure returns, the connection shuts down.

Example: Complete Agent

let transport = Stdio::new();

UntypedRole.builder()
    .name("my-agent")  // Optional: for debugging logs
    .on_receive_request(async |init: InitializeRequest, responder, cx| {
        let response: InitializeResponse = todo!();
        responder.respond(response)
    }, agent_client_protocol::on_receive_request!())
    .on_receive_request(async |prompt: PromptRequest, responder, cx| {
        // You can send notifications while processing a request
        let notif: SessionNotification = todo!();
        cx.send_notification(notif)?;

        // Then respond to the request
        let response: PromptResponse = todo!();
        responder.respond(response)
    }, agent_client_protocol::on_receive_request!())
    .connect_to(transport)
    .await?;

Implementations

impl<Host: Role> Builder<Host, NullHandler, NullRun, NullClose>

pub fn new(role: Host) -> Self

Create a new connection builder for the given role. This type follows a builder pattern; use other methods to configure and then invoke Self::connect_to (to use as a server) or Self::connect_with to use as a client.

impl<Host: Role, Handler> Builder<Host, Handler, NullRun, NullClose>

where Handler: HandleDispatchFrom<Host::Counterpart>,

pub fn new_with(role: Host, handler: Handler) -> Self

Create a new connection builder with the given handler.

impl<Host: Role, Handler: HandleDispatchFrom<Host::Counterpart>, Runner: RunWithConnectionTo<Host::Counterpart>, Close: HandleConnectionClose<Host::Counterpart>> Builder<Host, Handler, Runner, Close>

pub fn name(self, name: impl ToString) -> Self

Set the “name” of this connection – used only for debugging logs.

pub fn with_connection_builder( self, other: Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, impl RunWithConnectionTo<Host::Counterpart>, impl HandleConnectionClose<Host::Counterpart>>, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, impl RunWithConnectionTo<Host::Counterpart>, impl HandleConnectionClose<Host::Counterpart>>

Merge another Builder into this one.

Prefer Self::on_receive_request or Self::on_receive_notification. This is a low-level method that is not intended for general use.

pub fn with_handler( self, handler: impl HandleDispatchFrom<Host::Counterpart>, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

Add a new HandleDispatchFrom to the chain.

Prefer Self::on_receive_request or Self::on_receive_notification. This is a low-level method that is not intended for general use.

pub fn with_runner<Run1>( self, runner: Run1, ) -> Builder<Host, Handler, impl RunWithConnectionTo<Host::Counterpart>, Close>

where Run1: RunWithConnectionTo<Host::Counterpart>,

Add a new RunWithConnectionTo to the chain.

pub fn with_spawned<F, Fut>( self, task: F, ) -> Builder<Host, Handler, impl RunWithConnectionTo<Host::Counterpart>, Close>

where F: FnOnce(ConnectionTo<Host::Counterpart>) -> Fut + Send, Fut: Future<Output = Result<(), Error>> + Send,

Enqueue a task to run once the connection is actively serving traffic.

pub fn on_close<F, Fut>( self, callback: F, ) -> Builder<Host, Handler, Runner, impl HandleConnectionClose<Host::Counterpart>>

where F: FnOnce(ConnectionTo<Host::Counterpart>) -> Fut + Send, Fut: Future<Output = Result<(), Error>> + Send,

Run a callback when the incoming transport reaches clean EOF.

Each callback runs at most once and receives the connection context. A successful callback observes the close without otherwise changing the lifetime of connect_with. Returning an error shuts down the connection and cancels a still-running connect_with future.

Multiple callbacks run sequentially in registration order. All of them run even if an earlier callback fails, after which the first error is returned. Pending requests are failed before callbacks begin, while ConnectionTo::incoming_closed completes only after they finish. A callback must therefore not await that close future itself.

This separation lets applications choose their cancellation policy. A callback can notify application-owned tasks and return Ok(()) for graceful cleanup, or return an error to stop them immediately.

Client.builder()
    .on_close(async |_cx| {
        Err(Error::internal_error().data("agent transport closed"))
    })
    .connect_with(transport, async |_cx| {
        std::future::pending().await
    })
    .await?;

pub fn on_receive_dispatch<Req, Notif, F, T, ToFut>( self, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Host::Counterpart>, Req: JsonRpcRequest, Notif: JsonRpcNotification, F: AsyncFnMut(Dispatch<Req, Notif>, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<Dispatch<Req, Notif>>, ToFut: Fn(&mut F, Dispatch<Req, Notif>, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for requests, notifications, and responses.

Use this when you want to handle all JSON-RPC message kinds in one callback. Your handler receives a Dispatch<Req, Notif> with three variants:

• Dispatch::Request(request, responder) - A request with its response context
• Dispatch::Notification(notification) - A notification
• Dispatch::Response(result, router) - A response to a request we sent

Example

connection.on_receive_dispatch(async |message: Dispatch<MyRequest, StatusUpdate>, _cx| {
    match message {
        Dispatch::Request(req, responder) => {
            // Handle request and send response
            responder.respond(MyResponse { status: "ok".into() })
        }
        Dispatch::Notification(notif) => {
            // Handle notification (no response needed)
            Ok(())
        }
        Dispatch::Response(result, router) => {
            // Forward response to its destination
            router.route_with_result(result)
        }
    }
}, agent_client_protocol::on_receive_dispatch!())

For most use cases, prefer on_receive_request or on_receive_notification which provide cleaner APIs for handling requests or notifications separately.

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn on_receive_request<Req: JsonRpcRequest, F, T, ToFut>( self, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Host::Counterpart>, F: AsyncFnMut(Req, Responder<Req::Response>, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<(Req, Responder<Req::Response>)>, ToFut: Fn(&mut F, Req, Responder<Req::Response>, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for JSON-RPC requests of type Req.

Your handler receives three arguments:

1. The request (type Req)
2. A Responder<Req::Response> for sending the response
3. A ConnectionTo for the peer that sent the request

The request context allows you to:

• Send the response with Responder::respond
• Send notifications to the client with ConnectionTo::send_notification
• Send requests to the client with ConnectionTo::send_request

Example

Agent.builder().on_receive_request(async |request: PromptRequest, responder, cx| {
    // Send a notification while processing
    let notif: SessionNotification = todo!();
    cx.send_notification(notif)?;

    // Send the response
    let response: PromptResponse = todo!();
    responder.respond(response)
}, agent_client_protocol::on_receive_request!())
.connect_to(transport)
.await

Type Parameter

Req can be either a single request type or an enum of multiple request types. See the type-driven dispatch section for details.

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn on_receive_notification<Notif, F, T, ToFut>( self, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Host::Counterpart>, Notif: JsonRpcNotification, F: AsyncFnMut(Notif, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<(Notif, ConnectionTo<Host::Counterpart>)>, ToFut: Fn(&mut F, Notif, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for JSON-RPC notifications of type Notif.

Notifications are fire-and-forget messages that don’t expect a response. Your handler receives:

1. The notification (type Notif)
2. A ConnectionTo<R> for sending messages to the other side

Unlike request handlers, you cannot send a response (notifications don’t have IDs), but you can still send your own requests and notifications using the context.

Example

connection.on_receive_notification(async |notif: SessionUpdate, cx| {
    // Process the notification
    update_session_state(&notif)?;

    // Optionally send a notification back
    cx.send_notification(StatusUpdate {
        message: "Acknowledged".into(),
    })?;

    Ok(())
}, agent_client_protocol::on_receive_notification!())

Type Parameter

Notif can be either a single notification type or an enum of multiple notification types. See the type-driven dispatch section for details.

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn on_receive_dispatch_from<Req: JsonRpcRequest, Notif: JsonRpcNotification, Peer: Role, F, T, ToFut>( self, peer: Peer, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Peer>, F: AsyncFnMut(Dispatch<Req, Notif>, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<Dispatch<Req, Notif>>, ToFut: Fn(&mut F, Dispatch<Req, Notif>, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for messages from a specific peer.

This is similar to on_receive_dispatch, but allows specifying the source peer explicitly. This is useful when receiving messages from a peer that requires message transformation (e.g., unwrapping SuccessorMessage envelopes when receiving from an agent via a proxy).

For the common case of receiving from the default counterpart, use on_receive_dispatch instead.

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn on_receive_request_from<Req: JsonRpcRequest, Peer: Role, F, T, ToFut>( self, peer: Peer, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Peer>, F: AsyncFnMut(Req, Responder<Req::Response>, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<(Req, Responder<Req::Response>)>, ToFut: Fn(&mut F, Req, Responder<Req::Response>, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for JSON-RPC requests from a specific peer.

This is similar to on_receive_request, but allows specifying the source peer explicitly. This is useful when receiving messages from a peer that requires message transformation (e.g., unwrapping SuccessorRequest envelopes when receiving from an agent via a proxy).

For the common case of receiving from the default counterpart, use on_receive_request instead.

Example

use agent_client_protocol::Agent;
use agent_client_protocol::schema::v1::InitializeRequest;

// Conductor receiving from agent direction - messages will be unwrapped from SuccessorMessage
connection.on_receive_request_from(Agent, async |req: InitializeRequest, responder, cx| {
    // Handle the request
    responder.respond(InitializeResponse::make())
})

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn on_receive_notification_from<Notif: JsonRpcNotification, Peer: Role, F, T, ToFut>( self, peer: Peer, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner, Close>

where Host::Counterpart: HasPeer<Peer>, F: AsyncFnMut(Notif, ConnectionTo<Host::Counterpart>) -> Result<T, Error> + Send, T: IntoHandled<(Notif, ConnectionTo<Host::Counterpart>)>, ToFut: Fn(&mut F, Notif, ConnectionTo<Host::Counterpart>) -> BoxFuture<'_, Result<T, Error>> + Send + Sync,

Register a handler for JSON-RPC notifications from a specific peer.

This is similar to on_receive_notification, but allows specifying the source peer explicitly. This is useful when receiving messages from a peer that requires message transformation (e.g., unwrapping SuccessorNotification envelopes when receiving from an agent via a proxy).

For the common case of receiving from the default counterpart, use on_receive_notification instead.

Ordering

This callback runs inside the dispatch loop and blocks further message processing until it completes. See the ordering module for details on ordering guarantees and how to avoid deadlocks.

pub fn with_mcp_server( self, mcp_server: McpServer<Host::Counterpart, impl RunWithConnectionTo<Host::Counterpart>>, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, impl RunWithConnectionTo<Host::Counterpart>, Close>

where Host::Counterpart: HasPeer<Agent> + HasPeer<Client>,

Available on crate feature unstable_mcp_over_acp only.

Add an MCP server to session setup requests proxied through this connection.

The same native MCP server declaration is added to new, load, and resume requests, plus fork requests when unstable_session_fork is enabled.

Only applicable to proxies.

pub async fn connect_to( self, transport: impl ConnectTo<Host> + 'static, ) -> Result<(), Error>

Run in server mode with the provided transport.

This drives the connection by continuously processing messages from the transport and dispatching them to your registered handlers. The connection will run until:

• The transport closes (e.g., EOF on byte streams)
• An error occurs

Handler errors are normally contained: requests receive an Error Response, response-handler errors are routed to the pending local request, and notification errors are logged without a wire reply.

On clean EOF, messages already accepted by the outgoing queue—including handler responses and close-callback notifications—are drained through the transport sink before this returns Ok(()).

The transport boundary carries TransportFrame values. Physical stream adapters serialize and deserialize frames, while channel-based components relay them directly.

Use this mode when you only need to respond to incoming messages and don’t need to initiate your own requests. If you need to send requests to the other side, use connect_with instead.

Example: Byte Stream Transport

let transport = Stdio::new();

UntypedRole.builder()
    .on_receive_request(async |req: MyRequest, responder, cx| {
        responder.respond(MyResponse { status: "ok".into() })
    }, agent_client_protocol::on_receive_request!())
    .connect_to(transport)
    .await?;

pub async fn connect_with<R>( self, transport: impl ConnectTo<Host> + 'static, main_fn: impl AsyncFnOnce(ConnectionTo<Host::Counterpart>) -> Result<R, Error>, ) -> Result<R, Error>

Run the connection until the provided closure completes.

This drives the connection by:

1. Running your registered handlers in the background to process incoming messages
2. Executing your main_fn closure with a ConnectionTo<R> for sending requests/notifications

The connection stays active until your main_fn returns, then shuts down. Clean incoming EOF fails every pending request and makes future requests fail immediately. It does not cancel unrelated work in main_fn: that future may observe ConnectionTo::incoming_closed, or the builder can use on_close to notify it or return an error and stop it.

Use this mode when you need to initiate communication (send requests/notifications) in addition to responding to incoming messages. For server-only mode where you just respond to messages, use connect_to instead.

Example

let transport = Stdio::new();

UntypedRole.builder()
    .on_receive_request(async |req: MyRequest, responder, cx| {
        // Handle incoming requests in the background
        responder.respond(MyResponse { status: "ok".into() })
    }, agent_client_protocol::on_receive_request!())
    .connect_with(transport, async |cx| {
        // Initialize the protocol
        let init_response = cx.send_request(InitializeRequest::make())
            .block_task()
            .await?;

        // Send more requests...
        let result = cx.send_request(MyRequest {})
            .block_task()
            .await?;

        // When this closure returns, the connection shuts down
        Ok(())
    })
    .await?;

Parameters

• main_fn: Your client logic. Receives a ConnectionTo<R> for sending messages.

Errors

Returns an error if a handler, background task, transport, or close callback fails, or if main_fn returns an error. Clean incoming EOF is observable through ConnectionTo::incoming_closed and is not itself an error in this mode.

Trait Implementations

impl<R, H, Run, Close> ConnectTo<<R as Role>::Counterpart> for Builder<R, H, Run, Close>

where R: Role, H: HandleDispatchFrom<R::Counterpart> + 'static, Run: RunWithConnectionTo<R::Counterpart> + 'static, Close: HandleConnectionClose<R::Counterpart> + 'static,

async fn connect_to(self, client: impl ConnectTo<R>) -> Result<(), Error>

Connect this component to another component.

fn into_channel_and_future(self) -> (Channel, BoxFuture<'static, Result<()>>)

where Self: Sized,

Convert this component into a channel endpoint and connection future.

impl<Host: Debug + Role, Handler, Runner, Close> Debug for Builder<Host, Handler, Runner, Close>

where Handler: HandleDispatchFrom<Host::Counterpart> + Debug, Runner: RunWithConnectionTo<Host::Counterpart> + Debug, Close: HandleConnectionClose<Host::Counterpart> + Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.