Skip to main content

protocol_droid/adapters/secondary/mcp/
client.rs

1//! `McpStdioPeer` — the client-side transport for MCP.
2//!
3//! A `McpStdioPeer` connects to an MCP
4//! server over stdio. The wire form is
5//! newline-delimited JSON-RPC 2.0: the
6//! peer writes requests to the child's
7//! stdin and reads replies and
8//! notifications from the child's
9//! stdout. The peer owns the child
10//! process and the reader / writer
11//! tasks; dropping the peer kills the
12//! child and stops the tasks.
13//!
14//! # Why this peer exists
15//!
16//! An ACP agent is usually also an MCP
17//! client: the editor passes an
18//! `mcpServers` list to the agent in
19//! the `session/new` request, and the
20//! agent spawns one MCP client per
21//! server. The agent's tool calls
22//! then route to whichever MCP server
23//! advertises the tool. Without a
24//! typed MCP client, the agent's
25//! `session/new` handler is on the
26//! hook for hand-rolling the JSON-RPC
27//! envelope, the id allocation, and
28//! the reply dispatch — the same
29//! boilerplate every MCP client
30//! writes. `McpStdioPeer` is that
31//! boilerplate, typed, tested, and
32//! ready to spawn.
33//!
34//! # What the peer speaks
35//!
36//! The pinned MCP revision is the
37//! newest of the supported revisions
38//! the consumer's `version` enum
39//! allows. The peer's `methods`
40//! surface is the four most-used
41//! client-originating methods:
42//! `initialize`, `ping`, `tools/list`,
43//! `tools/call`. A future increment
44//! adds `resources/*` and
45//! `prompts/*`.
46//!
47//! # Differences from `AcpStdioPeer`
48//!
49//! The wire form is identical: the
50//! peer speaks newline-delimited
51//! JSON-RPC 2.0 in both directions.
52//! The differences are the
53//! protocol-level shapes (the typed
54//! request / response types) and
55//! the method table. The transport
56//! is the same shape; the
57//! application-level concerns differ.
58//! The duplication is deliberate: the
59//! two protocols may diverge in the
60//! future (e.g. MCP's Streamable
61//! HTTP) and a shared transport
62//! would couple their evolution.
63
64use std::collections::HashMap;
65use std::io::Write as _;
66use std::process::Stdio;
67use std::sync::Arc;
68use std::sync::atomic::{AtomicU64, Ordering};
69use std::time::Duration;
70
71use serde::Serialize;
72use serde::de::DeserializeOwned;
73use serde_json::Value;
74use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
75use tokio::process::{Child, ChildStdin, ChildStdout, Command};
76use tokio::sync::{Mutex, mpsc, oneshot};
77
78/// The MCP wire form's
79/// `initialize` request. Sent by
80/// the client as its first call.
81///
82/// Source of truth:
83/// `crates/protocol-droid-schema/spec/<pinned-revision>/schema.json`
84/// under `definitions.InitializeRequest`'s
85/// `params` shape (the wrapper is
86/// the JSON-RPC envelope; the typed
87/// shape is the `params` value).
88#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
89#[serde(rename_all = "camelCase")]
90pub struct InitializeRequest {
91    /// The MCP revision the client
92    /// supports. The schema names
93    /// every published revision; the
94    /// peer uses the revision the
95    /// consumer selected at
96    /// construction.
97    pub protocol_version: String,
98    /// The client's capabilities. The
99    /// schema defaults this to an
100    /// empty object; the typed
101    /// default is the all-off
102    /// capabilities block.
103    #[serde(default)]
104    pub capabilities: ClientCapabilities,
105    /// Information about the
106    /// client. The schema marks
107    /// this required in newer
108    /// revisions; the wire form
109    /// always carries it.
110    pub client_info: Implementation,
111}
112
113/// The MCP wire form's
114/// `initialize` reply. Sent by
115/// the server in response to
116/// `InitializeRequest`.
117#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
118#[serde(rename_all = "camelCase")]
119pub struct InitializeResult {
120    /// The MCP revision the server
121    /// supports. The negotiation
122    /// rule is the server picks
123    /// the client's revision if
124    /// it supports it; otherwise
125    /// the server's latest
126    /// supported revision.
127    pub protocol_version: String,
128    /// The server's capabilities.
129    pub capabilities: ServerCapabilities,
130    /// Information about the
131    /// server. The schema marks
132    /// this required.
133    pub server_info: Implementation,
134    /// Optional instructions the
135    /// server hands the client for
136    /// how to use the server. A
137    /// typed `None` is the default
138    /// for servers that do not
139    /// provide instructions.
140    #[serde(default)]
141    pub instructions: Option<String>,
142}
143
144/// The client's capabilities.
145///
146/// The schema's `ClientCapabilities`
147/// is an open object; the typed shape
148/// carries the canonical
149/// sub-capabilities (`roots`,
150/// `sampling`, `experimental`).
151/// Sub-capabilities the consumer
152/// does not enable default to
153/// `None`, which the wire form
154/// omits.
155#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
156pub struct ClientCapabilities {
157    /// The `roots` capability
158    /// (whether the client
159    /// advertises filesystem
160    /// roots to the server).
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub roots: Option<RootsCapability>,
163    /// The `sampling` capability
164    /// (whether the client
165    /// supports LLM sampling
166    /// on the server's behalf).
167    #[serde(skip_serializing_if = "Option::is_none")]
168    pub sampling: Option<SamplingCapability>,
169    /// Experimental
170    /// capabilities. The schema
171    /// does not constrain the
172    /// shape; the typed default
173    /// is an empty map.
174    #[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty", default)]
175    pub experimental: std::collections::BTreeMap<String, Value>,
176}
177
178/// The `roots` sub-capability.
179/// Whether the client advertises
180/// filesystem roots to the server
181/// and whether the list can change
182/// at runtime.
183#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
184#[serde(rename_all = "camelCase")]
185pub struct RootsCapability {
186    /// Whether the list of roots
187    /// can change after the
188    /// `initialize` round-trip.
189    #[serde(default)]
190    pub list_changed: bool,
191}
192
193/// The `sampling` sub-capability.
194/// A marker for the client's
195/// support of LLM sampling. The
196/// schema's `SamplingCapability`
197/// is an empty object.
198#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
199pub struct SamplingCapability {}
200
201/// The server's capabilities.
202///
203/// The schema's `ServerCapabilities`
204/// is an open object; the typed
205/// shape carries the canonical
206/// sub-capabilities
207/// (`prompts`, `resources`,
208/// `tools`, `logging`).
209/// Sub-capabilities the server
210/// does not advertise default to
211/// `None`, which the wire form
212/// omits.
213#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize)]
214pub struct ServerCapabilities {
215    /// The `prompts` capability.
216    #[serde(default)]
217    pub prompts: Option<PromptsCapability>,
218    /// The `resources` capability.
219    #[serde(default)]
220    pub resources: Option<ResourcesCapability>,
221    /// The `tools` capability.
222    #[serde(default)]
223    pub tools: Option<ToolsCapability>,
224    /// The `logging` capability.
225    #[serde(default)]
226    pub logging: Option<LoggingCapability>,
227    /// Experimental
228    /// capabilities. The schema
229    /// does not constrain the
230    /// shape; the typed default
231    /// is an empty map.
232    #[serde(default)]
233    pub experimental: std::collections::BTreeMap<String, Value>,
234}
235
236/// The `prompts` sub-capability.
237#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize)]
238#[serde(rename_all = "camelCase")]
239pub struct PromptsCapability {
240    /// Whether the list of
241    /// prompts can change at
242    /// runtime.
243    #[serde(default)]
244    pub list_changed: bool,
245}
246
247/// The `resources` sub-capability.
248#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize)]
249#[serde(rename_all = "camelCase")]
250pub struct ResourcesCapability {
251    /// Whether the list of
252    /// resources can change at
253    /// runtime.
254    #[serde(default)]
255    pub list_changed: bool,
256    /// Whether the client can
257    /// subscribe to individual
258    /// resource updates.
259    #[serde(default)]
260    pub subscribe: bool,
261}
262
263/// The `tools` sub-capability.
264#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize)]
265#[serde(rename_all = "camelCase")]
266pub struct ToolsCapability {
267    /// Whether the list of
268    /// tools can change at
269    /// runtime.
270    #[serde(default)]
271    pub list_changed: bool,
272}
273
274/// The `logging` sub-capability.
275/// A marker.
276#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize)]
277pub struct LoggingCapability {}
278
279/// The `Implementation` block on
280/// either side of `initialize`.
281/// The shape is identical on the
282/// client and server side.
283#[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)]
284#[serde(rename_all = "camelCase")]
285pub struct Implementation {
286    /// The implementation's
287    /// programmatic name.
288    pub name: String,
289    /// The implementation's
290    /// human-readable title.
291    #[serde(default, skip_serializing_if = "Option::is_none")]
292    pub title: Option<String>,
293    /// The implementation's
294    /// version string.
295    pub version: String,
296}
297
298/// The `tools/list` reply. The
299/// server returns the list of
300/// tools it advertises.
301#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
302#[serde(rename_all = "camelCase")]
303pub struct ListToolsResult {
304    /// The tools the server
305    /// advertises. The schema
306    /// requires this list.
307    pub tools: Vec<Tool>,
308    /// An optional pagination
309    /// cursor. The schema marks
310    /// this as the standard
311    /// JSON-RPC pagination shape;
312    /// the typed default is
313    /// `None` (no more pages).
314    #[serde(default)]
315    pub next_cursor: Option<String>,
316}
317
318/// A single tool. The server
319/// returns these from
320/// `tools/list` and the client
321/// invokes them via `tools/call`.
322#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
323#[serde(rename_all = "camelCase")]
324pub struct Tool {
325    /// The tool's programmatic
326    /// name. The schema requires
327    /// this and the server
328    /// guarantees it is unique
329    /// within the server's
330    /// advertised tool set.
331    pub name: String,
332    /// The tool's human-readable
333    /// description. Optional;
334    /// servers SHOULD provide
335    /// this for tool catalogs
336    /// shown to humans.
337    #[serde(default)]
338    pub description: Option<String>,
339    /// The JSON Schema for the
340    /// tool's input. The schema
341    /// requires this; the typed
342    /// shape carries it as a
343    /// raw `Value` because the
344    /// JSON Schema is
345    /// tool-specific.
346    pub input_schema: Value,
347}
348
349/// The `tools/call` request body.
350/// The client invokes a tool by
351/// name with a JSON object that
352/// matches the tool's
353/// `inputSchema`.
354#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
355#[serde(rename_all = "camelCase")]
356pub struct CallToolRequest {
357    /// The tool's name. Matches
358    /// the `name` field of a
359    /// `Tool` returned from
360    /// `tools/list`.
361    pub name: String,
362    /// The tool's arguments.
363    /// The shape is the tool's
364    /// `inputSchema`. A tool that
365    /// takes no arguments passes
366    /// an empty object.
367    #[serde(default)]
368    pub arguments: Value,
369}
370
371/// The `tools/call` reply. The
372/// server returns a list of
373/// content blocks (text, image,
374/// audio, or embedded resources)
375/// and an optional `is_error`
376/// flag.
377#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
378#[serde(rename_all = "camelCase")]
379pub struct CallToolResult {
380    /// The tool's result
381    /// content. The schema
382    /// requires at least one
383    /// block.
384    pub content: Vec<ContentBlock>,
385    /// Whether the tool
386    /// call failed. The
387    /// schema marks this
388    /// optional; the typed
389    /// default is `false`
390    /// (success).
391    #[serde(default)]
392    pub is_error: bool,
393}
394
395/// A single content block in a
396/// `CallToolResult`. The schema
397/// defines `text`, `image`,
398/// `audio`, and `resource`
399/// variants via a tagged union.
400/// The typed shape models the
401/// four shapes the schema names.
402#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
403#[serde(tag = "type", rename_all = "snake_case")]
404pub enum ContentBlock {
405    /// A text block.
406    Text {
407        /// The text content.
408        text: String,
409    },
410    /// An image block. The
411    /// `data` is base64-encoded.
412    Image {
413        /// The image's
414        /// MIME type.
415        #[serde(rename = "mimeType")]
416        mime_type: String,
417        /// The base64-encoded
418        /// image data.
419        data: String,
420    },
421    /// An audio block. The
422    /// `data` is base64-encoded.
423    Audio {
424        /// The audio's
425        /// MIME type.
426        #[serde(rename = "mimeType")]
427        mime_type: String,
428        /// The base64-encoded
429        /// audio data.
430        data: String,
431    },
432    /// An embedded resource
433    /// block. The resource is
434    /// inlined by URI.
435    Resource {
436        /// The embedded
437        /// resource's URI.
438        uri: String,
439        /// The embedded
440        /// resource's
441        /// MIME type.
442        #[serde(rename = "mimeType", default)]
443        mime_type: Option<String>,
444        /// The embedded
445        /// resource's text
446        /// content. The
447        /// schema does not
448        /// require this;
449        /// the typed
450        /// default is
451        /// `None`.
452        #[serde(default)]
453        text: Option<String>,
454    },
455}
456
457/// A pending reply. The reader
458/// task owns a clone of the
459/// `Arc<Mutex<HashMap>>`; the
460/// application layer owns the
461/// other clone. The reply is a
462/// oneshot: the reader task
463/// hands the result to the
464/// application once and the
465/// `HashMap` entry is removed.
466struct Pending {
467    tx: oneshot::Sender<Result<Value, McpPeerError>>,
468}
469
470/// The default request timeout.
471const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
472
473/// The MCP stdio peer.
474pub struct McpStdioPeer {
475    pending: Arc<Mutex<HashMap<u64, Pending>>>,
476    next_id: AtomicU64,
477    write_tx: mpsc::Sender<Vec<u8>>,
478    timeout: Duration,
479    /// The child process. Held
480    /// for `kill_on_drop`: when
481    /// the peer drops, the
482    /// `Child` is dropped, which
483    /// kills the child. The
484    /// application layer never
485    /// reads the field directly.
486    #[expect(dead_code, reason = "kill_on_drop machinery; never read directly")]
487    child: Child,
488}
489
490impl std::fmt::Debug for McpStdioPeer {
491    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
492        formatter
493            .debug_struct("McpStdioPeer")
494            .field("pending_len", &"*")
495            .field("next_id", &self.next_id)
496            .field("write_tx", &"*")
497            .field("timeout", &self.timeout)
498            .field("child", &"*")
499            .finish_non_exhaustive()
500    }
501}
502
503/// The specification for the
504/// child process. Mirrors
505/// `AcpStdioPeer::SpawnSpec` so
506/// the call sites that already
507/// know that shape can swap.
508#[derive(Debug, Clone)]
509pub struct SpawnSpec {
510    /// The executable to spawn.
511    pub command: String,
512    /// The arguments to pass.
513    pub args: Vec<String>,
514    /// The default request timeout.
515    pub timeout: Option<Duration>,
516}
517
518impl SpawnSpec {
519    /// Construct a new `SpawnSpec`
520    /// for a command and its
521    /// arguments. The timeout
522    /// defaults to 30 seconds; the
523    /// consumer overrides it with
524    /// [`with_timeout`][Self::with_timeout].
525    pub fn new(command: impl Into<String>, args: Vec<String>) -> Self {
526        Self {
527            command: command.into(),
528            args,
529            timeout: None,
530        }
531    }
532
533    /// Set the default request timeout.
534    #[must_use]
535    pub const fn with_timeout(mut self, timeout: Duration) -> Self {
536        self.timeout = Some(timeout);
537        self
538    }
539}
540
541/// The error type the peer
542/// returns. The variants are
543/// the failure modes the
544/// application layer can act on.
545#[derive(Debug, thiserror::Error)]
546pub enum McpPeerError {
547    /// The child process could
548    /// not be spawned. The
549    /// underlying `io::Error`
550    /// is the `Command::spawn`
551    /// failure (typically
552    /// `NotFound` for a missing
553    /// executable).
554    #[error("spawning the child process: {0}")]
555    Spawn(#[source] std::io::Error),
556
557    /// The JSON-RPC envelope
558    /// could not be serialised
559    /// or the reply could not be
560    /// deserialised into the
561    /// typed shape the
562    /// application requested.
563    /// The error is the
564    /// `serde_json` failure.
565    #[error("encoding or decoding the JSON-RPC envelope: {0}")]
566    Decode(#[source] serde_json::Error),
567
568    /// The reply never arrived
569    /// within the configured
570    /// timeout. The pending
571    /// oneshot is dropped, and
572    /// the reader task will see
573    /// a closed sender when (if)
574    /// the late reply does
575    /// come.
576    #[error("the reply did not arrive within {0:?}")]
577    Timeout(Duration),
578
579    /// The writer task is gone;
580    /// the child has exited and
581    /// the peer's pipe is
582    /// closed. A subsequent
583    /// `send_request` returns
584    /// this variant.
585    #[error("the writer task is gone")]
586    Write(#[source] std::io::Error),
587
588    /// The reader task is gone;
589    /// the child has exited and
590    /// the peer's pipe is
591    /// closed. A subsequent
592    /// `send_typed_request`
593    /// returns this variant.
594    #[error("the reader task is gone")]
595    ReadGone,
596
597    /// The reply was a
598    /// JSON-RPC 2.0 error. The
599    /// error code and message
600    /// are carried through.
601    #[error("the server returned a JSON-RPC error: code={code} message={message}")]
602    JsonRpc {
603        /// The JSON-RPC error
604        /// code.
605        code: i64,
606        /// The JSON-RPC error
607        /// message.
608        message: String,
609    },
610}
611
612impl McpStdioPeer {
613    /// Spawn a child process and
614    /// wire it up to the peer's
615    /// reader and writer tasks.
616    /// The peer owns the child
617    /// process; dropping the
618    /// peer kills the child
619    /// (the `Command` is
620    /// configured with
621    /// `kill_on_drop(true)`).
622    pub fn spawn(spec: &SpawnSpec) -> Result<Self, McpPeerError> {
623        let mut command = Command::new(&spec.command);
624        command
625            .args(&spec.args)
626            .stdin(Stdio::piped())
627            .stdout(Stdio::piped())
628            .stderr(Stdio::null())
629            .kill_on_drop(true);
630        let mut child = command.spawn().map_err(McpPeerError::Spawn)?;
631        let stdin = child.stdin.take().ok_or_else(|| {
632            McpPeerError::Spawn(std::io::Error::other("child stdin was not piped"))
633        })?;
634        let stdout = child.stdout.take().ok_or_else(|| {
635            McpPeerError::Spawn(std::io::Error::other("child stdout was not piped"))
636        })?;
637
638        let pending: Arc<Mutex<HashMap<u64, Pending>>> = Arc::new(Mutex::new(HashMap::new()));
639        let (write_tx, write_rx) = mpsc::channel::<Vec<u8>>(32);
640        spawn_reader(BufReader::new(stdout), Arc::clone(&pending));
641        spawn_writer(stdin, write_rx);
642
643        Ok(Self {
644            pending,
645            next_id: AtomicU64::new(1),
646            write_tx,
647            timeout: spec.timeout.unwrap_or(DEFAULT_TIMEOUT),
648            child,
649        })
650    }
651
652    /// Send a typed request and
653    /// decode the typed reply.
654    /// The two type parameters
655    /// are the consumer's call
656    /// to the server: `Req` is
657    /// what the consumer's
658    /// handler sends, `Resp` is
659    /// what the server replies
660    /// with.
661    ///
662    /// `params` is taken by
663    /// value (not by reference)
664    /// so the future the
665    /// function returns is
666    /// unconditionally `Send`:
667    /// a borrow of `Req` would
668    /// only be `Send` if
669    /// `Req: Sync`, which is a
670    /// sharper constraint than
671    /// the consumer wants.
672    pub async fn send_typed_request<Req, Resp>(
673        &self,
674        method: &str,
675        params: Req,
676    ) -> Result<Resp, McpPeerError>
677    where
678        Req: Serialize,
679        Resp: DeserializeOwned,
680    {
681        let value = self.send_request(method, params).await?;
682        serde_json::from_value(value).map_err(McpPeerError::Decode)
683    }
684
685    /// Send a request and await
686    /// the raw reply. The raw
687    /// `Value` is the consumer's
688    /// problem to decode; the
689    /// typed wrapper is
690    /// [`send_typed_request`][Self::send_typed_request].
691    pub async fn send_request<Req>(&self, method: &str, params: Req) -> Result<Value, McpPeerError>
692    where
693        Req: Serialize,
694    {
695        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
696        let params_value: Value = serde_json::to_value(&params).map_err(McpPeerError::Decode)?;
697        let envelope = serde_json::json!({
698            "jsonrpc": "2.0",
699            "id": id,
700            "method": method,
701            "params": params_value,
702        });
703        let bytes = serde_json::to_vec(&envelope).map_err(McpPeerError::Decode)?;
704        let mut framed = bytes;
705        framed.push(b'\n');
706        self.write_tx
707            .send(framed)
708            .await
709            .map_err(|_| McpPeerError::Write(io_other("the writer task is gone")))?;
710        let (tx, rx) = oneshot::channel();
711        self.pending.lock().await.insert(id, Pending { tx });
712        match tokio::time::timeout(self.timeout, rx).await {
713            Ok(Ok(result)) => result,
714            Ok(Err(_oneshot_dropped)) => {
715                self.pending.lock().await.remove(&id);
716                Err(McpPeerError::Timeout(self.timeout))
717            }
718            Err(_elapsed) => {
719                self.pending.lock().await.remove(&id);
720                Err(McpPeerError::Timeout(self.timeout))
721            }
722        }
723    }
724}
725
726fn io_other(message: &str) -> std::io::Error {
727    std::io::Error::other(message)
728}
729
730fn spawn_writer(mut stdin: ChildStdin, mut write_rx: mpsc::Receiver<Vec<u8>>) {
731    tokio::spawn(async move {
732        while let Some(mut bytes) = write_rx.recv().await {
733            // Belt-and-braces: ensure
734            // a trailing newline even
735            // if a future caller forgot
736            // to add one.
737            if !bytes.ends_with(b"\n") {
738                bytes.push(b'\n');
739            }
740            if let Err(error) = stdin.write_all(&bytes).await {
741                // The child has gone.
742                // Log to stderr; the
743                // application layer
744                // discovers the failure
745                // through the next
746                // `send_request` returning
747                // a `Write` error.
748                let _ = std::io::stderr()
749                    .lock()
750                    .write_all(format!("mcp writer task: write failed: {error}\n").as_bytes());
751                return;
752            }
753            if let Err(error) = stdin.flush().await {
754                let _ = std::io::stderr()
755                    .lock()
756                    .write_all(format!("mcp writer task: flush failed: {error}\n").as_bytes());
757                return;
758            }
759        }
760    });
761}
762
763fn spawn_reader(mut reader: BufReader<ChildStdout>, pending: Arc<Mutex<HashMap<u64, Pending>>>) {
764    tokio::spawn(async move {
765        let mut line = String::new();
766        loop {
767            line.clear();
768            match reader.read_line(&mut line).await {
769                Ok(0) | Err(_) => return,
770                Ok(_) => {}
771            }
772            let trimmed = line.trim_end_matches('\n');
773            if trimmed.is_empty() {
774                continue;
775            }
776            let value: Value = match serde_json::from_str(trimmed) {
777                Ok(value) => value,
778                Err(_) => continue,
779            };
780            let id = value.get("id").and_then(Value::as_u64).unwrap_or(0);
781            if id == 0 {
782                // Notification or
783                // malformed reply; the
784                // MCP client peer does
785                // not yet model
786                // notifications. A
787                // future increment
788                // adds the
789                // `notifications/*`
790                // stream (e.g.
791                // `notifications/tools/list_changed`).
792                continue;
793            }
794            let result = value.get("error").map_or_else(
795                || {
796                    value
797                        .get("result")
798                        .cloned()
799                        .ok_or_else(|| McpPeerError::JsonRpc {
800                            code: -32600,
801                            message: "the reply had no `result` and no `error`".to_owned(),
802                        })
803                },
804                |error_value| {
805                    let code = error_value.get("code").and_then(Value::as_i64).unwrap_or(0);
806                    let message = error_value
807                        .get("message")
808                        .and_then(Value::as_str)
809                        .unwrap_or("")
810                        .to_owned();
811                    Err(McpPeerError::JsonRpc { code, message })
812                },
813            );
814            let entry = pending.lock().await.remove(&id);
815            if let Some(entry) = entry {
816                let _ = entry.tx.send(result);
817            }
818        }
819    });
820}
821
822/// Default client capabilities.
823///
824/// The default form is the
825/// all-off block: no
826/// `roots`, no `sampling`,
827/// no `experimental`. A
828/// peer that wants any of
829/// these on starts from
830/// this default and sets
831/// the relevant field.
832///
833/// The function is `pub`
834/// so the library itself
835/// names the type in
836/// non-test code, which
837/// keeps the rust 1.98
838/// `dead_code_pub_in_binary`
839/// lint from flagging
840/// `ClientCapabilities` as
841/// "never constructed" in
842/// the library's own build
843/// (the lint is promoted to
844/// library crates in 1.98
845/// and fires when a
846/// `pub` type has no
847/// construction site in
848/// the lib's normal build,
849/// which `#[cfg(test)]`
850/// items do not satisfy).
851#[must_use]
852pub fn default_client_capabilities() -> ClientCapabilities {
853    ClientCapabilities::default()
854}
855
856/// A reference `initialize`
857/// request.
858///
859/// A consumer that wants
860/// to send an
861/// `initialize` request
862/// needs the revision
863/// string and a
864/// `client_info` block.
865/// The function returns
866/// a `2025-06-18`
867/// request — the newest
868/// revision the vendored
869/// schema has, and the
870/// one the integration
871/// test fixture speaks.
872///
873/// Like
874/// [`default_client_capabilities`],
875/// this function exists to
876/// give the library's own
877/// non-test build a
878/// construction site for
879/// the typed shapes
880/// `InitializeRequest` and
881/// `Implementation` name
882/// (see that function's
883/// note for the lint
884/// rationale).
885#[must_use]
886pub fn initialize_request_template() -> InitializeRequest {
887    InitializeRequest {
888        protocol_version: "2025-06-18".to_owned(),
889        capabilities: ClientCapabilities::default(),
890        client_info: Implementation {
891            name: "protocol-droid".to_owned(),
892            title: Some("protocol-droid MCP client".to_owned()),
893            version: "0.1.0".to_owned(),
894        },
895    }
896}
897
898#[cfg(test)]
899mod typed_shape_round_trip {
900    //! Wire-format property tests
901    //! for the typed MCP shapes.
902    //!
903    //! A property test over the
904    //! typed shapes ensures the
905    //! wire form does not drift
906    //! from the consumer's
907    //! expectations: a
908    //! `serde(rename_all)`
909    //! typo, a missing `default`,
910    //! or a wrong field name is
911    //! caught by the round-trip
912    //! property test. The
913    //! test also drives the
914    //! typed shapes through
915    //! the lib's own types so
916    //! the binary crate's
917    //! `dead_code` lint does
918    //! not flag them.
919    use super::*;
920
921    /// Smoke test: a
922    /// `ClientCapabilities`
923    /// constructs in the
924    /// default form, and
925    /// serialises without
926    /// panicking.
927    #[test]
928    fn client_capabilities_default_serialises() -> Result<(), serde_json::Error> {
929        let caps = ClientCapabilities::default();
930        let value = serde_json::to_value(&caps)?;
931        if let Some(object) = value.as_object() {
932            assert!(
933                !object.contains_key("roots"),
934                "the default form omits `roots`"
935            );
936        } else {
937            assert!(
938                value.is_object(),
939                "the serialised value should be an object"
940            );
941        }
942        Ok(())
943    }
944
945    /// `ClientCapabilities` omits
946    /// `roots` when it is `None`.
947    /// The wire form is camelCase:
948    /// a future migration to
949    /// `snake_case` would fail
950    /// this test.
951    #[test]
952    fn client_capabilities_omits_roots_when_none() -> Result<(), serde_json::Error> {
953        let caps = ClientCapabilities {
954            roots: None,
955            sampling: None,
956            experimental: std::collections::BTreeMap::new(),
957        };
958        let value = serde_json::to_value(&caps)?;
959        if let Some(object) = value.as_object() {
960            assert!(!object.contains_key("roots"));
961        } else {
962            assert!(
963                value.is_object(),
964                "the serialised value should be an object"
965            );
966        }
967        Ok(())
968    }
969
970    /// `Implementation`
971    /// round-trips through
972    /// `serde_json` with `title`
973    /// omitted when `None`.
974    #[test]
975    fn implementation_round_trip() -> Result<(), serde_json::Error> {
976        let impl_value = Implementation {
977            name: "test-client".to_owned(),
978            title: None,
979            version: "0.0.1".to_owned(),
980        };
981        let value = serde_json::to_value(&impl_value)?;
982        if let Some(object) = value.as_object() {
983            assert!(!object.contains_key("title"));
984        } else {
985            assert!(
986                value.is_object(),
987                "the serialised value should be an object"
988            );
989        }
990        let decoded: Implementation = serde_json::from_value(value)?;
991        assert_eq!(impl_value, decoded);
992        Ok(())
993    }
994
995    /// `CallToolRequest` uses
996    /// the `name` field
997    /// directly and the
998    /// `arguments` field is
999    /// optional via
1000    /// `#[serde(default)]`.
1001    #[test]
1002    fn call_tool_request_serialises_name() -> Result<(), serde_json::Error> {
1003        let request = CallToolRequest {
1004            name: "echo".to_owned(),
1005            arguments: serde_json::json!({}),
1006        };
1007        let value = serde_json::to_value(&request)?;
1008        assert_eq!(
1009            value.get("name").and_then(serde_json::Value::as_str),
1010            Some("echo"),
1011        );
1012        Ok(())
1013    }
1014
1015    /// The wire form of
1016    /// `InitializeRequest` uses
1017    /// `camelCase` for
1018    /// `protocolVersion` (the
1019    /// schema's wire form).
1020    #[test]
1021    fn initialize_request_uses_camel_case_on_the_wire() -> Result<(), serde_json::Error> {
1022        let request = InitializeRequest {
1023            protocol_version: "2025-06-18".to_owned(),
1024            capabilities: ClientCapabilities::default(),
1025            client_info: Implementation {
1026                name: "test-client".to_owned(),
1027                title: None,
1028                version: "0.0.1".to_owned(),
1029            },
1030        };
1031        let value = serde_json::to_value(&request)?;
1032        if let Some(object) = value.as_object() {
1033            assert!(object.contains_key("protocolVersion"));
1034            assert!(!object.contains_key("protocol_version"));
1035        } else {
1036            assert!(
1037                value.is_object(),
1038                "the serialised value should be an object"
1039            );
1040        }
1041        Ok(())
1042    }
1043}