soma_mcp_client/upstream/pool/
live.rs

1use std::process::Stdio;
2#[cfg(feature = "oauth")]
3use std::sync::Arc;
4
5use rmcp::model::{
6    CallToolRequestParams, CallToolResponse, ClientCapabilities, ClientInfo,
7    GetPromptRequestParams, GetPromptResponse, Implementation, ProtocolVersion,
8    ReadResourceRequestParams, ReadResourceResponse, RequestMetaObject,
9};
10use rmcp::service::{ClientInitializeError, ClientServiceExt, RunningService};
11use rmcp::transport::{
12    StreamableHttpClientTransport, TokioChildProcess,
13    streamable_http_client::StreamableHttpClientTransportConfig,
14};
15use rmcp::{ClientHandler, RoleClient};
16use serde_json::{Map, Value};
17use tokio::process::Command;
18
19use crate::config::UpstreamConfig;
20#[cfg(feature = "oauth")]
21use crate::oauth::UpstreamOAuthProvider;
22use crate::process::guard::SpawnGuard;
23use crate::upstream::http_body_cap::BodyCappedHttpClient;
24use crate::upstream::http_client::{HttpTransportDecision, decide_http_transport};
25use crate::upstream::transport::websocket::{
26    WebSocketTransportConfig, connect as connect_websocket_transport,
27};
28use crate::upstream::{
29    CapScope, McpRequestOutcome, McpRoundTrip, ResponseCaps, TransportKind, UpstreamError,
30    UpstreamSnapshot,
31};
32
33use super::lifecycle_compat::{LifecycleAttempt, compatibility_retry, log_fallback};
34
35#[path = "live_support.rs"]
36mod live_support;
37
38use live_support::{
39    bearer_token_from_env, capability_is_absent, drain_stderr, ensure_rustls_crypto_provider,
40    prompt_descriptor, resource_descriptor, stdio_env, tool_descriptor, websocket_authorization,
41};
42
43#[derive(Clone, Copy, Debug, Default)]
44struct UpstreamClientHandler;
45
46#[derive(Clone, Debug)]
47struct RequestScopedClientHandler {
48    info: ClientInfo,
49}
50
51impl ClientHandler for RequestScopedClientHandler {
52    fn get_info(&self) -> ClientInfo {
53        self.info.clone()
54    }
55}
56
57impl ClientHandler for UpstreamClientHandler {
58    fn get_info(&self) -> ClientInfo {
59        ClientInfo::new(
60            ClientCapabilities::builder().enable_tasks().build(),
61            Implementation::new("soma-upstream-client", env!("CARGO_PKG_VERSION")),
62        )
63        .with_protocol_version(ProtocolVersion::LATEST)
64    }
65}
66
67enum ConnectAttemptError {
68    Fatal(UpstreamError),
69    Lifecycle(Box<ClientInitializeError>),
70}
71
72impl ConnectAttemptError {
73    fn into_upstream(self, config: &UpstreamConfig, prefix: &str) -> UpstreamError {
74        match self {
75            Self::Fatal(error) => error,
76            Self::Lifecycle(error) => UpstreamError::connect(config, format!("{prefix}: {error}")),
77        }
78    }
79}
80
81#[derive(Clone)]
82pub(super) struct LiveConnectContext<'a> {
83    response_caps: &'a ResponseCaps,
84    #[cfg(feature = "oauth")]
85    oauth: Option<LiveOauthContext<'a>>,
86}
87
88#[cfg(feature = "oauth")]
89#[derive(Clone)]
90pub(super) struct LiveOauthContext<'a> {
91    pub subject: &'a str,
92    pub provider: Arc<dyn UpstreamOAuthProvider>,
93}
94
95impl<'a> LiveConnectContext<'a> {
96    pub(super) fn shared(response_caps: &'a ResponseCaps) -> Self {
97        Self {
98            response_caps,
99            #[cfg(feature = "oauth")]
100            oauth: None,
101        }
102    }
103
104    #[cfg(feature = "oauth")]
105    pub(super) fn oauth(
106        response_caps: &'a ResponseCaps,
107        subject: &'a str,
108        provider: Arc<dyn UpstreamOAuthProvider>,
109    ) -> Self {
110        Self {
111            response_caps,
112            oauth: Some(LiveOauthContext { subject, provider }),
113        }
114    }
115}
116
117#[derive(Debug, Clone, Copy, PartialEq, Eq)]
118pub(super) enum LiveKind {
119    Http(TransportKind),
120    Stdio,
121    WebSocket,
122}
123
124type LiveConnectionFor<H> = (
125    RunningService<RoleClient, H>,
126    rmcp::service::Peer<RoleClient>,
127    LiveKind,
128);
129
130pub(super) struct LiveUpstream {
131    _service: RunningService<RoleClient, UpstreamClientHandler>,
132    peer: rmcp::service::Peer<RoleClient>,
133}
134
135impl LiveUpstream {
136    pub(super) fn peer(&self) -> rmcp::service::Peer<RoleClient> {
137        self.peer.clone()
138    }
139}
140
141pub(super) async fn connect_live(
142    config: &UpstreamConfig,
143    guard: &SpawnGuard,
144    context: LiveConnectContext<'_>,
145) -> Result<(LiveUpstream, UpstreamSnapshot), UpstreamError> {
146    let (service, peer, kind) =
147        connect_with_handler(config, guard, context, UpstreamClientHandler).await?;
148
149    let tools = peer
150        .list_all_tools()
151        .await
152        .map_err(|error| UpstreamError::connect(config, error))?;
153    let resources = if config.proxy_resources {
154        list_resources_or_empty(config, &peer).await?
155    } else {
156        Vec::new()
157    };
158    let prompts = if config.proxy_prompts {
159        list_prompts_or_empty(config, &peer).await?
160    } else {
161        Vec::new()
162    };
163
164    let mut snapshot = UpstreamSnapshot::empty(
165        config.name.clone(),
166        match kind {
167            LiveKind::Http(transport) => transport,
168            LiveKind::Stdio => TransportKind::Stdio,
169            LiveKind::WebSocket => TransportKind::WebSocket,
170        },
171    );
172    snapshot.tools = tools.into_iter().map(tool_descriptor).collect();
173    snapshot.resources = resources.into_iter().map(resource_descriptor).collect();
174    snapshot.prompts = prompts.into_iter().map(prompt_descriptor).collect();
175    Ok((
176        LiveUpstream {
177            _service: service,
178            peer,
179        },
180        snapshot,
181    ))
182}
183
184async fn connect_with_handler<H>(
185    config: &UpstreamConfig,
186    guard: &SpawnGuard,
187    context: LiveConnectContext<'_>,
188    handler: H,
189) -> Result<LiveConnectionFor<H>, UpstreamError>
190where
191    H: ClientHandler + Clone,
192{
193    if let Some(url) = config.url.as_deref() {
194        return match decide_http_transport(url) {
195            HttpTransportDecision::WebSocket => {
196                connect_websocket_with_handler(config, url, handler).await
197            }
198            HttpTransportDecision::Json | HttpTransportDecision::Sse => {
199                connect_http_with_handler(config, url, context, handler).await
200            }
201        };
202    }
203    if let Some(command) = config.command.as_deref() {
204        return connect_stdio_with_handler(config, command, guard, handler).await;
205    }
206    Err(UpstreamError::Unsupported {
207        upstream: config.name.clone(),
208        capability: "transport",
209    })
210}
211
212pub(super) async fn call_live_tool(
213    upstream: &str,
214    peer: rmcp::service::Peer<RoleClient>,
215    tool: String,
216    params: Value,
217) -> Result<Value, UpstreamError> {
218    let Value::Object(args) = params else {
219        return Err(UpstreamError::ParamsMustBeObject);
220    };
221    let result = peer
222        .call_tool(CallToolRequestParams::new(tool).with_arguments(args))
223        .await
224        .map_err(|error| UpstreamError::LiveCall {
225            upstream: upstream.to_owned(),
226            operation: "tools/call",
227            message: error.to_string(),
228        })?;
229    if let Some(value) = result.structured_content.clone() {
230        return Ok(value);
231    }
232    Ok(serde_json::to_value(result).unwrap_or(Value::Null))
233}
234
235pub(super) async fn call_live_tool_once(
236    upstream: &str,
237    peer: rmcp::service::Peer<RoleClient>,
238    tool: String,
239    params: Value,
240    round_trip: McpRoundTrip,
241) -> Result<McpRequestOutcome, UpstreamError> {
242    let Value::Object(args) = params else {
243        return Err(UpstreamError::ParamsMustBeObject);
244    };
245    let mut request = CallToolRequestParams::new(tool).with_arguments(args);
246    request.input_responses = round_trip.input_responses;
247    request.request_state = round_trip.request_state;
248    let response = peer
249        .call_tool_once(request)
250        .await
251        .map_err(|error| UpstreamError::LiveCall {
252            upstream: upstream.to_owned(),
253            operation: "tools/call",
254            message: error.to_string(),
255        })?;
256    match response {
257        CallToolResponse::Complete(result) => {
258            serialize_live_outcome(upstream, "tools/call", result, McpRequestOutcome::Complete)
259        }
260        CallToolResponse::InputRequired(result) => serialize_live_outcome(
261            upstream,
262            "tools/call",
263            result,
264            McpRequestOutcome::InputRequired,
265        ),
266        CallToolResponse::Task(result) => {
267            serialize_live_outcome(upstream, "tools/call", result, McpRequestOutcome::Task)
268        }
269        _ => Err(unsupported_live_outcome(upstream, "tools/call")),
270    }
271}
272
273pub(super) async fn call_live_tool_once_scoped(
274    config: &UpstreamConfig,
275    context: LiveConnectContext<'_>,
276    tool: String,
277    params: Value,
278    mut round_trip: McpRoundTrip,
279) -> Result<McpRequestOutcome, UpstreamError> {
280    let meta = round_trip
281        .request_meta
282        .take()
283        .ok_or_else(|| UpstreamError::LiveCall {
284            upstream: config.name.clone(),
285            operation: "tools/call",
286            message: "request-scoped relay requires downstream MCP metadata".to_owned(),
287        })?;
288    let handler = request_scoped_handler(&config.name, "tools/call", meta)?;
289    let (mut service, peer, _) =
290        connect_with_handler(config, &SpawnGuard::default(), context, handler).await?;
291    let outcome = call_live_tool_once(
292        &config.name,
293        peer,
294        tool,
295        params,
296        McpRoundTrip {
297            input_responses: round_trip.input_responses,
298            request_state: round_trip.request_state,
299            request_meta: None,
300        },
301    )
302    .await;
303    if let Err(error) = service.close().await {
304        tracing::debug!(
305            upstream = %config.name,
306            error = %error,
307            "request-scoped upstream relay did not close cleanly"
308        );
309    }
310    outcome
311}
312
313pub(super) async fn read_live_resource(
314    upstream: &str,
315    peer: rmcp::service::Peer<RoleClient>,
316    uri: String,
317) -> Result<Value, UpstreamError> {
318    let result = peer
319        .read_resource(ReadResourceRequestParams::new(uri))
320        .await
321        .map_err(|error| UpstreamError::LiveCall {
322            upstream: upstream.to_owned(),
323            operation: "resources/read",
324            message: error.to_string(),
325        })?;
326    serde_json::to_value(result).map_err(|error| UpstreamError::LiveCall {
327        upstream: upstream.to_owned(),
328        operation: "resources/read",
329        message: error.to_string(),
330    })
331}
332
333pub(super) async fn read_live_resource_once(
334    upstream: &str,
335    peer: rmcp::service::Peer<RoleClient>,
336    uri: String,
337    round_trip: McpRoundTrip,
338) -> Result<McpRequestOutcome, UpstreamError> {
339    let mut request = ReadResourceRequestParams::new(uri);
340    request.input_responses = round_trip.input_responses;
341    request.request_state = round_trip.request_state;
342    let response =
343        peer.read_resource_once(request)
344            .await
345            .map_err(|error| UpstreamError::LiveCall {
346                upstream: upstream.to_owned(),
347                operation: "resources/read",
348                message: error.to_string(),
349            })?;
350    match response {
351        ReadResourceResponse::Complete(result) => serialize_live_outcome(
352            upstream,
353            "resources/read",
354            result,
355            McpRequestOutcome::Complete,
356        ),
357        ReadResourceResponse::InputRequired(result) => serialize_live_outcome(
358            upstream,
359            "resources/read",
360            result,
361            McpRequestOutcome::InputRequired,
362        ),
363        _ => Err(unsupported_live_outcome(upstream, "resources/read")),
364    }
365}
366
367pub(super) async fn get_live_prompt(
368    upstream: &str,
369    peer: rmcp::service::Peer<RoleClient>,
370    name: String,
371    arguments: Option<Map<String, Value>>,
372) -> Result<Value, UpstreamError> {
373    let mut params = GetPromptRequestParams::new(name);
374    params.arguments = arguments;
375    let result = peer
376        .get_prompt(params)
377        .await
378        .map_err(|error| UpstreamError::LiveCall {
379            upstream: upstream.to_owned(),
380            operation: "prompts/get",
381            message: error.to_string(),
382        })?;
383    serde_json::to_value(result).map_err(|error| UpstreamError::LiveCall {
384        upstream: upstream.to_owned(),
385        operation: "prompts/get",
386        message: error.to_string(),
387    })
388}
389
390pub(super) async fn get_live_prompt_once(
391    upstream: &str,
392    peer: rmcp::service::Peer<RoleClient>,
393    name: String,
394    arguments: Option<Map<String, Value>>,
395    round_trip: McpRoundTrip,
396) -> Result<McpRequestOutcome, UpstreamError> {
397    let mut request = GetPromptRequestParams::new(name);
398    request.arguments = arguments;
399    request.input_responses = round_trip.input_responses;
400    request.request_state = round_trip.request_state;
401    let response =
402        peer.get_prompt_once(request)
403            .await
404            .map_err(|error| UpstreamError::LiveCall {
405                upstream: upstream.to_owned(),
406                operation: "prompts/get",
407                message: error.to_string(),
408            })?;
409    match response {
410        GetPromptResponse::Complete(result) => {
411            serialize_live_outcome(upstream, "prompts/get", result, McpRequestOutcome::Complete)
412        }
413        GetPromptResponse::InputRequired(result) => serialize_live_outcome(
414            upstream,
415            "prompts/get",
416            result,
417            McpRequestOutcome::InputRequired,
418        ),
419        _ => Err(unsupported_live_outcome(upstream, "prompts/get")),
420    }
421}
422
423fn request_scoped_handler(
424    upstream: &str,
425    operation: &'static str,
426    value: Value,
427) -> Result<RequestScopedClientHandler, UpstreamError> {
428    let meta: RequestMetaObject =
429        serde_json::from_value(value).map_err(|error| UpstreamError::LiveCall {
430            upstream: upstream.to_owned(),
431            operation,
432            message: format!("invalid downstream MCP request metadata: {error}"),
433        })?;
434    let protocol_version = meta.protocol_version().unwrap_or(ProtocolVersion::LATEST);
435    let implementation = meta.client_info().unwrap_or_else(|| {
436        Implementation::new("soma-request-scoped-relay", env!("CARGO_PKG_VERSION"))
437    });
438    let capabilities = meta.client_capabilities().unwrap_or_default();
439    Ok(RequestScopedClientHandler {
440        info: ClientInfo::new(capabilities, implementation).with_protocol_version(protocol_version),
441    })
442}
443
444fn unsupported_live_outcome(upstream: &str, operation: &'static str) -> UpstreamError {
445    UpstreamError::LiveCall {
446        upstream: upstream.to_owned(),
447        operation,
448        message: "upstream returned an unsupported MCP response variant".to_owned(),
449    }
450}
451
452fn serialize_live_outcome<T: serde::Serialize>(
453    upstream: &str,
454    operation: &'static str,
455    result: T,
456    wrap: impl FnOnce(Value) -> McpRequestOutcome,
457) -> Result<McpRequestOutcome, UpstreamError> {
458    serde_json::to_value(result)
459        .map(wrap)
460        .map_err(|error| UpstreamError::LiveCall {
461            upstream: upstream.to_owned(),
462            operation,
463            message: error.to_string(),
464        })
465}
466
467async fn connect_http_with_handler<H>(
468    config: &UpstreamConfig,
469    url: &str,
470    context: LiveConnectContext<'_>,
471    handler: H,
472) -> Result<LiveConnectionFor<H>, UpstreamError>
473where
474    H: ClientHandler + Clone,
475{
476    match connect_http_once(
477        config,
478        url,
479        context.clone(),
480        LifecycleAttempt::Modern,
481        handler.clone(),
482    )
483    .await
484    {
485        Ok(connection) => Ok(connection),
486        Err(ConnectAttemptError::Fatal(error)) => Err(error),
487        Err(ConnectAttemptError::Lifecycle(error)) => {
488            let Some(attempt) = compatibility_retry(&error) else {
489                return Err(UpstreamError::connect(
490                    config,
491                    format!("http connect failed: {error}"),
492                ));
493            };
494            log_fallback(&config.name, "http", attempt, &error);
495            connect_http_once(config, url, context, attempt, handler)
496                .await
497                .map_err(|error| error.into_upstream(config, "http connect failed"))
498        }
499    }
500}
501
502async fn connect_http_once<H>(
503    config: &UpstreamConfig,
504    url: &str,
505    context: LiveConnectContext<'_>,
506    lifecycle: LifecycleAttempt,
507    handler: H,
508) -> Result<LiveConnectionFor<H>, ConnectAttemptError>
509where
510    H: ClientHandler + Clone,
511{
512    ensure_rustls_crypto_provider();
513    let transport_kind = match decide_http_transport(url) {
514        HttpTransportDecision::Json => TransportKind::HttpJson,
515        HttpTransportDecision::Sse => TransportKind::HttpSse,
516        HttpTransportDecision::WebSocket => TransportKind::WebSocket,
517    };
518    let mut transport_config = StreamableHttpClientTransportConfig::with_uri(url.to_owned());
519    #[cfg(feature = "oauth")]
520    if config.oauth.is_some() {
521        let oauth = context.oauth.ok_or_else(|| {
522            ConnectAttemptError::Fatal(UpstreamError::LiveConnect {
523                upstream: config.name.clone(),
524                message: "oauth upstream requires subject-scoped connection context".to_owned(),
525            })
526        })?;
527        let client = BodyCappedHttpClient::default_with_caps(
528            context.response_caps.limit_for(CapScope::HttpJson),
529            context.response_caps.limit_for(CapScope::HttpSseEvent),
530        );
531        let auth_client = oauth
532            .provider
533            .authenticated_http_client(config, oauth.subject, client)
534            .await
535            .map_err(|error| {
536                ConnectAttemptError::Fatal(UpstreamError::LiveConnect {
537                    upstream: config.name.clone(),
538                    message: error.to_string(),
539                })
540            })?;
541        let transport = StreamableHttpClientTransport::with_client(auth_client, transport_config);
542        let service = handler
543            .clone()
544            .serve_with_lifecycle(transport, lifecycle.mode())
545            .await
546            .map_err(|error| ConnectAttemptError::Lifecycle(Box::new(error)))?;
547        let peer = service.peer().clone();
548        return Ok((service, peer, LiveKind::Http(transport_kind)));
549    }
550    #[cfg(not(feature = "oauth"))]
551    if config.oauth.is_some() {
552        return Err(ConnectAttemptError::Fatal(UpstreamError::LiveConnect {
553            upstream: config.name.clone(),
554            message: "oauth upstream support is not compiled into soma-mcp-client".to_owned(),
555        }));
556    }
557    if let Some(token) = bearer_token_from_env(config) {
558        transport_config = transport_config.auth_header(token);
559    }
560    let client = BodyCappedHttpClient::default_with_caps(
561        context.response_caps.limit_for(CapScope::HttpJson),
562        context.response_caps.limit_for(CapScope::HttpSseEvent),
563    );
564    let transport = StreamableHttpClientTransport::with_client(client, transport_config);
565    let service = handler
566        .serve_with_lifecycle(transport, lifecycle.mode())
567        .await
568        .map_err(|error| ConnectAttemptError::Lifecycle(Box::new(error)))?;
569    let peer = service.peer().clone();
570    Ok((service, peer, LiveKind::Http(transport_kind)))
571}
572
573async fn connect_websocket_with_handler<H>(
574    config: &UpstreamConfig,
575    url: &str,
576    handler: H,
577) -> Result<LiveConnectionFor<H>, UpstreamError>
578where
579    H: ClientHandler + Clone,
580{
581    match connect_websocket_once(config, url, LifecycleAttempt::Modern, handler.clone()).await {
582        Ok(connection) => Ok(connection),
583        Err(error) => {
584            let Some(attempt) = compatibility_retry(&error) else {
585                return Err(UpstreamError::connect(
586                    config,
587                    format!("websocket connect failed: {error}"),
588                ));
589            };
590            log_fallback(&config.name, "websocket", attempt, &error);
591            connect_websocket_once(config, url, attempt, handler)
592                .await
593                .map_err(|error| {
594                    UpstreamError::connect(config, format!("websocket connect failed: {error}"))
595                })
596        }
597    }
598}
599
600async fn connect_websocket_once<H>(
601    config: &UpstreamConfig,
602    url: &str,
603    lifecycle: LifecycleAttempt,
604    handler: H,
605) -> Result<LiveConnectionFor<H>, ClientInitializeError>
606where
607    H: ClientHandler,
608{
609    ensure_rustls_crypto_provider();
610    let transport_config = WebSocketTransportConfig::new(url.to_owned())
611        .with_authorization(websocket_authorization(config));
612    let service = handler
613        .serve_with_lifecycle(
614            connect_websocket_transport(transport_config),
615            lifecycle.mode(),
616        )
617        .await?;
618    let peer = service.peer().clone();
619    Ok((service, peer, LiveKind::WebSocket))
620}
621
622async fn connect_stdio_with_handler<H>(
623    config: &UpstreamConfig,
624    command: &str,
625    guard: &SpawnGuard,
626    handler: H,
627) -> Result<LiveConnectionFor<H>, UpstreamError>
628where
629    H: ClientHandler + Clone,
630{
631    match connect_stdio_once(
632        config,
633        command,
634        guard,
635        LifecycleAttempt::Modern,
636        handler.clone(),
637    )
638    .await
639    {
640        Ok(connection) => Ok(connection),
641        Err(ConnectAttemptError::Fatal(error)) => Err(error),
642        Err(ConnectAttemptError::Lifecycle(error)) => {
643            let Some(attempt) = compatibility_retry(&error) else {
644                return Err(UpstreamError::connect(
645                    config,
646                    format!("stdio MCP handshake failed: {error}"),
647                ));
648            };
649            log_fallback(&config.name, "stdio", attempt, &error);
650            connect_stdio_once(config, command, guard, attempt, handler)
651                .await
652                .map_err(|error| error.into_upstream(config, "stdio MCP handshake failed"))
653        }
654    }
655}
656
657async fn connect_stdio_once<H>(
658    config: &UpstreamConfig,
659    command_name: &str,
660    guard: &SpawnGuard,
661    lifecycle: LifecycleAttempt,
662    handler: H,
663) -> Result<LiveConnectionFor<H>, ConnectAttemptError>
664where
665    H: ClientHandler,
666{
667    let spec = crate::upstream::pool::connect_stdio::plan_stdio_connection(config, guard).map_err(
668        |error| {
669            ConnectAttemptError::Fatal(UpstreamError::LiveConnect {
670                upstream: config.name.clone(),
671                message: error.to_string(),
672            })
673        },
674    )?;
675    let mut cmd = Command::new(command_name);
676    cmd.args(&spec.args)
677        .env_clear()
678        .envs(stdio_env())
679        .envs(spec.env.iter())
680        .stderr(Stdio::piped());
681    if let Some(env_name) = config.bearer_token_env.as_deref()
682        && let Ok(token) = std::env::var(env_name)
683    {
684        cmd.env(env_name, token);
685    }
686
687    #[cfg(unix)]
688    let command = {
689        use process_wrap::tokio::{CommandWrap, ProcessGroup};
690        let mut wrapped = CommandWrap::from(cmd);
691        wrapped.wrap(ProcessGroup::leader());
692        wrapped
693    };
694    #[cfg(not(unix))]
695    let command = cmd;
696
697    let (transport, stderr) = TokioChildProcess::builder(command)
698        .spawn()
699        .map_err(|error| {
700            ConnectAttemptError::Fatal(UpstreamError::LiveConnect {
701                upstream: config.name.clone(),
702                message: format!("stdio spawn failed: {error}"),
703            })
704        })?;
705    drain_stderr(config.name.clone(), stderr);
706    let service = handler
707        .serve_with_lifecycle(transport, lifecycle.mode())
708        .await
709        .map_err(|error| ConnectAttemptError::Lifecycle(Box::new(error)))?;
710    let peer = service.peer().clone();
711    Ok((service, peer, LiveKind::Stdio))
712}
713
714async fn list_resources_or_empty(
715    config: &UpstreamConfig,
716    peer: &rmcp::service::Peer<RoleClient>,
717) -> Result<Vec<rmcp::model::Resource>, UpstreamError> {
718    match peer.list_all_resources().await {
719        Ok(resources) => Ok(resources),
720        Err(error) if capability_is_absent(&error.to_string()) => Ok(Vec::new()),
721        Err(error) => Err(UpstreamError::LiveConnect {
722            upstream: config.name.clone(),
723            message: format!("resources/list failed: {error}"),
724        }),
725    }
726}
727
728async fn list_prompts_or_empty(
729    config: &UpstreamConfig,
730    peer: &rmcp::service::Peer<RoleClient>,
731) -> Result<Vec<rmcp::model::Prompt>, UpstreamError> {
732    match peer.list_all_prompts().await {
733        Ok(prompts) => Ok(prompts),
734        Err(error) if capability_is_absent(&error.to_string()) => Ok(Vec::new()),
735        Err(error) => Err(UpstreamError::LiveConnect {
736            upstream: config.name.clone(),
737            message: format!("prompts/list failed: {error}"),
738        }),
739    }
740}
741
742#[cfg(test)]
743#[path = "live_tests.rs"]
744mod tests;
soma_mcp_client/upstream/pool/live.rs

soma_mcp_client/upstream/pool/
live.rs
