use std::sync::Arc;
use crate::domain::{
NegotiationError, SpecError, SupportedVersions, SupportedVersionsError, TransportKind,
TransportSpec,
};
use super::handler::ConnectOptions;
use super::{Client, ClientError, ClientTransport, HandshakeError};
#[cfg(any(feature = "http", feature = "legacy-sse"))]
use super::{ClientConfig, handshake};
#[derive(Debug, thiserror::Error)]
pub enum ConnectError {
#[error("this build has no client adapter for `{kind}`: {reason}")]
Unsupported {
kind: TransportKind,
reason: &'static str,
},
#[error("the transport spec is not usable: {0}")]
Spec(#[from] SpecError),
#[error("the offered revisions are not usable: {0}")]
Config(#[from] ClientError),
#[error("this build's own revision table is unusable: {0}")]
Versions(#[from] SupportedVersionsError),
#[error("the handshake did not complete: {0}")]
Handshake(#[from] HandshakeError),
#[error("the client and the server share no revision: {0}")]
NoCommonVersion(#[from] NegotiationError),
#[cfg(feature = "http")]
#[error("the Streamable HTTP endpoint could not be prepared: {0}")]
HttpEndpoint(#[from] crate::adapters::secondary::transport::http::HttpClientError),
#[cfg(feature = "legacy-sse")]
#[error("the legacy HTTP+SSE endpoint could not be opened: {0}")]
LegacySse(#[from] crate::adapters::secondary::transport::legacy_sse::LegacySseError),
}
#[cfg(any(feature = "http", feature = "legacy-sse"))]
pub async fn open(
spec: &TransportSpec,
versions: &SupportedVersions,
) -> Result<(Client, Arc<dyn ClientTransport>), ConnectError> {
open_with(spec, versions, ConnectOptions::new()).await
}
#[cfg(any(feature = "http", feature = "legacy-sse"))]
pub(super) async fn open_with(
spec: &TransportSpec,
versions: &SupportedVersions,
options: ConnectOptions,
) -> Result<(Client, Arc<dyn ClientTransport>), ConnectError> {
let capabilities = options.capabilities;
#[cfg(feature = "http")]
let listener: Option<Arc<crate::adapters::secondary::transport::http::HttpClient>>;
let (transport, kind): (Arc<dyn ClientTransport>, TransportKind) = match spec.kind() {
#[cfg(feature = "http")]
TransportKind::StreamableHttp => {
use crate::adapters::secondary::transport::http::{HttpClient, HttpClientConfig};
let config = HttpClientConfig::new(spec.target())?;
let mut client = HttpClient::new(config)?;
if let Some(handler) = options.handler.clone() {
client = client.answering(handler);
}
if let Some(auth) = options.auth.clone() {
client = client.authorizing(auth);
}
let client = Arc::new(client);
listener = options.handler.is_some().then(|| Arc::clone(&client));
(client, TransportKind::StreamableHttp)
}
#[cfg(feature = "http")]
TransportKind::StatelessHttp => {
return open_stateless(spec, versions, options).await;
}
#[cfg(feature = "legacy-sse")]
TransportKind::LegacySse => {
use crate::adapters::secondary::transport::legacy_sse::{
LegacySseClient, LegacySseConfig,
};
let config = LegacySseConfig::new(spec.target())?;
let client = LegacySseClient::connect(config).await?;
listener = None;
(Arc::new(client), TransportKind::LegacySse)
}
kind => {
return Err(ConnectError::Unsupported {
kind,
reason: missing_reason(kind),
});
}
};
let mut offered: Vec<crate::domain::ProtocolVersion> = versions
.as_slice()
.iter()
.copied()
.filter(|version| crate::domain::handshake_defined_in(*version))
.collect();
offered.sort_by(|left, right| right.cmp(left));
if offered.is_empty() {
return Err(ConnectError::Unsupported {
kind,
reason: "none of the offered revisions has an `initialize` handshake; \
use an `http-stateless:` spec for 2026-07-28",
});
}
let config = ClientConfig::speaking(offered, kind)?.with_capabilities(capabilities);
let session = handshake(transport.as_ref(), &config).await?;
#[cfg(feature = "http")]
if let Some(listener) = &listener {
listener.spawn_listener();
}
Ok((
Client::from_session(&session, Arc::clone(&transport)),
transport,
))
}
#[cfg(not(any(feature = "http", feature = "legacy-sse")))]
pub async fn open(
spec: &TransportSpec,
versions: &SupportedVersions,
) -> Result<(Client, Arc<dyn ClientTransport>), ConnectError> {
open_with(spec, versions, ConnectOptions::new()).await
}
#[cfg(not(any(feature = "http", feature = "legacy-sse")))]
pub(super) async fn open_with(
spec: &TransportSpec,
versions: &SupportedVersions,
options: ConnectOptions,
) -> Result<(Client, Arc<dyn ClientTransport>), ConnectError> {
let _ = (versions, options);
Err(ConnectError::Unsupported {
kind: spec.kind(),
reason: missing_reason(spec.kind()),
})
}
#[cfg(feature = "http")]
async fn open_stateless(
spec: &TransportSpec,
versions: &SupportedVersions,
options: ConnectOptions,
) -> Result<(Client, Arc<dyn ClientTransport>), ConnectError> {
use crate::adapters::secondary::transport::http::{HttpClientConfig, StatelessHttpClient};
let capabilities = options.capabilities;
let mut stateless: Vec<_> = versions
.as_slice()
.iter()
.copied()
.filter(|version| !crate::domain::handshake_defined_in(*version))
.collect();
stateless.sort_by(|left, right| right.cmp(left));
if stateless.is_empty() {
return Err(ConnectError::Unsupported {
kind: TransportKind::StatelessHttp,
reason: "none of the offered revisions has a stateless lifecycle",
});
}
let mut attempts: Vec<(crate::domain::ProtocolVersion, u8)> =
stateless.iter().map(|version| (*version, 0)).collect();
let mut candidate = stateless.first().copied();
loop {
let Some(current) = candidate else {
return Err(ConnectError::Unsupported {
kind: TransportKind::StatelessHttp,
reason: "the server shares no stateless revision with this client",
});
};
if let Some((_, tried)) = attempts.iter_mut().find(|(version, _)| *version == current) {
*tried = tried.saturating_add(1);
}
let config = ClientConfig::speaking(vec![current], TransportKind::StatelessHttp)?
.with_capabilities(capabilities);
let capabilities_json =
serde_json::Value::Object(super::handshake::client_capabilities(&config));
let mut client = StatelessHttpClient::new(
HttpClientConfig::new(spec.target())?,
current,
capabilities_json,
)?;
if let Some(auth) = options.auth.clone() {
client = client.authorizing(auth);
}
let transport: Arc<dyn ClientTransport> = Arc::new(client);
match handshake(transport.as_ref(), &config).await {
Ok(session) => {
return Ok((
Client::from_session(&session, Arc::clone(&transport)),
transport,
));
}
Err(HandshakeError::Transport(refused)) if refused.supported_versions().is_some() => {
let supported = refused.supported_versions().unwrap_or_default();
candidate = attempts
.iter()
.find(|(version, tried)| {
*tried < 2 && supported.iter().any(|name| *name == version.as_wire())
})
.map(|(version, _)| *version);
}
Err(other) => return Err(other.into()),
}
}
}
const fn missing_reason(kind: TransportKind) -> &'static str {
match kind {
TransportKind::Stdio => {
"a stdio upstream needs a program to run, and a spec carries no \
command line"
}
TransportKind::StatelessHttp => {
"the stateless HTTP client is compiled in with the `http` feature"
}
_ => "no client adapter for this transport is compiled in",
}
}
pub fn default_versions() -> Result<SupportedVersions, SupportedVersionsError> {
SupportedVersions::all()
}
#[cfg(test)]
mod tests {
use crate::domain::{ProtocolVersion, TransportKind, TransportSpec};
use super::{ConnectError, default_versions, open};
#[derive(Debug)]
enum Failure {
Expectation(&'static str),
Connect(ConnectError),
}
impl core::fmt::Display for Failure {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Expectation(what) => write!(formatter, "expected {what}"),
Self::Connect(error) => write!(formatter, "connecting failed: {error}"),
}
}
}
impl std::error::Error for Failure {}
impl From<ConnectError> for Failure {
fn from(error: ConnectError) -> Self {
Self::Connect(error)
}
}
impl From<crate::domain::SupportedVersionsError> for Failure {
fn from(error: crate::domain::SupportedVersionsError) -> Self {
Self::Expectation(match error {
crate::domain::SupportedVersionsError::Empty => "a non-empty default",
})
}
}
fn spec(kind: TransportKind) -> Result<TransportSpec, Failure> {
TransportSpec::new(kind, "http://127.0.0.1:1/mcp")
.map_err(|_| Failure::Expectation("a usable spec"))
}
#[cfg(not(feature = "http"))]
#[tokio::test]
async fn a_kind_with_no_adapter_is_refused_by_name() -> Result<(), Failure> {
let refused = open(&spec(TransportKind::StatelessHttp)?, &default_versions()?).await;
match refused {
Err(ConnectError::Unsupported { kind, reason }) => {
if kind != TransportKind::StatelessHttp {
return Err(Failure::Expectation("the refused kind reported back"));
}
if reason.is_empty() {
return Err(Failure::Expectation("a reason that says what to do"));
}
Ok(())
}
Err(other) => Err(Failure::Connect(other)),
Ok(_) => Err(Failure::Expectation("a refusal rather than an adapter")),
}
}
#[cfg(feature = "http")]
#[tokio::test]
async fn a_stateless_spec_with_no_stateless_revision_is_refused() -> Result<(), Failure> {
let versions = crate::domain::SupportedVersions::from_slice(&[
crate::domain::ProtocolVersion::V2025_11_25,
])?;
match open(&spec(TransportKind::StatelessHttp)?, &versions).await {
Err(ConnectError::Unsupported {
kind: TransportKind::StatelessHttp,
..
}) => Ok(()),
Err(other) => Err(Failure::Connect(other)),
Ok(_) => Err(Failure::Expectation("a refusal")),
}
}
#[tokio::test]
async fn a_stdio_spec_is_refused_for_wanting_a_command_line() -> Result<(), Failure> {
match open(&spec(TransportKind::Stdio)?, &default_versions()?).await {
Err(ConnectError::Unsupported { reason, .. }) => {
if !reason.contains("command") {
return Err(Failure::Expectation(
"a reason naming the command line that is missing",
));
}
Ok(())
}
Err(other) => Err(Failure::Connect(other)),
Ok(_) => Err(Failure::Expectation("a refusal rather than a session")),
}
}
#[test]
fn the_default_offers_every_known_revision() -> Result<(), Failure> {
let offered = default_versions()?;
if offered.len() != ProtocolVersion::KNOWN.len() {
return Err(Failure::Expectation(
"the default to offer every revision with wire types",
));
}
for known in ProtocolVersion::KNOWN {
if !offered.contains(&known) {
return Err(Failure::Expectation("every known revision to be offered"));
}
}
Ok(())
}
}