Skip to main content

max / makenotwork

10.8 KB · 309 lines History Blame Raw
1 //! Resources — the read side of the MCP object surface.
2 //!
3 //! Where [`crate::tool`] exposes an app's *actions*, resources expose its live
4 //! *objects*: things a client reads and (in the push layer) subscribes to. deox
5 //! exposes editor buffers this way; a content-addressed store could expose its
6 //! samples.
7 //!
8 //! Unlike tools, resources are **dynamic** — a buffer opens and closes at
9 //! runtime — so [`ResourceRegistry`] is interior-mutable and cloneable:
10 //! [`register`](ResourceRegistry::register) and [`remove`](ResourceRegistry::remove)
11 //! take `&self` and any clone observes the change. Each resource carries an
12 //! async read handler the registry invokes on `resources/read`; the app owns the
13 //! underlying state (for deox, behind the main-loop channel).
14 //!
15 //! The subscription + `notifications/resources/updated` push layer builds on
16 //! this registry; see the server module.
17
18 use std::collections::HashMap;
19 use std::future::Future;
20 use std::pin::Pin;
21 use std::sync::{Arc, RwLock};
22
23 use serde::Serialize;
24
25 use crate::error::{Error, Result};
26
27 type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
28 type ReadHandler = Arc<dyn Fn(String) -> BoxFuture<Result<ResourceContents>> + Send + Sync>;
29
30 /// Metadata for a single resource, as it appears in `resources/list`.
31 ///
32 /// `uri` is the stable identifier the client passes back to `resources/read`
33 /// and `resources/subscribe`. Choose a scheme the app owns (deox uses
34 /// `buffer://<path>`); it is an addressing contract, not a filesystem path.
35 #[derive(Debug, Clone, Serialize)]
36 pub struct ResourceDescriptor {
37 pub uri: String,
38 pub name: String,
39 #[serde(skip_serializing_if = "Option::is_none")]
40 pub description: Option<String>,
41 #[serde(rename = "mimeType", skip_serializing_if = "Option::is_none")]
42 pub mime_type: Option<String>,
43 }
44
45 impl ResourceDescriptor {
46 pub fn new(uri: impl Into<String>, name: impl Into<String>) -> Self {
47 Self {
48 uri: uri.into(),
49 name: name.into(),
50 description: None,
51 mime_type: None,
52 }
53 }
54
55 #[must_use]
56 pub fn with_description(mut self, description: impl Into<String>) -> Self {
57 self.description = Some(description.into());
58 self
59 }
60
61 #[must_use]
62 pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
63 self.mime_type = Some(mime_type.into());
64 self
65 }
66 }
67
68 /// The contents of one resource, as returned inside a `resources/read` result.
69 ///
70 /// Mirrors MCP's resource-contents shape: a `uri`, an optional `mimeType`, and
71 /// exactly one of `text` (UTF-8) or `blob` (base64). Use [`text`](Self::text)
72 /// for buffers and other textual resources; [`blob`](Self::blob) for binary.
73 #[derive(Debug, Clone, Serialize)]
74 pub struct ResourceContents {
75 pub uri: String,
76 #[serde(rename = "mimeType", skip_serializing_if = "Option::is_none")]
77 pub mime_type: Option<String>,
78 #[serde(skip_serializing_if = "Option::is_none")]
79 pub text: Option<String>,
80 #[serde(skip_serializing_if = "Option::is_none")]
81 pub blob: Option<String>,
82 }
83
84 impl ResourceContents {
85 /// Textual contents (the common case: a buffer, a note, a config file).
86 pub fn text(uri: impl Into<String>, text: impl Into<String>) -> Self {
87 Self {
88 uri: uri.into(),
89 mime_type: None,
90 text: Some(text.into()),
91 blob: None,
92 }
93 }
94
95 /// Binary contents, already base64-encoded.
96 pub fn blob(uri: impl Into<String>, blob_base64: impl Into<String>) -> Self {
97 Self {
98 uri: uri.into(),
99 mime_type: None,
100 text: None,
101 blob: Some(blob_base64.into()),
102 }
103 }
104
105 #[must_use]
106 pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
107 self.mime_type = Some(mime_type.into());
108 self
109 }
110 }
111
112 struct Entry {
113 descriptor: ResourceDescriptor,
114 handler: ReadHandler,
115 }
116
117 /// Registry of resources a client can list and read.
118 ///
119 /// Cheaply cloneable and interior-mutable: the map lives behind `Arc<RwLock>`,
120 /// so the app can [`register`](Self::register) and [`remove`](Self::remove)
121 /// resources at runtime while the server holds its own clone. This is the
122 /// deliberate contrast with [`crate::tool::ToolRegistry`], whose surface is
123 /// fixed before serving.
124 ///
125 /// With the `server` feature, [`notify_updated`](Self::notify_updated) pushes a
126 /// change to subscribed clients over the SSE stream.
127 #[derive(Clone)]
128 pub struct ResourceRegistry {
129 entries: Arc<RwLock<HashMap<String, Entry>>>,
130 /// Fan-out of updated URIs to every connected SSE session, which filters by
131 /// its own subscription set. Capacity bounds how far a slow session may lag
132 /// before it drops updates (it then re-reads current state on next use).
133 #[cfg(feature = "server")]
134 updates: tokio::sync::broadcast::Sender<String>,
135 }
136
137 impl Default for ResourceRegistry {
138 fn default() -> Self {
139 Self::new()
140 }
141 }
142
143 impl ResourceRegistry {
144 pub fn new() -> Self {
145 Self {
146 entries: Arc::new(RwLock::new(HashMap::new())),
147 #[cfg(feature = "server")]
148 updates: tokio::sync::broadcast::channel(256).0,
149 }
150 }
151
152 /// Register (or replace) a resource. `handler` is invoked on each
153 /// `resources/read` for this uri and returns the resource's current
154 /// contents — it should fetch live state, not a snapshot taken at
155 /// registration. Registering an existing uri replaces its descriptor and
156 /// handler.
157 pub fn register<F, Fut>(&self, descriptor: ResourceDescriptor, handler: F)
158 where
159 F: Fn(String) -> Fut + Send + Sync + 'static,
160 Fut: Future<Output = Result<ResourceContents>> + Send + 'static,
161 {
162 let handler: ReadHandler = Arc::new(move |uri| Box::pin(handler(uri)));
163 let uri = descriptor.uri.clone();
164 self.entries.write().unwrap().insert(
165 uri,
166 Entry {
167 descriptor,
168 handler,
169 },
170 );
171 }
172
173 /// Remove a resource. Returns whether it was present.
174 pub fn remove(&self, uri: &str) -> bool {
175 self.entries.write().unwrap().remove(uri).is_some()
176 }
177
178 pub fn contains(&self, uri: &str) -> bool {
179 self.entries.read().unwrap().contains_key(uri)
180 }
181
182 /// Descriptors for every registered resource, for `resources/list`.
183 pub fn list(&self) -> Vec<ResourceDescriptor> {
184 self.entries
185 .read()
186 .unwrap()
187 .values()
188 .map(|e| e.descriptor.clone())
189 .collect()
190 }
191
192 /// Read one resource's current contents by uri. The read handler runs
193 /// without the registry lock held, so a handler may itself touch the
194 /// registry.
195 pub async fn read(&self, uri: &str) -> Result<ResourceContents> {
196 let handler = {
197 let map = self.entries.read().unwrap();
198 map.get(uri).map(|e| e.handler.clone())
199 };
200 match handler {
201 Some(handler) => handler(uri.to_string()).await,
202 None => Err(Error::ResourceNotFound(uri.to_string())),
203 }
204 }
205 }
206
207 #[cfg(feature = "server")]
208 impl ResourceRegistry {
209 /// Announce that a resource changed. Subscribed clients receive a
210 /// `notifications/resources/updated` for `uri` over their SSE stream; they
211 /// then re-`read` it to get the new contents (the notification carries the
212 /// uri, not the body). The app calls this after mutating the underlying
213 /// state — for deox, on a buffer or selection change. A no-op when no
214 /// session is subscribed.
215 pub fn notify_updated(&self, uri: impl Into<String>) {
216 // `send` errors only when there are zero receivers; that is the normal
217 // "nobody is streaming" case, so the error is deliberately ignored.
218 let _ = self.updates.send(uri.into());
219 }
220
221 /// A receiver for the update fan-out, used by each SSE session.
222 pub(crate) fn subscribe_updates(&self) -> tokio::sync::broadcast::Receiver<String> {
223 self.updates.subscribe()
224 }
225 }
226
227 /// `resources/read` param object: `{ "uri": "..." }`.
228 #[cfg(feature = "server")]
229 #[derive(Debug, serde::Deserialize)]
230 pub(crate) struct ReadResourceParams {
231 pub uri: String,
232 }
233
234 /// Wrap a single [`ResourceContents`] into the `resources/read` result shape
235 /// (`{ "contents": [ ... ] }`).
236 #[cfg(feature = "server")]
237 // contents is consumed into the returned JSON value.
238 #[allow(clippy::needless_pass_by_value)]
239 pub(crate) fn read_result(contents: ResourceContents) -> serde_json::Value {
240 serde_json::json!({ "contents": [contents] })
241 }
242
243 #[cfg(test)]
244 mod tests {
245 use super::*;
246 use serde_json::json;
247
248 #[tokio::test]
249 async fn register_list_read_roundtrip() {
250 let reg = ResourceRegistry::new();
251 reg.register(
252 ResourceDescriptor::new("buffer://main.rs", "main.rs").with_mime_type("text/x-rust"),
253 |uri| async move { Ok(ResourceContents::text(uri, "fn main() {}")) },
254 );
255
256 let list = reg.list();
257 assert_eq!(list.len(), 1);
258 assert_eq!(list[0].uri, "buffer://main.rs");
259
260 let contents = reg.read("buffer://main.rs").await.unwrap();
261 assert_eq!(contents.text.as_deref(), Some("fn main() {}"));
262
263 // Serialized read result matches the MCP `contents` array shape.
264 let v = read_result(contents);
265 assert_eq!(v["contents"][0]["uri"], "buffer://main.rs");
266 assert_eq!(v["contents"][0]["mimeType"], json!(null)); // not set on this content
267 assert_eq!(v["contents"][0]["text"], "fn main() {}");
268 }
269
270 #[tokio::test]
271 async fn read_unknown_uri_is_not_found() {
272 let reg = ResourceRegistry::new();
273 let err = reg.read("buffer://gone.rs").await.unwrap_err();
274 assert!(matches!(err, Error::ResourceNotFound(_)));
275 }
276
277 #[tokio::test]
278 async fn remove_is_observed_through_a_clone() {
279 let reg = ResourceRegistry::new();
280 let clone = reg.clone();
281 reg.register(
282 ResourceDescriptor::new("buffer://a", "a"),
283 |uri| async move { Ok(ResourceContents::text(uri, "x")) },
284 );
285 assert!(clone.contains("buffer://a"));
286 assert!(reg.remove("buffer://a"));
287 assert!(!clone.contains("buffer://a"));
288 assert!(!reg.remove("buffer://a")); // second remove reports absent
289 }
290
291 #[tokio::test]
292 async fn register_replaces_existing_uri() {
293 let reg = ResourceRegistry::new();
294 reg.register(
295 ResourceDescriptor::new("buffer://a", "a"),
296 |uri| async move { Ok(ResourceContents::text(uri, "old")) },
297 );
298 reg.register(
299 ResourceDescriptor::new("buffer://a", "a"),
300 |uri| async move { Ok(ResourceContents::text(uri, "new")) },
301 );
302 assert_eq!(reg.list().len(), 1);
303 assert_eq!(
304 reg.read("buffer://a").await.unwrap().text.as_deref(),
305 Some("new")
306 );
307 }
308 }
309