pmcp 2.15.0

High-quality Rust SDK for Model Context Protocol (MCP) with full TypeScript SDK compatibility
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

//! Shared in-process duplex transport + call helpers for dispatcher tests.
//!
//! Extracted from the (previously copy-pasted) harness in
//! `tests/tool_output_passthrough.rs` / `tests/tool_with_result.rs` et al.:
//! an mpsc-backed client<->server [`Transport`] pair, plus helpers that drive
//! a `tools/call` through a real [`Client`] against either a high-level
//! [`Server`] (own transport loop) or a `ServerCore` [`ProtocolHandler`]
//! (request/response pump).
//!
//! Each file in `tests/` is compiled as a separate integration crate; include
//! this module per-crate via `#[path = "common/duplex.rs"] mod duplex;`.

#![allow(dead_code)]
#![cfg(not(target_arch = "wasm32"))]

use std::sync::Arc;

use async_trait::async_trait;
use pmcp::server::core::ProtocolHandler;
use pmcp::shared::{Transport, TransportMessage};
use pmcp::types::{CallToolResult, ClientCapabilities};
use pmcp::{Client, Error, Result, Server};
use serde_json::Value;
use tokio::sync::mpsc;

/// One half of an in-process duplex transport. The client side sends Requests
/// and receives Responses; the server side does the reverse.
#[derive(Debug)]
pub struct DuplexTransport {
    tx: mpsc::UnboundedSender<TransportMessage>,
    rx: mpsc::UnboundedReceiver<TransportMessage>,
    connected: bool,
}

impl DuplexTransport {
    /// Create a connected client/server transport pair.
    pub fn pair() -> (Self, Self) {
        let (client_tx, server_rx) = mpsc::unbounded_channel();
        let (server_tx, client_rx) = mpsc::unbounded_channel();
        (
            Self {
                tx: client_tx,
                rx: client_rx,
                connected: true,
            },
            Self {
                tx: server_tx,
                rx: server_rx,
                connected: true,
            },
        )
    }
}

#[async_trait]
impl Transport for DuplexTransport {
    async fn send(&mut self, message: TransportMessage) -> Result<()> {
        self.tx
            .send(message)
            .map_err(|_| Error::internal("duplex peer dropped"))
    }

    async fn receive(&mut self) -> Result<TransportMessage> {
        self.rx
            .recv()
            .await
            .ok_or_else(|| Error::internal("duplex peer closed"))
    }

    async fn close(&mut self) -> Result<()> {
        self.connected = false;
        Ok(())
    }

    fn is_connected(&self) -> bool {
        self.connected
    }

    fn transport_type(&self) -> &'static str {
        "in-process-duplex"
    }
}

/// Drive a `tools/call` through a real [`Client`] against a high-level
/// [`Server`] running its own transport loop.
pub async fn call_via_server(server: Server, name: &str, args: Value) -> CallToolResult {
    let (client_t, server_t) = DuplexTransport::pair();
    tokio::spawn(async move {
        let _ = server.run(server_t).await;
    });
    let mut client = Client::new(client_t);
    client
        .initialize(ClientCapabilities::default())
        .await
        .expect("client initializes against server");
    client
        .call_tool(name.to_string(), args)
        .await
        .expect("tools/call succeeds against server")
}

/// Drive a `tools/call` through a real [`Client`] against a `ServerCore`
/// served by a request/response pump over the duplex transport.
pub async fn call_via_core(
    core: Arc<dyn ProtocolHandler>,
    name: &str,
    args: Value,
) -> CallToolResult {
    let (client_t, mut server_t) = DuplexTransport::pair();
    tokio::spawn(async move {
        while let Ok(message) = server_t.receive().await {
            if let TransportMessage::Request { id, request } = message {
                let response = core.handle_request(id, request, None).await;
                if server_t
                    .send(TransportMessage::Response(response))
                    .await
                    .is_err()
                {
                    break;
                }
            }
        }
    });
    let mut client = Client::new(client_t);
    client
        .initialize(ClientCapabilities::default())
        .await
        .expect("client initializes against core");
    client
        .call_tool(name.to_string(), args)
        .await
        .expect("tools/call succeeds against core")
}