diff --git a/Cargo.lock b/Cargo.lock index e48c243..56a0950 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -128,6 +128,24 @@ dependencies = [ "tokio", ] +[[package]] +name = "agntcy-a2a-stdio" +version = "0.1.0" +dependencies = [ + "a2a-client-lf", + "a2a-lf", + "a2a-server-lf", + "async-trait", + "futures", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tracing", + "uuid", +] + [[package]] name = "agntcy-slim" version = "1.3.0" @@ -175,7 +193,7 @@ dependencies = [ "parking_lot", "pin-project", "prost-types 0.14.3", - "rand 0.9.4", + "rand 0.9.2", "reqwest", "schemars", "serde", @@ -292,7 +310,7 @@ dependencies = [ "prost 0.14.3", "prost-types 0.14.3", "protoc-bin-vendored", - "rand 0.9.4", + "rand 0.9.2", "serde", "serde_json", "thiserror 2.0.18", @@ -324,7 +342,7 @@ dependencies = [ "parking_lot", "prost 0.14.3", "protoc-bin-vendored", - "rand 0.9.4", + "rand 0.9.2", "semver", "serde", "serde_json", @@ -399,7 +417,7 @@ dependencies = [ "futures", "futures-timer", "parking_lot", - "rand 0.9.4", + "rand 0.9.2", "serde", "thiserror 2.0.18", "tokio", @@ -708,9 +726,9 @@ dependencies = [ [[package]] name = "axum" -version = "0.8.9" +version = "0.8.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" dependencies = [ "axum-core", "axum-macros", @@ -761,9 +779,9 @@ dependencies = [ [[package]] name = "axum-macros" -version = "0.5.1" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" +checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" dependencies = [ "proc-macro2", "quote", @@ -831,9 +849,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" [[package]] name = "bitflags" -version = "2.11.1" +version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" [[package]] name = "block-buffer" @@ -901,9 +919,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.60" +version = "1.2.59" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43c5703da9466b66a946814e1adf53ea2c90f10063b86290cc9eb67ce3478a20" +checksum = "b7a4d3ec6524d28a329fc53654bbadc9bdd7b0431f5d65f1a56ffb28a1ee5283" dependencies = [ "find-msvc-tools", "jobserver", @@ -1272,9 +1290,9 @@ dependencies = [ [[package]] name = "fastrand" -version = "2.4.1" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" [[package]] name = "find-msvc-tools" @@ -1539,9 +1557,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.17.0" +version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" [[package]] name = "headers" @@ -1676,14 +1694,15 @@ dependencies = [ [[package]] name = "hyper-rustls" -version = "0.27.8" +version = "0.27.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2b52f86d1d4bc0d6b4e6826d960b1b333217e07d36b882dca570a5e1c48895b" +checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58" dependencies = [ "http", "hyper", "hyper-util", "rustls", + "rustls-pki-types", "tokio", "tokio-rustls", "tower-service", @@ -1885,12 +1904,12 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.14.0" +version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff" dependencies = [ "equivalent", - "hashbrown 0.17.0", + "hashbrown 0.16.1", "serde", "serde_core", ] @@ -1901,7 +1920,7 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bd5b3eaf1a28b758ac0faa5a4254e8ab2705605496f1b1f3fbbc3988ad73d199" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "inotify-sys", "libc", ] @@ -1993,9 +2012,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.95" +version = "0.3.94" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2964e92d1d9dc3364cae4d718d93f227e3abb088e747d92e0395bfdedf1c12ca" +checksum = "2e04e2ef80ce82e13552136fabeef8a5ed1f985a96805761cbb9a2c34e7664d9" dependencies = [ "cfg-if", "futures-util", @@ -2069,9 +2088,9 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "libc" -version = "0.2.185" +version = "0.2.184" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52ff2c0fe9bc6cb6b14a0592c2ff4fa9ceb83eea9db979b0487cd054946a2b8f" +checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" [[package]] name = "linux-raw-sys" @@ -2384,7 +2403,7 @@ version = "8.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "fsevent-sys", "inotify", "kqueue", @@ -2402,7 +2421,7 @@ version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", ] [[package]] @@ -2501,11 +2520,11 @@ checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" [[package]] name = "openssl" -version = "0.10.77" +version = "0.10.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfe4646e360ec77dff7dde40ed3d6c5fee52d156ef4a62f53973d38294dad87f" +checksum = "951c002c75e16ea2c65b8c7e4d3d51d5530d8dfa7d060b4776828c88cfb18ecf" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "cfg-if", "foreign-types", "libc", @@ -2533,9 +2552,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "openssl-sys" -version = "0.9.113" +version = "0.9.112" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ad2f2c0eba47118757e4c6d2bff2838f3e0523380021356e7875e858372ce644" +checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb" dependencies = [ "cc", "libc", @@ -2630,7 +2649,7 @@ dependencies = [ "futures-util", "opentelemetry", "percent-encoding", - "rand 0.9.4", + "rand 0.9.2", "thiserror 2.0.18", "tokio", "tokio-stream", @@ -2771,9 +2790,9 @@ dependencies = [ [[package]] name = "pkg-config" -version = "0.3.33" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" [[package]] name = "plain" @@ -3083,7 +3102,7 @@ version = "0.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c3a14896dfa883796f1cb410461aef38810ea05f2b2c33c5aded3649095fdad" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "memchr", "unicase", ] @@ -3126,7 +3145,7 @@ dependencies = [ "bytes", "getrandom 0.3.4", "lru-slab", - "rand 0.9.4", + "rand 0.9.2", "ring", "rustc-hash", "rustls", @@ -3186,9 +3205,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.9.4" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" dependencies = [ "rand_chacha 0.9.0", "rand_core 0.9.5", @@ -3234,9 +3253,9 @@ dependencies = [ [[package]] name = "rayon" -version = "1.12.0" +version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f" dependencies = [ "either", "rayon-core", @@ -3258,7 +3277,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", ] [[package]] @@ -3402,7 +3421,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "errno", "libc", "linux-raw-sys", @@ -3411,9 +3430,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.38" +version = "0.23.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69f9466fb2c14ea04357e91413efb882e2a6d4a406e625449bc0a5d360d53a21" +checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" dependencies = [ "aws-lc-rs", "log", @@ -3449,9 +3468,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.12" +version = "0.103.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8279bb85272c9f10811ae6a6c547ff594d6a7f3c6c6b02ee9726d1d0dcfcdd06" +checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" dependencies = [ "aws-lc-rs", "ring", @@ -3546,7 +3565,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -3565,9 +3584,9 @@ dependencies = [ [[package]] name = "semver" -version = "1.0.28" +version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" +checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" dependencies = [ "serde", "serde_core", @@ -3886,7 +3905,7 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "core-foundation 0.9.4", "system-configuration-sys", ] @@ -4036,9 +4055,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.51.1" +version = "1.51.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f66bf9585cda4b724d3e78ab34b73fb2bbaba9011b9bfdf69dc836382ea13b8c" +checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" dependencies = [ "bytes", "libc", @@ -4281,7 +4300,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ "base64 0.22.1", - "bitflags 2.11.1", + "bitflags 2.11.0", "bytes", "futures-util", "http", @@ -4417,7 +4436,7 @@ version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" dependencies = [ - "rand 0.9.4", + "rand 0.9.2", ] [[package]] @@ -4721,9 +4740,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.118" +version = "0.2.117" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0bf938a0bacb0469e83c1e148908bd7d5a6010354cf4fb73279b7447422e3a89" +checksum = "0551fc1bb415591e3372d0bc4780db7e587d84e2a7e79da121051c5c4b89d0b0" dependencies = [ "cfg-if", "once_cell", @@ -4734,9 +4753,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.68" +version = "0.4.67" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f371d383f2fb139252e0bfac3b81b265689bf45b6874af544ffa4c975ac1ebf8" +checksum = "03623de6905b7206edd0a75f69f747f134b7f0a2323392d664448bf2d3c5d87e" dependencies = [ "js-sys", "wasm-bindgen", @@ -4744,9 +4763,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.118" +version = "0.2.117" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eeff24f84126c0ec2db7a449f0c2ec963c6a49efe0698c4242929da037ca28ed" +checksum = "7fbdf9a35adf44786aecd5ff89b4563a90325f9da0923236f6104e603c7e86be" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -4754,9 +4773,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.118" +version = "0.2.117" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d08065faf983b2b80a79fd87d8254c409281cf7de75fc4b773019824196c904" +checksum = "dca9693ef2bab6d4e6707234500350d8dad079eb508dca05530c85dc3a529ff2" dependencies = [ "bumpalo", "proc-macro2", @@ -4767,9 +4786,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.118" +version = "0.2.117" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5fd04d9e306f1907bd13c6361b5c6bfc7b3b3c095ed3f8a9246390f8dbdee129" +checksum = "39129a682a6d2d841b6c429d0c51e5cb0ed1a03829d8b3d1e69a011e62cb3d3b" dependencies = [ "unicode-ident", ] @@ -4815,7 +4834,7 @@ version = "0.244.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.11.0", "hashbrown 0.15.5", "indexmap", "semver", @@ -4823,9 +4842,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.95" +version = "0.3.94" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f2dfbb17949fa2088e5d39408c48368947b86f7834484e87b73de55bc14d97d" +checksum = "cd70027e39b12f0849461e08ffc50b9cd7688d942c1c8e3c7b22273236b4dd0a" dependencies = [ "js-sys", "wasm-bindgen", @@ -5184,7 +5203,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", - "bitflags 2.11.1", + "bitflags 2.11.0", "indexmap", "log", "serde", diff --git a/Cargo.toml b/Cargo.toml index eaf524e..7c51835 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "a2a-pb", "a2a-grpc", "a2a-slimrpc", + "a2a-stdio", "a2acli", "examples/helloworld", ] @@ -25,6 +26,7 @@ a2a-server = { package = "a2a-server-lf", path = "a2a-server", version = "0.2.6" a2a-pb = { package = "a2a-pb", path = "a2a-pb", version = "0.1.6" } a2a-grpc = { package = "a2a-grpc", path = "a2a-grpc", version = "0.1.11" } a2a-slimrpc = { package = "a2a-slimrpc", path = "a2a-slimrpc", version = "0.1.8" } +a2a-stdio = { package = "agntcy-a2a-stdio", path = "a2a-stdio", version = "0.1.0" } # Serialization serde = { version = "1", features = ["derive"] } diff --git a/a2a-client/src/rest.rs b/a2a-client/src/rest.rs index a9cf64d..308022a 100644 --- a/a2a-client/src/rest.rs +++ b/a2a-client/src/rest.rs @@ -267,7 +267,7 @@ fn parse_rest_error(status: reqwest::StatusCode, body: &str) -> A2AError { } if let Value::Object(values) = raw_detail { - details.extend(values.into_iter()); + details.extend(values); } } diff --git a/a2a-stdio/Cargo.toml b/a2a-stdio/Cargo.toml new file mode 100644 index 0000000..10cfc56 --- /dev/null +++ b/a2a-stdio/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "agntcy-a2a-stdio" +version = "0.1.0" +description = "A2A v1 STDIO transport binding for client and server" +readme = "README.md" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true + +[lib] +name = "a2a_stdio" + +# Test-only helper subprocess used by integration tests in tests/spawned.rs. +# Not part of the public surface; kept under tests/bin/ to make that explicit. +[[bin]] +name = "stdio_test_helper" +path = "tests/bin/stdio_test_helper.rs" + +[dependencies] +a2a = { workspace = true } +a2a-client = { workspace = true } +a2a-server = { workspace = true } +tokio = { workspace = true } +tokio-stream = { workspace = true } +futures = { workspace = true } +async-trait = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true } diff --git a/a2a-stdio/README.md b/a2a-stdio/README.md new file mode 100644 index 0000000..b626707 --- /dev/null +++ b/a2a-stdio/README.md @@ -0,0 +1,66 @@ +# a2a-stdio + +STDIO transport binding for A2A v1 client and server implementations. + +This crate is published as `agntcy-a2a-stdio` and imported in Rust as `a2a_stdio`. + +## What It Provides + +- `StdioTransport` — `Transport` implementation that spawns an agent + subprocess and speaks A2A JSON-RPC over its stdin/stdout +- `StdioTransportFactory` — `TransportFactory` integration for agent cards + that advertise the `STDIO` protocol +- `StdioServer` — server-side dispatcher that wraps an + `a2a_server::RequestHandler` and serves requests over a pair of + asynchronous reader/writer streams + +## Wire Format + +Messages on the wire use LSP-style framing followed by a JSON-RPC 2.0 body: + +```text +Content-Length: \r\n +Content-Type: application/json\r\n +\r\n + +``` + +Method names use the slash-separated A2A convention +(`message/send`, `tasks/get`, `tasks/subscribe`, ...). + +## Handshake + +Before serving requests, the server emits a `handshake` message advertising +its supported variants and features. The client replies with a +`handshakeAck` that either accepts (selecting a variant) or rejects the +session. + +A session identifier may be passed via the `A2A_SESSION_ID` environment +variable; otherwise the server generates a UUIDv7. + +## Agent Card Target Format + +`StdioTransportFactory` interprets `supportedInterfaces[].url` as one of: + +- `stdio:///path/to/binary` — spawn the binary with no arguments +- `stdio:///path/to/binary?arg1&arg2` — spawn with the given arguments +- `/path/to/binary arg1 arg2` — plain command line, split on whitespace + +Arguments in the `stdio://` form are split on `&` and the path is split on +the first `?`, so individual arguments cannot contain `&` or `?`. Use the +plain command-line form when those characters are required. + +## Install + +```toml +[dependencies] +a2a = { package = "a2a-lf", version = "0.2" } +a2a-stdio = { package = "agntcy-a2a-stdio", version = "0.1" } +``` + +## Workspace + +This crate is part of the `a2a-rs` workspace. + +- Repository: https://github.com/a2aproject/a2a-rs +- Workspace README: https://github.com/a2aproject/a2a-rs/blob/main/README.md diff --git a/a2a-stdio/src/client.rs b/a2a-stdio/src/client.rs new file mode 100644 index 0000000..a5f9936 --- /dev/null +++ b/a2a-stdio/src/client.rs @@ -0,0 +1,565 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! STDIO client transport — spawns a subprocess and communicates over stdin/stdout. + +use a2a::event::StreamResponse; +use a2a::*; +use a2a_client::transport::{ServiceParams, Transport, TransportFactory}; +use async_trait::async_trait; +use futures::stream::BoxStream; +use serde::Serialize; +use serde::de::DeserializeOwned; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::{AsyncWrite, AsyncWriteExt, BufReader}; +use tokio::process::{Child, Command}; +use tokio::sync::Mutex; +use tokio_stream::wrappers::ReceiverStream; + +use crate::errors::StdioError; +use crate::framing; +use crate::handshake::{self, HandshakeAck}; + +/// How long to wait for the subprocess to send its initial handshake. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + +/// How long to wait for the subprocess to exit after closing its stdin +/// before forcibly killing it. +const CLOSE_TIMEOUT: Duration = Duration::from_secs(5); + +// --------------------------------------------------------------------------- +// Stdio method names (slash-separated, per maintainer design doc) +// --------------------------------------------------------------------------- + +mod stdio_methods { + pub const SEND_MESSAGE: &str = "message/send"; + pub const SEND_STREAMING_MESSAGE: &str = "message/stream"; + pub const GET_TASK: &str = "tasks/get"; + pub const LIST_TASKS: &str = "tasks/list"; + pub const CANCEL_TASK: &str = "tasks/cancel"; + pub const SUBSCRIBE_TO_TASK: &str = "tasks/subscribe"; + pub const CREATE_PUSH_CONFIG: &str = "tasks/pushNotificationConfig/create"; + pub const GET_PUSH_CONFIG: &str = "tasks/pushNotificationConfig/get"; + pub const LIST_PUSH_CONFIGS: &str = "tasks/pushNotificationConfig/list"; + pub const DELETE_PUSH_CONFIG: &str = "tasks/pushNotificationConfig/delete"; + pub const GET_EXTENDED_AGENT_CARD: &str = "agent/extendedCard"; +} + +// --------------------------------------------------------------------------- +// StdioTransport +// --------------------------------------------------------------------------- + +/// A transport that communicates with an agent subprocess over stdio pipes. +/// +/// The client spawns the child process, performs the handshake, then sends +/// JSON-RPC requests over the child's stdin and reads responses from stdout. +/// A transport that communicates with an agent subprocess over stdio pipes. +/// +/// The client spawns the child process, performs the handshake, then sends +/// JSON-RPC requests over the child's stdin and reads responses from stdout. +/// +/// # Concurrency +/// +/// Each call to [`Transport`] takes a request-level lock that is held across +/// the full write/read cycle. Concurrent calls therefore block rather than +/// race on the underlying pipes — responses are not multiplexed by `id`. +pub struct StdioTransport { + /// The child process. Held for lifetime management. + child: Arc>, + /// Writer to the child's stdin (framed JSON-RPC). + writer: Arc>>, + /// Reader from the child's stdout (framed JSON-RPC). + reader: Arc>>, + /// Serializes full request/response cycles so concurrent callers do not + /// interleave on the underlying pipes. + request_lock: Arc>, +} + +impl StdioTransport { + /// Spawn a subprocess and perform the handshake. + /// + /// `program` and `args` define the command to run. An optional + /// `session_id` is passed via the `A2A_SESSION_ID` environment variable. + pub async fn spawn( + program: &str, + args: &[&str], + session_id: Option<&str>, + ) -> Result { + let mut cmd = Command::new(program); + cmd.args(args) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::inherit()); + + if let Some(sid) = session_id { + cmd.env("A2A_SESSION_ID", sid); + } + + let mut child = cmd.spawn()?; + + let stdin = child + .stdin + .take() + .ok_or_else(|| StdioError::HandshakeFailed("failed to open child stdin".into()))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| StdioError::HandshakeFailed("failed to open child stdout".into()))?; + + let mut reader = BufReader::new(stdout); + let mut writer: Box = Box::new(stdin); + + // Server sends handshake first. Bound the wait so a hung subprocess + // does not block the spawn indefinitely. + let hs = + match tokio::time::timeout(HANDSHAKE_TIMEOUT, handshake::read_handshake(&mut reader)) + .await + { + Ok(res) => res?, + Err(_) => { + // Best-effort: kill the subprocess so it does not linger. + let _ = child.kill().await; + return Err(StdioError::HandshakeFailed(format!( + "timed out waiting for handshake after {:?}", + HANDSHAKE_TIMEOUT + ))); + } + }; + + // Select the first supported variant. + let selected = hs.supported_variants.first().cloned().unwrap_or_default(); + + let ack = HandshakeAck::accept(selected); + handshake::write_handshake_ack(&mut writer, &ack).await?; + + // Handshake content (selected variant, features) is currently unused + // beyond the accept/reject decision; drop it. + let _ = hs; + + Ok(StdioTransport { + child: Arc::new(Mutex::new(child)), + writer: Arc::new(Mutex::new(writer)), + reader: Arc::new(Mutex::new(reader)), + request_lock: Arc::new(Mutex::new(())), + }) + } + + /// Send a JSON-RPC request and read the response. + async fn call( + &self, + method: &str, + params: &Req, + ) -> Result { + let id = uuid::Uuid::now_v7().to_string(); + let request = JsonRpcRequest::new( + JsonRpcId::String(id), + method, + Some( + serde_json::to_value(params) + .map_err(|e| A2AError::internal(format!("serialize params: {e}")))?, + ), + ); + + let body = serde_json::to_vec(&request) + .map_err(|e| A2AError::internal(format!("serialize request: {e}")))?; + + // Hold the request lock across the full write+read cycle so concurrent + // callers do not interleave on the underlying pipes. + let _guard = self.request_lock.lock().await; + + // Write request. + { + let mut writer = self.writer.lock().await; + framing::write_frame(&mut *writer, &body) + .await + .map_err(|e| A2AError::internal(format!("write frame: {e}")))?; + } + + // Read response. + let frame = { + let mut reader = self.reader.lock().await; + framing::read_frame(&mut *reader) + .await + .map_err(|e| A2AError::internal(format!("read frame: {e}")))? + .ok_or_else(|| A2AError::internal("subprocess closed stdout unexpectedly"))? + }; + + let rpc_response: JsonRpcResponse = serde_json::from_slice(&frame) + .map_err(|e| A2AError::internal(format!("parse response: {e}")))?; + + if let Some(err) = rpc_response.error { + return Err(A2AError::new(err.code, err.message)); + } + + // For methods returning `()` the JSON-RPC response carries + // `"result": null`, which serde deserializes into `None` for + // `Option`. Treat a missing/null result as `Value::Null` + // so unit responses round-trip correctly while non-null results + // still flow through normal deserialization. + let result = rpc_response.result.unwrap_or(serde_json::Value::Null); + + serde_json::from_value(result) + .map_err(|e| A2AError::internal(format!("deserialize result: {e}"))) + } + + /// Send a JSON-RPC request for a streaming method. + /// + /// Returns a true async stream that yields events as they arrive. + /// Notifications (no `id`) are yielded as streaming events. + /// The stream ends when the final response (with `id`) is received. + async fn call_streaming( + &self, + method: &str, + params: &Req, + ) -> Result>, A2AError> { + let id = uuid::Uuid::now_v7().to_string(); + let request = JsonRpcRequest::new( + JsonRpcId::String(id.clone()), + method, + Some( + serde_json::to_value(params) + .map_err(|e| A2AError::internal(format!("serialize params: {e}")))?, + ), + ); + + let body = serde_json::to_vec(&request) + .map_err(|e| A2AError::internal(format!("serialize request: {e}")))?; + + // Acquire the request lock for the entire stream lifetime so concurrent + // callers do not interleave on the underlying pipes. The owned guard is + // moved into the background task and dropped when the stream ends. + let request_guard = Arc::clone(&self.request_lock).lock_owned().await; + + // Write request. + { + let mut writer = self.writer.lock().await; + framing::write_frame(&mut *writer, &body) + .await + .map_err(|e| A2AError::internal(format!("write frame: {e}")))?; + } + + // Spawn a background task that reads frames and sends events through + // a channel, yielding them to the caller as they arrive. + let (tx, rx) = tokio::sync::mpsc::channel::>(32); + let reader = Arc::clone(&self.reader); + + tokio::spawn(async move { + // Tie the lock guard to this task so it is released when the stream + // terminates (either at the final response or on error). + let _request_guard = request_guard; + loop { + let frame = { + let mut r = reader.lock().await; + framing::read_frame(&mut *r).await + }; + + let data = match frame { + Ok(Some(data)) => data, + Ok(None) => break, // EOF + Err(e) => { + let _ = tx + .send(Err(A2AError::internal(format!("read frame: {e}")))) + .await; + break; + } + }; + + let value: serde_json::Value = match serde_json::from_slice(&data) { + Ok(v) => v, + Err(e) => { + let _ = tx + .send(Err(A2AError::internal(format!("parse frame: {e}")))) + .await; + break; + } + }; + + // Final response — has a non-null "id" field. + if value.get("id").is_some() && !value["id"].is_null() { + let rpc_response: JsonRpcResponse = match serde_json::from_value(value) { + Ok(r) => r, + Err(e) => { + let _ = tx + .send(Err(A2AError::internal(format!("parse response: {e}")))) + .await; + break; + } + }; + + if let Some(err) = rpc_response.error { + let _ = tx.send(Err(A2AError::new(err.code, err.message))).await; + } else if let Some(result) = rpc_response.result { + if !result.is_null() { + match serde_json::from_value::(result) { + Ok(sr) => { + if tx.send(Ok(sr)).await.is_err() { + // Receiver dropped, no need to continue. + break; + } + } + Err(e) => { + let _ = tx + .send(Err(A2AError::internal(format!( + "failed to parse final stream response: {e}" + )))) + .await; + } + } + } + } + break; + } + + // Notification (no id) — parse params as StreamResponse. + if let Some(params_value) = value.get("params") { + match serde_json::from_value::(params_value.clone()) { + Ok(sr) => { + if tx.send(Ok(sr)).await.is_err() { + break; // Receiver dropped. + } + } + Err(e) => { + let _ = tx + .send(Err(A2AError::internal(format!( + "parse streaming event: {e}" + )))) + .await; + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } +} + +#[async_trait] +impl Transport for StdioTransport { + async fn send_message( + &self, + _params: &ServiceParams, + req: &SendMessageRequest, + ) -> Result { + self.call(stdio_methods::SEND_MESSAGE, req).await + } + + async fn send_streaming_message( + &self, + _params: &ServiceParams, + req: &SendMessageRequest, + ) -> Result>, A2AError> { + self.call_streaming(stdio_methods::SEND_STREAMING_MESSAGE, req) + .await + } + + async fn get_task( + &self, + _params: &ServiceParams, + req: &GetTaskRequest, + ) -> Result { + self.call(stdio_methods::GET_TASK, req).await + } + + async fn list_tasks( + &self, + _params: &ServiceParams, + req: &ListTasksRequest, + ) -> Result { + self.call(stdio_methods::LIST_TASKS, req).await + } + + async fn cancel_task( + &self, + _params: &ServiceParams, + req: &CancelTaskRequest, + ) -> Result { + self.call(stdio_methods::CANCEL_TASK, req).await + } + + async fn subscribe_to_task( + &self, + _params: &ServiceParams, + req: &SubscribeToTaskRequest, + ) -> Result>, A2AError> { + self.call_streaming(stdio_methods::SUBSCRIBE_TO_TASK, req) + .await + } + + async fn create_push_config( + &self, + _params: &ServiceParams, + req: &CreateTaskPushNotificationConfigRequest, + ) -> Result { + self.call(stdio_methods::CREATE_PUSH_CONFIG, req).await + } + + async fn get_push_config( + &self, + _params: &ServiceParams, + req: &GetTaskPushNotificationConfigRequest, + ) -> Result { + self.call(stdio_methods::GET_PUSH_CONFIG, req).await + } + + async fn list_push_configs( + &self, + _params: &ServiceParams, + req: &ListTaskPushNotificationConfigsRequest, + ) -> Result { + self.call(stdio_methods::LIST_PUSH_CONFIGS, req).await + } + + async fn delete_push_config( + &self, + _params: &ServiceParams, + req: &DeleteTaskPushNotificationConfigRequest, + ) -> Result<(), A2AError> { + self.call(stdio_methods::DELETE_PUSH_CONFIG, req).await + } + + async fn get_extended_agent_card( + &self, + _params: &ServiceParams, + req: &GetExtendedAgentCardRequest, + ) -> Result { + self.call(stdio_methods::GET_EXTENDED_AGENT_CARD, req).await + } + + async fn destroy(&self) -> Result<(), A2AError> { + // Close stdin to signal the child to shut down. + { + let mut writer = self.writer.lock().await; + writer + .shutdown() + .await + .map_err(|e| A2AError::internal(format!("shutdown stdin: {e}")))?; + } + + // Wait for the child to exit, but do not block forever — fall back + // to killing it if it does not exit within `CLOSE_TIMEOUT`. + let mut child = self.child.lock().await; + match tokio::time::timeout(CLOSE_TIMEOUT, child.wait()).await { + Ok(res) => { + res.map_err(|e| A2AError::internal(format!("wait for child: {e}")))?; + } + Err(_) => { + let _ = child.kill().await; + // Reap the killed child so it does not become a zombie. + let _ = child.wait().await; + return Err(A2AError::internal(format!( + "subprocess did not exit within {:?}; killed", + CLOSE_TIMEOUT + ))); + } + } + + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// StdioTransportFactory +// --------------------------------------------------------------------------- + +/// Factory that creates [`StdioTransport`] instances. +/// +/// The agent card's interface URL is expected to contain the command to spawn, +/// e.g. `stdio://path/to/agent --flag`. +pub struct StdioTransportFactory; + +#[async_trait] +impl TransportFactory for StdioTransportFactory { + fn protocol(&self) -> &str { + a2a::TRANSPORT_PROTOCOL_STDIO + } + + async fn create( + &self, + _card: &AgentCard, + iface: &AgentInterface, + ) -> Result, A2AError> { + // Parse the URL to extract the command. + // Expected format: "stdio:///path/to/binary?arg1&arg2" or just the path. + let url = &iface.url; + let (program, args) = parse_stdio_url(url)?; + + let transport = StdioTransport::spawn( + &program, + &args.iter().map(|s| s.as_str()).collect::>(), + None, + ) + .await + .map_err(|e| A2AError::internal(format!("failed to spawn stdio transport: {e}")))?; + + Ok(Box::new(transport)) + } +} + +/// Parse a stdio URL into program and args. +/// +/// Supports formats: +/// - `stdio:///path/to/binary` → program="/path/to/binary", args=[] +/// - `stdio:///path/to/binary?arg1&arg2` → program="/path/to/binary", args=["arg1", "arg2"] +/// - `/path/to/binary arg1 arg2` → program="/path/to/binary", args=["arg1", "arg2"] +/// +/// # Limitations +/// +/// Arguments in the `stdio://` form are split on `&` and the path is split on +/// the first `?`. As a result, individual arguments cannot themselves contain +/// the `&` or `?` characters. For commands that need such arguments, use the +/// plain `program arg1 arg2` form instead. +fn parse_stdio_url(url: &str) -> Result<(String, Vec), A2AError> { + if let Some(rest) = url.strip_prefix("stdio://") { + let (path, query) = rest.split_once('?').unwrap_or((rest, "")); + let program = path.to_string(); + let args: Vec = if query.is_empty() { + vec![] + } else { + query.split('&').map(|s| s.to_string()).collect() + }; + Ok((program, args)) + } else { + // Plain command string — split on whitespace. + let parts: Vec<&str> = url.split_whitespace().collect(); + if parts.is_empty() { + return Err(A2AError::internal("empty stdio command")); + } + let program = parts[0].to_string(); + let args = parts[1..].iter().map(|s| s.to_string()).collect(); + Ok((program, args)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_stdio_url_with_scheme() { + let (prog, args) = parse_stdio_url("stdio:///usr/bin/agent").unwrap(); + assert_eq!(prog, "/usr/bin/agent"); + assert!(args.is_empty()); + } + + #[test] + fn test_parse_stdio_url_with_args() { + let (prog, args) = parse_stdio_url("stdio:///usr/bin/agent?--port&8080").unwrap(); + assert_eq!(prog, "/usr/bin/agent"); + assert_eq!(args, vec!["--port", "8080"]); + } + + #[test] + fn test_parse_stdio_url_plain_command() { + let (prog, args) = parse_stdio_url("/usr/bin/agent --port 8080").unwrap(); + assert_eq!(prog, "/usr/bin/agent"); + assert_eq!(args, vec!["--port", "8080"]); + } + + #[test] + fn test_parse_stdio_url_empty() { + let err = parse_stdio_url("").unwrap_err(); + assert!(err.message.contains("empty")); + } +} diff --git a/a2a-stdio/src/errors.rs b/a2a-stdio/src/errors.rs new file mode 100644 index 0000000..d8d3ac1 --- /dev/null +++ b/a2a-stdio/src/errors.rs @@ -0,0 +1,22 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 +use thiserror::Error; + +/// Errors specific to the STDIO transport. +#[derive(Debug, Error)] +pub enum StdioError { + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + + #[error("JSON serialization error: {0}")] + Json(#[from] serde_json::Error), + + #[error("invalid frame header: {0}")] + InvalidHeader(String), + + #[error("handshake failed: {0}")] + HandshakeFailed(String), + + #[error("transport closed")] + Closed, +} diff --git a/a2a-stdio/src/framing.rs b/a2a-stdio/src/framing.rs new file mode 100644 index 0000000..b3cd4bf --- /dev/null +++ b/a2a-stdio/src/framing.rs @@ -0,0 +1,176 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! LSP-style Content-Length framing for the STDIO transport. +//! +//! Every message on the wire is prefixed with HTTP-style headers: +//! +//! ```text +//! Content-Length: \r\n +//! Content-Type: application/json\r\n +//! \r\n +//! +//! ``` +//! +//! `Content-Type` is optional on read (always written for clarity). + +use crate::errors::StdioError; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +const CONTENT_LENGTH_PREFIX: &str = "Content-Length: "; + +/// Maximum size (in bytes) accepted for a single framed message body. +/// +/// Bounds memory allocated from a single peer-supplied `Content-Length` header +/// so a buggy or malicious subprocess cannot trigger a huge allocation by +/// advertising an unrealistic body size. +pub const MAX_FRAME_SIZE: usize = 64 * 1024 * 1024; // 64 MiB + +/// Read a single framed message from the given reader. +/// +/// Returns the raw JSON bytes of the message body. +/// Returns `Ok(None)` when the reader reaches EOF (pipe closed). +pub async fn read_frame( + reader: &mut R, +) -> Result>, StdioError> { + let mut content_length: Option = None; + + // Parse headers until we hit the empty \r\n separator line. + loop { + let mut line = String::new(); + let bytes_read = reader.read_line(&mut line).await?; + if bytes_read == 0 { + // EOF — pipe closed. + return Ok(None); + } + + let trimmed = line.trim_end_matches(['\r', '\n']); + + // Empty line = end of headers. + if trimmed.is_empty() { + break; + } + + if let Some(value) = trimmed.strip_prefix(CONTENT_LENGTH_PREFIX) { + content_length = Some(value.parse::().map_err(|_| { + StdioError::InvalidHeader(format!("invalid Content-Length value: {value}")) + })?); + } + // Ignore unknown headers (e.g. Content-Type) for forward compatibility. + } + + let length = content_length + .ok_or_else(|| StdioError::InvalidHeader("missing Content-Length header".to_string()))?; + + if length > MAX_FRAME_SIZE { + return Err(StdioError::InvalidHeader(format!( + "Content-Length {length} exceeds maximum frame size {MAX_FRAME_SIZE}" + ))); + } + + // Read exactly `length` bytes of body. + let mut body = vec![0u8; length]; + reader.read_exact(&mut body).await?; + + Ok(Some(body)) +} + +/// Write a single framed message to the given writer. +/// +/// Prepends `Content-Length` and `Content-Type` headers, then flushes. +pub async fn write_frame( + writer: &mut W, + body: &[u8], +) -> Result<(), StdioError> { + let header = format!( + "Content-Length: {}\r\nContent-Type: application/json\r\n\r\n", + body.len() + ); + writer.write_all(header.as_bytes()).await?; + writer.write_all(body).await?; + writer.flush().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + use tokio::io::BufReader; + + #[tokio::test] + async fn test_write_then_read_roundtrip() { + let message = b"{\"id\":\"1\",\"method\":\"message/send\"}"; + + // Write a frame into a buffer. + let mut buf = Vec::new(); + write_frame(&mut buf, message).await.unwrap(); + + // Read it back. + let mut reader = BufReader::new(Cursor::new(buf)); + let result = read_frame(&mut reader).await.unwrap().unwrap(); + assert_eq!(result, message); + } + + #[tokio::test] + async fn test_read_eof_returns_none() { + let mut reader = BufReader::new(Cursor::new(Vec::::new())); + let result = read_frame(&mut reader).await.unwrap(); + assert!(result.is_none()); + } + + #[tokio::test] + async fn test_read_missing_content_length() { + let raw = b"Content-Type: application/json\r\n\r\n{}"; + let mut reader = BufReader::new(Cursor::new(raw.to_vec())); + let err = read_frame(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::InvalidHeader(_))); + } + + #[tokio::test] + async fn test_read_invalid_content_length() { + let raw = b"Content-Length: abc\r\n\r\n"; + let mut reader = BufReader::new(Cursor::new(raw.to_vec())); + let err = read_frame(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::InvalidHeader(_))); + } + + #[tokio::test] + async fn test_multiple_frames() { + let mut buf = Vec::new(); + write_frame(&mut buf, b"first").await.unwrap(); + write_frame(&mut buf, b"second").await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let first = read_frame(&mut reader).await.unwrap().unwrap(); + assert_eq!(first, b"first"); + let second = read_frame(&mut reader).await.unwrap().unwrap(); + assert_eq!(second, b"second"); + + // Third read should be EOF. + let eof = read_frame(&mut reader).await.unwrap(); + assert!(eof.is_none()); + } + + #[tokio::test] + async fn test_ignores_unknown_headers() { + let raw = b"Content-Length: 2\r\nX-Custom: foo\r\nContent-Type: application/json\r\n\r\nhi"; + let mut reader = BufReader::new(Cursor::new(raw.to_vec())); + let result = read_frame(&mut reader).await.unwrap().unwrap(); + assert_eq!(result, b"hi"); + } + + #[tokio::test] + async fn test_rejects_oversized_content_length() { + let too_big = MAX_FRAME_SIZE + 1; + let raw = format!("Content-Length: {too_big}\r\n\r\n"); + let mut reader = BufReader::new(Cursor::new(raw.into_bytes())); + let err = read_frame(&mut reader).await.unwrap_err(); + match err { + StdioError::InvalidHeader(msg) => { + assert!(msg.contains("exceeds maximum frame size"), "got: {msg}"); + } + other => panic!("expected InvalidHeader, got {other:?}"), + } + } +} diff --git a/a2a-stdio/src/handshake.rs b/a2a-stdio/src/handshake.rs new file mode 100644 index 0000000..5e4cec2 --- /dev/null +++ b/a2a-stdio/src/handshake.rs @@ -0,0 +1,363 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! Handshake types and logic for the STDIO transport. +//! +//! After the server process starts, it sends a `Handshake` message to the +//! client. The client replies with a `HandshakeAck` to accept or reject +//! the connection and select a protocol variant. + +use crate::errors::StdioError; +use crate::framing; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use tokio::io::{AsyncBufRead, AsyncWrite}; + +/// Features the server advertises during the handshake. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct HandshakeFeatures { + /// Whether the server supports heartbeat pings. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub heartbeat: Option, + + /// Heartbeat interval in seconds (default 30). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub heartbeat_interval_secs: Option, + + /// Additional feature flags for forward compatibility. + #[serde(flatten)] + pub extra: HashMap, +} + +/// Server → Client handshake message, sent immediately after startup. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct Handshake { + /// Must be `"handshake"`. + #[serde(rename = "type")] + pub msg_type: String, + + /// Session identifier (matches `A2A_SESSION_ID` env var if set). + pub session_id: String, + + /// Protocol variants the server supports (e.g. `["a2a/v1"]`). + pub supported_variants: Vec, + + /// Server process ID. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pid: Option, + + /// Optional feature negotiation. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub features: Option, +} + +/// Client → Server handshake acknowledgment. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct HandshakeAck { + /// Must be `"handshakeAck"`. + #[serde(rename = "type")] + pub msg_type: String, + + /// The variant the client selected from `supported_variants`. + pub selected_variant: String, + + /// Whether the client accepts the connection. + pub accept: bool, +} + +impl Handshake { + /// Create a new handshake message. + pub fn new(session_id: String, supported_variants: Vec) -> Self { + Handshake { + msg_type: "handshake".to_string(), + session_id, + supported_variants, + pid: None, + features: None, + } + } +} + +impl HandshakeAck { + /// Create an accepting handshake ack. + pub fn accept(selected_variant: String) -> Self { + HandshakeAck { + msg_type: "handshakeAck".to_string(), + selected_variant, + accept: true, + } + } + + /// Create a rejecting handshake ack. + pub fn reject() -> Self { + HandshakeAck { + msg_type: "handshakeAck".to_string(), + selected_variant: String::new(), + accept: false, + } + } +} + +/// Read a `Handshake` message from the server's stdout. +pub async fn read_handshake( + reader: &mut R, +) -> Result { + let frame = framing::read_frame(reader) + .await? + .ok_or_else(|| StdioError::HandshakeFailed("EOF before handshake".to_string()))?; + + // Parse as generic JSON first for better error messages. + let value: serde_json::Value = serde_json::from_slice(&frame) + .map_err(|e| StdioError::HandshakeFailed(format!("invalid JSON: {e}")))?; + + let msg_type = value.get("type").and_then(|v| v.as_str()).unwrap_or(""); + + if msg_type != "handshake" { + return Err(StdioError::HandshakeFailed(format!( + "expected type \"handshake\", got \"{}\"", + msg_type + ))); + } + + let handshake: Handshake = serde_json::from_value(value) + .map_err(|e| StdioError::HandshakeFailed(format!("invalid handshake: {e}")))?; + + if handshake.supported_variants.is_empty() { + return Err(StdioError::HandshakeFailed( + "server offered no supported variants".to_string(), + )); + } + + Ok(handshake) +} + +/// Write a `Handshake` message (server side). +pub async fn write_handshake( + writer: &mut W, + handshake: &Handshake, +) -> Result<(), StdioError> { + let body = serde_json::to_vec(handshake)?; + framing::write_frame(writer, &body).await +} + +/// Read a `HandshakeAck` from the client's response. +pub async fn read_handshake_ack( + reader: &mut R, +) -> Result { + let frame = framing::read_frame(reader) + .await? + .ok_or_else(|| StdioError::HandshakeFailed("EOF before handshake ack".to_string()))?; + + let value: serde_json::Value = serde_json::from_slice(&frame) + .map_err(|e| StdioError::HandshakeFailed(format!("invalid JSON: {e}")))?; + + let msg_type = value.get("type").and_then(|v| v.as_str()).unwrap_or(""); + + if msg_type != "handshakeAck" { + return Err(StdioError::HandshakeFailed(format!( + "expected type \"handshakeAck\", got \"{}\"", + msg_type + ))); + } + + let ack: HandshakeAck = serde_json::from_value(value) + .map_err(|e| StdioError::HandshakeFailed(format!("invalid handshake ack: {e}")))?; + + Ok(ack) +} + +/// Write a `HandshakeAck` message (client side). +pub async fn write_handshake_ack( + writer: &mut W, + ack: &HandshakeAck, +) -> Result<(), StdioError> { + let body = serde_json::to_vec(ack)?; + framing::write_frame(writer, &body).await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + use tokio::io::BufReader; + + #[test] + fn test_handshake_serde() { + let hs = Handshake::new("sess-1".into(), vec!["a2a/v1".into()]); + let json = serde_json::to_string(&hs).unwrap(); + let back: Handshake = serde_json::from_str(&json).unwrap(); + assert_eq!(hs, back); + assert_eq!(back.msg_type, "handshake"); + } + + #[test] + fn test_handshake_ack_accept_serde() { + let ack = HandshakeAck::accept("a2a/v1".into()); + let json = serde_json::to_string(&ack).unwrap(); + let back: HandshakeAck = serde_json::from_str(&json).unwrap(); + assert!(back.accept); + assert_eq!(back.selected_variant, "a2a/v1"); + } + + #[test] + fn test_handshake_ack_reject_serde() { + let ack = HandshakeAck::reject(); + assert!(!ack.accept); + } + + #[test] + fn test_handshake_with_features() { + let hs = Handshake { + msg_type: "handshake".into(), + session_id: "s1".into(), + supported_variants: vec!["a2a/v1".into()], + pid: Some(1234), + features: Some(HandshakeFeatures { + heartbeat: Some(true), + heartbeat_interval_secs: Some(30), + extra: HashMap::new(), + }), + }; + let json = serde_json::to_string(&hs).unwrap(); + let back: Handshake = serde_json::from_str(&json).unwrap(); + assert_eq!(back.pid, Some(1234)); + assert_eq!(back.features.unwrap().heartbeat, Some(true)); + } + + #[tokio::test] + async fn test_handshake_roundtrip_over_frames() { + let hs = Handshake::new("sess-1".into(), vec!["a2a/v1".into()]); + + // Server writes handshake. + let mut buf = Vec::new(); + write_handshake(&mut buf, &hs).await.unwrap(); + + // Client reads handshake. + let mut reader = BufReader::new(Cursor::new(buf)); + let received = read_handshake(&mut reader).await.unwrap(); + assert_eq!(received, hs); + } + + #[tokio::test] + async fn test_handshake_ack_roundtrip_over_frames() { + let ack = HandshakeAck::accept("a2a/v1".into()); + + // Client writes ack. + let mut buf = Vec::new(); + write_handshake_ack(&mut buf, &ack).await.unwrap(); + + // Server reads ack. + let mut reader = BufReader::new(Cursor::new(buf)); + let received = read_handshake_ack(&mut reader).await.unwrap(); + assert_eq!(received, ack); + } + + #[tokio::test] + async fn test_read_handshake_rejects_wrong_type() { + // Send a handshakeAck where a handshake is expected. + let ack = HandshakeAck::accept("a2a/v1".into()); + let mut buf = Vec::new(); + write_handshake_ack(&mut buf, &ack).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("expected type \"handshake\"")); + } + + #[tokio::test] + async fn test_read_handshake_rejects_missing_fields() { + // Send JSON with correct type but missing required fields. + let json = br#"{"type":"handshake"}"#; + let mut buf = Vec::new(); + framing::write_frame(&mut buf, json).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("invalid handshake")); + } + + #[tokio::test] + async fn test_read_handshake_eof() { + let mut reader = BufReader::new(Cursor::new(Vec::::new())); + let err = read_handshake(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + } + + #[tokio::test] + async fn test_read_handshake_rejects_invalid_json() { + let raw = b"not-json-at-all"; + let mut buf = Vec::new(); + framing::write_frame(&mut buf, raw).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("invalid JSON")); + } + + #[tokio::test] + async fn test_read_handshake_rejects_empty_supported_variants() { + let json = br#"{"type":"handshake","sessionId":"s","supportedVariants":[]}"#; + let mut buf = Vec::new(); + framing::write_frame(&mut buf, json).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake(&mut reader).await.unwrap_err(); + match err { + StdioError::HandshakeFailed(msg) => { + assert!(msg.contains("no supported variants"), "got: {msg}"); + } + other => panic!("expected HandshakeFailed, got {other:?}"), + } + } + + #[tokio::test] + async fn test_read_handshake_ack_eof() { + let mut reader = BufReader::new(Cursor::new(Vec::::new())); + let err = read_handshake_ack(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("EOF")); + } + + #[tokio::test] + async fn test_read_handshake_ack_rejects_invalid_json() { + let raw = b"\xff\xfe garbage"; + let mut buf = Vec::new(); + framing::write_frame(&mut buf, raw).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake_ack(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("invalid JSON")); + } + + #[tokio::test] + async fn test_read_handshake_ack_rejects_wrong_type() { + // Send a handshake where an ack is expected. + let hs = Handshake::new("s".into(), vec!["a2a/v1".into()]); + let mut buf = Vec::new(); + write_handshake(&mut buf, &hs).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake_ack(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("expected type \"handshakeAck\"")); + } + + #[tokio::test] + async fn test_read_handshake_ack_rejects_missing_fields() { + let json = br#"{"type":"handshakeAck"}"#; + let mut buf = Vec::new(); + framing::write_frame(&mut buf, json).await.unwrap(); + + let mut reader = BufReader::new(Cursor::new(buf)); + let err = read_handshake_ack(&mut reader).await.unwrap_err(); + assert!(matches!(err, StdioError::HandshakeFailed(_))); + assert!(err.to_string().contains("invalid handshake ack")); + } +} diff --git a/a2a-stdio/src/lib.rs b/a2a-stdio/src/lib.rs new file mode 100644 index 0000000..19240c7 --- /dev/null +++ b/a2a-stdio/src/lib.rs @@ -0,0 +1,10 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 +pub mod client; +pub mod errors; +pub mod framing; +pub mod handshake; +pub mod server; + +pub use client::{StdioTransport, StdioTransportFactory}; +pub use server::{StdioServer, serve}; diff --git a/a2a-stdio/src/server.rs b/a2a-stdio/src/server.rs new file mode 100644 index 0000000..3d86817 --- /dev/null +++ b/a2a-stdio/src/server.rs @@ -0,0 +1,361 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! STDIO server — reads JSON-RPC requests from stdin, dispatches to a +//! `RequestHandler`, and writes responses to stdout. + +use std::sync::Arc; + +use a2a::*; +use a2a_server::RequestHandler; +use a2a_server::middleware::ServiceParams; +use futures::StreamExt; +use tokio::io::{AsyncWrite, BufReader}; + +use crate::errors::StdioError; +use crate::framing; +use crate::handshake::{self, Handshake, HandshakeFeatures}; + +/// Dispatch a unary (request → response) JSON-RPC method. +/// +/// Deserialises `$params` into `$Req`, calls `$handler.$method(…)`, +/// serialises the response or error, and writes the frame. +/// Uses `continue` to skip to the next request on parse errors, +/// so this macro must be invoked inside the request loop. +macro_rules! dispatch_unary { + ($writer:expr, $id:expr, $handler:expr, $sp:expr, $params:expr, + $Req:ty, $method:ident) => {{ + let req: $Req = match serde_json::from_value($params) { + Ok(r) => r, + Err(e) => { + write_error( + &mut $writer, + $id, + error_code::INVALID_PARAMS, + &format!("invalid params: {e}"), + ) + .await?; + continue; + } + }; + match $handler.$method(&$sp, req).await { + Ok(value) => { + let result_value = serde_json::to_value(&value).map_err(StdioError::Json)?; + let resp = JsonRpcResponse::success($id, result_value); + let body = serde_json::to_vec(&resp)?; + framing::write_frame(&mut $writer, &body).await?; + } + Err(e) => { + write_error(&mut $writer, $id, e.code, &e.message).await?; + } + } + }}; +} + +/// Dispatch a streaming JSON-RPC method. +/// +/// Like `dispatch_unary!` but the handler returns a `BoxStream` of events. +/// Each event is sent as a JSON-RPC notification, followed by a final +/// success response carrying the original request id. +macro_rules! dispatch_streaming { + ($writer:expr, $id:expr, $handler:expr, $sp:expr, $params:expr, + $Req:ty, $method:ident) => {{ + let req: $Req = match serde_json::from_value($params) { + Ok(r) => r, + Err(e) => { + write_error(&mut $writer, $id, error_code::INVALID_PARAMS, &format!("invalid params: {e}")).await?; + continue; + } + }; + match $handler.$method(&$sp, req).await { + Ok(mut stream) => { + while let Some(event) = stream.next().await { + let notification = match &event { + Ok(sr) => { + let params = serde_json::to_value(sr).map_err(StdioError::Json)?; + serde_json::json!({ + "jsonrpc": "2.0", + "method": "event/streamResponse", + "params": params + }) + } + Err(e) => { + serde_json::json!({ + "jsonrpc": "2.0", + "method": "event/streamError", + "params": { + "code": e.code, + "message": e.message + } + }) + } + }; + let body = serde_json::to_vec(¬ification)?; + framing::write_frame(&mut $writer, &body).await?; + } + // Final response indicating stream complete. + let resp = JsonRpcResponse::success($id, serde_json::Value::Null); + let body = serde_json::to_vec(&resp)?; + framing::write_frame(&mut $writer, &body).await?; + } + Err(e) => { + write_error(&mut $writer, $id, e.code, &e.message).await?; + } + } + }}; +} + +/// STDIO server that reads from stdin and writes to stdout. +/// +/// Wraps a `RequestHandler` and dispatches incoming JSON-RPC requests +/// to the appropriate handler method based on the method name. +pub struct StdioServer { + handler: Arc, +} + +impl StdioServer { + pub fn new(handler: Arc) -> Self { + StdioServer { handler } + } + + /// Run the server loop on the given stdin/stdout streams. + /// + /// 1. Sends the handshake to the client. + /// 2. Reads the handshake ack. + /// 3. Enters the main request loop: read request → dispatch → write response. + /// 4. Exits when stdin is closed (EOF). + pub async fn run(&self, reader: R, writer: W) -> Result<(), StdioError> + where + R: tokio::io::AsyncRead + Unpin + Send, + W: AsyncWrite + Unpin + Send, + { + let mut reader = BufReader::new(reader); + let mut writer = writer; + + // Step 1: Send handshake. + let session_id = + std::env::var("A2A_SESSION_ID").unwrap_or_else(|_| uuid::Uuid::now_v7().to_string()); + + let hs = Handshake { + msg_type: "handshake".to_string(), + session_id, + supported_variants: vec!["a2a/v1".to_string()], + pid: Some(std::process::id()), + features: Some(HandshakeFeatures::default()), + }; + handshake::write_handshake(&mut writer, &hs).await?; + + // Step 2: Read handshake ack. + let ack = handshake::read_handshake_ack(&mut reader).await?; + if !ack.accept { + return Err(StdioError::HandshakeFailed( + "client rejected handshake".to_string(), + )); + } + + // Step 3: Main request loop. + let handler = &*self.handler; + loop { + let frame = match framing::read_frame(&mut reader).await? { + Some(f) => f, + None => break, // EOF — client closed stdin. + }; + + let value: serde_json::Value = match serde_json::from_slice(&frame) { + Ok(v) => v, + Err(e) => { + write_error( + &mut writer, + JsonRpcId::Null, + error_code::PARSE_ERROR, + &format!("invalid json: {e}"), + ) + .await?; + continue; + } + }; + + // Parse as a JSON-RPC request. + let rpc_request: JsonRpcRequest = match serde_json::from_value(value) { + Ok(r) => r, + Err(e) => { + write_error( + &mut writer, + JsonRpcId::Null, + error_code::PARSE_ERROR, + &format!("invalid request: {e}"), + ) + .await?; + continue; + } + }; + + let id = rpc_request.id.clone(); + let method = rpc_request.method.clone(); + let params_value = rpc_request.params.unwrap_or(serde_json::Value::Null); + let sp = ServiceParams::new(); + + match method.as_str() { + "message/send" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + SendMessageRequest, + send_message + ); + } + "message/stream" => { + dispatch_streaming!( + writer, + id, + handler, + sp, + params_value, + SendMessageRequest, + send_streaming_message + ); + } + "tasks/get" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + GetTaskRequest, + get_task + ); + } + "tasks/list" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + ListTasksRequest, + list_tasks + ); + } + "tasks/cancel" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + CancelTaskRequest, + cancel_task + ); + } + "tasks/subscribe" => { + dispatch_streaming!( + writer, + id, + handler, + sp, + params_value, + SubscribeToTaskRequest, + subscribe_to_task + ); + } + "tasks/pushNotificationConfig/create" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + CreateTaskPushNotificationConfigRequest, + create_push_config + ); + } + "tasks/pushNotificationConfig/get" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + GetTaskPushNotificationConfigRequest, + get_push_config + ); + } + "tasks/pushNotificationConfig/list" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + ListTaskPushNotificationConfigsRequest, + list_push_configs + ); + } + "tasks/pushNotificationConfig/delete" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + DeleteTaskPushNotificationConfigRequest, + delete_push_config + ); + } + "agent/extendedCard" => { + dispatch_unary!( + writer, + id, + handler, + sp, + params_value, + GetExtendedAgentCardRequest, + get_extended_agent_card + ); + } + _ => { + write_error( + &mut writer, + id, + error_code::METHOD_NOT_FOUND, + &format!("unknown method: {method}"), + ) + .await?; + } + } + } + + Ok(()) + } +} + +/// Write a JSON-RPC error response. +async fn write_error( + writer: &mut (impl AsyncWrite + Unpin + Send), + id: JsonRpcId, + code: i32, + message: &str, +) -> Result<(), StdioError> { + let resp = JsonRpcResponse::error( + id, + JsonRpcError { + code, + message: message.to_string(), + data: None, + }, + ); + let body = serde_json::to_vec(&resp)?; + framing::write_frame(writer, &body).await +} + +/// Convenience function to run a `StdioServer` on the actual process stdin/stdout. +pub async fn serve(handler: Arc) -> Result<(), StdioError> { + let stdin = tokio::io::stdin(); + let stdout = tokio::io::stdout(); + let server = StdioServer::new(handler); + server.run(stdin, stdout).await +} diff --git a/a2a-stdio/tests/bin/stdio_test_helper.rs b/a2a-stdio/tests/bin/stdio_test_helper.rs new file mode 100644 index 0000000..d958fac --- /dev/null +++ b/a2a-stdio/tests/bin/stdio_test_helper.rs @@ -0,0 +1,402 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! Test-only subprocess helper for `agntcy-a2a-stdio` integration tests. +//! +//! This binary is spawned by tests in `tests/spawned.rs` to exercise paths +//! that require a real child process. It is not part of the public API. +//! +//! Modes (selected via the first CLI argument): +//! +//! - `hang-no-handshake` — block forever; never write the handshake. +//! - `handshake-then-hang` — write handshake, read ack, then block forever +//! (ignores stdin EOF so `child.wait()` hangs). +//! - `full-echo` — run a `StdioServer` that implements all 11 +//! `RequestHandler` methods with canned data. `send_message` sleeps +//! `STDIO_TEST_DELAY_MS` ms (default 200) before responding so the +//! request-serialization test can race two calls. +//! - `bad-response` — manually write a JSON-RPC response whose `result` +//! cannot be deserialized into the expected type. Used to exercise the +//! client-side `deserialize result` error branch. +//! - `stream-bad-frame` — handshake, then on the first request write a +//! non-JSON byte payload as a single frame. Exercises the streaming +//! `parse frame` branch in the client background reader. +//! - `stream-bad-notification` — handshake, then on the first request write +//! one streaming notification whose `params` cannot be parsed as a +//! `StreamResponse`, followed by a valid final response. Exercises the +//! streaming `parse streaming event` branch. +//! - `stream-bad-final` — handshake, then on the first request write a +//! final JSON-RPC response (with id) whose `result` is not a +//! `StreamResponse`. Exercises the streaming `failed to parse final +//! stream response` branch. + +use std::env; +use std::sync::Arc; +use std::time::Duration; + +use a2a::event::{StreamResponse, TaskStatusUpdateEvent}; +use a2a::*; +use a2a_server::RequestHandler; +use a2a_server::middleware::ServiceParams; +use a2a_stdio::framing; +use a2a_stdio::handshake::{Handshake, read_handshake_ack}; +use a2a_stdio::{handshake, serve}; +use async_trait::async_trait; +use futures::stream::{self, BoxStream}; +use tokio::io::{BufReader, stdin, stdout}; +use tokio::time::sleep; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let mode = env::args().nth(1).unwrap_or_default(); + match mode.as_str() { + "hang-no-handshake" => hang_no_handshake().await, + "handshake-then-hang" => handshake_then_hang().await, + "full-echo" => full_echo().await, + "bad-response" => bad_response().await, + "stream-bad-frame" => stream_bad_frame().await, + "stream-bad-notification" => stream_bad_notification().await, + "stream-bad-final" => stream_bad_final().await, + other => { + eprintln!("stdio_test_helper: unknown mode {other:?}"); + std::process::exit(2); + } + } +} + +async fn hang_no_handshake() { + std::future::pending::<()>().await; +} + +async fn handshake_then_hang() { + let mut writer = stdout(); + let mut reader = BufReader::new(stdin()); + + let hs = Handshake::new("test-session".to_string(), vec!["a2a/v1".to_string()]); + handshake::write_handshake(&mut writer, &hs).await.unwrap(); + let _ = read_handshake_ack(&mut reader).await; + + std::future::pending::<()>().await; +} + +async fn full_echo() { + let delay_ms: u64 = env::var("STDIO_TEST_DELAY_MS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(200); + let handler = Arc::new(EchoHandler { + send_delay: Duration::from_millis(delay_ms), + }); + // Drive the server through the public `serve()` convenience function so + // that path is covered by the spawned-subprocess tests. + let _ = serve(handler).await; +} + +/// Performs the handshake, reads one request, then writes a single frame +/// containing non-JSON bytes. The client streaming reader hits the +/// `parse frame` error branch. +async fn stream_bad_frame() { + let mut writer = stdout(); + let mut reader = BufReader::new(stdin()); + + let hs = Handshake::new("test-session".to_string(), vec!["a2a/v1".to_string()]); + handshake::write_handshake(&mut writer, &hs).await.unwrap(); + let _ = read_handshake_ack(&mut reader).await; + + let _ = framing::read_frame(&mut reader).await.unwrap().unwrap(); + framing::write_frame(&mut writer, b"not-json-at-all") + .await + .unwrap(); +} + +/// Performs the handshake, reads one request, then writes one streaming +/// notification whose `params` cannot be parsed as a `StreamResponse`, +/// followed by a valid final response. Exercises the +/// `parse streaming event` branch in the client background reader. +async fn stream_bad_notification() { + let mut writer = stdout(); + let mut reader = BufReader::new(stdin()); + + let hs = Handshake::new("test-session".to_string(), vec!["a2a/v1".to_string()]); + handshake::write_handshake(&mut writer, &hs).await.unwrap(); + let _ = read_handshake_ack(&mut reader).await; + + let frame = framing::read_frame(&mut reader).await.unwrap().unwrap(); + let value: serde_json::Value = serde_json::from_slice(&frame).unwrap(); + let id = value.get("id").cloned().unwrap_or(serde_json::Value::Null); + + // Streaming notification (no id) with garbage params. + let notif = serde_json::json!({ + "jsonrpc": "2.0", + "method": "stream/event", + "params": "not-a-stream-response", + }); + framing::write_frame(&mut writer, &serde_json::to_vec(¬if).unwrap()) + .await + .unwrap(); + + // Final response that ends the stream cleanly. + let final_resp = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "result": serde_json::Value::Null, + }); + framing::write_frame(&mut writer, &serde_json::to_vec(&final_resp).unwrap()) + .await + .unwrap(); +} + +/// Performs the handshake, reads one request, then writes a final JSON-RPC +/// response (with id) whose `result` is not a valid `StreamResponse`. +/// Exercises the `failed to parse final stream response` branch. +async fn stream_bad_final() { + let mut writer = stdout(); + let mut reader = BufReader::new(stdin()); + + let hs = Handshake::new("test-session".to_string(), vec!["a2a/v1".to_string()]); + handshake::write_handshake(&mut writer, &hs).await.unwrap(); + let _ = read_handshake_ack(&mut reader).await; + + let frame = framing::read_frame(&mut reader).await.unwrap().unwrap(); + let value: serde_json::Value = serde_json::from_slice(&frame).unwrap(); + let id = value.get("id").cloned().unwrap_or(serde_json::Value::Null); + + let resp = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "result": "not-a-stream-response", + }); + framing::write_frame(&mut writer, &serde_json::to_vec(&resp).unwrap()) + .await + .unwrap(); +} + +/// Performs the handshake, then on the first request writes back a JSON-RPC +/// response whose `result` is a JSON string instead of the expected object. +/// Exercises the client's `deserialize result` error path. +async fn bad_response() { + let mut writer = stdout(); + let mut reader = BufReader::new(stdin()); + + let hs = Handshake::new("test-session".to_string(), vec!["a2a/v1".to_string()]); + handshake::write_handshake(&mut writer, &hs).await.unwrap(); + let _ = read_handshake_ack(&mut reader).await; + + let frame = framing::read_frame(&mut reader).await.unwrap().unwrap(); + let value: serde_json::Value = serde_json::from_slice(&frame).unwrap(); + let id = value.get("id").cloned().unwrap_or(serde_json::Value::Null); + + let resp = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "result": "this-is-not-a-task-object", + }); + let body = serde_json::to_vec(&resp).unwrap(); + framing::write_frame(&mut writer, &body).await.unwrap(); +} + +// --------------------------------------------------------------------------- +// EchoHandler — canned responses for all 11 RequestHandler methods. +// --------------------------------------------------------------------------- + +struct EchoHandler { + send_delay: Duration, +} + +fn echo_task(id: &str, state: TaskState) -> Task { + Task { + id: id.to_string(), + context_id: "ctx".to_string(), + status: TaskStatus { + state, + message: None, + timestamp: None, + }, + artifacts: None, + history: None, + metadata: None, + } +} + +#[async_trait] +impl RequestHandler for EchoHandler { + async fn send_message( + &self, + _params: &ServiceParams, + req: SendMessageRequest, + ) -> Result { + sleep(self.send_delay).await; + let task_id = req + .message + .task_id + .clone() + .unwrap_or_else(|| "unknown".to_string()); + Ok(SendMessageResponse::Task(echo_task( + &task_id, + TaskState::Completed, + ))) + } + + async fn send_streaming_message( + &self, + _params: &ServiceParams, + req: SendMessageRequest, + ) -> Result>, A2AError> { + let task_id = req + .message + .task_id + .clone() + .unwrap_or_else(|| "stream-task".to_string()); + Ok(Box::pin(stream::iter(vec![ + Ok(StreamResponse::StatusUpdate(TaskStatusUpdateEvent { + task_id: task_id.clone(), + context_id: "ctx".to_string(), + status: TaskStatus { + state: TaskState::Working, + message: None, + timestamp: None, + }, + metadata: None, + })), + Ok(StreamResponse::Task(echo_task( + &task_id, + TaskState::Completed, + ))), + ]))) + } + + async fn get_task( + &self, + _params: &ServiceParams, + req: GetTaskRequest, + ) -> Result { + if req.id == "missing" { + return Err(A2AError::task_not_found(&req.id)); + } + Ok(echo_task(&req.id, TaskState::Completed)) + } + + async fn list_tasks( + &self, + _params: &ServiceParams, + _req: ListTasksRequest, + ) -> Result { + Ok(ListTasksResponse { + tasks: vec![echo_task("task-1", TaskState::Completed)], + next_page_token: "next".to_string(), + page_size: 1, + total_size: 1, + }) + } + + async fn cancel_task( + &self, + _params: &ServiceParams, + req: CancelTaskRequest, + ) -> Result { + Ok(echo_task(&req.id, TaskState::Canceled)) + } + + async fn subscribe_to_task( + &self, + _params: &ServiceParams, + req: SubscribeToTaskRequest, + ) -> Result>, A2AError> { + Ok(Box::pin(stream::iter(vec![ + Ok(StreamResponse::StatusUpdate(TaskStatusUpdateEvent { + task_id: req.id.clone(), + context_id: "ctx".to_string(), + status: TaskStatus { + state: TaskState::Working, + message: None, + timestamp: None, + }, + metadata: None, + })), + Ok(StreamResponse::Task(echo_task( + &req.id, + TaskState::Completed, + ))), + ]))) + } + + async fn create_push_config( + &self, + _params: &ServiceParams, + req: CreateTaskPushNotificationConfigRequest, + ) -> Result { + Ok(TaskPushNotificationConfig { + task_id: req.task_id, + tenant: req.tenant, + config: req.config, + }) + } + + async fn get_push_config( + &self, + _params: &ServiceParams, + req: GetTaskPushNotificationConfigRequest, + ) -> Result { + Ok(TaskPushNotificationConfig { + task_id: req.task_id, + tenant: req.tenant, + config: PushNotificationConfig { + url: "https://example.com/cb".to_string(), + id: Some(req.id), + token: None, + authentication: None, + }, + }) + } + + async fn list_push_configs( + &self, + _params: &ServiceParams, + req: ListTaskPushNotificationConfigsRequest, + ) -> Result { + Ok(ListTaskPushNotificationConfigsResponse { + configs: vec![TaskPushNotificationConfig { + task_id: req.task_id, + tenant: req.tenant, + config: PushNotificationConfig { + url: "https://example.com/cb".to_string(), + id: Some("cfg-1".to_string()), + token: None, + authentication: None, + }, + }], + next_page_token: None, + }) + } + + async fn delete_push_config( + &self, + _params: &ServiceParams, + _req: DeleteTaskPushNotificationConfigRequest, + ) -> Result<(), A2AError> { + Ok(()) + } + + async fn get_extended_agent_card( + &self, + _params: &ServiceParams, + _req: GetExtendedAgentCardRequest, + ) -> Result { + Ok(AgentCard { + name: "echo-agent".to_string(), + description: "test".to_string(), + version: "0.0.1".to_string(), + supported_interfaces: vec![], + capabilities: AgentCapabilities::default(), + default_input_modes: vec!["text/plain".to_string()], + default_output_modes: vec!["text/plain".to_string()], + skills: vec![], + provider: None, + documentation_url: None, + icon_url: None, + security_schemes: None, + security_requirements: None, + signatures: None, + }) + } +} diff --git a/a2a-stdio/tests/e2e.rs b/a2a-stdio/tests/e2e.rs new file mode 100644 index 0000000..507669e --- /dev/null +++ b/a2a-stdio/tests/e2e.rs @@ -0,0 +1,794 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! End-to-end tests for the STDIO transport. +//! +//! These tests wire the `StdioServer` to an in-process `tokio::io::duplex` +//! pair so we can exercise the full request→dispatch→response path without +//! spawning a real subprocess. + +use std::sync::Arc; + +use a2a::event::{StreamResponse, TaskStatusUpdateEvent}; +use a2a::*; +use a2a_server::RequestHandler; +use a2a_server::middleware::ServiceParams; +use a2a_stdio::StdioServer; +use a2a_stdio::errors::StdioError; +use a2a_stdio::framing; +use a2a_stdio::handshake::{self, HandshakeAck}; +use async_trait::async_trait; +use futures::stream::{self, BoxStream}; +use tokio::io::BufReader; + +// --------------------------------------------------------------------------- +// Helpers (same canned data used across the workspace's e2e tests) +// --------------------------------------------------------------------------- + +fn sample_message(role: Role, text: &str, task_id: &str) -> Message { + Message { + message_id: format!("msg-{text}"), + context_id: Some("ctx-1".to_string()), + task_id: Some(task_id.to_string()), + role, + parts: vec![Part::text(text)], + metadata: None, + extensions: None, + reference_task_ids: None, + } +} + +fn sample_task(id: &str, state: TaskState) -> Task { + Task { + id: id.to_string(), + context_id: "ctx-1".to_string(), + status: TaskStatus { + state, + message: Some(sample_message(Role::Agent, "done", id)), + timestamp: None, + }, + artifacts: None, + history: None, + metadata: None, + } +} + +fn sample_push_config(id: &str) -> TaskPushNotificationConfig { + TaskPushNotificationConfig { + task_id: "task-1".to_string(), + tenant: None, + config: PushNotificationConfig { + url: "https://example.com/callback".to_string(), + id: Some(id.to_string()), + token: Some("tok-1".to_string()), + authentication: None, + }, + } +} + +fn sample_agent_card() -> AgentCard { + AgentCard { + name: "STDIO Test Agent".to_string(), + description: "Integration test agent".to_string(), + version: VERSION.to_string(), + supported_interfaces: vec![AgentInterface::new( + "stdio://test-agent", + TRANSPORT_PROTOCOL_STDIO, + )], + capabilities: AgentCapabilities::default(), + default_input_modes: vec!["text/plain".to_string()], + default_output_modes: vec!["text/plain".to_string()], + skills: vec![], + provider: None, + documentation_url: None, + icon_url: None, + security_schemes: None, + security_requirements: None, + signatures: None, + } +} + +fn send_message_request(task_id: &str) -> SendMessageRequest { + SendMessageRequest { + message: sample_message(Role::User, "hello", task_id), + configuration: None, + metadata: None, + tenant: None, + } +} + +// --------------------------------------------------------------------------- +// TestHandler — implements all 11 RequestHandler methods with canned data +// --------------------------------------------------------------------------- + +struct TestHandler; + +#[async_trait] +impl RequestHandler for TestHandler { + async fn send_message( + &self, + _params: &ServiceParams, + req: SendMessageRequest, + ) -> Result { + let task_id = req + .message + .task_id + .clone() + .unwrap_or_else(|| "task-1".to_string()); + Ok(SendMessageResponse::Task(sample_task( + &task_id, + TaskState::Completed, + ))) + } + + async fn send_streaming_message( + &self, + _params: &ServiceParams, + _req: SendMessageRequest, + ) -> Result>, A2AError> { + Ok(Box::pin(stream::iter(vec![ + Ok(StreamResponse::StatusUpdate(TaskStatusUpdateEvent { + task_id: "task-1".to_string(), + context_id: "ctx-1".to_string(), + status: TaskStatus { + state: TaskState::Working, + message: None, + timestamp: None, + }, + metadata: None, + })), + Ok(StreamResponse::Task(sample_task( + "task-1", + TaskState::Completed, + ))), + ]))) + } + + async fn get_task( + &self, + _params: &ServiceParams, + req: GetTaskRequest, + ) -> Result { + if req.id == "missing" { + return Err(A2AError::task_not_found(&req.id)); + } + Ok(sample_task(&req.id, TaskState::Completed)) + } + + async fn list_tasks( + &self, + _params: &ServiceParams, + _req: ListTasksRequest, + ) -> Result { + Ok(ListTasksResponse { + tasks: vec![sample_task("task-1", TaskState::Completed)], + next_page_token: "next-page".to_string(), + page_size: 1, + total_size: 1, + }) + } + + async fn cancel_task( + &self, + _params: &ServiceParams, + req: CancelTaskRequest, + ) -> Result { + if req.id == "missing" { + return Err(A2AError::task_not_found(&req.id)); + } + Ok(sample_task(&req.id, TaskState::Canceled)) + } + + async fn subscribe_to_task( + &self, + _params: &ServiceParams, + req: SubscribeToTaskRequest, + ) -> Result>, A2AError> { + if req.id == "missing" { + return Err(A2AError::task_not_found(&req.id)); + } + Ok(Box::pin(stream::once(async move { + Ok(StreamResponse::StatusUpdate(TaskStatusUpdateEvent { + task_id: req.id, + context_id: "ctx-1".to_string(), + status: TaskStatus { + state: TaskState::Working, + message: None, + timestamp: None, + }, + metadata: None, + })) + }))) + } + + async fn create_push_config( + &self, + _params: &ServiceParams, + req: CreateTaskPushNotificationConfigRequest, + ) -> Result { + Ok(TaskPushNotificationConfig { + task_id: req.task_id, + config: req.config, + tenant: req.tenant, + }) + } + + async fn get_push_config( + &self, + _params: &ServiceParams, + req: GetTaskPushNotificationConfigRequest, + ) -> Result { + Ok(TaskPushNotificationConfig { + task_id: req.task_id, + tenant: req.tenant, + config: PushNotificationConfig { + url: "https://example.com/callback".to_string(), + id: Some(req.id), + token: Some("tok-1".to_string()), + authentication: None, + }, + }) + } + + async fn list_push_configs( + &self, + _params: &ServiceParams, + req: ListTaskPushNotificationConfigsRequest, + ) -> Result { + Ok(ListTaskPushNotificationConfigsResponse { + configs: vec![TaskPushNotificationConfig { + task_id: req.task_id, + tenant: req.tenant, + ..sample_push_config("cfg-1") + }], + next_page_token: Some("next".to_string()), + }) + } + + async fn delete_push_config( + &self, + _params: &ServiceParams, + req: DeleteTaskPushNotificationConfigRequest, + ) -> Result<(), A2AError> { + if req.id == "missing" { + return Err(A2AError::task_not_found(&req.id)); + } + Ok(()) + } + + async fn get_extended_agent_card( + &self, + _params: &ServiceParams, + _req: GetExtendedAgentCardRequest, + ) -> Result { + Ok(sample_agent_card()) + } +} + +// --------------------------------------------------------------------------- +// Test client helper — speaks the STDIO protocol over raw reader/writer +// --------------------------------------------------------------------------- + +/// A minimal test client that performs the handshake and sends/receives +/// JSON-RPC frames. This avoids depending on `StdioTransport` (which +/// requires a real subprocess) and lets us test the server in-process. +struct TestClient { + reader: BufReader, + writer: W, +} + +impl TestClient +where + R: tokio::io::AsyncRead + Unpin + Send, + W: tokio::io::AsyncWrite + Unpin + Send, +{ + /// Connect by reading the server's handshake and sending an ack. + async fn connect(reader: R, writer: W) -> Result { + let mut reader = BufReader::new(reader); + let mut writer = writer; + + // Read server's handshake. + let hs = handshake::read_handshake(&mut reader).await?; + assert_eq!(hs.msg_type, "handshake"); + assert!(hs.supported_variants.contains(&"a2a/v1".to_string())); + + // Send ack. + let ack = HandshakeAck::accept("a2a/v1".to_string()); + handshake::write_handshake_ack(&mut writer, &ack).await?; + + Ok(Self { reader, writer }) + } + + /// Send a JSON-RPC request and read a single response frame. + async fn call(&mut self, method: &str, params: serde_json::Value) -> JsonRpcResponse { + let id = JsonRpcId::String(uuid::Uuid::now_v7().to_string()); + let request = JsonRpcRequest::new(id, method, Some(params)); + + let body = serde_json::to_vec(&request).unwrap(); + framing::write_frame(&mut self.writer, &body).await.unwrap(); + + let frame = framing::read_frame(&mut self.reader) + .await + .unwrap() + .expect("expected response frame"); + serde_json::from_slice(&frame).unwrap() + } + + /// Send a JSON-RPC request and collect all notification frames + /// until a response with an `id` arrives (the final frame). + async fn call_streaming( + &mut self, + method: &str, + params: serde_json::Value, + ) -> (Vec, JsonRpcResponse) { + let id = JsonRpcId::String(uuid::Uuid::now_v7().to_string()); + let request = JsonRpcRequest::new(id, method, Some(params)); + + let body = serde_json::to_vec(&request).unwrap(); + framing::write_frame(&mut self.writer, &body).await.unwrap(); + + let mut notifications = Vec::new(); + loop { + let frame = framing::read_frame(&mut self.reader) + .await + .unwrap() + .expect("expected frame"); + let value: serde_json::Value = serde_json::from_slice(&frame).unwrap(); + + // Notifications have "method" but no "id". + // The final response has "id". + if value.get("id").is_some() { + let resp: JsonRpcResponse = serde_json::from_value(value).unwrap(); + return (notifications, resp); + } + notifications.push(value); + } + } +} + +// --------------------------------------------------------------------------- +// Helper to start a server + client pair over duplex channels +// --------------------------------------------------------------------------- + +async fn setup() -> ( + tokio::task::JoinHandle>, + TestClient, +) { + // duplex(1) = "server_to_client": server writes, client reads + // duplex(2) = "client_to_server": client writes, server reads + let (server_writer, client_reader) = tokio::io::duplex(64 * 1024); + let (client_writer, server_reader) = tokio::io::duplex(64 * 1024); + + let handler = Arc::new(TestHandler); + let server = StdioServer::new(handler); + + let server_handle = tokio::spawn(async move { server.run(server_reader, server_writer).await }); + + let client = TestClient::connect(client_reader, client_writer) + .await + .expect("handshake should succeed"); + + (server_handle, client) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn stdio_unary_send_message() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(send_message_request("task-1")).unwrap(); + let resp = client.call("message/send", params).await; + + assert!(resp.error.is_none(), "expected success: {:?}", resp.error); + let result: SendMessageResponse = serde_json::from_value(resp.result.unwrap()).unwrap(); + match result { + SendMessageResponse::Task(t) => { + assert_eq!(t.id, "task-1"); + assert_eq!(t.status.state, TaskState::Completed); + } + other => panic!("expected Task variant, got: {other:?}"), + } +} + +#[tokio::test] +async fn stdio_unary_get_task() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&GetTaskRequest { + id: "task-42".to_string(), + history_length: None, + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/get", params).await; + + assert!(resp.error.is_none()); + let task: Task = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(task.id, "task-42"); +} + +#[tokio::test] +async fn stdio_unary_get_task_not_found() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&GetTaskRequest { + id: "missing".to_string(), + history_length: None, + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/get", params).await; + + assert!(resp.error.is_some()); + let err = resp.error.unwrap(); + assert_eq!(err.code, error_code::TASK_NOT_FOUND); +} + +#[tokio::test] +async fn stdio_unary_list_tasks() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&ListTasksRequest { + context_id: Some("ctx-1".to_string()), + status: None, + page_size: None, + page_token: None, + history_length: None, + status_timestamp_after: None, + include_artifacts: None, + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/list", params).await; + + assert!(resp.error.is_none()); + let list: ListTasksResponse = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(list.tasks.len(), 1); + assert_eq!(list.tasks[0].id, "task-1"); +} + +#[tokio::test] +async fn stdio_unary_cancel_task() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&CancelTaskRequest { + id: "task-1".to_string(), + metadata: None, + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/cancel", params).await; + + assert!(resp.error.is_none()); + let task: Task = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(task.status.state, TaskState::Canceled); +} + +#[tokio::test] +async fn stdio_unary_push_config_crud() { + let (_server, mut client) = setup().await; + + // Create + let params = serde_json::to_value(&CreateTaskPushNotificationConfigRequest { + task_id: "task-1".to_string(), + config: sample_push_config("cfg-1").config, + tenant: None, + }) + .unwrap(); + let resp = client + .call("tasks/pushNotificationConfig/create", params) + .await; + assert!(resp.error.is_none()); + let cfg: TaskPushNotificationConfig = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(cfg.task_id, "task-1"); + + // Get + let params = serde_json::to_value(&GetTaskPushNotificationConfigRequest { + task_id: "task-1".to_string(), + id: "cfg-1".to_string(), + tenant: None, + }) + .unwrap(); + let resp = client + .call("tasks/pushNotificationConfig/get", params) + .await; + assert!(resp.error.is_none()); + + // List + let params = serde_json::to_value(&ListTaskPushNotificationConfigsRequest { + task_id: "task-1".to_string(), + page_size: None, + page_token: None, + tenant: None, + }) + .unwrap(); + let resp = client + .call("tasks/pushNotificationConfig/list", params) + .await; + assert!(resp.error.is_none()); + let list: ListTaskPushNotificationConfigsResponse = + serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(list.configs.len(), 1); + + // Delete + let params = serde_json::to_value(&DeleteTaskPushNotificationConfigRequest { + task_id: "task-1".to_string(), + id: "cfg-1".to_string(), + tenant: None, + }) + .unwrap(); + let resp = client + .call("tasks/pushNotificationConfig/delete", params) + .await; + assert!(resp.error.is_none()); +} + +#[tokio::test] +async fn stdio_unary_get_extended_agent_card() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&GetExtendedAgentCardRequest { tenant: None }).unwrap(); + let resp = client.call("agent/extendedCard", params).await; + + assert!(resp.error.is_none()); + let card: AgentCard = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(card.name, "STDIO Test Agent"); +} + +#[tokio::test] +async fn stdio_streaming_send_message() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(send_message_request("task-1")).unwrap(); + let (notifications, final_resp) = client.call_streaming("message/stream", params).await; + + // We expect 2 notifications (Working + Completed) then a final success. + assert_eq!(notifications.len(), 2, "expected 2 stream events"); + + // First event should be a status update (Working). + let first_method = notifications[0]["method"].as_str().unwrap(); + assert_eq!(first_method, "event/streamResponse"); + + // Final response should be success. + assert!(final_resp.error.is_none()); +} + +#[tokio::test] +async fn stdio_streaming_subscribe_to_task() { + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&SubscribeToTaskRequest { + id: "task-1".to_string(), + tenant: None, + }) + .unwrap(); + let (notifications, final_resp) = client.call_streaming("tasks/subscribe", params).await; + + assert_eq!(notifications.len(), 1); + assert!(final_resp.error.is_none()); +} + +#[tokio::test] +async fn stdio_unknown_method_returns_error() { + let (_server, mut client) = setup().await; + + let resp = client.call("unknown/method", serde_json::json!({})).await; + + assert!(resp.error.is_some()); + let err = resp.error.unwrap(); + assert_eq!(err.code, error_code::METHOD_NOT_FOUND); + assert!(err.message.contains("unknown/method")); +} + +#[tokio::test] +async fn stdio_invalid_params_returns_error() { + let (_server, mut client) = setup().await; + + // Send garbage params to a typed method. + let resp = client + .call("tasks/get", serde_json::json!({"wrong_field": true})) + .await; + + assert!(resp.error.is_some()); + let err = resp.error.unwrap(); + assert_eq!(err.code, error_code::INVALID_PARAMS); +} + +#[tokio::test] +async fn stdio_server_exits_on_eof() { + let (server_writer, client_reader) = tokio::io::duplex(64 * 1024); + let (client_writer, server_reader) = tokio::io::duplex(64 * 1024); + + let handler = Arc::new(TestHandler); + let server = StdioServer::new(handler); + + let server_handle = tokio::spawn(async move { server.run(server_reader, server_writer).await }); + + // Complete the handshake, then immediately drop the client. + { + let _client = TestClient::connect(client_reader, client_writer) + .await + .expect("handshake should succeed"); + } + // client is dropped here — server should see EOF and return Ok(()). + + let result = server_handle.await.unwrap(); + assert!( + result.is_ok(), + "server should exit cleanly on EOF: {result:?}" + ); +} + +#[tokio::test] +async fn stdio_handshake_reject() { + let (server_writer, client_reader) = tokio::io::duplex(64 * 1024); + let (client_writer, server_reader) = tokio::io::duplex(64 * 1024); + + let handler = Arc::new(TestHandler); + let server = StdioServer::new(handler); + + let server_handle = tokio::spawn(async move { server.run(server_reader, server_writer).await }); + + // Read the server's handshake, then send a reject ack. + let mut reader = BufReader::new(client_reader); + let mut writer = client_writer; + + let _hs = handshake::read_handshake(&mut reader).await.unwrap(); + let ack = HandshakeAck { + msg_type: "handshakeAck".to_string(), + selected_variant: String::new(), + accept: false, + }; + handshake::write_handshake_ack(&mut writer, &ack) + .await + .unwrap(); + + let result = server_handle.await.unwrap(); + assert!(result.is_err(), "server should fail when client rejects"); + match result.unwrap_err() { + StdioError::HandshakeFailed(msg) => { + assert!(msg.contains("rejected"), "msg: {msg}"); + } + other => panic!("expected HandshakeFailed, got: {other:?}"), + } +} + +#[tokio::test] +async fn stdio_multiple_requests_in_sequence() { + let (_server, mut client) = setup().await; + + // Send multiple requests on the same connection. + for i in 0..5 { + let task_id = format!("task-{i}"); + let params = serde_json::to_value(&GetTaskRequest { + id: task_id.clone(), + history_length: None, + tenant: None, + }) + .unwrap(); + + let resp = client.call("tasks/get", params).await; + assert!(resp.error.is_none()); + let task: Task = serde_json::from_value(resp.result.unwrap()).unwrap(); + assert_eq!(task.id, task_id); + } +} + +// --------------------------------------------------------------------------- +// Server-side coverage: error branches in the request loop and dispatchers +// --------------------------------------------------------------------------- + +/// Send a raw frame to the server bypassing TestClient::call, then read the +/// next response frame back. Returns the parsed JsonRpcResponse. +async fn send_raw_and_read( + client: &mut TestClient, + frame: &[u8], +) -> JsonRpcResponse { + framing::write_frame(&mut client.writer, frame) + .await + .unwrap(); + let resp_frame = framing::read_frame(&mut client.reader) + .await + .unwrap() + .expect("expected response frame"); + serde_json::from_slice(&resp_frame).unwrap() +} + +#[tokio::test] +async fn stdio_server_replies_parse_error_for_malformed_json() { + let (_server, mut client) = setup().await; + + // Body that frames cleanly but is not valid JSON. + let resp = send_raw_and_read(&mut client, b"not-json-at-all").await; + let err = resp.error.expect("server should send a JSON-RPC error"); + assert_eq!(err.code, error_code::PARSE_ERROR); + assert!(err.message.to_lowercase().contains("invalid json")); +} + +#[tokio::test] +async fn stdio_server_replies_parse_error_for_non_jsonrpc_object() { + let (_server, mut client) = setup().await; + + // Valid JSON object that does not deserialize into JsonRpcRequest. + let resp = send_raw_and_read(&mut client, br#"{"hello":"world"}"#).await; + let err = resp.error.expect("server should send a JSON-RPC error"); + assert_eq!(err.code, error_code::PARSE_ERROR); + assert!(err.message.to_lowercase().contains("invalid request")); +} + +#[tokio::test] +async fn stdio_server_recovers_after_malformed_json() { + let (_server, mut client) = setup().await; + + // First frame: garbage. Second frame: a valid request. The server must + // continue serving after the parse-error response instead of exiting. + let resp1 = send_raw_and_read(&mut client, b"").await; + assert_eq!(resp1.error.unwrap().code, error_code::PARSE_ERROR); + + let params = serde_json::to_value(&GetTaskRequest { + id: "task-after-garbage".to_string(), + history_length: None, + tenant: None, + }) + .unwrap(); + let resp2 = client.call("tasks/get", params).await; + assert!(resp2.error.is_none(), "server should still serve requests"); + let task: Task = serde_json::from_value(resp2.result.unwrap()).unwrap(); + assert_eq!(task.id, "task-after-garbage"); +} + +#[tokio::test] +async fn stdio_streaming_invalid_params_returns_error() { + let (_server, mut client) = setup().await; + + // Hits the dispatch_streaming! invalid-params branch. + let resp = client + .call("message/stream", serde_json::json!({"foo": 1})) + .await; + let err = resp.error.expect("expected INVALID_PARAMS"); + assert_eq!(err.code, error_code::INVALID_PARAMS); +} + +#[tokio::test] +async fn stdio_subscribe_invalid_params_returns_error() { + let (_server, mut client) = setup().await; + + let resp = client.call("tasks/subscribe", serde_json::json!({})).await; + let err = resp.error.expect("expected INVALID_PARAMS"); + assert_eq!(err.code, error_code::INVALID_PARAMS); +} + +#[tokio::test] +async fn stdio_subscribe_handler_error_returns_error_response() { + // Hits the dispatch_streaming! handler-error branch (TestHandler returns + // task_not_found for id == "missing"). + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&SubscribeToTaskRequest { + id: "missing".to_string(), + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/subscribe", params).await; + let err = resp.error.expect("expected handler error"); + assert!(err.message.to_lowercase().contains("missing") || err.code != 0); +} + +#[tokio::test] +async fn stdio_get_task_handler_error_returns_error_response() { + // Hits the dispatch_unary! handler-error branch. + let (_server, mut client) = setup().await; + + let params = serde_json::to_value(&GetTaskRequest { + id: "missing".to_string(), + history_length: None, + tenant: None, + }) + .unwrap(); + let resp = client.call("tasks/get", params).await; + let err = resp.error.expect("expected handler error"); + assert!(err.message.to_lowercase().contains("missing") || err.code != 0); +} diff --git a/a2a-stdio/tests/spawned.rs b/a2a-stdio/tests/spawned.rs new file mode 100644 index 0000000..6280ac4 --- /dev/null +++ b/a2a-stdio/tests/spawned.rs @@ -0,0 +1,608 @@ +// Copyright AGNTCY Contributors (https://github.com/agntcy) +// SPDX-License-Identifier: Apache-2.0 + +//! Integration tests that spawn a real subprocess via `StdioTransport`. +//! +//! These tests exercise paths that cannot be reached through the in-process +//! `tokio::io::duplex` setup used in `tests/e2e.rs`: +//! +//! - `HANDSHAKE_TIMEOUT` firing when the subprocess never sends a handshake. +//! - `CLOSE_TIMEOUT` + kill fallback when the subprocess ignores stdin EOF. +//! - The request-level lock that serializes concurrent client calls. +//! - Every method on the `Transport` trait against a real subprocess +//! (so all of `client.rs` is hit at least once). +//! - `StdioTransportFactory::create` end-to-end. +//! - Client-side error branches (`deserialize result`). +//! +//! All tests cap their own runtime with `tokio::time::timeout` so a +//! regression cannot hang CI. + +use std::time::{Duration, Instant}; + +use a2a::*; +use a2a_client::transport::{ServiceParams, Transport, TransportFactory}; +use a2a_stdio::{StdioTransport, StdioTransportFactory}; +use futures::StreamExt; + +/// Path to the helper binary, resolved at compile time by Cargo. +const HELPER: &str = env!("CARGO_BIN_EXE_stdio_test_helper"); + +// --------------------------------------------------------------------------- +// Sample request builders +// --------------------------------------------------------------------------- + +fn sample_send_message_request(task_id: &str) -> SendMessageRequest { + SendMessageRequest { + message: Message { + message_id: format!("msg-{task_id}"), + context_id: Some("ctx".to_string()), + task_id: Some(task_id.to_string()), + role: Role::User, + parts: vec![Part::text("hello")], + metadata: None, + extensions: None, + reference_task_ids: None, + }, + configuration: None, + metadata: None, + tenant: None, + } +} + +async fn spawn_full_echo() -> StdioTransport { + StdioTransport::spawn(HELPER, &["full-echo"], Some("session-x")) + .await + .expect("full-echo spawn should succeed") +} + +async fn shutdown(transport: StdioTransport) { + let _ = tokio::time::timeout(Duration::from_secs(10), transport.destroy()).await; +} + +// --------------------------------------------------------------------------- +// Robustness: handshake timeout, close timeout/kill, request serialization +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn spawn_fails_when_handshake_never_arrives() { + let res = tokio::time::timeout( + Duration::from_secs(30), + StdioTransport::spawn(HELPER, &["hang-no-handshake"], None), + ) + .await + .expect("spawn should return within HANDSHAKE_TIMEOUT, not hang"); + + let err = match res { + Err(e) => e, + Ok(_) => panic!("spawn should fail when handshake never arrives"), + }; + let msg = err.to_string().to_lowercase(); + assert!( + msg.contains("handshake") || msg.contains("timed out"), + "unexpected error message: {msg}" + ); +} + +#[tokio::test] +async fn destroy_kills_subprocess_that_ignores_stdin_close() { + let transport = StdioTransport::spawn(HELPER, &["handshake-then-hang"], None) + .await + .expect("handshake should succeed for handshake-then-hang"); + + let started = Instant::now(); + let res = tokio::time::timeout(Duration::from_secs(20), transport.destroy()) + .await + .expect("destroy should return within CLOSE_TIMEOUT, not hang"); + + let err = match res { + Err(e) => e, + Ok(_) => panic!("destroy should report a timeout when child ignores stdin close"), + }; + assert!( + err.to_string().to_lowercase().contains("did not exit"), + "unexpected error message: {err}" + ); + assert!( + started.elapsed() < Duration::from_secs(15), + "destroy took too long: {:?}", + started.elapsed() + ); +} + +#[tokio::test] +async fn concurrent_calls_are_serialized_and_responses_match_requests() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + let req_a = sample_send_message_request("task-A"); + let req_b = sample_send_message_request("task-B"); + + let (res_a, res_b) = tokio::join!( + transport.send_message(&svc, &req_a), + transport.send_message(&svc, &req_b), + ); + + let resp_a = res_a.expect("call A should succeed"); + let resp_b = res_b.expect("call B should succeed"); + + match resp_a { + SendMessageResponse::Task(t) => assert_eq!(t.id, "task-A"), + other => panic!("expected Task for A, got {other:?}"), + } + match resp_b { + SendMessageResponse::Task(t) => assert_eq!(t.id, "task-B"), + other => panic!("expected Task for B, got {other:?}"), + } + + shutdown(transport).await; +} + +// --------------------------------------------------------------------------- +// Coverage: each Transport method against a real subprocess +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn transport_send_message() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let resp = transport + .send_message(&svc, &sample_send_message_request("t1")) + .await + .unwrap(); + match resp { + SendMessageResponse::Task(t) => assert_eq!(t.id, "t1"), + other => panic!("unexpected response: {other:?}"), + } + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_send_streaming_message() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let mut stream = transport + .send_streaming_message(&svc, &sample_send_message_request("stream-1")) + .await + .unwrap(); + + let mut count = 0; + while let Some(item) = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + item.expect("stream item should be Ok"); + count += 1; + } + assert!(count >= 2, "expected at least 2 stream events, got {count}"); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_get_task() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let task = transport + .get_task( + &svc, + &GetTaskRequest { + id: "found".to_string(), + history_length: None, + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(task.id, "found"); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_get_task_returns_server_error() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let err = transport + .get_task( + &svc, + &GetTaskRequest { + id: "missing".to_string(), + history_length: None, + tenant: None, + }, + ) + .await + .unwrap_err(); + // Server returns task_not_found; client should surface it as A2AError. + assert!(err.message.to_lowercase().contains("missing") || err.code != 0); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_list_tasks() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let resp = transport + .list_tasks( + &svc, + &ListTasksRequest { + context_id: None, + status: None, + page_size: None, + page_token: None, + history_length: None, + status_timestamp_after: None, + include_artifacts: None, + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(resp.tasks.len(), 1); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_cancel_task() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let task = transport + .cancel_task( + &svc, + &CancelTaskRequest { + id: "t1".to_string(), + metadata: None, + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(task.id, "t1"); + assert!(matches!(task.status.state, TaskState::Canceled)); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_subscribe_to_task() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let mut stream = transport + .subscribe_to_task( + &svc, + &SubscribeToTaskRequest { + id: "sub-1".to_string(), + tenant: None, + }, + ) + .await + .unwrap(); + + let mut count = 0; + while let Some(item) = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + item.expect("stream item should be Ok"); + count += 1; + } + assert!(count >= 2, "expected at least 2 stream events, got {count}"); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_push_config_crud() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let cfg = transport + .create_push_config( + &svc, + &CreateTaskPushNotificationConfigRequest { + task_id: "t1".to_string(), + config: PushNotificationConfig { + url: "https://x".to_string(), + id: Some("cfg-1".to_string()), + token: None, + authentication: None, + }, + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(cfg.task_id, "t1"); + + let got = transport + .get_push_config( + &svc, + &GetTaskPushNotificationConfigRequest { + task_id: "t1".to_string(), + id: "cfg-1".to_string(), + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(got.config.id.as_deref(), Some("cfg-1")); + + let list = transport + .list_push_configs( + &svc, + &ListTaskPushNotificationConfigsRequest { + task_id: "t1".to_string(), + page_size: None, + page_token: None, + tenant: None, + }, + ) + .await + .unwrap(); + assert_eq!(list.configs.len(), 1); + + transport + .delete_push_config( + &svc, + &DeleteTaskPushNotificationConfigRequest { + task_id: "t1".to_string(), + id: "cfg-1".to_string(), + tenant: None, + }, + ) + .await + .unwrap(); + + shutdown(transport).await; +} + +#[tokio::test] +async fn transport_get_extended_agent_card() { + let transport = spawn_full_echo().await; + let svc = ServiceParams::default(); + + let card = transport + .get_extended_agent_card(&svc, &GetExtendedAgentCardRequest { tenant: None }) + .await + .unwrap(); + assert_eq!(card.name, "echo-agent"); + + shutdown(transport).await; +} + +// --------------------------------------------------------------------------- +// Coverage: error paths in `call` +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn call_reports_deserialize_error_for_malformed_result() { + // bad-response helper completes the handshake then sends back a JSON-RPC + // response whose `result` is a string, not a Task. The client should + // surface this as an internal "deserialize result" error rather than + // panicking. + let transport = StdioTransport::spawn(HELPER, &["bad-response"], None) + .await + .expect("bad-response handshake should succeed"); + + let svc = ServiceParams::default(); + let err = transport + .get_task( + &svc, + &GetTaskRequest { + id: "t1".to_string(), + history_length: None, + tenant: None, + }, + ) + .await + .unwrap_err(); + assert!( + err.message.to_lowercase().contains("deserialize"), + "unexpected error message: {err:?}" + ); + + // Helper exits after writing the bad response; destroy should be quick. + let _ = tokio::time::timeout(Duration::from_secs(10), transport.destroy()).await; +} + +// --------------------------------------------------------------------------- +// Coverage: StdioTransportFactory +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn factory_protocol_is_stdio() { + let factory = StdioTransportFactory; + assert_eq!(factory.protocol(), a2a::TRANSPORT_PROTOCOL_STDIO); +} + +#[tokio::test] +async fn factory_creates_transport_from_plain_command_url() { + // The factory parses the AgentInterface URL, spawns the subprocess via + // StdioTransport::spawn, and returns a Box. + let factory = StdioTransportFactory; + + // Use the plain (non-stdio://) form so we can pass an absolute path + // followed by the helper mode. This avoids stdio:// URL escaping. + let url = format!("{HELPER} full-echo"); + + let card = AgentCard { + name: "x".to_string(), + description: "x".to_string(), + version: "0".to_string(), + supported_interfaces: vec![], + capabilities: AgentCapabilities::default(), + default_input_modes: vec![], + default_output_modes: vec![], + skills: vec![], + provider: None, + documentation_url: None, + icon_url: None, + security_schemes: None, + security_requirements: None, + signatures: None, + }; + let iface = AgentInterface { + url, + protocol_binding: a2a::TRANSPORT_PROTOCOL_STDIO.to_string(), + protocol_version: "1.0".to_string(), + tenant: None, + }; + + let transport = factory + .create(&card, &iface) + .await + .expect("factory should spawn the helper"); + + let svc = ServiceParams::default(); + let resp = transport + .send_message(&svc, &sample_send_message_request("via-factory")) + .await + .unwrap(); + match resp { + SendMessageResponse::Task(t) => assert_eq!(t.id, "via-factory"), + other => panic!("unexpected response: {other:?}"), + } + + let _ = tokio::time::timeout(Duration::from_secs(10), transport.destroy()).await; +} + +#[tokio::test] +async fn factory_propagates_spawn_failure_for_missing_program() { + let factory = StdioTransportFactory; + let card = AgentCard { + name: "x".to_string(), + description: "x".to_string(), + version: "0".to_string(), + supported_interfaces: vec![], + capabilities: AgentCapabilities::default(), + default_input_modes: vec![], + default_output_modes: vec![], + skills: vec![], + provider: None, + documentation_url: None, + icon_url: None, + security_schemes: None, + security_requirements: None, + signatures: None, + }; + let iface = AgentInterface { + url: "/this/program/does/not/exist-xyz123".to_string(), + protocol_binding: a2a::TRANSPORT_PROTOCOL_STDIO.to_string(), + protocol_version: "1.0".to_string(), + tenant: None, + }; + + let res = factory.create(&card, &iface).await; + let err = match res { + Err(e) => e, + Ok(_) => panic!("factory should fail when the program cannot be spawned"), + }; + assert!( + err.message.to_lowercase().contains("spawn") + || err.message.to_lowercase().contains("stdio"), + "unexpected error message: {err:?}" + ); +} + +// --------------------------------------------------------------------------- +// Streaming background-reader error paths +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn streaming_parse_frame_error_is_surfaced() { + let transport = StdioTransport::spawn(HELPER, &["stream-bad-frame"], None) + .await + .expect("stream-bad-frame spawn should succeed"); + let svc = ServiceParams::default(); + + let mut stream = transport + .send_streaming_message(&svc, &sample_send_message_request("s")) + .await + .unwrap(); + + let item = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("stream should yield within timeout") + .expect("stream should not end before yielding the parse error"); + let err = item.expect_err("first item should be a parse-frame error"); + assert!( + err.message.to_lowercase().contains("parse frame"), + "unexpected error: {err:?}" + ); + + shutdown(transport).await; +} + +#[tokio::test] +async fn streaming_bad_notification_is_surfaced_then_stream_completes() { + let transport = StdioTransport::spawn(HELPER, &["stream-bad-notification"], None) + .await + .expect("stream-bad-notification spawn should succeed"); + let svc = ServiceParams::default(); + + let mut stream = transport + .send_streaming_message(&svc, &sample_send_message_request("s")) + .await + .unwrap(); + + let mut saw_parse_event_err = false; + while let Some(item) = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + if let Err(e) = item { + if e.message.to_lowercase().contains("parse streaming event") { + saw_parse_event_err = true; + } + } + } + assert!( + saw_parse_event_err, + "expected to observe a 'parse streaming event' error" + ); + + shutdown(transport).await; +} + +#[tokio::test] +async fn streaming_bad_final_response_is_surfaced() { + let transport = StdioTransport::spawn(HELPER, &["stream-bad-final"], None) + .await + .expect("stream-bad-final spawn should succeed"); + let svc = ServiceParams::default(); + + let mut stream = transport + .send_streaming_message(&svc, &sample_send_message_request("s")) + .await + .unwrap(); + + let mut saw_final_parse_err = false; + while let Some(item) = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + if let Err(e) = item { + if e.message + .to_lowercase() + .contains("failed to parse final stream response") + { + saw_final_parse_err = true; + } + } + } + assert!( + saw_final_parse_err, + "expected to observe a 'failed to parse final stream response' error" + ); + + shutdown(transport).await; +} diff --git a/a2a/src/types.rs b/a2a/src/types.rs index 6b37080..a346d23 100644 --- a/a2a/src/types.rs +++ b/a2a/src/types.rs @@ -704,6 +704,8 @@ pub const TRANSPORT_PROTOCOL_GRPC: &str = "GRPC"; pub const TRANSPORT_PROTOCOL_HTTP_JSON: &str = "HTTP+JSON"; /// SLIMRPC transport protocol constant. pub const TRANSPORT_PROTOCOL_SLIMRPC: &str = "SLIMRPC"; +/// STDIO transport protocol constant. +pub const TRANSPORT_PROTOCOL_STDIO: &str = "STDIO"; /// Protocol version string (e.g., "1.0"). pub type ProtocolVersion = String;