soma_mcp_client/upstream/pool/
resources.rs

1use crate::upstream::{
2    CapScope, McpRequestOutcome, McpRoundTrip, ResourceDescriptor, UpstreamError,
3};
4
5use super::tools::matches_filter;
6
7impl super::UpstreamPool {
8    pub async fn list_resources(
9        &self,
10        upstream: &str,
11    ) -> Result<Vec<ResourceDescriptor>, UpstreamError> {
12        self.ensure_connected(upstream).await?;
13        self.with_entry(upstream, |entry| {
14            if !entry.config.proxy_resources {
15                return Ok(Vec::new());
16            }
17            let resources: Vec<ResourceDescriptor> = entry
18                .snapshot
19                .resources
20                .iter()
21                .filter(|resource| {
22                    matches_filter(entry.config.expose_resources.as_deref(), &resource.uri)
23                })
24                .cloned()
25                .collect();
26            let bytes = serde_json::to_vec(&resources).map_or(usize::MAX, |bytes| bytes.len());
27            self.response_caps()
28                .enforce(CapScope::ResourcesList, bytes)?;
29            Ok(resources)
30        })
31    }
32
33    pub async fn read_resource(
34        &self,
35        upstream: &str,
36        uri: &str,
37    ) -> Result<serde_json::Value, UpstreamError> {
38        self.ensure_connected(upstream).await?;
39        let peer =
40            self.with_entry(upstream, |entry| {
41                if !entry.config.proxy_resources {
42                    return Err(UpstreamError::Unsupported {
43                        upstream: upstream.to_owned(),
44                        capability: "resources/read",
45                    });
46                }
47                entry.live.as_ref().map(|live| live.peer()).ok_or_else(|| {
48                    UpstreamError::Unsupported {
49                        upstream: upstream.to_owned(),
50                        capability: "resources/read",
51                    }
52                })
53            })?;
54        let value = super::live::read_live_resource(upstream, peer, uri.to_owned()).await?;
55        let bytes = serde_json::to_vec(&value).map_or(usize::MAX, |bytes| bytes.len());
56        self.response_caps()
57            .enforce(CapScope::ResourcesRead, bytes)?;
58        Ok(value)
59    }
60
61    pub async fn read_resource_once(
62        &self,
63        upstream: &str,
64        uri: &str,
65        round_trip: McpRoundTrip,
66    ) -> Result<McpRequestOutcome, UpstreamError> {
67        self.ensure_connected(upstream).await?;
68        let peer =
69            self.with_entry(upstream, |entry| {
70                if !entry.config.proxy_resources {
71                    return Err(UpstreamError::Unsupported {
72                        upstream: upstream.to_owned(),
73                        capability: "resources/read",
74                    });
75                }
76                entry.live.as_ref().map(|live| live.peer()).ok_or_else(|| {
77                    UpstreamError::Unsupported {
78                        upstream: upstream.to_owned(),
79                        capability: "resources/read",
80                    }
81                })
82            })?;
83        let outcome =
84            super::live::read_live_resource_once(upstream, peer, uri.to_owned(), round_trip)
85                .await?;
86        let bytes = serde_json::to_vec(outcome.payload()).map_or(usize::MAX, |bytes| bytes.len());
87        self.response_caps()
88            .enforce(CapScope::ResourcesRead, bytes)?;
89        Ok(outcome)
90    }
91}
92
93#[cfg(test)]
94#[path = "resources_tests.rs"]
95mod tests;
soma_mcp_client/upstream/pool/resources.rs

soma_mcp_client/upstream/pool/
resources.rs
