soma_mcp_client/
upstream.rs

1//! Self-contained upstream runtime primitives.
2
3pub mod http_body_cap;
4pub mod http_client;
5pub mod pool;
6pub mod relay;
7pub mod transport;
8
9use std::collections::BTreeMap;
10
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13use thiserror::Error;
14
15/// Protocol-neutral representation of a modern MCP request outcome.
16///
17/// The payload contains the serialized result object for the corresponding
18/// `resultType`, allowing product layers to preserve MRTR and task responses
19/// without depending directly on rmcp's protocol models.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub enum McpRequestOutcome {
22    Complete(Value),
23    InputRequired(Value),
24    Task(Value),
25}
26
27impl McpRequestOutcome {
28    #[must_use]
29    pub fn payload(&self) -> &Value {
30        match self {
31            Self::Complete(value) | Self::InputRequired(value) | Self::Task(value) => value,
32        }
33    }
34}
35
36/// Follow-up data echoed by clients during a modern MCP multi-round trip.
37#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
38pub struct McpRoundTrip {
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub input_responses: Option<BTreeMap<String, Value>>,
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub request_state: Option<String>,
43    /// Complete per-request MCP metadata from the downstream client.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub request_meta: Option<Value>,
46}
47
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct ToolDescriptor {
50    pub name: String,
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    pub description: Option<String>,
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub input_schema: Option<Value>,
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub output_schema: Option<Value>,
57    #[serde(default)]
58    pub destructive: bool,
59}
60
61impl ToolDescriptor {
62    #[must_use]
63    pub fn new(name: impl Into<String>) -> Self {
64        Self {
65            name: name.into(),
66            description: None,
67            input_schema: None,
68            output_schema: None,
69            destructive: true,
70        }
71    }
72}
73
74#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
75pub struct ResourceDescriptor {
76    pub uri: String,
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub name: Option<String>,
79}
80
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82pub struct PromptDescriptor {
83    pub name: String,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub description: Option<String>,
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
89pub enum TransportKind {
90    InProcess,
91    HttpJson,
92    HttpSse,
93    Stdio,
94    WebSocket,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub enum UpstreamHealth {
99    Connected,
100    Degraded {
101        consecutive_failures: u32,
102        error: Option<String>,
103    },
104    Disabled,
105    Unsupported {
106        reason: String,
107    },
108}
109
110impl UpstreamHealth {
111    #[must_use]
112    pub fn is_routable(&self) -> bool {
113        matches!(self, Self::Connected)
114    }
115}
116
117#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118pub struct UpstreamSnapshot {
119    pub name: String,
120    pub transport: TransportKind,
121    pub health: UpstreamHealth,
122    pub tools: Vec<ToolDescriptor>,
123    pub resources: Vec<ResourceDescriptor>,
124    pub prompts: Vec<PromptDescriptor>,
125    #[serde(default)]
126    pub stale: bool,
127}
128
129impl UpstreamSnapshot {
130    #[must_use]
131    pub fn empty(name: impl Into<String>, transport: TransportKind) -> Self {
132        Self {
133            name: name.into(),
134            transport,
135            health: UpstreamHealth::Connected,
136            tools: Vec::new(),
137            resources: Vec::new(),
138            prompts: Vec::new(),
139            stale: false,
140        }
141    }
142}
143
144#[derive(Debug, Clone, Copy, PartialEq, Eq)]
145pub enum CapScope {
146    ToolsList,
147    ToolsCall,
148    ResourcesList,
149    ResourcesRead,
150    PromptsList,
151    PromptsGet,
152    RelayCall,
153    HttpJson,
154    HttpSseEvent,
155    WebSocketFrame,
156    StdioMessage,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub struct ResponseCaps {
161    pub tools_list_bytes: usize,
162    pub tools_call_bytes: usize,
163    pub resources_list_bytes: usize,
164    pub resources_read_bytes: usize,
165    pub prompts_list_bytes: usize,
166    pub prompts_get_bytes: usize,
167    pub relay_call_bytes: usize,
168    pub http_json_bytes: usize,
169    pub http_sse_event_bytes: usize,
170    pub websocket_frame_bytes: usize,
171    pub stdio_message_bytes: usize,
172}
173
174impl Default for ResponseCaps {
175    fn default() -> Self {
176        Self {
177            tools_list_bytes: 512 * 1024,
178            tools_call_bytes: 2 * 1024 * 1024,
179            resources_list_bytes: 512 * 1024,
180            resources_read_bytes: 2 * 1024 * 1024,
181            prompts_list_bytes: 256 * 1024,
182            prompts_get_bytes: 512 * 1024,
183            relay_call_bytes: 4 * 1024 * 1024,
184            http_json_bytes: 2 * 1024 * 1024,
185            http_sse_event_bytes: 512 * 1024,
186            websocket_frame_bytes: 512 * 1024,
187            stdio_message_bytes: 2 * 1024 * 1024,
188        }
189    }
190}
191
192impl ResponseCaps {
193    #[must_use]
194    pub fn limit_for(&self, scope: CapScope) -> usize {
195        match scope {
196            CapScope::ToolsList => self.tools_list_bytes,
197            CapScope::ToolsCall => self.tools_call_bytes,
198            CapScope::ResourcesList => self.resources_list_bytes,
199            CapScope::ResourcesRead => self.resources_read_bytes,
200            CapScope::PromptsList => self.prompts_list_bytes,
201            CapScope::PromptsGet => self.prompts_get_bytes,
202            CapScope::RelayCall => self.relay_call_bytes,
203            CapScope::HttpJson => self.http_json_bytes,
204            CapScope::HttpSseEvent => self.http_sse_event_bytes,
205            CapScope::WebSocketFrame => self.websocket_frame_bytes,
206            CapScope::StdioMessage => self.stdio_message_bytes,
207        }
208    }
209
210    pub fn enforce(&self, scope: CapScope, observed_bytes: usize) -> Result<(), UpstreamError> {
211        let limit = self.limit_for(scope);
212        if observed_bytes > limit {
213            return Err(UpstreamError::ResponseTooLarge {
214                scope,
215                observed_bytes,
216                limit,
217            });
218        }
219        Ok(())
220    }
221}
222
223#[derive(Debug, Error, PartialEq, Eq)]
224pub enum UpstreamError {
225    #[error("upstream `{upstream}` is not configured")]
226    UnknownUpstream { upstream: String },
227    #[error("upstream `{upstream}` is not routable: {reason}")]
228    NotRoutable { upstream: String, reason: String },
229    #[error("upstream `{upstream}` does not expose `{item}`")]
230    NotExposed { upstream: String, item: String },
231    #[error("upstream `{upstream}` does not support `{capability}`")]
232    Unsupported {
233        upstream: String,
234        capability: &'static str,
235    },
236    #[error("upstream `{upstream}` failed to connect: {message}")]
237    LiveConnect { upstream: String, message: String },
238    #[error("upstream `{upstream}` {operation} failed: {message}")]
239    LiveCall {
240        upstream: String,
241        operation: &'static str,
242        message: String,
243    },
244    #[error("{scope:?} payload was {observed_bytes} bytes, exceeding {limit} bytes")]
245    ResponseTooLarge {
246        scope: CapScope,
247        observed_bytes: usize,
248        limit: usize,
249    },
250    #[error("tool params must be a JSON object")]
251    ParamsMustBeObject,
252}
253
254impl UpstreamError {
255    pub(crate) fn connect(
256        config: &crate::config::UpstreamConfig,
257        error: impl std::fmt::Display,
258    ) -> Self {
259        Self::LiveConnect {
260            upstream: config.name.clone(),
261            message: error.to_string(),
262        }
263    }
264}
265
266#[cfg(test)]
267#[path = "upstream_tests.rs"]
268mod tests;
soma_mcp_client/upstream.rs

soma_mcp_client/
upstream.rs
