protocol_droid/adapters/secondary/transport/
connect.rs

1//! Connecting to a server that has not been told which transport it speaks.
2//!
3//! [`detect`] answers "which transport is this"; this answers "give me a client
4//! to talk to it with", which is the question a caller actually has. It is the
5//! two joined, plus the handshake that turns a transport into a session.
6//!
7//! ## The probe is the handshake, on one path and not the other
8//!
9//! `detect` needs a request to find out, and the request it sends is an
10//! `initialize`. If the peer answers, that request *was* the handshake: its
11//! result carries the negotiated version and the server's capabilities, and
12//! sending `initialize` a second time would open a second session on a server
13//! that is counting them.
14//!
15//! If the peer answers `4xx`, the probe was never processed — it was a request
16//! to a path that does not take requests — and the legacy client that comes
17//! back has not been sent anything at all. So that path runs the full
18//! the handshake over the connected stream.
19//!
20//! Both paths end in the same place. What differs between them is not the
21//! negotiation but whether the request already happened.
22
23use std::sync::Arc;
24
25use crate::adapters::secondary::transport::detect::{DetectConfig, DetectError, Detected, detect};
26use crate::adapters::secondary::transport::http::HttpClientError;
27use crate::application::client::{
28    Client, ClientConfig, ClientTransport, HandshakeError, INITIALIZED, initialize_params,
29    initialize_session, offer,
30};
31use crate::domain::NegotiatedSession;
32use crate::domain::ir::JsonObject;
33
34/// Why a connection could not be established.
35#[derive(Debug, thiserror::Error)]
36pub enum ConnectError {
37    /// The client was built offering no revision, so there was nothing to
38    /// negotiate.
39    #[error("this client was offered no revision, so there is nothing to negotiate")]
40    NothingOffered,
41
42    /// The probe could not be sent, or neither transport answered.
43    #[error(transparent)]
44    Detect(#[from] DetectError),
45
46    /// The handshake failed.
47    ///
48    /// Distinct from [`Detect`](Self::Detect) because the peer *did* answer: a
49    /// caller who collapsed the two would be sent to check their URL when the
50    /// real problem is a revision the two sides do not share.
51    #[error(transparent)]
52    Handshake(#[from] HandshakeError),
53
54    /// The probe's answer was not an object, so it could not be read as a
55    /// handshake result.
56    #[error("the server answered the probe with something that is not an object")]
57    UnreadableProbe,
58
59    /// The URL could not be turned into a client.
60    #[error(transparent)]
61    Http(#[from] HttpClientError),
62
63    /// The closing notification could not be delivered.
64    ///
65    /// Its own variant rather than folded into [`Handshake`](Self::Handshake)
66    /// because it happens *after* a session exists. A caller told the handshake
67    /// failed would go looking for a version mismatch, and the version
68    /// matched fine — the notification is what did not get through.
69    #[error("the session was negotiated but the client could not announce it: {0}")]
70    NotAnnounced(#[from] crate::application::client::TransportError),
71}
72
73/// What a connection ended up speaking, once the transport is known.
74///
75/// Separate from [`Client`] because the transport is the part a caller cannot
76/// get back out of one, and it is the part worth logging: "connected" does not
77/// say whether the peer is running the revision this build prefers or the one
78/// it fell back to.
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum Spoke {
81    /// The peer speaks Streamable HTTP.
82    StreamableHttp,
83    /// The peer speaks the `2024-11-05` transport.
84    LegacyHttpSse,
85}
86
87impl Spoke {
88    /// The name a log line or a JSON report wants.
89    #[must_use]
90    pub const fn as_str(self) -> &'static str {
91        match self {
92            Self::StreamableHttp => "streamable-http",
93            Self::LegacyHttpSse => "legacy-http-sse",
94        }
95    }
96}
97
98/// Connects to `url` and negotiates a session, whichever transport answers.
99///
100/// # Errors
101///
102/// [`ConnectError::Detect`] if neither transport answered,
103/// [`ConnectError::Handshake`] if the peer's answer could not be negotiated into
104/// a session, and [`ConnectError::NothingOffered`] for a client configured with
105/// no revision.
106///
107/// # Examples
108///
109/// ```no_run
110/// use protocol_droid::adapters::secondary::transport::connect::{Spoke, connect};
111/// use protocol_droid::application::client::ClientConfig;
112/// use protocol_droid::domain::{ProtocolVersion, TransportKind};
113///
114/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
115/// let config = ClientConfig::new(ProtocolVersion::V2025_06_18, TransportKind::StreamableHttp);
116/// let (client, spoke) = connect("https://example.test/mcp", &config).await?;
117/// let tools = client.list_tools().await?;
118/// assert!(matches!(spoke, Spoke::StreamableHttp | Spoke::LegacyHttpSse));
119/// # Ok(())
120/// # }
121/// ```
122pub async fn connect(url: &str, config: &ClientConfig) -> Result<(Client, Spoke), ConnectError> {
123    let (preferred, supported) = offer(config).ok_or(ConnectError::NothingOffered)?;
124
125    // The probe *is* the `initialize` request, so it is built by the same code
126    // a handshake would use. A probe hand-built here would be a second
127    // definition of what a client's offer looks like.
128    let probe = initialize_params(preferred, config)?;
129    let detected = detect(DetectConfig::new(url)?, probe.as_str().as_bytes()).await?;
130
131    match detected {
132        Detected::Streamable { client, result } => {
133            // The probe already asked, so this reads that answer rather than
134            // asking a second time.
135            let answer = JsonObject::new(String::from_utf8_lossy(&result).into_owned())
136                .map_err(|_| ConnectError::UnreadableProbe)?;
137            let session = initialize_session(&answer, preferred, &supported, config)?;
138            let client = finish(client, session).await?;
139            Ok((client, Spoke::StreamableHttp))
140        }
141        Detected::Legacy(client) => {
142            // The probe was refused, so nothing has been sent over the stream
143            // yet and the whole handshake still has to happen.
144            let session = crate::application::client::handshake(&client, config).await?;
145            let client = finish(client, session).await?;
146            Ok((client, Spoke::LegacyHttpSse))
147        }
148    }
149}
150
151/// Sends the closing notification and binds the transport to the session.
152///
153/// The notification goes out on both paths and is not optional: a server that
154/// has not been told the client is ready will not answer anything else, so a
155/// client that skipped it would look connected and then hang on its first call.
156async fn finish<T>(transport: T, session: NegotiatedSession) -> Result<Client, ConnectError>
157where
158    T: ClientTransport + 'static,
159{
160    let params = JsonObject::new("{}").map_err(|error| HandshakeError::Assembling {
161        method: INITIALIZED,
162        source: error,
163    })?;
164    transport.notify(INITIALIZED, &params).await?;
165    Ok(Client::from_session(&session, Arc::new(transport)))
166}
167
168#[cfg(all(test, feature = "http"))]
169#[path = "connect/tests.rs"]
170mod tests;
protocol_droid/adapters/secondary/transport/connect.rs

protocol_droid/adapters/secondary/transport/
connect.rs
