protocol_droid/adapters/secondary/transport/http/
stateless.rs

1//! The client half of the `2026-07-28` stateless lifecycle.
2//!
3//! SEP-2575 removed the handshake and the session. Every request now carries
4//! its revision and the client's capabilities in `params._meta`, and on
5//! Streamable HTTP the transport mirrors part of the body into headers so an
6//! intermediary can route without parsing it:
7//!
8//! - `MCP-Protocol-Version`, equal to `_meta["io.modelcontextprotocol/protocolVersion"]`;
9//! - `Mcp-Method`, equal to the JSON-RPC `method`;
10//! - `Mcp-Name`, equal to `params.name` (`tools/call`, `prompts/get`) or
11//!   `params.uri` (`resources/read`);
12//! - `Mcp-Param-{Name}`, for each tool argument the tool's schema annotates
13//!   with `x-mcp-header` (SEP-2243).
14//!
15//! All of it is a property of one body, so it is built per request here
16//! rather than configured once on the connection. Source:
17//! <https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#request-metadata>
18//!
19//! Learning the `x-mcp-header` annotations needs the tool's schema, so this
20//! client keeps the schemas from the last `tools/list` it relayed. The
21//! specification also has a client drop a tool whose annotations are invalid
22//! from that listing, which happens on the same pass.
23
24use core::future::Future;
25use core::pin::Pin;
26use std::collections::HashMap;
27use std::sync::{Arc, Mutex, PoisonError};
28
29use serde_json::{Map, Value};
30
31use super::client::{HttpClient, HttpClientConfig};
32use super::error::HttpClientError;
33use crate::application::client::bearer::BearerAuth;
34use crate::application::client::{ClientTransport, TransportError};
35use crate::application::jsonrpc::document::{
36    RequestError, notification_document, reply_result, request_document,
37};
38use crate::application::mcp::param_headers;
39use crate::application::mcp::request_meta::{
40    CLIENT_CAPABILITIES_KEY, CLIENT_INFO_KEY, PROTOCOL_VERSION_KEY,
41};
42use crate::domain::ProtocolVersion;
43use crate::domain::ir::{JsonObject, MethodName};
44
45/// A Streamable HTTP client that speaks the stateless lifecycle.
46#[derive(Debug)]
47pub struct StatelessHttpClient {
48    inner: HttpClient,
49    version: ProtocolVersion,
50    capabilities: Value,
51    client_info: Value,
52    /// Each tool's `inputSchema`, from the last `tools/list` relayed.
53    schemas: Mutex<HashMap<String, Value>>,
54}
55
56impl StatelessHttpClient {
57    /// A client speaking `version`, declaring `capabilities` on every request.
58    ///
59    /// `capabilities` is a `ClientCapabilities` object. It is sent on every
60    /// request because the server may not remember it: "Servers MUST NOT infer
61    /// capabilities from prior requests".
62    ///
63    /// # Errors
64    ///
65    /// [`HttpClientError`] if the HTTP client cannot be built.
66    pub fn new(
67        config: HttpClientConfig,
68        version: ProtocolVersion,
69        capabilities: Value,
70    ) -> Result<Self, HttpClientError> {
71        Ok(Self {
72            inner: HttpClient::new(config)?,
73            version,
74            capabilities,
75            client_info: serde_json::json!({
76                "name": "protocol-droid",
77                "version": env!("CARGO_PKG_VERSION"),
78            }),
79            schemas: Mutex::new(HashMap::new()),
80        })
81    }
82
83    /// This client, authenticating through `auth`.
84    #[must_use]
85    pub fn authorizing(mut self, auth: Arc<dyn BearerAuth>) -> Self {
86        self.inner = self.inner.authorizing(auth);
87        self
88    }
89
90    /// The revision every request is sent at.
91    #[must_use]
92    pub const fn version(&self) -> ProtocolVersion {
93        self.version
94    }
95
96    /// `params` with this client's `_meta` keys added, keeping any the caller
97    /// set itself (a progress token, a log level).
98    fn with_meta(&self, params: &JsonObject) -> Value {
99        let mut object: Map<String, Value> = match serde_json::from_str(params.as_str()) {
100            Ok(Value::Object(object)) => object,
101            _ => Map::new(),
102        };
103        let meta = object
104            .entry("_meta")
105            .or_insert_with(|| Value::Object(Map::new()));
106        if let Some(meta) = meta.as_object_mut() {
107            meta.insert(
108                PROTOCOL_VERSION_KEY.to_owned(),
109                Value::from(self.version.as_wire().into_owned()),
110            );
111            meta.entry(CLIENT_CAPABILITIES_KEY)
112                .or_insert_with(|| self.capabilities.clone());
113            meta.entry(CLIENT_INFO_KEY)
114                .or_insert_with(|| self.client_info.clone());
115        }
116        Value::Object(object)
117    }
118
119    /// The request-metadata headers for one body.
120    fn headers(&self, method: &str, params: &Value) -> Vec<(String, String)> {
121        let mut headers = vec![
122            (
123                "MCP-Protocol-Version".to_owned(),
124                self.version.as_wire().into_owned(),
125            ),
126            ("Mcp-Method".to_owned(), method.to_owned()),
127        ];
128        let named = match method {
129            "tools/call" | "prompts/get" => params.get("name"),
130            "resources/read" => params.get("uri"),
131            _ => None,
132        };
133        if let Some(name) = named.and_then(Value::as_str) {
134            headers.push(("Mcp-Name".to_owned(), param_headers::encode(name)));
135        }
136        if method == "tools/call"
137            && let Some(tool) = params.get("name").and_then(Value::as_str)
138        {
139            let schema = self
140                .schemas
141                .lock()
142                .unwrap_or_else(PoisonError::into_inner)
143                .get(tool)
144                .cloned();
145            if let Some(schema) = schema {
146                let arguments = params.get("arguments").cloned().unwrap_or(Value::Null);
147                headers.extend(param_headers::headers_for(&schema, &arguments));
148            }
149        }
150        headers
151    }
152
153    /// Remembers each listed tool's schema and drops the tools whose
154    /// `x-mcp-header` annotations are invalid.
155    fn learn_tools(&self, result: JsonObject) -> JsonObject {
156        let Ok(Value::Object(mut members)) = serde_json::from_str::<Value>(result.as_str()) else {
157            return result;
158        };
159        let Some(Value::Array(tools)) = members.get_mut("tools") else {
160            return result;
161        };
162        let mut schemas = self.schemas.lock().unwrap_or_else(PoisonError::into_inner);
163        tools.retain(|tool| {
164            let schema = tool.get("inputSchema").cloned().unwrap_or(Value::Null);
165            if param_headers::has_invalid_annotation(&schema) {
166                return false;
167            }
168            if let Some(name) = tool.get("name").and_then(Value::as_str) {
169                schemas.insert(name.to_owned(), schema);
170            }
171            true
172        });
173        drop(schemas);
174        JsonObject::from_value(Value::Object(members)).unwrap_or(result)
175    }
176}
177
178fn unavailable(error: impl core::fmt::Display) -> TransportError {
179    TransportError::Unavailable(error.to_string())
180}
181
182/// Reads an HTTP refusal whose body is a JSON-RPC error as the error it is.
183///
184/// A `2026-07-28` server answers `-32022`, `-32021`, `-32020` and `-32601`
185/// with a `4xx` status and the error in the body. The HTTP client reports that
186/// as a refused `POST`; a caller wants the code and the `data`.
187fn refusal(error: HttpClientError) -> TransportError {
188    if let HttpClientError::Refused { preview, .. } = &error
189        && let Ok(Value::Object(body)) = serde_json::from_str::<Value>(preview)
190        && let Some(Value::Object(failure)) = body.get("error")
191        && let Some(code) = failure.get("code").and_then(Value::as_i64)
192    {
193        return TransportError::Refused {
194            code,
195            message: failure
196                .get("message")
197                .and_then(Value::as_str)
198                .unwrap_or_default()
199                .to_owned(),
200            data: failure.get("data").map(Value::to_string),
201        };
202    }
203    unavailable(error)
204}
205
206impl ClientTransport for StatelessHttpClient {
207    /// POSTs one self-describing request and returns its `result`.
208    ///
209    /// An `InputRequiredResult` is a result like any other and is returned as
210    /// one: answering the input requests it carries is the caller's decision —
211    /// it may need a person — and a transport that answered on the caller's
212    /// behalf would be inventing consent.
213    ///
214    /// # Errors
215    ///
216    /// [`TransportError::Unavailable`] for a frame that cannot be assembled, a
217    /// refused `POST`, or a reply carrying an `error`.
218    fn send(
219        &self,
220        method: &str,
221        params: &JsonObject,
222    ) -> Pin<Box<dyn Future<Output = Result<JsonObject, TransportError>> + Send + '_>> {
223        let prepared = (|| {
224            let id = self.inner.next_request_id();
225            let name = MethodName::new(method).map_err(unavailable)?;
226            let params = self.with_meta(params);
227            let headers = self.headers(method, &params);
228            let params = JsonObject::from_value(params).map_err(unavailable)?;
229            let document = request_document(&id, &name, &params).map_err(unavailable)?;
230            let envelope = document
231                .as_object()
232                .cloned()
233                .ok_or_else(|| unavailable("the request is not a JSON object"))?;
234            Ok::<_, TransportError>((id, envelope, headers))
235        })();
236        let listing = method == "tools/list";
237
238        Box::pin(async move {
239            let (id, envelope, headers) = prepared?;
240            let reply = self
241                .inner
242                .post_envelope_with(&envelope, &headers)
243                .await
244                .map_err(refusal)?
245                .ok_or_else(|| unavailable("a request was answered with no body at all"))?;
246            let text = serde_json::to_string(&Value::Object(reply)).map_err(unavailable)?;
247            let result = reply_result(&text, &id).map_err(|error| match error {
248                RequestError::PeerError {
249                    code,
250                    message,
251                    data,
252                } => TransportError::Refused {
253                    code: code.code(),
254                    message,
255                    data,
256                },
257                other => unavailable(other),
258            })?;
259            Ok(if listing {
260                self.learn_tools(result)
261            } else {
262                result
263            })
264        })
265    }
266
267    /// POSTs a notification. The core protocol defines none a client sends
268    /// over this transport at this revision, but an extension may.
269    ///
270    /// # Errors
271    ///
272    /// [`TransportError::Unavailable`] if the frame cannot be assembled or the
273    /// `POST` is refused.
274    fn notify(
275        &self,
276        method: &str,
277        params: &JsonObject,
278    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + '_>> {
279        let prepared = (|| {
280            let name = MethodName::new(method).map_err(unavailable)?;
281            let params = self.with_meta(params);
282            let headers = self.headers(method, &params);
283            let params = JsonObject::from_value(params).map_err(unavailable)?;
284            let document = notification_document(&name, &params).map_err(unavailable)?;
285            let envelope = document
286                .as_object()
287                .cloned()
288                .ok_or_else(|| unavailable("the notification is not a JSON object"))?;
289            Ok::<_, TransportError>((envelope, headers))
290        })();
291        Box::pin(async move {
292            let (envelope, headers) = prepared?;
293            self.inner
294                .post_envelope_with(&envelope, &headers)
295                .await
296                .map(|_| ())
297                .map_err(unavailable)
298        })
299    }
300}
301
302#[cfg(test)]
303mod tests;
protocol_droid/adapters/secondary/transport/http/stateless.rs

protocol_droid/adapters/secondary/transport/http/
stateless.rs
