rust-mcp-transport 2.0.0

Transport implementations for the MCP (Model Context Protocol) within the rust-mcp-sdk ecosystem, enabling asynchronous data exchange and efficient message handling between MCP clients and servers.
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
134
135
136
137
138
139

mod cancellation_token;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
mod http_utils;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
mod readable_channel;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
mod sse_event;

#[cfg(feature = "streamable-http")]
mod sse_parser;

#[cfg(feature = "sse")]
mod sse_stream;

#[cfg(feature = "streamable-http")]
mod streamable_http_stream;

mod time_utils;

#[cfg(feature = "sse")]
mod writable_channel;

use crate::error::{TransportError, TransportResult};
use crate::schema::schema_utils::SdkError;
pub(crate) use cancellation_token::*;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
use crate::SessionId;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
pub(crate) use http_utils::*;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
pub(crate) use readable_channel::*;

#[cfg(any(feature = "sse", feature = "streamable-http"))]
pub use sse_event::*;

#[cfg(feature = "streamable-http")]
pub(crate) use sse_parser::*;

#[cfg(feature = "sse")]
pub(crate) use sse_stream::*;

#[cfg(feature = "streamable-http")]
pub(crate) use streamable_http_stream::*;

// Re-exported for upcoming request-signing work; currently unused within the crate.
#[allow(unused_imports)]
pub use time_utils::*;
use tokio::time::{timeout, Duration};
#[cfg(feature = "sse")]
pub(crate) use writable_channel::*;

pub async fn await_timeout<F, T, E>(operation: F, timeout_duration: Duration) -> TransportResult<T>
where
    F: std::future::Future<Output = Result<T, E>>, // The operation returns a Result
    E: Into<TransportError>,
{
    match timeout(timeout_duration, operation).await {
        Ok(result) => result.map_err(|err| err.into()),
        Err(_) => Err(SdkError::request_timeout(timeout_duration.as_millis()).into()), // Timeout error
    }
}

/// Adds a session ID as a query parameter to a given endpoint URL.
///
/// # Arguments
/// * `endpoint` - The base URL or endpoint (e.g., "/messages")
/// * `session_id` - The session ID to append as a query parameter
///
/// # Returns
/// A String containing the endpoint with the session ID added as a query parameter
///
#[cfg(any(feature = "sse", feature = "streamable-http"))]
pub(crate) fn endpoint_with_session_id(endpoint: &str, session_id: &SessionId) -> String {
    // Handle empty endpoint
    let base = if endpoint.is_empty() { "/" } else { endpoint };

    // Split fragment if it exists
    let (path_and_query, fragment) = if let Some((p, f)) = base.split_once('#') {
        (p, Some(f))
    } else {
        (base, None)
    };

    // Split path and query
    let (path, query) = if let Some((p, q)) = path_and_query.split_once('?') {
        (p, Some(q))
    } else {
        (path_and_query, None)
    };

    // Build the query string
    let new_query = match query {
        Some(q) if !q.is_empty() => format!("{q}&sessionId={session_id}"),
        _ => format!("sessionId={session_id}"),
    };

    // Construct final URL
    match fragment {
        Some(f) => format!("{path}?{new_query}#{f}"),
        None => format!("{path}?{new_query}"),
    }
}

#[cfg(feature = "sse")]
#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_endpoint_with_session_id() {
        let session_id: SessionId = "AAA".to_string();
        assert_eq!(
            endpoint_with_session_id("/messages", &session_id),
            "/messages?sessionId=AAA"
        );
        assert_eq!(
            endpoint_with_session_id("/messages?foo=bar&baz=qux", &session_id),
            "/messages?foo=bar&baz=qux&sessionId=AAA"
        );
        assert_eq!(
            endpoint_with_session_id("/messages#section1", &session_id),
            "/messages?sessionId=AAA#section1"
        );
        assert_eq!(
            endpoint_with_session_id("/messages?key=value#section2", &session_id),
            "/messages?key=value&sessionId=AAA#section2"
        );
        assert_eq!(
            endpoint_with_session_id("/", &session_id),
            "/?sessionId=AAA"
        );
        assert_eq!(endpoint_with_session_id("", &session_id), "/?sessionId=AAA");
    }
}