use std::sync::Arc;
use crate::domain::TransportSpec;
use crate::domain::feature::{Feature, FeatureProfile, Negotiation};
use crate::domain::ir::JsonObject;
use crate::domain::{
CapabilitySet, NegotiatedSession, ProtocolVersion, SupportedVersions, TransportKind,
};
pub mod bearer;
mod connect;
pub mod handler;
mod handshake;
pub mod inputs;
pub use connect::{ConnectError, default_versions, open};
pub use handshake::{
DISCOVER, HandshakeError, INITIALIZE, INITIALIZED, discover_session, handshake,
initialize_params, initialize_session, offer,
};
pub trait ClientTransport: Send + Sync {
fn send(
&self,
method: &str,
params: &JsonObject,
) -> std::pin::Pin<Box<dyn Future<Output = Result<JsonObject, TransportError>> + Send + '_>>;
fn notify(
&self,
method: &str,
params: &JsonObject,
) -> std::pin::Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + '_>>;
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum TransportError {
#[error("{0}")]
Unavailable(String),
#[error("the peer refused with {code}: {message}")]
Refused {
code: i64,
message: String,
data: Option<String>,
},
}
impl TransportError {
#[must_use]
pub fn supported_versions(&self) -> Option<Vec<String>> {
let Self::Refused {
code: -32022,
data: Some(data),
..
} = self
else {
return None;
};
let parsed: serde_json::Value = serde_json::from_str(data).ok()?;
Some(
parsed
.get("supported")?
.as_array()?
.iter()
.filter_map(|version| version.as_str().map(ToOwned::to_owned))
.collect(),
)
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ClientError {
#[error("{feature} was not negotiated (session is {negotiated})")]
FeatureUnavailable {
feature: Feature,
negotiated: ProtocolVersion,
},
#[error("the request did not complete: {0}")]
Transport(#[from] TransportError),
#[error("the server asked for `{method}` input, which no resolver provided")]
InputUnresolved {
method: String,
},
#[error("the server asked for input more than {0} times")]
TooManyRounds(u32),
}
impl ClientError {
#[must_use]
pub const fn feature(&self) -> Option<Feature> {
match self {
Self::FeatureUnavailable { feature, .. } => Some(*feature),
Self::Transport(_) | Self::InputUnresolved { .. } | Self::TooManyRounds(_) => None,
}
}
}
#[cfg(feature = "stdio")]
#[derive(Debug, thiserror::Error)]
pub enum StdioConnectError {
#[error("this build's own revision table is unusable: {0}")]
Versions(#[from] crate::domain::SupportedVersionsError),
#[error("the transport config is not usable: {0}")]
Config(#[from] ClientError),
#[error("the child could not be spawned: {0}")]
Spawn(#[from] crate::adapters::secondary::stdio::SpawnError),
#[error("the handshake did not complete: {0}")]
Handshake(#[from] HandshakeError),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientConfig {
versions: Vec<ProtocolVersion>,
capabilities: CapabilitySet,
transport: TransportKind,
}
impl ClientConfig {
#[must_use]
pub fn new(version: ProtocolVersion, transport: TransportKind) -> Self {
Self {
versions: vec![version],
capabilities: CapabilitySet::empty(),
transport,
}
}
pub fn speaking(
versions: impl IntoIterator<Item = ProtocolVersion>,
transport: TransportKind,
) -> Result<Self, ClientError> {
let versions: Vec<ProtocolVersion> = versions.into_iter().collect();
if versions.is_empty() {
return Err(ClientError::Transport(TransportError::Unavailable(
"a client must offer at least one revision".to_string(),
)));
}
Ok(Self {
versions,
capabilities: CapabilitySet::empty(),
transport,
})
}
#[must_use]
pub const fn with_capabilities(mut self, capabilities: CapabilitySet) -> Self {
self.capabilities = capabilities;
self
}
#[must_use]
pub fn versions(&self) -> &[ProtocolVersion] {
&self.versions
}
#[must_use]
pub const fn capabilities(&self) -> CapabilitySet {
self.capabilities
}
#[must_use]
pub const fn transport(&self) -> TransportKind {
self.transport
}
#[must_use]
pub const fn negotiated_with(
&self,
version: ProtocolVersion,
server_caps: CapabilitySet,
) -> Negotiation {
Negotiation {
version,
client_caps: self.capabilities,
server_caps,
transport: self.transport,
}
}
}
#[derive(Clone)]
pub struct Client {
session: Arc<NegotiatedSession>,
transport: Arc<dyn ClientTransport>,
}
impl core::fmt::Debug for Client {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Client")
.field("negotiated", &self.session.version)
.field("transport", &self.session.transport)
.field("feature_count", &self.session.profile.len())
.finish_non_exhaustive()
}
}
impl Client {
#[must_use]
pub fn connect(negotiation: Negotiation, transport: Arc<dyn ClientTransport>) -> Self {
Self {
session: Arc::new(NegotiatedSession::new(
negotiation.version,
negotiation.transport,
negotiation.client_caps,
negotiation.server_caps,
)),
transport,
}
}
#[must_use]
pub fn from_session(session: &NegotiatedSession, transport: Arc<dyn ClientTransport>) -> Self {
Self {
session: Arc::new(session.clone()),
transport,
}
}
pub async fn connect_to(spec: &TransportSpec) -> Result<Self, ConnectError> {
Ok(connect::open(spec, &default_versions()?).await?.0)
}
pub async fn connect_to_speaking(
spec: &TransportSpec,
versions: &SupportedVersions,
) -> Result<Self, ConnectError> {
Ok(connect::open(spec, versions).await?.0)
}
pub async fn connect_with(
spec: &TransportSpec,
versions: &SupportedVersions,
options: handler::ConnectOptions,
) -> Result<Self, ConnectError> {
Ok(connect::open_with(spec, versions, options).await?.0)
}
#[cfg(feature = "stdio")]
pub async fn connect_stdio(
spec: &crate::adapters::secondary::stdio::ChildSpec,
) -> Result<Self, StdioConnectError> {
use crate::adapters::primary::stdio::Diagnostics;
use crate::adapters::secondary::stdio::StdioClient;
let diagnostics = Diagnostics::stderr();
let transport_config = crate::adapters::secondary::stdio::ClientConfig::default();
let transport = StdioClient::spawn(spec, transport_config, diagnostics)
.await
.map_err(StdioConnectError::Spawn)?;
let transport: Arc<dyn ClientTransport> = Arc::new(transport);
let versions = default_versions().map_err(StdioConnectError::Versions)?;
let config = ClientConfig::speaking(versions.as_slice().to_vec(), TransportKind::Stdio)
.map_err(StdioConnectError::Config)?;
let session = handshake(transport.as_ref(), &config)
.await
.map_err(StdioConnectError::Handshake)?;
Ok(Self::from_session(&session, transport))
}
#[must_use]
pub fn negotiated(&self) -> ProtocolVersion {
self.session.version
}
#[must_use]
pub fn profile(&self) -> FeatureProfile {
self.session.profile
}
#[must_use]
pub fn session(&self) -> &NegotiatedSession {
&self.session
}
#[must_use]
pub fn supports(&self, feature: Feature) -> bool {
self.session.profile.allows(feature)
}
pub async fn send_unchecked(
&self,
method: &str,
params: &JsonObject,
) -> Result<JsonObject, ClientError> {
Ok(self.transport.send(method, params).await?)
}
pub async fn list_tools(&self) -> Result<JsonObject, ClientError> {
self.require(Feature::Tools)?;
self.request("tools/list", "{}").await
}
pub async fn call_tool(
&self,
name: &str,
arguments: &JsonObject,
) -> Result<JsonObject, ClientError> {
self.require(Feature::Tools)?;
let nested = serde_json::from_str::<serde_json::Value>(arguments.as_str())
.unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::new()));
let params = JsonObject::from_value(serde_json::json!({
"name": name,
"arguments": nested,
}))
.unwrap_or_default();
self.transport
.send("tools/call", ¶ms)
.await
.map_err(ClientError::from)
}
pub async fn call_tool_resolving(
&self,
name: &str,
arguments: &JsonObject,
resolve: &(impl inputs::InputResolver + ?Sized),
) -> Result<JsonObject, ClientError> {
self.require(Feature::Tools)?;
let nested = serde_json::from_str::<serde_json::Value>(arguments.as_str())
.unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::new()));
let params =
JsonObject::from_value(serde_json::json!({ "name": name, "arguments": nested }))
.unwrap_or_default();
inputs::call_resolving(self.transport.as_ref(), "tools/call", ¶ms, resolve).await
}
pub async fn list_resources(&self) -> Result<JsonObject, ClientError> {
self.require(Feature::Resources)?;
self.request("resources/list", "{}").await
}
pub async fn list_prompts(&self) -> Result<JsonObject, ClientError> {
self.require(Feature::Prompts)?;
self.request("prompts/list", "{}").await
}
pub async fn list_resource_templates(&self) -> Result<JsonObject, ClientError> {
self.require(Feature::Resources)?;
self.request("resources/templates/list", "{}").await
}
pub async fn read_resource(&self, uri: &str) -> Result<JsonObject, ClientError> {
self.require(Feature::Resources)?;
let params = JsonObject::from_value(serde_json::json!({ "uri": uri })).unwrap_or_default();
Ok(self.transport.send("resources/read", ¶ms).await?)
}
pub async fn get_prompt(
&self,
name: &str,
arguments: &JsonObject,
) -> Result<JsonObject, ClientError> {
self.require(Feature::Prompts)?;
let nested = serde_json::from_str::<serde_json::Value>(arguments.as_str())
.unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::new()));
let params =
JsonObject::from_value(serde_json::json!({ "name": name, "arguments": nested }))
.unwrap_or_default();
Ok(self.transport.send("prompts/get", ¶ms).await?)
}
fn require(&self, feature: Feature) -> Result<(), ClientError> {
if self.session.profile.allows(feature) {
return Ok(());
}
Err(ClientError::FeatureUnavailable {
feature,
negotiated: self.session.version,
})
}
async fn request(&self, method: &str, params: &str) -> Result<JsonObject, ClientError> {
let owned = match JsonObject::new(params) {
Ok(object) => object,
Err(error) => {
return Err(ClientError::Transport(TransportError::Unavailable(
format!("{method} was given parameters that are not a JSON object: {error}"),
)));
}
};
self.transport
.send(method, &owned)
.await
.map_err(ClientError::from)
}
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use super::{Client, ClientConfig, ClientError, ClientTransport, TransportError};
use crate::domain::feature::Feature;
use crate::domain::ir::{JsonObject, JsonObjectError};
use crate::domain::{CapabilitySet, ProtocolVersion, TransportKind};
#[derive(Debug)]
enum Failure {
Message(String),
Fixture(JsonObjectError),
}
impl core::fmt::Display for Failure {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Message(text) => f.write_str(text),
Self::Fixture(error) => write!(f, "bad fixture: {error}"),
}
}
}
impl std::error::Error for Failure {}
impl From<String> for Failure {
fn from(text: String) -> Self {
Self::Message(text)
}
}
impl From<&str> for Failure {
fn from(text: &str) -> Self {
Self::Message(text.to_string())
}
}
impl From<JsonObjectError> for Failure {
fn from(error: JsonObjectError) -> Self {
Self::Fixture(error)
}
}
type TestError = Failure;
struct Recorder {
calls: Mutex<Vec<String>>,
answer: Option<JsonObject>,
}
impl Recorder {
fn answering(answer: &str) -> Result<Self, TestError> {
Ok(Self {
calls: Mutex::new(Vec::new()),
answer: Some(JsonObject::new(answer)?),
})
}
fn dead() -> Self {
Self {
calls: Mutex::new(Vec::new()),
answer: None,
}
}
fn called(&self) -> Vec<String> {
let calls = self
.calls
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
calls.clone()
}
}
impl ClientTransport for Recorder {
fn send(
&self,
method: &str,
_params: &JsonObject,
) -> std::pin::Pin<Box<dyn Future<Output = Result<JsonObject, TransportError>> + Send + '_>>
{
{
let mut calls = self
.calls
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
calls.push(method.to_string());
}
let outcome = self
.answer
.clone()
.ok_or_else(|| TransportError::Unavailable("nothing listening".to_string()));
Box::pin(std::future::ready(outcome))
}
fn notify(
&self,
method: &str,
_params: &JsonObject,
) -> std::pin::Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + '_>>
{
self.called();
{
let mut calls = self
.calls
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
calls.push(method.to_string());
}
Box::pin(std::future::ready(Ok(())))
}
}
fn client_at(version: ProtocolVersion, transport: Arc<Recorder>) -> Client {
let caps = CapabilitySet::all_for(version);
let config = ClientConfig::new(version, TransportKind::Stdio);
Client::connect(config.negotiated_with(version, caps), transport)
}
#[tokio::test]
async fn a_gated_call_on_a_session_without_it_never_reaches_the_transport()
-> Result<(), TestError> {
let transport = Arc::new(Recorder::dead());
let version = ProtocolVersion::V2024_11_05;
let config = ClientConfig::new(version, TransportKind::Stdio);
let client = Client::connect(
config.negotiated_with(version, CapabilitySet::empty()),
transport.clone(),
);
let refusal = match client.list_prompts().await {
Ok(answer) => {
return Err(Failure::from(format!("expected a refusal, got {answer}")));
}
Err(error) => error,
};
assert_eq!(
refusal,
ClientError::FeatureUnavailable {
feature: Feature::Prompts,
negotiated: ProtocolVersion::V2024_11_05,
}
);
assert_eq!(refusal.feature(), Some(Feature::Prompts));
assert!(transport.called().is_empty());
Ok(())
}
#[tokio::test]
async fn a_permitted_call_reaches_the_transport() -> Result<(), TestError> {
let version = ProtocolVersion::V2025_06_18;
let transport = Arc::new(Recorder::answering(r#"{"tools":[]}"#)?);
let client = client_at(version, transport.clone());
let answer = match client.list_tools().await {
Ok(answer) => answer,
Err(error) => return Err(Failure::from(error.to_string())),
};
assert_eq!(answer.as_str(), r#"{"tools":[]}"#);
assert_eq!(transport.called(), vec!["tools/list".to_string()]);
Ok(())
}
#[tokio::test]
async fn each_typed_method_sends_its_own_method() -> Result<(), TestError> {
let version = ProtocolVersion::V2025_06_18;
let transport = Arc::new(Recorder::answering("{}")?);
let client = client_at(version, transport.clone());
for outcome in [
client.list_tools().await.map(|_| ()),
client.list_resources().await.map(|_| ()),
client.list_prompts().await.map(|_| ()),
] {
if let Err(error) = outcome {
return Err(Failure::from(error.to_string()));
}
}
assert_eq!(
transport.called(),
vec![
"tools/list".to_string(),
"resources/list".to_string(),
"prompts/list".to_string(),
]
);
Ok(())
}
#[test]
fn a_client_offering_no_revisions_is_refused() -> Result<(), TestError> {
let empty: Vec<ProtocolVersion> = Vec::new();
let config = ClientConfig::speaking(empty, TransportKind::Stdio);
match config {
Ok(_) => Err(Failure::from("a client offering no revisions was accepted")),
Err(_) => Ok(()),
}
}
#[test]
fn the_client_reports_the_negotiations_profile() {
let version = ProtocolVersion::V2025_06_18;
let transport = Arc::new(Recorder::dead());
let config = ClientConfig::new(version, TransportKind::Stdio);
let generous = Client::connect(
config.negotiated_with(version, CapabilitySet::all_for(version)),
transport.clone(),
);
let stingy = Client::connect(
config.negotiated_with(version, CapabilitySet::empty()),
transport,
);
assert_eq!(generous.negotiated(), version);
assert!(
generous.supports(Feature::Resources),
"both sides declared everything {version} defines"
);
assert!(
!stingy.supports(Feature::Resources),
"a server that declared nothing offers no resources"
);
}
#[test]
fn a_client_from_a_session_keeps_the_sessions_profile() {
let version = ProtocolVersion::V2025_06_18;
let session = crate::domain::NegotiatedSession::new(
version,
TransportKind::Stdio,
CapabilitySet::all_for(version),
CapabilitySet::all_for(version),
);
let expected = session.profile;
let client = Client::from_session(&session, Arc::new(Recorder::dead()));
assert_eq!(client.negotiated(), version);
assert_eq!(
client.profile(),
expected,
"the client's profile must be the session's, not a second derivation of it"
);
assert!(client.supports(Feature::Resources));
}
#[test]
fn a_client_reports_what_the_handshake_agreed() {
let version = ProtocolVersion::V2025_06_18;
let caps = CapabilitySet::all_for(version);
let session =
crate::domain::NegotiatedSession::new(version, TransportKind::LegacySse, caps, caps)
.with_client_info(crate::domain::PeerInfo::new("droid-cli", "0.4.1"))
.with_server_info(crate::domain::PeerInfo::new("weather", "2.11.0"));
let client = Client::from_session(&session, Arc::new(Recorder::dead()));
let seen = client.session();
assert_eq!(seen.version, version);
assert_eq!(
seen.transport,
TransportKind::LegacySse,
"the transport comes off the session, not off the URL the caller passed"
);
assert_eq!(seen.client_caps, caps);
assert_eq!(seen.server_caps, caps);
assert_eq!(
seen.server_info.as_ref().map(|i| i.name.as_str()),
Some("weather"),
"a peer that announced itself must still be able to say so"
);
assert_eq!(
seen.client_info.as_ref().map(|i| i.version.as_str()),
Some("0.4.1"),
"our own identity is as reportable as the server's"
);
assert!(
!seen.is_stateless,
"a 2025-06-18 session retains state; only 2026-07-28 does not"
);
assert_eq!(client.profile(), session.profile);
}
#[test]
fn a_client_built_by_hand_keeps_the_transport_it_was_given() {
let version = ProtocolVersion::V2025_03_26;
let caps = CapabilitySet::all_for(version);
let client = Client::connect(
crate::domain::Negotiation {
version,
client_caps: caps,
server_caps: caps,
transport: TransportKind::Stdio,
},
Arc::new(Recorder::dead()),
);
assert_eq!(client.session().transport, TransportKind::Stdio);
assert_eq!(client.session().client_caps, caps);
assert_eq!(
client.session().server_info,
None,
"a bare Negotiation carries no peer names, and inventing them is worse than none"
);
}
#[cfg(feature = "stdio")]
#[test]
fn stdio_connect_error_names_the_stage() {
use super::StdioConnectError;
let err = StdioConnectError::Versions(crate::domain::SupportedVersionsError::Empty);
let rendered = format!("{err}");
assert!(
rendered.contains("revision table"),
"Versions variant must name the stage: {rendered}"
);
}
}