@ondeinference / ed / commits / 5f0a2a3

Add ed-mcp: MCP client glue over rmcp, with streamable HTTP

Server configuration for stdio and streamable HTTP (with an optional bearer token), tool namespacing, timeouts, progress and cancellation. The agent feature adds McpExecutor, which serves a toolset to Ed with risk taken from readOnlyHint plus a host override list. The acp feature maps ACP server lists. Tests run against an in-process rmcp HTTP server. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-Authored-By: siGit Code v1.6.3 <noreply@sigit.si>

Seto Elkahfi committed Oct 9, 2026 at 00:12 UTC 5f0a2a3b198b3e50f4555ce612e88fe786c997d2
8 files changed +1962 -1
Cargo.lock
+440
@@ -18,6 +18,56 @@ version = "2.0.1"
18 source = "registry+https://github.com/rust-lang/crates.io-index"
19 checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
20
21 +[[package]]
22 +name = "agent-client-protocol"
23 +version = "2.2.0"
24 +source = "registry+https://github.com/rust-lang/crates.io-index"
25 +checksum = "13f780597b2c88ab83d02509bcc3e86366751f86fa12114005e35c55ed691763"
26 +dependencies = [
27 + "agent-client-protocol-derive",
28 + "agent-client-protocol-schema",
29 + "async-io",
30 + "async-process",
31 + "blocking",
32 + "futures",
33 + "futures-concurrency",
34 + "rustc-hash 2.1.3",
35 + "rustix",
36 + "schemars 1.2.2",
37 + "serde",
38 + "serde_json",
39 + "shell-words",
40 + "tracing",
41 + "uuid 1.27.0",
42 + "windows-sys 0.61.2",
43 +]
44 +
45 +[[package]]
46 +name = "agent-client-protocol-derive"
47 +version = "2.2.0"
48 +source = "registry+https://github.com/rust-lang/crates.io-index"
49 +checksum = "f4893bfb11e0e193aa5a64611a126ee87567acb1879556519c902722899229de"
50 +dependencies = [
51 + "quote",
52 + "syn 3.0.6",
53 +]
54 +
55 +[[package]]
56 +name = "agent-client-protocol-schema"
57 +version = "1.9.1"
58 +source = "registry+https://github.com/rust-lang/crates.io-index"
59 +checksum = "1d595a79e1665d02c91c7dd1bedd118501e75cc0a3d5b1651b5150ec4c5c1e0d"
60 +dependencies = [
61 + "anyhow",
62 + "derive_more",
63 + "schemars 1.2.2",
64 + "serde",
65 + "serde_json",
66 + "serde_with",
67 + "strum 0.28.0",
68 + "tracing",
69 +]
70 +
71 [[package]]
72 name = "ahash"
73 version = "0.8.12"
@@ -272,6 +322,18 @@ dependencies = [
322 "winnow 0.7.15",
323 ]
324
325 +[[package]]
326 +name = "async-channel"
327 +version = "2.5.0"
328 +source = "registry+https://github.com/rust-lang/crates.io-index"
329 +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2"
330 +dependencies = [
331 + "concurrent-queue",
332 + "event-listener-strategy",
333 + "futures-core",
334 + "pin-project-lite",
335 +]
336 +
337 [[package]]
338 name = "async-compat"
339 version = "0.2.6"
@@ -285,6 +347,77 @@ dependencies = [
347 "tokio",
348 ]
349
350 +[[package]]
351 +name = "async-io"
352 +version = "2.6.0"
353 +source = "registry+https://github.com/rust-lang/crates.io-index"
354 +checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc"
355 +dependencies = [
356 + "autocfg",
357 + "cfg-if",
358 + "concurrent-queue",
359 + "futures-io",
360 + "futures-lite",
361 + "parking",
362 + "polling",
363 + "rustix",
364 + "slab",
365 + "windows-sys 0.61.2",
366 +]
367 +
368 +[[package]]
369 +name = "async-lock"
370 +version = "3.4.2"
371 +source = "registry+https://github.com/rust-lang/crates.io-index"
372 +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
373 +dependencies = [
374 + "event-listener",
375 + "event-listener-strategy",
376 + "pin-project-lite",
377 +]
378 +
379 +[[package]]
380 +name = "async-process"
381 +version = "2.5.0"
382 +source = "registry+https://github.com/rust-lang/crates.io-index"
383 +checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75"
384 +dependencies = [
385 + "async-channel",
386 + "async-io",
387 + "async-lock",
388 + "async-signal",
389 + "async-task",
390 + "blocking",
391 + "cfg-if",
392 + "event-listener",
393 + "futures-lite",
394 + "rustix",
395 +]
396 +
397 +[[package]]
398 +name = "async-signal"
399 +version = "0.2.14"
400 +source = "registry+https://github.com/rust-lang/crates.io-index"
401 +checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485"
402 +dependencies = [
403 + "async-io",
404 + "async-lock",
405 + "atomic-waker",
406 + "cfg-if",
407 + "futures-core",
408 + "futures-io",
409 + "rustix",
410 + "signal-hook-registry",
411 + "slab",
412 + "windows-sys 0.61.2",
413 +]
414 +
415 +[[package]]
416 +name = "async-task"
417 +version = "4.7.1"
418 +source = "registry+https://github.com/rust-lang/crates.io-index"
419 +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
420 +
421 [[package]]
422 name = "async-trait"
423 version = "0.1.92"
@@ -408,6 +541,58 @@ dependencies = [
541 "pkg-config",
542 ]
543
544 +[[package]]
545 +name = "axum"
546 +version = "0.8.9"
547 +source = "registry+https://github.com/rust-lang/crates.io-index"
548 +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
549 +dependencies = [
550 + "axum-core",
551 + "bytes",
552 + "form_urlencoded",
553 + "futures-util",
554 + "http",
555 + "http-body",
556 + "http-body-util",
557 + "hyper",
558 + "hyper-util",
559 + "itoa",
560 + "matchit",
561 + "memchr",
562 + "mime",
563 + "percent-encoding",
564 + "pin-project-lite",
565 + "serde_core",
566 + "serde_json",
567 + "serde_path_to_error",
568 + "serde_urlencoded",
569 + "sync_wrapper",
570 + "tokio",
571 + "tower",
572 + "tower-layer",
573 + "tower-service",
574 + "tracing",
575 +]
576 +
577 +[[package]]
578 +name = "axum-core"
579 +version = "0.5.6"
580 +source = "registry+https://github.com/rust-lang/crates.io-index"
581 +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
582 +dependencies = [
583 + "bytes",
584 + "futures-core",
585 + "http",
586 + "http-body",
587 + "http-body-util",
588 + "mime",
589 + "pin-project-lite",
590 + "sync_wrapper",
591 + "tower-layer",
592 + "tower-service",
593 + "tracing",
594 +]
595 +
596 [[package]]
597 name = "base64"
598 version = "0.13.1"
@@ -525,6 +710,19 @@ dependencies = [
710 "objc2",
711 ]
712
713 +[[package]]
714 +name = "blocking"
715 +version = "1.7.0"
716 +source = "registry+https://github.com/rust-lang/crates.io-index"
717 +checksum = "a70e4329df6cb94385eed412ec92375c3cdd8a6e502493d1229b6414e4036dfa"
718 +dependencies = [
719 + "async-channel",
720 + "async-task",
721 + "futures-io",
722 + "futures-lite",
723 + "piper",
724 +]
725 +
726 [[package]]
727 name = "bm25"
728 version = "2.3.2"
@@ -961,6 +1159,15 @@ dependencies = [
1159 "static_assertions",
1160 ]
1161
1162 +[[package]]
1163 +name = "concurrent-queue"
1164 +version = "2.5.0"
1165 +source = "registry+https://github.com/rust-lang/crates.io-index"
1166 +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
1167 +dependencies = [
1168 + "crossbeam-utils",
1169 +]
1170 +
1171 [[package]]
1172 name = "console"
1173 version = "0.15.11"
@@ -1530,6 +1737,7 @@ dependencies = [
1737 "quote",
1738 "rustc_version",
1739 "syn 2.0.119",
1740 + "unicode-xid",
1741 ]
1742
1743 [[package]]
@@ -1769,6 +1977,25 @@ dependencies = [
1977 "uuid 1.27.0",
1978 ]
1979
1980 +[[package]]
1981 +name = "ed-mcp"
1982 +version = "1.0.1"
1983 +dependencies = [
1984 + "agent-client-protocol",
1985 + "anyhow",
1986 + "async-trait",
1987 + "axum",
1988 + "ed-agent",
1989 + "ed-mcp",
1990 + "futures",
1991 + "http",
1992 + "rmcp",
1993 + "serde_json",
1994 + "tokio",
1995 + "tokio-util",
1996 + "tracing",
1997 +]
1998 +
1999 [[package]]
2000 name = "ed-uniffi-bindgen"
2001 version = "0.1.0"
@@ -1931,6 +2158,26 @@ version = "0.1.10"
2158 source = "registry+https://github.com/rust-lang/crates.io-index"
2159 checksum = "d817e038c30374a4bcb22f94d0a8a0e216958d4c3dcde369b1439fec4bdda6e6"
2160
2161 +[[package]]
2162 +name = "event-listener"
2163 +version = "5.4.2"
2164 +source = "registry+https://github.com/rust-lang/crates.io-index"
2165 +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
2166 +dependencies = [
2167 + "parking",
2168 + "pin-project-lite",
2169 +]
2170 +
2171 +[[package]]
2172 +name = "event-listener-strategy"
2173 +version = "0.5.4"
2174 +source = "registry+https://github.com/rust-lang/crates.io-index"
2175 +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
2176 +dependencies = [
2177 + "event-listener",
2178 + "pin-project-lite",
2179 +]
2180 +
2181 [[package]]
2182 name = "exr"
2183 version = "1.74.2"
@@ -2013,6 +2260,12 @@ version = "0.1.14"
2260 source = "registry+https://github.com/rust-lang/crates.io-index"
2261 checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484"
2262
2263 +[[package]]
2264 +name = "fixedbitset"
2265 +version = "0.5.7"
2266 +source = "registry+https://github.com/rust-lang/crates.io-index"
2267 +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
2268 +
2269 [[package]]
2270 name = "flate2"
2271 version = "1.1.10"
@@ -2140,6 +2393,19 @@ dependencies = [
2393 "futures-sink",
2394 ]
2395
2396 +[[package]]
2397 +name = "futures-concurrency"
2398 +version = "7.7.1"
2399 +source = "registry+https://github.com/rust-lang/crates.io-index"
2400 +checksum = "175cd8cca9e1d45b87f18ffa75088f2099e3c4fe5e2f83e42de112560bea8ea6"
2401 +dependencies = [
2402 + "fixedbitset",
2403 + "futures-core",
2404 + "futures-lite",
2405 + "pin-project",
2406 + "smallvec 1.16.2",
2407 +]
2408 +
2409 [[package]]
2410 name = "futures-core"
2411 version = "0.3.34"
@@ -2163,6 +2429,19 @@ version = "0.3.34"
2429 source = "registry+https://github.com/rust-lang/crates.io-index"
2430 checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
2431
2432 +[[package]]
2433 +name = "futures-lite"
2434 +version = "2.6.1"
2435 +source = "registry+https://github.com/rust-lang/crates.io-index"
2436 +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad"
2437 +dependencies = [
2438 + "fastrand",
2439 + "futures-core",
2440 + "futures-io",
2441 + "parking",
2442 + "pin-project-lite",
2443 +]
2444 +
2445 [[package]]
2446 name = "futures-macro"
2447 version = "0.3.34"
@@ -3032,6 +3311,12 @@ version = "1.10.1"
3311 source = "registry+https://github.com/rust-lang/crates.io-index"
3312 checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
3313
3314 +[[package]]
3315 +name = "httpdate"
3316 +version = "1.0.3"
3317 +source = "registry+https://github.com/rust-lang/crates.io-index"
3318 +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
3319 +
3320 [[package]]
3321 name = "hyper"
3322 version = "1.11.1"
@@ -3046,6 +3331,7 @@ dependencies = [
3331 "http",
3332 "http-body",
3333 "httparse",
3334 + "httpdate",
3335 "itoa",
3336 "pin-project-lite",
3337 "smallvec 1.16.2",
@@ -3871,6 +4157,12 @@ dependencies = [
4157 "regex-automata",
4158 ]
4159
4160 +[[package]]
4161 +name = "matchit"
4162 +version = "0.8.4"
4163 +source = "registry+https://github.com/rust-lang/crates.io-index"
4164 +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
4165 +
4166 [[package]]
4167 name = "matrixmultiply"
4168 version = "0.3.11"
@@ -5045,6 +5337,12 @@ dependencies = [
5337 "system-deps",
5338 ]
5339
5340 +[[package]]
5341 +name = "parking"
5342 +version = "2.2.1"
5343 +source = "registry+https://github.com/rust-lang/crates.io-index"
5344 +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
5345 +
5346 [[package]]
5347 name = "parking_lot"
5348 version = "0.12.5"
@@ -5145,12 +5443,43 @@ dependencies = [
5443 "siphasher",
5444 ]
5445
5446 +[[package]]
5447 +name = "pin-project"
5448 +version = "1.1.13"
5449 +source = "registry+https://github.com/rust-lang/crates.io-index"
5450 +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924"
5451 +dependencies = [
5452 + "pin-project-internal",
5453 +]
5454 +
5455 +[[package]]
5456 +name = "pin-project-internal"
5457 +version = "1.1.13"
5458 +source = "registry+https://github.com/rust-lang/crates.io-index"
5459 +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b"
5460 +dependencies = [
5461 + "proc-macro2",
5462 + "quote",
5463 + "syn 2.0.119",
5464 +]
5465 +
5466 [[package]]
5467 name = "pin-project-lite"
5468 version = "0.2.17"
5469 source = "registry+https://github.com/rust-lang/crates.io-index"
5470 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
5471
5472 +[[package]]
5473 +name = "piper"
5474 +version = "0.2.5"
5475 +source = "registry+https://github.com/rust-lang/crates.io-index"
5476 +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1"
5477 +dependencies = [
5478 + "atomic-waker",
5479 + "fastrand",
5480 + "futures-io",
5481 +]
5482 +
5483 [[package]]
5484 name = "pkg-config"
5485 version = "0.3.34"
@@ -5202,6 +5531,20 @@ dependencies = [
5531 "miniz_oxide 0.8.9",
5532 ]
5533
5534 +[[package]]
5535 +name = "polling"
5536 +version = "3.11.0"
5537 +source = "registry+https://github.com/rust-lang/crates.io-index"
5538 +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218"
5539 +dependencies = [
5540 + "cfg-if",
5541 + "concurrent-queue",
5542 + "hermit-abi 0.5.3",
5543 + "pin-project-lite",
5544 + "rustix",
5545 + "windows-sys 0.61.2",
5546 +]
5547 +
5548 [[package]]
5549 name = "portable-atomic"
5550 version = "1.15.0"
@@ -5318,6 +5661,20 @@ dependencies = [
5661 "unicode-ident",
5662 ]
5663
5664 +[[package]]
5665 +name = "process-wrap"
5666 +version = "10.0.1"
5667 +source = "registry+https://github.com/rust-lang/crates.io-index"
5668 +checksum = "1f21b97672d2dc848e7b25701ab4535618b92f4861c13cc3f7f7bed52ad3c8da"
5669 +dependencies = [
5670 + "futures",
5671 + "indexmap 2.14.2",
5672 + "nix",
5673 + "tokio",
5674 + "tracing",
5675 + "windows 0.62.2",
5676 +]
5677 +
5678 [[package]]
5679 name = "profiling"
5680 version = "1.0.18"
@@ -5908,6 +6265,39 @@ dependencies = [
6265 "windows-sys 0.52.0",
6266 ]
6267
6268 +[[package]]
6269 +name = "rmcp"
6270 +version = "3.5.1"
6271 +source = "registry+https://github.com/rust-lang/crates.io-index"
6272 +checksum = "ab5023804cbe0304cbe06197ac16e645a0a0d0ffa2953bf735c15c2fa0213350"
6273 +dependencies = [
6274 + "async-trait",
6275 + "base64 0.23.1",
6276 + "bytes",
6277 + "chrono",
6278 + "futures",
6279 + "http",
6280 + "http-body",
6281 + "http-body-util",
6282 + "indexmap 2.14.2",
6283 + "pastey 0.2.3",
6284 + "pin-project-lite",
6285 + "process-wrap",
6286 + "rand 0.10.3",
6287 + "reqwest 0.13.5",
6288 + "schemars 1.2.2",
6289 + "serde",
6290 + "serde_json",
6291 + "sse-stream",
6292 + "thiserror 2.0.21",
6293 + "tokio",
6294 + "tokio-stream",
6295 + "tokio-util",
6296 + "tower-service",
6297 + "tracing",
6298 + "uuid 1.27.0",
6299 +]
6300 +
6301 [[package]]
6302 name = "rubato"
6303 version = "0.16.2"
@@ -6171,6 +6561,7 @@ version = "1.2.2"
6561 source = "registry+https://github.com/rust-lang/crates.io-index"
6562 checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"
6563 dependencies = [
6564 + "chrono",
6565 "dyn-clone",
6566 "ref-cast",
6567 "schemars_derive 1.2.2",
@@ -6442,6 +6833,17 @@ dependencies = [
6833 "zmij",
6834 ]
6835
6836 +[[package]]
6837 +name = "serde_path_to_error"
6838 +version = "0.1.20"
6839 +source = "registry+https://github.com/rust-lang/crates.io-index"
6840 +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
6841 +dependencies = [
6842 + "itoa",
6843 + "serde",
6844 + "serde_core",
6845 +]
6846 +
6847 [[package]]
6848 name = "serde_plain"
6849 version = "1.0.2"
@@ -6587,6 +6989,12 @@ dependencies = [
6989 "lazy_static",
6990 ]
6991
6992 +[[package]]
6993 +name = "shell-words"
6994 +version = "1.1.1"
6995 +source = "registry+https://github.com/rust-lang/crates.io-index"
6996 +checksum = "dc6fe69c597f9c37bfeeeeeb33da3530379845f10be461a66d16d03eca2ded77"
6997 +
6998 [[package]]
6999 name = "shlex"
7000 version = "2.0.1"
@@ -6850,6 +7258,19 @@ dependencies = [
7258 "unicode-segmentation",
7259 ]
7260
7261 +[[package]]
7262 +name = "sse-stream"
7263 +version = "0.2.6"
7264 +source = "registry+https://github.com/rust-lang/crates.io-index"
7265 +checksum = "c25ac7aff0abd1dbc474536e40416e1102c7dd9bfba0b9861c6d357f835dcfb4"
7266 +dependencies = [
7267 + "bytes",
7268 + "futures-util",
7269 + "http-body",
7270 + "http-body-util",
7271 + "pin-project-lite",
7272 +]
7273 +
7274 [[package]]
7275 name = "stable_deref_trait"
7276 version = "1.2.1"
@@ -7860,6 +8281,17 @@ dependencies = [
8281 "tokio",
8282 ]
8283
8284 +[[package]]
8285 +name = "tokio-stream"
8286 +version = "0.1.19"
8287 +source = "registry+https://github.com/rust-lang/crates.io-index"
8288 +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b"
8289 +dependencies = [
8290 + "futures-core",
8291 + "pin-project-lite",
8292 + "tokio",
8293 +]
8294 +
8295 [[package]]
8296 name = "tokio-tungstenite"
8297 version = "0.28.0"
@@ -8046,6 +8478,7 @@ dependencies = [
8478 "tokio",
8479 "tower-layer",
8480 "tower-service",
8481 + "tracing",
8482 ]
8483
8484 [[package]]
@@ -8084,6 +8517,7 @@ version = "0.1.44"
8517 source = "registry+https://github.com/rust-lang/crates.io-index"
8518 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
8519 dependencies = [
8520 + "log",
8521 "pin-project-lite",
8522 "tracing-attributes",
8523 "tracing-core",
@@ -8314,6 +8748,12 @@ version = "0.2.2"
8748 source = "registry+https://github.com/rust-lang/crates.io-index"
8749 checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254"
8750
8751 +[[package]]
8752 +name = "unicode-xid"
8753 +version = "0.2.6"
8754 +source = "registry+https://github.com/rust-lang/crates.io-index"
8755 +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
8756 +
8757 [[package]]
8758 name = "unicode_categories"
8759 version = "0.1.1"
Cargo.toml
+23 -1
@@ -1,6 +1,12 @@
1 [workspace]
2 resolver = "2"
3 -members = ["crates/ed-agent", "crates/ed-agent-ffi", "crates/ed-agent-tauri", "uniffi-bindgen"]
3 +members = [
4 + "crates/ed-agent",
5 + "crates/ed-agent-ffi",
6 + "crates/ed-agent-tauri",
7 + "crates/ed-mcp",
8 + "uniffi-bindgen",
9 +]
10
11 # Shared across every crate in the workspace so a release moves one number.
12 # `sync-version` tooling (see the Makefile) reads these.
@@ -30,10 +36,26 @@ tauri = { version = "2", features = [] }
36 # versions break extraction with "Invalid string data" errors.
37 uniffi = { version = "=0.31.0", features = ["tokio"] }
38 smbcloud-gresiq-sdk = "0.4.7"
39 +
40 +# Onde Agent Platform v0.1 baseline (onde-web .agents/specs/onde-agent-platform, §5.4).
41 +# The ACP crates (ed-mcp, ed-acp, ed-acp-tui, ed-acp-testkit) use these; ed-agent does not.
42 +agent-client-protocol = "2.2"
43 +rmcp = { version = "3.5", default-features = false }
44 +reqwest = { version = "0.13", default-features = false, features = ["rustls", "json", "stream"] }
45 +ratatui = { version = "0.30.2", default-features = false, features = ["crossterm", "layout-cache", "unstable-rendered-line-info"] }
46 +crossterm = { version = "0.29", features = ["event-stream"] }
47 +tokio-util = "0.7"
48 +futures = "0.3"
49 +anyhow = "1"
50 +tracing = "0.1"
51 +tracing-subscriber = { version = "0.3", features = ["env-filter"] }
52 uuid = { version = "1", features = ["v4"] }
53 +directories = "6"
54 +base64 = "0.23"
55
56 # Internal
57 ed-agent = { version = "1.0.1", path = "crates/ed-agent" }
58 +ed-mcp = { version = "1.0.1", path = "crates/ed-mcp" }
59
60 # gemm-common (via candle, via onde) depends on sysctl for every
61 # `target_vendor = "apple"`, but sysctl 0.6.0 only has a `sys` module behind
crates/ed-mcp/Cargo.toml new
+42
@@ -0,0 +1,42 @@
1 +[package]
2 +name = "ed-mcp"
3 +description = "MCP for Onde agents: server configuration, tool namespacing, timeouts, progress and cancellation over rmcp, the official Rust SDK."
4 +keywords = ["mcp", "llm", "agent", "onde"]
5 +categories = ["asynchronous"]
6 +version.workspace = true
7 +edition = "2024"
8 +rust-version.workspace = true
9 +license.workspace = true
10 +authors.workspace = true
11 +repository.workspace = true
12 +homepage.workspace = true
13 +
14 +[dependencies]
15 +agent-client-protocol = { workspace = true, optional = true }
16 +ed-agent = { workspace = true, optional = true }
17 +async-trait = { workspace = true, optional = true }
18 +http = "1"
19 +rmcp = { workspace = true, features = ["client", "transport-child-process", "transport-streamable-http-client-reqwest", "reqwest"] }
20 +tokio = { workspace = true, features = ["sync", "time", "process", "macros", "io-std", "io-util", "rt"] }
21 +tokio-util.workspace = true
22 +futures.workspace = true
23 +serde_json.workspace = true
24 +anyhow.workspace = true
25 +tracing.workspace = true
26 +
27 +[features]
28 +# Map ACP `session/new` MCP server lists (`McpToolset::connect`, `ServerSpec::from_acp`).
29 +# Pulls in `agent-client-protocol`; `ed-acp` enables it.
30 +acp = ["dep:agent-client-protocol"]
31 +# `McpExecutor`: an `McpToolset` as Ed tools. Pulls in `ed-agent` (and with it `onde`).
32 +agent = ["dep:ed-agent", "dep:async-trait"]
33 +# A scriptable MCP server for tests: `echo`, `slow_progress`, `wait_for_cancel`, and the
34 +# demucs-shaped `list_models` / `separate_stems`.
35 +test-server = []
36 +
37 +[dev-dependencies]
38 +rmcp = { workspace = true, features = ["server", "transport-streamable-http-server"] }
39 +axum = "0.8"
40 +tokio = { workspace = true, features = ["net", "rt-multi-thread"] }
41 +# Run the feature-gated code's tests under a plain `cargo test`.
42 +ed-mcp = { path = ".", features = ["acp", "agent", "test-server"] }
crates/ed-mcp/README.md new
+33
@@ -0,0 +1,33 @@
1 +# ed-mcp
2 +
3 +MCP client glue over [`rmcp`](https://crates.io/crates/rmcp), the official Rust SDK, for
4 +[Ed](https://crates.io/crates/ed-agent) agents.
5 +
6 +`rmcp` is the client. This crate owns what every agent was writing around it: server
7 +configuration (stdio child processes and streamable HTTP with an optional bearer token),
8 +`mcp__<server>__<tool>` namespacing, per-server timeouts, progress and cancellation, and read-only
9 +gating from `readOnlyHint`. A server that fails to start is logged and left out.
10 +
11 +```rust
12 +use ed_mcp::{ClientInfo, McpToolset, ServerSpec, ToolsetConfig};
13 +
14 +let spec = ServerSpec::http("mail", "https://api.example.com/v1/mcp").with_bearer(token);
15 +let tools = McpToolset::connect_specs(
16 + ToolsetConfig::new(ClientInfo::new("my-app", "1.0")),
17 + &[spec],
18 + &[],
19 +)
20 +.await;
21 +```
22 +
23 +## Features
24 +
25 +- `acp`: map `McpServer` lists from an Agent Client Protocol `session/new` request.
26 +- `agent`: `McpExecutor` serves a toolset to Ed as tools, with risk from `readOnlyHint` plus a
27 + host override list, and optional unprefixed names for single-server apps.
28 +- `test-server`: a scriptable MCP server for tests.
29 +
30 +## Copyright
31 +
32 +© 2026 [Splitfire AB](https://5mb.app) ([Onde Inference](https://ondeinference.com)).
33 +Licensed under MIT or Apache-2.0.
crates/ed-mcp/src/executor.rs new
+199
@@ -0,0 +1,199 @@
1 +//! An [`McpToolset`] as [Ed](ed_agent) tools. Behind the `agent` feature.
2 +
3 +use std::collections::{HashMap, HashSet};
4 +use std::sync::Arc;
5 +
6 +use async_trait::async_trait;
7 +use ed_agent::{AgentToolDefinition, Ed, EventSink, ToolExecutor, ToolRisk};
8 +use serde_json::Value;
9 +use tokio::sync::{Mutex, RwLock};
10 +use tokio_util::sync::CancellationToken;
11 +
12 +use crate::{McpTool, McpToolset};
13 +
14 +/// How MCP tools are presented to Ed.
15 +#[derive(Debug, Clone, Default)]
16 +pub struct ToolPolicy {
17 + read_only: HashSet<String>,
18 + unprefixed: bool,
19 +}
20 +
21 +impl ToolPolicy {
22 + /// Default policy: tool names are `mcp__<server>__<tool>`, and a tool is read-only only when
23 + /// its server sets `readOnlyHint: true`.
24 + pub fn new() -> Self {
25 + Self::default()
26 + }
27 +
28 + /// Treat these tools as read-only (no approval) whatever their server says. Names are the
29 + /// server's own tool names, or the names the model sees. For servers that do not set
30 + /// `readOnlyHint`.
31 + pub fn read_only(mut self, names: &[&str]) -> Self {
32 + self.read_only.extend(names.iter().map(|n| (*n).to_owned()));
33 + self
34 + }
35 +
36 + /// Show the model the server's own tool names, without the `mcp__<server>__` prefix. For apps
37 + /// with one server whose tool names are already in the system prompt. If two servers use the
38 + /// same tool name the first one wins.
39 + pub fn unprefixed(mut self) -> Self {
40 + self.unprefixed = true;
41 + self
42 + }
43 +
44 + fn risk(&self, tool: &McpTool, exposed: &str) -> ToolRisk {
45 + if tool.read_only || self.read_only.contains(exposed) || self.read_only.contains(&tool.tool)
46 + {
47 + ToolRisk::ReadOnly
48 + } else {
49 + ToolRisk::Mutating
50 + }
51 + }
52 +}
53 +
54 +struct Loaded {
55 + toolset: Arc<McpToolset>,
56 + /// Name the model sees -> qualified name in the toolset.
57 + names: HashMap<String, String>,
58 + definitions: Vec<AgentToolDefinition>,
59 +}
60 +
61 +/// Runs the tools of an [`McpToolset`] for Ed.
62 +///
63 +/// Cheap to clone. Hand one clone to [`Ed::with_agent`] as the executor, and keep another to
64 +/// connect and disconnect servers as the user signs in and out:
65 +///
66 +/// ```ignore
67 +/// let mcp = McpExecutor::new(ToolPolicy::new().read_only(&["list_messages"]).unprefixed());
68 +/// let ed = Ed::with_agent(sink, Arc::new(mcp.clone()), approvals, config);
69 +/// mcp.set_toolset(Some(toolset));
70 +/// mcp.sync(&ed).await;
71 +/// ```
72 +#[derive(Clone)]
73 +pub struct McpExecutor {
74 + policy: Arc<ToolPolicy>,
75 + loaded: Arc<RwLock<Option<Loaded>>>,
76 + registered: Arc<Mutex<Vec<String>>>,
77 +}
78 +
79 +impl McpExecutor {
80 + pub fn new(policy: ToolPolicy) -> Self {
81 + Self {
82 + policy: Arc::new(policy),
83 + loaded: Arc::default(),
84 + registered: Arc::default(),
85 + }
86 + }
87 +
88 + /// Use `toolset`, or none, and shut the previous toolset down. Call
89 + /// [`sync`](McpExecutor::sync) afterwards. A tool call still running on the old toolset
90 + /// finishes first; its connections close when it does.
91 + pub async fn set_toolset(&self, toolset: Option<McpToolset>) {
92 + let new = toolset.map(|toolset| {
93 + let mut names = HashMap::new();
94 + let mut definitions = Vec::new();
95 + for tool in toolset.tools() {
96 + let exposed = if self.policy.unprefixed {
97 + tool.tool.clone()
98 + } else {
99 + tool.name.clone()
100 + };
101 + if names.contains_key(&exposed) {
102 + tracing::warn!(
103 + "duplicate tool name `{exposed}` after prefix removal; keeping the first"
104 + );
105 + continue;
106 + }
107 + let description = if self.policy.unprefixed {
108 + tool.description
109 + .strip_prefix(&format!("[{} MCP server] ", tool.server))
110 + .unwrap_or(&tool.description)
111 + .to_owned()
112 + } else {
113 + tool.description.clone()
114 + };
115 + let description = if description.trim().is_empty() {
116 + format!("The `{}` tool.", tool.tool)
117 + } else {
118 + description
119 + };
120 + let schema = tool.parameters.to_string();
121 + definitions.push(match self.policy.risk(tool, &exposed) {
122 + ToolRisk::ReadOnly => {
123 + AgentToolDefinition::read_only(exposed.clone(), description, schema)
124 + }
125 + ToolRisk::Mutating => {
126 + AgentToolDefinition::mutating(exposed.clone(), description, schema)
127 + }
128 + });
129 + names.insert(exposed, tool.name.clone());
130 + }
131 + Loaded {
132 + toolset: Arc::new(toolset),
133 + names,
134 + definitions,
135 + }
136 + });
137 + let old = std::mem::replace(&mut *self.loaded.write().await, new);
138 + if let Some(old) = old
139 + && let Some(toolset) = Arc::into_inner(old.toolset)
140 + {
141 + toolset.shutdown().await;
142 + }
143 + }
144 +
145 + /// The tools Ed should offer, for the connected toolset (empty when none).
146 + pub async fn definitions(&self) -> Vec<AgentToolDefinition> {
147 + self.loaded
148 + .read()
149 + .await
150 + .as_ref()
151 + .map(|l| l.definitions.clone())
152 + .unwrap_or_default()
153 + }
154 +
155 + /// Make `ed`'s registered tools match the connected toolset: the ones this executor registered
156 + /// before are removed, the current ones added.
157 + pub async fn sync<S: EventSink>(&self, ed: &Ed<S>) {
158 + let mut registered = self.registered.lock().await;
159 + for name in registered.drain(..) {
160 + ed.unregister_tool(&name).await;
161 + }
162 + for definition in self.definitions().await {
163 + let name = definition.name.clone();
164 + match ed.register_tool(definition).await {
165 + Ok(()) => registered.push(name),
166 + Err(error) => tracing::warn!("could not register MCP tool `{name}`: {error}"),
167 + }
168 + }
169 + }
170 +}
171 +
172 +#[async_trait]
173 +impl ToolExecutor for McpExecutor {
174 + async fn execute(&self, tool_name: &str, arguments: &str) -> Result<String, String> {
175 + let args: Value =
176 + serde_json::from_str(arguments).map_err(|e| format!("bad arguments: {e}"))?;
177 + // Clone out of the lock: a call can run for a long time and must not hold up a
178 + // disconnect.
179 + let (toolset, qualified) = {
180 + let guard = self.loaded.read().await;
181 + let loaded = guard.as_ref().ok_or("no MCP server is connected")?;
182 + let qualified = loaded
183 + .names
184 + .get(tool_name)
185 + .ok_or_else(|| format!("unknown tool `{tool_name}`"))?
186 + .clone();
187 + (loaded.toolset.clone(), qualified)
188 + };
189 + let outcome = toolset
190 + .call(&qualified, args, &CancellationToken::new(), &|_| {})
191 + .await
192 + .map_err(|e| e.to_string())?;
193 + if outcome.failed {
194 + Err(outcome.text)
195 + } else {
196 + Ok(outcome.text)
197 + }
198 + }
199 +}
crates/ed-mcp/src/lib.rs new
+821
@@ -0,0 +1,821 @@
1 +//! MCP for Onde agents, over [`rmcp`], the official Rust SDK.
2 +//!
3 +//! This crate is not an MCP client: `rmcp` is. It owns what every Onde agent was writing for
4 +//! itself around one:
5 +//!
6 +//! - **Server configuration.** Stdio servers supplied by an ACP client in
7 +//! `session/new|load|resume`, plus process-wide built-in servers ([`SharedServer`]) that start
8 +//! at most once, on first need. A client-supplied server with the same name as a built-in wins.
9 +//! - **Tool namespacing.** Tools are exposed to the model as `mcp__<server>__<tool>`, sanitised
10 +//! and capped at [`MAX_TOOL_NAME`] bytes.
11 +//! - **Read-only gating.** A tool is read-only only when its server sets `readOnlyHint: true`.
12 +//! - **Calls.** Progress notifications are handed to a callback, session cancellation is
13 +//! forwarded as `notifications/cancelled`, and every call has a deadline.
14 +//!
15 +//! A server that fails to start or to list its tools is never fatal: it is logged and its tools
16 +//! are absent.
17 +//!
18 +//! Servers are reached over stdio (a child process) or streamable HTTP with an optional bearer
19 +//! token, described by a [`ServerSpec`].
20 +//!
21 +//! Features: `acp` maps ACP `McpServer` lists, `agent` adds [`McpExecutor`] to expose a toolset
22 +//! as [Ed](ed_agent) tools, and `test-server` adds a scriptable server for tests.
23 +
24 +use std::collections::HashMap;
25 +use std::path::{Path, PathBuf};
26 +use std::sync::Arc;
27 +use std::time::Duration;
28 +
29 +use anyhow::{Context, Result, anyhow, bail};
30 +use futures::StreamExt;
31 +use rmcp::handler::client::progress::ProgressDispatcher;
32 +use rmcp::model::{
33 + CallToolRequestParams, ClientCapabilities, ClientRequest, ContentBlock, Implementation,
34 + ProgressNotificationParam, Request, ServerResult, Tool,
35 +};
36 +use rmcp::service::{NotificationContext, PeerRequestOptions, RunningService};
37 +use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
38 +use rmcp::transport::{StreamableHttpClientTransport, TokioChildProcess};
39 +use rmcp::{ClientHandler, RoleClient, ServiceExt};
40 +use serde_json::{Map, Value, json};
41 +use tokio::sync::Mutex;
42 +use tokio_util::sync::CancellationToken;
43 +
44 +pub use rmcp;
45 +
46 +#[cfg(feature = "agent")]
47 +mod executor;
48 +#[cfg(feature = "agent")]
49 +pub use executor::{McpExecutor, ToolPolicy};
50 +
51 +#[cfg(feature = "test-server")]
52 +pub mod test_server;
53 +
54 +/// Prefix of every MCP tool name the model sees.
55 +pub const PREFIX: &str = "mcp__";
56 +/// Longest tool name sent to the model; OpenAI-compatible endpoints reject longer ones.
57 +pub const MAX_TOOL_NAME: usize = 64;
58 +/// Time a server has to finish the `initialize` handshake and answer `tools/list`.
59 +pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
60 +/// Upper bound for one tool call. Long, since some tools (stem separation) take many minutes.
61 +pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(60 * 60);
62 +
63 +/// Split `mcp__<server>__<tool>` into its parts.
64 +pub fn split_qualified(name: &str) -> Option<(&str, &str)> {
65 + let rest = name.strip_prefix(PREFIX)?;
66 + let (server, tool) = rest.split_once("__")?;
67 + (!server.is_empty() && !tool.is_empty()).then_some((server, tool))
68 +}
69 +
70 +/// The name the model sees for `tool` on `server`.
71 +pub fn qualified(server: &str, tool: &str) -> String {
72 + let mut name = format!("{PREFIX}{}__{}", sanitize(server), sanitize(tool));
73 + name.truncate(MAX_TOOL_NAME);
74 + name
75 +}
76 +
77 +fn sanitize(part: &str) -> String {
78 + let mut out: String = part
79 + .chars()
80 + .map(|c| {
81 + if c.is_ascii_alphanumeric() || c == '-' {
82 + c
83 + } else {
84 + '_'
85 + }
86 + })
87 + .collect();
88 + while out.contains("__") {
89 + out = out.replace("__", "_");
90 + }
91 + out.trim_matches('_').to_string()
92 +}
93 +
94 +/// Who is connecting: sent to every server in `initialize` as `clientInfo`.
95 +#[derive(Debug, Clone)]
96 +pub struct ClientInfo {
97 + pub name: String,
98 + pub version: String,
99 +}
100 +
101 +impl ClientInfo {
102 + pub fn new(name: impl Into<String>, version: impl Into<String>) -> Self {
103 + Self {
104 + name: name.into(),
105 + version: version.into(),
106 + }
107 + }
108 +}
109 +
110 +/// How to reach a server.
111 +#[derive(Clone)]
112 +pub enum Transport {
113 + /// Start `command args…` with extra environment variables and talk over its stdio.
114 + Stdio {
115 + command: PathBuf,
116 + args: Vec<String>,
117 + env: Vec<(String, String)>,
118 + },
119 + /// Streamable HTTP. `bearer` is the token without the `Bearer ` prefix.
120 + Http {
121 + url: String,
122 + bearer: Option<String>,
123 + headers: Vec<(String, String)>,
124 + },
125 +}
126 +
127 +impl std::fmt::Debug for Transport {
128 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129 + match self {
130 + Self::Stdio { command, args, .. } => f
131 + .debug_struct("Stdio")
132 + .field("command", command)
133 + .field("args", args)
134 + .finish_non_exhaustive(),
135 + // The token and header values are credentials; never print them.
136 + Self::Http { url, .. } => f
137 + .debug_struct("Http")
138 + .field("url", url)
139 + .finish_non_exhaustive(),
140 + }
141 + }
142 +}
143 +
144 +/// A server to connect to.
145 +#[derive(Debug, Clone)]
146 +pub struct ServerSpec {
147 + pub name: String,
148 + pub transport: Transport,
149 +}
150 +
151 +impl ServerSpec {
152 + /// A stdio server.
153 + pub fn stdio(
154 + name: impl Into<String>,
155 + command: impl Into<PathBuf>,
156 + args: Vec<String>,
157 + env: Vec<(String, String)>,
158 + ) -> Self {
159 + Self {
160 + name: name.into(),
161 + transport: Transport::Stdio {
162 + command: command.into(),
163 + args,
164 + env,
165 + },
166 + }
167 + }
168 +
169 + /// A streamable HTTP server.
170 + pub fn http(name: impl Into<String>, url: impl Into<String>) -> Self {
171 + Self {
172 + name: name.into(),
173 + transport: Transport::Http {
174 + url: url.into(),
175 + bearer: None,
176 + headers: Vec::new(),
177 + },
178 + }
179 + }
180 +
181 + /// Send `Authorization: Bearer <token>`. No effect on a stdio server.
182 + pub fn with_bearer(mut self, token: impl Into<String>) -> Self {
183 + if let Transport::Http { bearer, .. } = &mut self.transport {
184 + *bearer = Some(token.into());
185 + }
186 + self
187 + }
188 +
189 + /// Send an extra header on every request. No effect on a stdio server.
190 + pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
191 + if let Transport::Http { headers, .. } = &mut self.transport {
192 + headers.push((name.into(), value.into()));
193 + }
194 + self
195 + }
196 +
197 + /// The spec for an ACP-supplied server.
198 + #[cfg(feature = "acp")]
199 + pub fn from_acp(server: &agent_client_protocol::schema::v1::McpServer) -> Option<Self> {
200 + use agent_client_protocol::schema::v1::McpServer;
201 + match server {
202 + McpServer::Stdio(s) => Some(Self::stdio(
203 + s.name.clone(),
204 + s.command.clone(),
205 + s.args.clone(),
206 + s.env
207 + .iter()
208 + .map(|e| (e.name.clone(), e.value.clone()))
209 + .collect(),
210 + )),
211 + McpServer::Http(h) => Some(Self {
212 + name: h.name.clone(),
213 + transport: Transport::Http {
214 + url: h.url.clone(),
215 + bearer: None,
216 + headers: h
217 + .headers
218 + .iter()
219 + .map(|h| (h.name.clone(), h.value.clone()))
220 + .collect(),
221 + },
222 + }),
223 + _ => None,
224 + }
225 + }
226 +}
227 +
228 +struct Handler {
229 + progress: ProgressDispatcher,
230 + info: ClientInfo,
231 +}
232 +
233 +impl ClientHandler for Handler {
234 + async fn on_progress(
235 + &self,
236 + params: ProgressNotificationParam,
237 + _context: NotificationContext<RoleClient>,
238 + ) {
239 + self.progress.handle_notification(params).await;
240 + }
241 +
242 + fn get_info(&self) -> rmcp::model::ClientConfig {
243 + rmcp::model::ClientConfig::new(
244 + ClientCapabilities::default(),
245 + Implementation::new(self.info.name.clone(), self.info.version.clone()),
246 + )
247 + }
248 +}
249 +
250 +/// One live connection to an MCP server.
251 +pub struct Connection {
252 + name: String,
253 + service: RunningService<RoleClient, Handler>,
254 + tools: Vec<Tool>,
255 +}
256 +
257 +impl Connection {
258 + /// Connect to `spec`, run the handshake and list its tools, each within `timeout`.
259 + pub async fn connect(spec: &ServerSpec, info: &ClientInfo, timeout: Duration) -> Result<Self> {
260 + let handler = Handler {
261 + progress: ProgressDispatcher::new(),
262 + info: info.clone(),
263 + };
264 + let handshake = match &spec.transport {
265 + Transport::Stdio { command, args, env } => {
266 + let mut cmd = tokio::process::Command::new(command);
267 + cmd.args(args).envs(env.iter().map(|(k, v)| (k, v)));
268 + let transport = TokioChildProcess::new(cmd)
269 + .with_context(|| format!("starting {}", command.display()))?;
270 + tokio::time::timeout(timeout, handler.serve(transport)).await
271 + }
272 + Transport::Http {
273 + url,
274 + bearer,
275 + headers,
276 + } => {
277 + let mut config = StreamableHttpClientTransportConfig::with_uri(url.as_str());
278 + if let Some(token) = bearer {
279 + config = config.auth_header(token.as_str());
280 + }
281 + if !headers.is_empty() {
282 + let mut map = HashMap::new();
283 + for (name, value) in headers {
284 + map.insert(
285 + http::HeaderName::from_bytes(name.as_bytes())
286 + .with_context(|| format!("invalid header name `{name}`"))?,
287 + http::HeaderValue::from_str(value)
288 + .with_context(|| format!("invalid value for header `{name}`"))?,
289 + );
290 + }
291 + config = config.custom_headers(map);
292 + }
293 + let transport = StreamableHttpClientTransport::from_config(config);
294 + tokio::time::timeout(timeout, handler.serve(transport)).await
295 + }
296 + };
297 + let service = handshake
298 + .map_err(|_| anyhow!("no MCP handshake within {}s", timeout.as_secs()))?
299 + .map_err(|e| anyhow!("MCP handshake failed: {e}"))?;
300 + let tools = tokio::time::timeout(timeout, service.peer().list_all_tools())
301 + .await
302 + .map_err(|_| anyhow!("tools/list timed out"))?
303 + .map_err(|e| anyhow!("tools/list failed: {e}"))?;
304 + Ok(Self {
305 + name: spec.name.clone(),
306 + service,
307 + tools,
308 + })
309 + }
310 +
311 + pub fn name(&self) -> &str {
312 + &self.name
313 + }
314 +
315 + /// Call a tool on this server directly, outside any turn (for example to build a
316 + /// permission prompt). Answers `None` on any failure or after `timeout`.
317 + pub async fn call_raw(&self, tool: &str, args: Value, timeout: Duration) -> Option<Value> {
318 + let mut params = CallToolRequestParams::new(tool.to_string());
319 + if let Value::Object(map) = args {
320 + params = params.with_arguments(map);
321 + }
322 + let result = tokio::time::timeout(timeout, self.service.peer().call_tool(params))
323 + .await
324 + .ok()?
325 + .ok()?;
326 + serde_json::to_value(result).ok()
327 + }
328 +
329 + async fn close(mut self) {
330 + let _ = self
331 + .service
332 + .close_with_timeout(Duration::from_secs(2))
333 + .await;
334 + }
335 +}
336 +
337 +enum SharedState {
338 + NotStarted,
339 + Unavailable,
340 + Running(Arc<Connection>),
341 +}
342 +
343 +/// A built-in server shared by every session in the process, started at most once, on first
344 +/// need. `resolve` decides at that moment whether it is available: `None` (disabled, binary
345 +/// missing) is logged once and the server stays absent for the life of the process.
346 +///
347 +/// ```ignore
348 +/// static DEMUCS: SharedServer = SharedServer::new("demucs", demucs_spec);
349 +/// ```
350 +pub struct SharedServer {
351 + name: &'static str,
352 + resolve: fn() -> Option<ServerSpec>,
353 + state: Mutex<SharedState>,
354 +}
355 +
356 +impl SharedServer {
357 + pub const fn new(name: &'static str, resolve: fn() -> Option<ServerSpec>) -> Self {
358 + Self {
359 + name,
360 + resolve,
361 + state: Mutex::const_new(SharedState::NotStarted),
362 + }
363 + }
364 +
365 + pub fn name(&self) -> &'static str {
366 + self.name
367 + }
368 +
369 + async fn get(&self, info: &ClientInfo, timeout: Duration) -> Option<Arc<Connection>> {
370 + let mut state = self.state.lock().await;
371 + match &*state {
372 + SharedState::Running(conn) => return Some(conn.clone()),
373 + SharedState::Unavailable => return None,
374 + SharedState::NotStarted => {}
375 + }
376 + *state = SharedState::Unavailable;
377 + let Some(spec) = (self.resolve)() else {
378 + tracing::info!("built-in MCP server `{}` is not available", self.name);
379 + return None;
380 + };
381 + match Connection::connect(&spec, info, timeout).await {
382 + Ok(conn) => {
383 + tracing::info!("built-in MCP server `{}` available", self.name);
384 + let conn = Arc::new(conn);
385 + *state = SharedState::Running(conn.clone());
386 + Some(conn)
387 + }
388 + Err(e) => {
389 + tracing::info!("built-in MCP server `{}` unavailable: {e:#}", self.name);
390 + None
391 + }
392 + }
393 + }
394 +
395 + /// Stop the server if it was started. Call once every session's [`McpToolset`] has been
396 + /// shut down, so this is the last owner.
397 + pub async fn shutdown(&self) {
398 + let state = std::mem::replace(&mut *self.state.lock().await, SharedState::Unavailable);
399 + if let SharedState::Running(conn) = state
400 + && let Some(conn) = Arc::into_inner(conn)
401 + {
402 + conn.close().await;
403 + }
404 + }
405 +}
406 +
407 +/// How to connect a session's servers.
408 +#[derive(Clone)]
409 +pub struct ToolsetConfig {
410 + pub client: ClientInfo,
411 + pub connect_timeout: Duration,
412 + pub call_timeout: Duration,
413 +}
414 +
415 +impl ToolsetConfig {
416 + pub fn new(client: ClientInfo) -> Self {
417 + Self {
418 + client,
419 + connect_timeout: DEFAULT_CONNECT_TIMEOUT,
420 + call_timeout: DEFAULT_CALL_TIMEOUT,
421 + }
422 + }
423 +}
424 +
425 +/// One MCP tool as the model sees it.
426 +#[derive(Debug, Clone)]
427 +pub struct McpTool {
428 + /// `mcp__<server>__<tool>`.
429 + pub name: String,
430 + pub server: String,
431 + /// The tool's name on its server.
432 + pub tool: String,
433 + /// The server's description, prefixed with `[<server> MCP server]`.
434 + pub description: String,
435 + /// JSON Schema for the arguments; always has `"type": "object"`.
436 + pub parameters: Value,
437 + pub read_only: bool,
438 +}
439 +
440 +struct Entry {
441 + conn: Arc<Connection>,
442 + tool: McpTool,
443 +}
444 +
445 +/// What a call produced, flattened for the model.
446 +#[derive(Debug, Clone)]
447 +pub struct CallOutcome {
448 + /// Text content joined by newlines; non-text blocks are described in brackets, and
449 + /// `structuredContent` is used when there is no content at all.
450 + pub text: String,
451 + /// The server reported `isError: true`, or the call failed, was cancelled or timed out.
452 + pub failed: bool,
453 + /// `structuredContent`, when the server sent one.
454 + pub structured: Option<Value>,
455 +}
456 +
457 +impl CallOutcome {
458 + fn failed(text: impl Into<String>) -> Self {
459 + Self {
460 + text: text.into(),
461 + failed: true,
462 + structured: None,
463 + }
464 + }
465 +}
466 +
467 +/// The MCP tools available to one session.
468 +pub struct McpToolset {
469 + config: ToolsetConfig,
470 + tools: HashMap<String, Entry>,
471 + conns: Vec<Arc<Connection>>,
472 +}
473 +
474 +impl McpToolset {
475 + /// A toolset with no servers.
476 + pub fn empty(config: ToolsetConfig) -> Self {
477 + Self {
478 + config,
479 + tools: HashMap::new(),
480 + conns: Vec::new(),
481 + }
482 + }
483 +
484 + /// Connect the client-supplied ACP servers (stdio and HTTP) and every built-in the client
485 + /// did not supply a server of the same name for.
486 + #[cfg(feature = "acp")]
487 + pub async fn connect(
488 + config: ToolsetConfig,
489 + client_servers: &[agent_client_protocol::schema::v1::McpServer],
490 + builtins: &[&'static SharedServer],
491 + ) -> Self {
492 + let specs: Vec<ServerSpec> = client_servers
493 + .iter()
494 + .filter_map(|server| {
495 + let spec = ServerSpec::from_acp(server);
496 + if spec.is_none() {
497 + tracing::info!("ignoring MCP server of an unsupported transport");
498 + }
499 + spec
500 + })
501 + .collect();
502 + Self::connect_specs(config, &specs, builtins).await
503 + }
504 +
505 + /// Connect `servers` and every built-in no server of the same name was supplied for. A server
506 + /// that fails is logged and left out.
507 + pub async fn connect_specs(
508 + config: ToolsetConfig,
509 + servers: &[ServerSpec],
510 + builtins: &[&'static SharedServer],
511 + ) -> Self {
512 + let mut set = Self::empty(config);
513 + for spec in servers {
514 + match Connection::connect(spec, &set.config.client, set.config.connect_timeout).await {
515 + Ok(conn) => set.add(Arc::new(conn)),
516 + Err(e) => tracing::warn!("MCP server `{}` unavailable: {e:#}", spec.name),
517 + }
518 + }
519 + for builtin in builtins {
520 + if servers.iter().any(|s| s.name == builtin.name()) {
521 + continue;
522 + }
523 + if let Some(conn) = builtin
524 + .get(&set.config.client, set.config.connect_timeout)
525 + .await
526 + {
527 + set.add(conn);
528 + }
529 + }
530 + set
531 + }
532 +
533 + fn add(&mut self, conn: Arc<Connection>) {
534 + for tool in &conn.tools {
535 + let name = qualified(&conn.name, &tool.name);
536 + if self.tools.contains_key(&name) {
537 + tracing::warn!("duplicate MCP tool name {name}; keeping the first");
538 + continue;
539 + }
540 + let mut parameters = Value::Object((*tool.input_schema).clone());
541 + if parameters.get("type").is_none() {
542 + parameters["type"] = json!("object");
543 + }
544 + let read_only = tool
545 + .annotations
546 + .as_ref()
547 + .and_then(|a| a.read_only_hint)
548 + .unwrap_or(false);
549 + let entry = McpTool {
550 + name: name.clone(),
551 + server: conn.name.clone(),
552 + tool: tool.name.to_string(),
553 + description: format!(
554 + "[{} MCP server] {}",
555 + conn.name,
556 + tool.description.as_deref().unwrap_or("")
557 + ),
558 + parameters,
559 + read_only,
560 + };
561 + self.tools.insert(
562 + name,
563 + Entry {
564 + conn: conn.clone(),
565 + tool: entry,
566 + },
567 + );
568 + }
569 + self.conns.push(conn);
570 + }
571 +
572 + /// Every tool, sorted by name.
573 + pub fn tools(&self) -> Vec<&McpTool> {
574 + let mut tools: Vec<&McpTool> = self.tools.values().map(|e| &e.tool).collect();
575 + tools.sort_by(|a, b| a.name.cmp(&b.name));
576 + tools
577 + }
578 +
579 + /// The tool the model calls `name`, if any.
580 + pub fn tool(&self, name: &str) -> Option<&McpTool> {
581 + self.tools.get(name).map(|e| &e.tool)
582 + }
583 +
584 + /// Whether `tool` on `server` is connected.
585 + pub fn has_tool(&self, server: &str, tool: &str) -> bool {
586 + self.tools.contains_key(&qualified(server, tool))
587 + }
588 +
589 + /// The connection serving the model-facing tool `name`.
590 + pub fn connection(&self, name: &str) -> Option<&Connection> {
591 + self.tools.get(name).map(|e| e.conn.as_ref())
592 + }
593 +
594 + /// Call `name` with `args`. `on_progress` gets one line per MCP progress notification
595 + /// (`"Separating (25%)"`). Cancelling `cancel` sends `notifications/cancelled` and returns
596 + /// a failed outcome. Only an unknown tool or malformed arguments are an `Err`.
597 + pub async fn call(
598 + &self,
599 + name: &str,
600 + args: Value,
601 + cancel: &CancellationToken,
602 + on_progress: &(dyn Fn(String) + Send + Sync),
603 + ) -> Result<CallOutcome> {
604 + let entry = self
605 + .tools
606 + .get(name)
607 + .ok_or_else(|| anyhow!("unknown tool `{name}`"))?;
608 + let arguments: Map<String, Value> = match args {
609 + Value::Object(o) => o,
610 + Value::Null => Map::new(),
611 + _ => bail!("tool arguments must be a JSON object"),
612 + };
613 + let params = CallToolRequestParams::new(entry.tool.tool.clone()).with_arguments(arguments);
614 + let mut handle = match entry
615 + .conn
616 + .service
617 + .peer()
618 + .send_cancellable_request(
619 + ClientRequest::CallToolRequest(Request::new(params)),
620 + PeerRequestOptions::no_options(),
621 + )
622 + .await
623 + {
624 + Ok(handle) => handle,
625 + Err(e) => {
626 + return Ok(CallOutcome::failed(format!(
627 + "MCP server `{}` is not responding: {e}",
628 + entry.conn.name
629 + )));
630 + }
631 + };
632 + let mut progress = entry
633 + .conn
634 + .service
635 + .service()
636 + .progress
637 + .subscribe(handle.progress_token.clone())
638 + .await;
639 + let deadline = tokio::time::sleep(self.config.call_timeout);
640 + tokio::pin!(deadline);
641 +
642 + enum End {
643 + Done(Box<Result<ServerResult, rmcp::ServiceError>>),
644 + Cancelled,
645 + TimedOut,
646 + }
647 + let end = loop {
648 + tokio::select! {
649 + r = &mut handle.rx => {
650 + break End::Done(Box::new(r.unwrap_or(Err(rmcp::ServiceError::TransportClosed))));
651 + }
652 + Some(p) = progress.next() => on_progress(progress_line(&p)),
653 + () = cancel.cancelled() => break End::Cancelled,
654 + () = &mut deadline => break End::TimedOut,
655 + }
656 + };
657 + Ok(match end {
658 + End::Done(done) => match *done {
659 + Ok(ServerResult::CallToolResult(r)) => outcome_from(r),
660 + Ok(_) => CallOutcome::failed("The MCP server sent an unexpected response."),
661 + Err(e) => CallOutcome::failed(format!("MCP call failed: {e}")),
662 + },
663 + End::Cancelled => {
664 + let _ = handle.cancel(Some("cancelled by user".into())).await;
665 + CallOutcome::failed("Cancelled by user.")
666 + }
667 + End::TimedOut => {
668 + let _ = handle.cancel(Some("timed out".into())).await;
669 + CallOutcome::failed(format!(
670 + "Timed out after {} minutes.",
671 + self.config.call_timeout.as_secs() / 60
672 + ))
673 + }
674 + })
675 + }
676 +
677 + /// Close this session's own servers. Built-ins stay up for other sessions.
678 + pub async fn shutdown(self) {
679 + drop(self.tools);
680 + for conn in self.conns {
681 + if let Some(conn) = Arc::into_inner(conn) {
682 + conn.close().await;
683 + }
684 + }
685 + }
686 +}
687 +
688 +/// One line for an MCP progress notification: its message, and a percentage when it has a
689 +/// total.
690 +pub fn progress_line(p: &ProgressNotificationParam) -> String {
691 + let pct = p
692 + .total
693 + .filter(|t| *t > 0.0)
694 + .map(|t| format!(" ({:.0}%)", 100.0 * p.progress / t));
695 + format!(
696 + "{}{}",
697 + p.message.as_deref().unwrap_or("Working…"),
698 + pct.unwrap_or_default()
699 + )
700 +}
701 +
702 +fn outcome_from(r: rmcp::model::CallToolResult) -> CallOutcome {
703 + let mut parts = Vec::new();
704 + for block in &r.content {
705 + match block {
706 + ContentBlock::Text(t) => parts.push(t.text.clone()),
707 + ContentBlock::Image(i) => parts.push(format!("[image: {}]", i.mime_type)),
708 + ContentBlock::Audio(a) => parts.push(format!("[audio: {}]", a.mime_type)),
709 + ContentBlock::ResourceLink(l) => parts.push(format!("[resource: {}]", l.uri)),
710 + ContentBlock::Resource(_) => parts.push("[embedded resource]".into()),
711 + _ => parts.push("[unsupported content]".into()),
712 + }
713 + }
714 + let mut text = parts.join("\n");
715 + if text.is_empty() {
716 + text = r
717 + .structured_content
718 + .as_ref()
719 + .map(|v| v.to_string())
720 + .unwrap_or_else(|| "(no output)".into());
721 + }
722 + CallOutcome {
723 + text,
724 + failed: r.is_error == Some(true),
725 + structured: r.structured_content,
726 + }
727 +}
728 +
729 +/// `true` when an `*_OFF`-style switch is set to `off`, `0`, `false` or `no`.
730 +pub fn switched_off(var: &str) -> bool {
731 + std::env::var(var).is_ok_and(|v| {
732 + matches!(
733 + v.to_ascii_lowercase().as_str(),
734 + "off" | "0" | "false" | "no"
735 + )
736 + })
737 +}
738 +
739 +/// `override_var` when set and non-empty, else `binary` found on `PATH`.
740 +pub fn find_binary(override_var: &str, binary: &str) -> Option<PathBuf> {
741 + if let Some(bin) = std::env::var_os(override_var).filter(|v| !v.is_empty()) {
742 + return Some(bin.into());
743 + }
744 + let path = std::env::var_os("PATH")?;
745 + std::env::split_paths(&path)
746 + .map(|dir| dir.join(binary))
747 + .find(|candidate| is_file(candidate))
748 +}
749 +
750 +fn is_file(path: &Path) -> bool {
751 + path.is_file()
752 +}
753 +
754 +#[cfg(test)]
755 +mod tests {
756 + use super::*;
757 +
758 + #[test]
759 + fn qualified_names_round_trip() {
760 + let q = qualified("My Server!", "do.thing");
761 + assert_eq!(q, "mcp__My_Server__do_thing");
762 + assert_eq!(split_qualified(&q), Some(("My_Server", "do_thing")));
763 + assert_eq!(split_qualified("read_file"), None);
764 + assert_eq!(split_qualified("mcp__x"), None);
765 + assert_eq!(split_qualified("mcp____t"), None);
766 + assert!(qualified("s", &"t".repeat(100)).len() <= MAX_TOOL_NAME);
767 + }
768 +
769 + #[test]
770 + fn progress_text() {
771 + let p = ProgressNotificationParam::new(
772 + rmcp::model::ProgressToken(rmcp::model::NumberOrString::Number(1)),
773 + 25.0,
774 + )
775 + .with_total(100.0)
776 + .with_message("Separating");
777 + assert_eq!(progress_line(&p), "Separating (25%)");
778 + let bare = ProgressNotificationParam::new(
779 + rmcp::model::ProgressToken(rmcp::model::NumberOrString::Number(2)),
780 + 3.0,
781 + );
782 + assert_eq!(progress_line(&bare), "Working…");
783 + }
784 +
785 + #[test]
786 + fn structured_content_stands_in_for_empty_content() {
787 + let r: rmcp::model::CallToolResult = serde_json::from_value(json!({
788 + "content": [],
789 + "structuredContent": {"stems": []},
790 + "isError": false
791 + }))
792 + .unwrap();
793 + let out = outcome_from(r);
794 + assert_eq!(out.text, r#"{"stems":[]}"#);
795 + assert!(!out.failed);
796 + assert_eq!(out.structured, Some(json!({"stems": []})));
797 + }
798 +
799 + #[test]
800 + fn errors_are_reported_as_failed() {
801 + let r: rmcp::model::CallToolResult = serde_json::from_value(json!({
802 + "content": [{"type": "text", "text": "busy"}],
803 + "isError": true
804 + }))
805 + .unwrap();
806 + let out = outcome_from(r);
807 + assert_eq!(out.text, "busy");
808 + assert!(out.failed);
809 + }
810 +
811 + #[test]
812 + fn off_switches() {
813 + // SAFETY: only this test touches this variable.
814 + unsafe { std::env::set_var("ED_MCP_TEST_SWITCH", "OFF") };
815 + assert!(switched_off("ED_MCP_TEST_SWITCH"));
816 + unsafe { std::env::set_var("ED_MCP_TEST_SWITCH", "on") };
817 + assert!(!switched_off("ED_MCP_TEST_SWITCH"));
818 + unsafe { std::env::remove_var("ED_MCP_TEST_SWITCH") };
819 + assert!(!switched_off("ED_MCP_TEST_SWITCH"));
820 + }
821 +}
crates/ed-mcp/src/test_server.rs new
+185
@@ -0,0 +1,185 @@
1 +//! A test MCP server over stdio (newline-delimited JSON-RPC). Stands in for third-party servers
2 +//! and for built-ins such as `demucs --mcp`: it ignores its arguments. A product exposes it as a
3 +//! binary whose `main` calls [`run`] and points `ed_acp_testkit::Target::mcp_stub` at it. Behind the
4 +//! `test-server` feature.
5 +//!
6 +//! Set `STUB_PID_FILE` to have it write its process id, so a test can check it was stopped.
7 +//!
8 +//! Tools: `echo` (read-only), `slow_progress` (three progress notifications),
9 +//! `wait_for_cancel` (answers only after `notifications/cancelled`), and the demucs-shaped
10 +//! `list_models` / `separate_stems`.
11 +
12 +use std::collections::HashSet;
13 +use std::sync::{Arc, Mutex};
14 +use std::time::Duration;
15 +
16 +use serde_json::{Value, json};
17 +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
18 +
19 +fn tools() -> Value {
20 + json!([
21 + {
22 + "name": "echo",
23 + "description": "Echo the text back.",
24 + "inputSchema": {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]},
25 + "annotations": {"readOnlyHint": true}
26 + },
27 + {
28 + "name": "slow_progress",
29 + "description": "Report progress three times, then finish.",
30 + "inputSchema": {"type": "object", "properties": {}}
31 + },
32 + {
33 + "name": "wait_for_cancel",
34 + "description": "Blocks until the call is cancelled.",
35 + "inputSchema": {"type": "object", "properties": {}}
36 + },
37 + {
38 + "name": "list_models",
39 + "description": "List stem separation models.",
40 + "inputSchema": {"type": "object", "properties": {}},
41 + "annotations": {"readOnlyHint": true}
42 + },
43 + {
44 + "name": "separate_stems",
45 + "description": "Pretend to separate stems.",
46 + "inputSchema": {"type": "object", "properties": {"input": {"type": "string"}, "model": {"type": "string"}}, "required": ["input"]}
47 + }
48 + ])
49 +}
50 +
51 +/// Serve on stdin/stdout until stdin closes.
52 +pub async fn run() {
53 + // Lets tests find out whether this process is still alive after the agent closes it.
54 + if let Some(path) = std::env::var_os("STUB_PID_FILE") {
55 + let _ = std::fs::write(path, std::process::id().to_string());
56 + }
57 + let out = Arc::new(tokio::sync::Mutex::new(tokio::io::stdout()));
58 + let cancelled: Arc<Mutex<HashSet<String>>> = Arc::default();
59 + let mut lines = BufReader::new(tokio::io::stdin()).lines();
60 +
61 + let send = {
62 + let out = out.clone();
63 + move |msg: Value| {
64 + let out = out.clone();
65 + async move {
66 + let mut o = out.lock().await;
67 + let _ = o.write_all(format!("{msg}\n").as_bytes()).await;
68 + let _ = o.flush().await;
69 + }
70 + }
71 + };
72 +
73 + while let Ok(Some(line)) = lines.next_line().await {
74 + let Ok(msg) = serde_json::from_str::<Value>(&line) else {
75 + continue;
76 + };
77 + let method = msg
78 + .get("method")
79 + .and_then(Value::as_str)
80 + .unwrap_or("")
81 + .to_string();
82 + let id = msg.get("id").cloned();
83 + let params = msg.get("params").cloned().unwrap_or(Value::Null);
84 + let Some(id) = id else {
85 + if method == "notifications/cancelled"
86 + && let Some(rid) = params.get("requestId")
87 + {
88 + cancelled.lock().unwrap().insert(rid.to_string());
89 + }
90 + continue;
91 + };
92 + match method.as_str() {
93 + "initialize" => {
94 + let version = params
95 + .get("protocolVersion")
96 + .cloned()
97 + .unwrap_or(json!("2025-06-18"));
98 + send(json!({"jsonrpc": "2.0", "id": id, "result": {
99 + "protocolVersion": version,
100 + "capabilities": {"tools": {}},
101 + "serverInfo": {"name": "stub", "version": "0.0.0"}
102 + }}))
103 + .await;
104 + }
105 + "ping" => send(json!({"jsonrpc": "2.0", "id": id, "result": {}})).await,
106 + "tools/list" => {
107 + send(json!({"jsonrpc": "2.0", "id": id, "result": {"tools": tools()}})).await
108 + }
109 + "tools/call" => {
110 + let send = send.clone();
111 + let cancelled = cancelled.clone();
112 + tokio::spawn(async move {
113 + let name = params
114 + .get("name")
115 + .and_then(Value::as_str)
116 + .unwrap_or("")
117 + .to_string();
118 + let args = params.get("arguments").cloned().unwrap_or(json!({}));
119 + let token = params.pointer("/_meta/progressToken").cloned();
120 + let progress = |n: u32, msg: &str| {
121 + token.as_ref().map(|t| {
122 + json!({"jsonrpc": "2.0", "method": "notifications/progress",
123 + "params": {"progressToken": t, "progress": n, "total": 100, "message": msg}})
124 + })
125 + };
126 + let result = match name.as_str() {
127 + "echo" => {
128 + let text = args.get("text").and_then(Value::as_str).unwrap_or("");
129 + json!({"content": [{"type": "text", "text": format!("echo: {text}")}], "isError": false})
130 + }
131 + "slow_progress" | "separate_stems" => {
132 + for (n, m) in [(10, "starting"), (50, "halfway"), (90, "almost")] {
133 + if let Some(p) = progress(n, m) {
134 + send(p).await;
135 + }
136 + tokio::time::sleep(Duration::from_millis(40)).await;
137 + }
138 + if name == "slow_progress" {
139 + json!({"content": [{"type": "text", "text": "done"}], "isError": false})
140 + } else {
141 + let input = args
142 + .get("input")
143 + .and_then(Value::as_str)
144 + .unwrap_or("/tmp/x.wav");
145 + let dir = format!("{}_stems", input.trim_end_matches(".wav"));
146 + let stems: Vec<Value> = ["drums", "bass", "vocals", "other"]
147 + .iter()
148 + .map(|s| json!({"id": s, "path": format!("{dir}/{s}.wav")}))
149 + .collect();
150 + json!({
151 + "content": [{"type": "text", "text": format!("Separated into {} stems", stems.len())}],
152 + "structuredContent": {"stems": stems},
153 + "isError": false
154 + })
155 + }
156 + }
157 + "list_models" => json!({
158 + "content": [{"type": "text", "text": "htdemucs 80 MB"}],
159 + "structuredContent": {"models": [
160 + {"id": "htdemucs", "size_mb": 80, "cached": false},
161 + {"id": "htdemucs_6s", "size_mb": 55, "cached": true}
162 + ]},
163 + "isError": false
164 + }),
165 + "wait_for_cancel" => {
166 + let key = id.to_string();
167 + while !cancelled.lock().unwrap().contains(&key) {
168 + tokio::time::sleep(Duration::from_millis(20)).await;
169 + }
170 + json!({"content": [{"type": "text", "text": "cancelled"}], "isError": true})
171 + }
172 + other => {
173 + send(json!({"jsonrpc": "2.0", "id": id, "error": {"code": -32602, "message": format!("unknown tool {other}")}})).await;
174 + return;
175 + }
176 + };
177 + send(json!({"jsonrpc": "2.0", "id": id, "result": result})).await;
178 + });
179 + }
180 + other => {
181 + send(json!({"jsonrpc": "2.0", "id": id, "error": {"code": -32601, "message": format!("method not found: {other}")}})).await;
182 + }
183 + }
184 + }
185 +}
crates/ed-mcp/tests/http.rs new
+219
@@ -0,0 +1,219 @@
1 +//! Streamable HTTP against an in-process rmcp server, and the Ed executor on top of it.
2 +
3 +use std::sync::Arc;
4 +use std::time::Duration;
5 +
6 +use axum::extract::Request;
7 +use axum::http::StatusCode;
8 +use axum::middleware::{self, Next};
9 +use axum::response::Response;
10 +use ed_agent::{ToolExecutor, ToolRisk};
11 +use ed_mcp::{
12 + ClientInfo, Connection, McpExecutor, McpToolset, ServerSpec, ToolPolicy, ToolsetConfig,
13 +};
14 +use rmcp::RoleServer;
15 +use rmcp::ServerHandler;
16 +use rmcp::model::{
17 + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ListToolsResult,
18 + ServerCapabilities, ServerConfig, Tool, ToolAnnotations,
19 +};
20 +use rmcp::service::RequestContext;
21 +use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
22 +use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
23 +use serde_json::json;
24 +use tokio_util::sync::CancellationToken;
25 +
26 +const TOKEN: &str = "s3cret";
27 +
28 +#[derive(Clone)]
29 +struct Mailbox;
30 +
31 +impl ServerHandler for Mailbox {
32 + fn get_info(&self) -> ServerConfig {
33 + ServerConfig::new(ServerCapabilities::builder().enable_tools().build())
34 + }
35 +
36 + async fn list_tools(
37 + &self,
38 + _request: Option<rmcp::model::PaginatedRequestParams>,
39 + _context: RequestContext<RoleServer>,
40 + ) -> Result<ListToolsResult, rmcp::ErrorData> {
41 + let schema = Arc::new(json!({"type": "object"}).as_object().unwrap().clone());
42 + Ok(ListToolsResult {
43 + tools: vec![
44 + // Declares itself read-only.
45 + Tool::new("whoami", "Who is signed in.", schema.clone())
46 + .annotate(ToolAnnotations::new().read_only(true)),
47 + // Says nothing, like a server that predates annotations.
48 + Tool::new("list_messages", "Recent messages.", schema.clone()),
49 + Tool::new("send_message", "Send mail.", schema),
50 + ],
51 + ..Default::default()
52 + })
53 + }
54 +
55 + async fn call_tool(
56 + &self,
57 + request: CallToolRequestParams,
58 + _context: RequestContext<RoleServer>,
59 + ) -> Result<CallToolResponse, rmcp::ErrorData> {
60 + Ok(
61 + CallToolResult::success(vec![ContentBlock::text(format!("ran {}", request.name))])
62 + .into(),
63 + )
64 + }
65 +}
66 +
67 +async fn require_bearer(request: Request, next: Next) -> Result<Response, StatusCode> {
68 + let ok = request
69 + .headers()
70 + .get("authorization")
71 + .and_then(|v| v.to_str().ok())
72 + == Some(&format!("Bearer {TOKEN}"));
73 + if ok {
74 + Ok(next.run(request).await)
75 + } else {
76 + Err(StatusCode::UNAUTHORIZED)
77 + }
78 +}
79 +
80 +/// Serve [`Mailbox`] behind bearer auth; returns its URL.
81 +async fn serve() -> String {
82 + let config = StreamableHttpServerConfig::default()
83 + .with_legacy_session_mode(true)
84 + .with_sse_keep_alive(None)
85 + .with_cancellation_token(CancellationToken::new());
86 + let service: StreamableHttpService<Mailbox, LocalSessionManager> =
87 + StreamableHttpService::new(|| Ok(Mailbox), Default::default(), config);
88 + let router = axum::Router::new()
89 + .nest_service("/mcp", service)
90 + .layer(middleware::from_fn(require_bearer));
91 + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
92 + let addr = listener.local_addr().unwrap();
93 + tokio::spawn(async move {
94 + let _ = axum::serve(listener, router).await;
95 + });
96 + format!("http://{addr}/mcp")
97 +}
98 +
99 +fn client() -> ClientInfo {
100 + ClientInfo::new("ed-mcp-test", "0")
101 +}
102 +
103 +#[tokio::test]
104 +async fn connects_lists_and_calls_with_a_bearer_token() {
105 + let url = serve().await;
106 + let spec = ServerSpec::http("mail", url).with_bearer(TOKEN);
107 +
108 + let set = McpToolset::connect_specs(ToolsetConfig::new(client()), &[spec], &[]).await;
109 +
110 + let names: Vec<_> = set.tools().iter().map(|t| t.name.clone()).collect();
111 + assert_eq!(
112 + names,
113 + [
114 + "mcp__mail__list_messages",
115 + "mcp__mail__send_message",
116 + "mcp__mail__whoami"
117 + ]
118 + );
119 + let out = set
120 + .call(
121 + "mcp__mail__whoami",
122 + json!({}),
123 + &CancellationToken::new(),
124 + &|_| {},
125 + )
126 + .await
127 + .unwrap();
128 + assert_eq!(out.text, "ran whoami");
129 + assert!(!out.failed);
130 + set.shutdown().await;
131 +}
132 +
133 +#[tokio::test]
134 +async fn a_wrong_bearer_token_is_rejected() {
135 + let url = serve().await;
136 + let spec = ServerSpec::http("mail", url).with_bearer("nope");
137 +
138 + let err = Connection::connect(&spec, &client(), Duration::from_secs(5))
139 + .await
140 + .err()
141 + .expect("the server answers 401");
142 + assert!(err.to_string().contains("handshake failed"), "{err:#}");
143 +
144 + // A rejected server is left out, not fatal.
145 + let set = McpToolset::connect_specs(ToolsetConfig::new(client()), &[spec], &[]).await;
146 + assert!(set.tools().is_empty());
147 +}
148 +
149 +#[tokio::test]
150 +async fn a_missing_bearer_token_is_rejected() {
151 + let url = serve().await;
152 + let spec = ServerSpec::http("mail", url);
153 + assert!(
154 + Connection::connect(&spec, &client(), Duration::from_secs(5))
155 + .await
156 + .is_err()
157 + );
158 +}
159 +
160 +async fn executor(policy: ToolPolicy) -> McpExecutor {
161 + let url = serve().await;
162 + let spec = ServerSpec::http("mail", url).with_bearer(TOKEN);
163 + let set = McpToolset::connect_specs(ToolsetConfig::new(client()), &[spec], &[]).await;
164 + let exec = McpExecutor::new(policy);
165 + exec.set_toolset(Some(set)).await;
166 + exec
167 +}
168 +
169 +fn risk(defs: &[ed_agent::AgentToolDefinition], name: &str) -> ToolRisk {
170 + defs.iter().find(|d| d.name == name).unwrap().risk
171 +}
172 +
173 +#[tokio::test]
174 +async fn risk_follows_the_read_only_hint() {
175 + let defs = executor(ToolPolicy::new()).await.definitions().await;
176 +
177 + assert_eq!(risk(&defs, "mcp__mail__whoami"), ToolRisk::ReadOnly);
178 + // No hint means it may change things.
179 + assert_eq!(risk(&defs, "mcp__mail__list_messages"), ToolRisk::Mutating);
180 + assert_eq!(risk(&defs, "mcp__mail__send_message"), ToolRisk::Mutating);
181 +}
182 +
183 +#[tokio::test]
184 +async fn the_host_can_override_risk_and_drop_the_prefix() {
185 + let policy = ToolPolicy::new()
186 + .read_only(&["list_messages", "whoami"])
187 + .unprefixed();
188 + let exec = executor(policy).await;
189 + let defs = exec.definitions().await;
190 +
191 + assert_eq!(risk(&defs, "list_messages"), ToolRisk::ReadOnly);
192 + assert_eq!(risk(&defs, "whoami"), ToolRisk::ReadOnly);
193 + assert_eq!(risk(&defs, "send_message"), ToolRisk::Mutating);
194 + assert_eq!(
195 + defs.iter()
196 + .find(|d| d.name == "send_message")
197 + .unwrap()
198 + .description,
199 + "Send mail."
200 + );
201 +
202 + // The executor answers to the names the model sees.
203 + assert_eq!(
204 + exec.execute("send_message", "{}").await.unwrap(),
205 + "ran send_message"
206 + );
207 + assert!(exec.execute("mcp__mail__send_message", "{}").await.is_err());
208 +}
209 +
210 +#[tokio::test]
211 +async fn disconnecting_removes_the_tools() {
212 + let exec = executor(ToolPolicy::new()).await;
213 + assert!(!exec.definitions().await.is_empty());
214 +
215 + exec.set_toolset(None).await;
216 +
217 + assert!(exec.definitions().await.is_empty());
218 + assert!(exec.execute("mcp__mail__whoami", "{}").await.is_err());
219 +}