Struct Builder 

pub struct Builder<Host: Role, Handler = NullHandler, Runner = NullRun>
where
    Handler: HandleDispatchFrom<Host::Counterpart>,
    Runner: RunWithConnectionTo<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

For enums containing both requests AND notifications, use on_receive_dispatch:

// on_receive_dispatch receives Dispatch which can be either a request or notification
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.respond_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 enums containing both requests and notifications
• 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!())
    .on_receive_dispatch(async |msg: Dispatch, cx| {
        // This runs for any message not handled above
        msg.respond_with_error(agent_client_protocol::util::internal_error("unknown method"), cx)
    }, agent_client_protocol::on_receive_dispatch!())

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 = ByteStreams::new(
    tokio::io::stdout().compat_write(),
    tokio::io::stdin().compat(),
);

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>

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>

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>> Builder<Host, Handler, Runner>

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>>, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, impl RunWithConnectionTo<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>

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_responder<Run1>( self, responder: Run1, ) -> Builder<Host, Handler, impl RunWithConnectionTo<Host::Counterpart>>

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

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_receive_dispatch<Req, Notif, F, T, ToFut>( self, op: F, to_future_hack: ToFut, ) -> Builder<Host, impl HandleDispatchFrom<Host::Counterpart>, Runner>

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 messages that can be either requests OR notifications.

Use this when you want to handle an enum type that contains both request and notification variants. Your handler receives a Dispatch<Req, Notif> which is an enum with two 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.respond_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>

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 two arguments:

1. The request (type Req)
2. A Responder<R, Req::Response> for sending the response

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

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

    // Do some work...
    let result = todo!("process the prompt");

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

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>

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>

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>

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::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>

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

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

Add an MCP server that will be added to all new sessions that are proxied through this connection.

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
• One of your handlers returns an error

The transport is responsible for serializing and deserializing jsonrpcmsg::Message values to/from the underlying I/O mechanism (byte streams, channels, etc.).

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 = ByteStreams::new(
    tokio::io::stdout().compat_write(),
    tokio::io::stdin().compat(),
);

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 gracefully. If the connection closes unexpectedly before main_fn completes, this returns an error.

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 = ByteStreams::new(
    tokio::io::stdout().compat_write(),
    tokio::io::stdin().compat(),
);

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 the connection closes before main_fn completes.

Trait Implementations

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

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

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

Serve this component by forwarding to a client component.

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

where Self: Sized,

Convert this component into a channel endpoint and server future.

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

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

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

Formats the value using the given formatter.