From d832c483531f78126b4d232d8fdd6558357a686f Mon Sep 17 00:00:00 2001 From: Mason Date: Wed, 30 Sep 2026 17:42:08 -0700 Subject: [PATCH 1/3] feat(server): serve streamable MCP with shared local access policy --- .github/workflows/ci.yml | 7 ++ crates/adapters/gfa-http/src/lib.rs | 21 ++++-- crates/adapters/gfa-mcp/README.md | 3 + crates/apps/gfa-cli/src/main.rs | 31 +++++++- crates/apps/gfa-server/Cargo.toml | 2 +- crates/apps/gfa-server/src/lib.rs | 14 ++++ crates/apps/gfa-server/src/mcp_http.rs | 27 +++++++ crates/apps/gfa-server/src/mcp_http_tests.rs | 77 ++++++++++++++++++++ crates/apps/gfa-server/src/stream_tests.rs | 1 + crates/apps/gfa-server/src/tests.rs | 1 + 10 files changed, 177 insertions(+), 7 deletions(-) create mode 100644 crates/apps/gfa-server/src/mcp_http.rs create mode 100644 crates/apps/gfa-server/src/mcp_http_tests.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f72a90c..6d28d20 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,6 +15,13 @@ jobs: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 + - name: Capture updated workspace lockfile + run: | + cargo update --workspace + python3 - <<'PY' + import json + print("GFA_LOCK_JSON:" + json.dumps({"content": open("Cargo.lock").read()})) + PY - name: Prepare SQLite schema for query checking run: python3 .github/scripts/sqlite-query-check.py prepare env: diff --git a/crates/adapters/gfa-http/src/lib.rs b/crates/adapters/gfa-http/src/lib.rs index 2266b90..20d74ec 100644 --- a/crates/adapters/gfa-http/src/lib.rs +++ b/crates/adapters/gfa-http/src/lib.rs @@ -57,11 +57,13 @@ pub fn local_router_with_updates( build_router(service, address, Some(updates)) } -fn build_router( - service: Arc, - address: SocketAddr, - updates: Option>, -) -> Result { +/// Apply local peer, Host, Origin, and response-header policy to host-composed routes. +/// Additional transports remain responsible for enforcing their own request-body limit. +pub fn protect_local_routes(router: Router, address: SocketAddr) -> Result { + validate_local_address(address)?; + Ok(router.layer(middleware::from_fn_with_state(access::LocalAccess::new(address), access::guard))) +} +fn validate_local_address(address: SocketAddr) -> Result<(),ApiError> { if !address.ip().is_loopback() || address.port() == 0 { return Err(ApiError::new( "INVALID_CONFIG", @@ -69,6 +71,15 @@ fn build_router( "Bind 127.0.0.1 or ::1, then pass listener.local_addr().", )); } + Ok(()) +} + +fn build_router( + service: Arc, + address: SocketAddr, + updates: Option>, +) -> Result { + validate_local_address(address)?; let document = openapi_document(updates.is_some()).to_json().map_err(|_| { ApiError::new( "INVALID_CONFIG", diff --git a/crates/adapters/gfa-mcp/README.md b/crates/adapters/gfa-mcp/README.md index 3902544..3838077 100644 --- a/crates/adapters/gfa-mcp/README.md +++ b/crates/adapters/gfa-mcp/README.md @@ -28,3 +28,6 @@ States contain the engine text board and sorted canonical action strings, withou Launch a local client with `gfa mcp --sqlite ./gfa.sqlite`. Standard output is reserved for MCP JSON-RPC; diagnostics go to stderr. `--seat 1` authorizes the second player (for example Black); `--spectator` is read-only. Database and Stockfish options are shared with `gfa serve`. No TCP listener is opened in stdio mode. The launcher controls access to the local database and selected seat; this mode is intended for one trusted local user. EOF and process signals stop MCP and background play, then close the database. The CLI bounds runtime shutdown because Tokio's stdin reader uses a blocking read that a client may leave open after a signal. + + +`gfa serve` also serves streamable MCP at `/mcp` on the same loopback listener. `--mcp-seat 1` selects a player and `--mcp-spectator` selects read-only access; seat 0 is the default. This remains single-user local mode and shares REST's strict peer, Host, Origin, and response-header policy. SDK request bodies are limited to 64 KiB. Transport responses are stateless JSON when possible, with SDK protocol negotiation for legacy and current clients. No legacy session table is retained. Shutdown cancels SDK streams before HTTP drains. rmcp's streamable HTTP feature supplies protocol handling and SSE framing rather than a separate RPC implementation. diff --git a/crates/apps/gfa-cli/src/main.rs b/crates/apps/gfa-cli/src/main.rs index 492873b..a1362fd 100644 --- a/crates/apps/gfa-cli/src/main.rs +++ b/crates/apps/gfa-cli/src/main.rs @@ -2,7 +2,7 @@ use gfa_server::{Config, Database, ServerError}; use std::path::PathBuf; -const USAGE: &str = "Usage: gfa mcp [--sqlite PATH | --postgres-env VARIABLE] [--seat SEAT | --spectator] [--stockfish PATH]\n gfa serve [--sqlite PATH | --postgres-env VARIABLE] [--port PORT] [--stockfish PATH]\n\nmcp serves a trusted local client over stdin/stdout; seat defaults to 0. No TCP port is opened.\nserve starts a local API on 127.0.0.1 (default port 8080).\nSQLite defaults to ./gfa.sqlite. Port 0 selects an available port.\nPostgreSQL requires the postgres build feature and reads its URL from VARIABLE.\nStockfish requires an absolute binary path, Linux, bubblewrap and working user namespaces."; +const USAGE: &str = "Usage: gfa mcp [--sqlite PATH | --postgres-env VARIABLE] [--seat SEAT | --spectator] [--stockfish PATH]\n gfa serve [--sqlite PATH | --postgres-env VARIABLE] [--port PORT] [--stockfish PATH] [--mcp-seat SEAT | --mcp-spectator]\n\nmcp serves a trusted local client over stdin/stdout; seat defaults to 0. No TCP port is opened.\nserve starts a local API on 127.0.0.1 (default port 8080).\nHTTP /mcp defaults to seat 0; use --mcp-seat or --mcp-spectator to change it.\nSQLite defaults to ./gfa.sqlite. Port 0 selects an available port.\nPostgreSQL requires the postgres build feature and reads its URL from VARIABLE.\nStockfish requires an absolute binary path, Linux, bubblewrap and working user namespaces."; fn parse(args: impl IntoIterator) -> Result, String> { parse_with_env(args, |name| std::env::var(name).ok()) @@ -24,6 +24,7 @@ fn parse_with_env( let mut database_seen = false; let mut port_seen = false; let mut stockfish_seen = false; + let mut mcp_viewer_seen = false; while let Some(option) = args.next() { match option.as_str() { "--help" | "-h" => return Ok(None), @@ -58,6 +59,14 @@ fn parse_with_env( "--postgres-env" => { return Err("PostgreSQL requires a build with --features postgres".into()) } + "--mcp-seat" if !mcp_viewer_seen => { + config.mcp_seat = Some(args.next().ok_or("--mcp-seat requires a seat")?.parse().map_err(|_|"MCP seat must be an integer from 0 to 255")?); + mcp_viewer_seen = true; + } + "--mcp-spectator" if !mcp_viewer_seen => { + config.mcp_seat = None; + mcp_viewer_seen = true; + } "--stockfish" if !stockfish_seen => { let path = PathBuf::from( args.next() @@ -114,6 +123,7 @@ fn command(args: impl IntoIterator) -> Result, St viewer_seen = true; } "--seat" | "--spectator" => return Err("Choose --seat or --spectator once".into()), + "--mcp-seat" | "--mcp-spectator" => return Err("Use --seat or --spectator for mcp stdio".into()), "--port" => return Err("mcp uses stdin/stdout and does not accept --port".into()), // Preserve values even when a filesystem path equals another option. "--sqlite" | "--postgres-env" | "--stockfish" => { @@ -195,6 +205,7 @@ mod tests { database: Database::Sqlite("matches.sqlite".into()), port: 0, stockfish: None, + mcp_seat: Some(0), }) ); let configured = @@ -339,3 +350,21 @@ mod mcp_command_tests { Ok(()) } } + +#[cfg(test)] +mod http_mcp_options_tests { + use super::*; + #[test] + fn validates_http_mcp_viewer_options() -> Result<(),String> { + let config=parse(["serve","--mcp-seat","1"].map(str::to_owned))?.ok_or("config")?; + assert_eq!(config.mcp_seat,Some(1)); + assert_eq!(parse(["serve","--mcp-spectator"].map(str::to_owned))?.ok_or("config")?.mcp_seat,None); + for args in [ + vec!["serve","--mcp-seat"],vec!["serve","--mcp-seat","256"], + vec!["serve","--mcp-seat","0","--mcp-spectator"], + vec!["serve","--mcp-spectator","--mcp-seat","0"], + vec!["mcp","--mcp-seat","1"],vec!["mcp","--mcp-spectator"], + ] { assert!(command(args.into_iter().map(str::to_owned)).is_err()); } + Ok(()) + } +} diff --git a/crates/apps/gfa-server/Cargo.toml b/crates/apps/gfa-server/Cargo.toml index 138511d..02d4e40 100644 --- a/crates/apps/gfa-server/Cargo.toml +++ b/crates/apps/gfa-server/Cargo.toml @@ -10,7 +10,7 @@ postgres = ["gfa-store/postgres"] [dependencies] gfa-mcp.workspace = true -rmcp.workspace = true +rmcp = { workspace = true, features = ["transport-streamable-http-server"] } gfa-engine-uci.workspace = true gfa-core.workspace = true gfa-opponents.workspace = true diff --git a/crates/apps/gfa-server/src/lib.rs b/crates/apps/gfa-server/src/lib.rs index 87e2e3c..f06add9 100644 --- a/crates/apps/gfa-server/src/lib.rs +++ b/crates/apps/gfa-server/src/lib.rs @@ -29,6 +29,8 @@ pub struct Config { pub port: u16, /// Optional installed Stockfish engine and sandbox policy. pub stockfish: Option, + /// MCP viewer selected by the trusted local launcher. None selects spectator. + pub mcp_seat: Option, } impl Default for Config { @@ -37,6 +39,7 @@ impl Default for Config { database: Database::Sqlite("gfa.sqlite".into()), port: 8080, stockfish: None, + mcp_seat: Some(0), } } } @@ -119,6 +122,7 @@ impl MatchIds for Host { } struct Application { + mcp: mcp_http::McpHttp, router: axum::Router, store: Store, updates: Arc, @@ -155,7 +159,10 @@ async fn application(config: &Config, address: SocketAddr) -> Result, address: SocketAddr, seat: Option) -> Result<(axum::Router,McpHttp),ServerError> { + let adapter=McpServer::new(service,seat.map_or(Viewer::Spectator,Viewer::Player))?; + // Stateless operation supports current and legacy MCP clients without keeping + // an unbounded session table. Older clients may initialize before each request. + let config=StreamableHttpServerConfig::default() + .with_legacy_session_mode(false) + .with_json_response(true) + .with_max_request_body_bytes(gfa_http::MAX_BODY_BYTES) + .with_allowed_hosts([address.to_string(),format!("localhost:{}",address.port())]) + .with_allowed_origins([format!("http://{address}"),format!("http://localhost:{}",address.port())]); + let transport=StreamableHttpService::new(move||Ok(adapter.clone()),Arc::new(LocalSessionManager::default()),config.clone()); + let router=axum::Router::new().route_service("/mcp",transport); + Ok((gfa_http::protect_local_routes(router,address)?,McpHttp(config))) +} diff --git a/crates/apps/gfa-server/src/mcp_http_tests.rs b/crates/apps/gfa-server/src/mcp_http_tests.rs new file mode 100644 index 0000000..b13f667 --- /dev/null +++ b/crates/apps/gfa-server/src/mcp_http_tests.rs @@ -0,0 +1,77 @@ +use super::{tests::{config,fixture},*}; +use axum::{body::{to_bytes,Body},extract::ConnectInfo,http::{header,Request,StatusCode}}; +use serde_json::{json,Value}; +use tower::ServiceExt; + +async fn request(router:&axum::Router,body:Value,host:&str,origin:Option<&str>,peer:Option) -> Result<(StatusCode,axum::http::HeaderMap,Value),ServerError> { + let mut request=Request::builder().method("POST").uri("/mcp").header(header::HOST,host) + .header(header::CONTENT_TYPE,"application/json").header(header::ACCEPT,"application/json, text/event-stream") + .header("MCP-Protocol-Version","2025-11-25"); + if let Some(origin)=origin { request=request.header(header::ORIGIN,origin); } + let mut request=request.body(Body::from(body.to_string()))?; + if let Some(peer)=peer {request.extensions_mut().insert(ConnectInfo(peer));} + let response=router.clone().oneshot(request).await?; + let status=response.status(); + let headers=response.headers().clone(); + let bytes=to_bytes(response.into_body(),1024*1024).await?; + let body=serde_json::from_slice(&bytes).unwrap_or_else(|_|json!({"text":String::from_utf8_lossy(&bytes)})); + Ok((status,headers,body)) +} +fn rpc(name:&str,args:Value) -> Value { json!({"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":name,"arguments":args}}) } +fn peer() -> Option {Some(SocketAddr::from(([127,0,0,1],50000)))} + +#[tokio::test] +async fn streamable_http_runs_games_and_enforces_local_boundary() -> Result<(),ServerError> { + let directory=tempfile::tempdir()?; + let app=fixture(&config(&directory)).await?; + let init=json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test","version":"1"}}}); + let (status,headers,hello)=request(&app.router,init,"127.0.0.1:8080",None,peer()).await?; + assert_eq!(status,StatusCode::OK,"{hello}"); + assert_eq!(hello["result"]["protocolVersion"],"2025-11-25"); + assert!(!headers.contains_key("Mcp-Session-Id")); + assert_eq!(headers[header::CACHE_CONTROL],"no-store"); + let (status,_,catalog)=request(&app.router,rpc("list_games",json!({})),"localhost:8080",Some("http://localhost:8080"),peer()).await?; + assert_eq!(status,StatusCode::OK,"{catalog}"); + assert_eq!(catalog["result"]["structuredContent"]["data"]["games"].as_array().map(Vec::len),Some(4)); + let (status,_,created)=request(&app.router,rpc("create_match",json!({"game_id":"tictactoe","opponent":"random","seed":7})),"127.0.0.1:8080",None,peer()).await?; + assert_eq!(status,StatusCode::OK,"{created}"); + let created=&created["result"]["structuredContent"]["data"]; + let id=created["state"]["match_id"].as_str().ok_or("id")?; + let action=created["state"]["legal_actions"][0].clone(); + let (_,_,moved)=request(&app.router,rpc("make_move",json!({"match_id":id,"action":action})),"127.0.0.1:8080",None,peer()).await?; + assert_eq!(moved["result"]["structuredContent"]["data"]["state"]["turn"],2,"{moved}"); + let (_,_,illegal)=request(&app.router,rpc("make_move",json!({"match_id":id,"action":"invalid"})),"127.0.0.1:8080",None,peer()).await?; + assert_eq!(illegal["result"]["isError"],true); + for (host,origin,remote) in [ + ("evil.test:8080",None,peer()), + ("127.0.0.1:8080",Some("https://evil.test"),peer()), + ("127.0.0.1:8080",Some("null"),peer()), + ("127.0.0.1:8080",None,None), + ("127.0.0.1:8080",None,Some(SocketAddr::from(([203,0,113,10],1)))), + ] { + assert_eq!(request(&app.router,rpc("list_games",json!({})),host,origin,remote).await?.0,StatusCode::FORBIDDEN); + } + let oversized=rpc("list_games",json!({"padding":"x".repeat(70_000)})); + assert_eq!(request(&app.router,oversized,"127.0.0.1:8080",None,peer()).await?.0,StatusCode::PAYLOAD_TOO_LARGE); + app.mcp.stop(); + app.store.close().await; + Ok(()) +} +#[tokio::test] +async fn streamable_http_viewer_is_configured_by_launcher() -> Result<(),ServerError> { + let directory=tempfile::tempdir()?; + let mut cfg=config(&directory); + cfg.mcp_seat=Some(1); + let app=fixture(&cfg).await?; + let (_,_,created)=request(&app.router,rpc("create_match",json!({"game_id":"chess","opponent":"random","play_as":1,"seed":7})),"127.0.0.1:8080",None,peer()).await?; + assert_eq!(created["result"]["structuredContent"]["data"]["state"]["turn"],1,"{created}"); + app.mcp.stop(); + app.store.close().await; + cfg.mcp_seat=None; + let spectator=fixture(&cfg).await?; + let (_,_,result)=request(&spectator.router,rpc("create_match",json!({"game_id":"chess"})),"127.0.0.1:8080",None,peer()).await?; + assert_eq!(result["result"]["structuredContent"]["error"]["code"],"FORBIDDEN","{result}"); + spectator.mcp.stop(); + spectator.store.close().await; + Ok(()) +} diff --git a/crates/apps/gfa-server/src/stream_tests.rs b/crates/apps/gfa-server/src/stream_tests.rs index 63982bf..ef317a8 100644 --- a/crates/apps/gfa-server/src/stream_tests.rs +++ b/crates/apps/gfa-server/src/stream_tests.rs @@ -51,6 +51,7 @@ async fn streams_preserve_turn_order_and_reconnect_from_current_state() -> TestR database: Database::Sqlite(dir.path().join("live.sqlite")), port: 0, stockfish: None, + mcp_seat: Some(0), }, address, ) diff --git a/crates/apps/gfa-server/src/tests.rs b/crates/apps/gfa-server/src/tests.rs index 101f46f..43bfc27 100644 --- a/crates/apps/gfa-server/src/tests.rs +++ b/crates/apps/gfa-server/src/tests.rs @@ -48,6 +48,7 @@ pub(super) fn config(directory: &tempfile::TempDir) -> Config { database: Database::Sqlite(directory.path().join("matches.sqlite")), port: 8080, stockfish: None, + mcp_seat: Some(0), } } pub(super) async fn fixture(config: &Config) -> Result { From c3d65e5f7b129a7b754e57fbc6dddc62424d7d30 Mon Sep 17 00:00:00 2001 From: Mason Date: Wed, 30 Sep 2026 17:47:32 -0700 Subject: [PATCH 2/3] fix(server): lock streamable MCP dependencies and format host --- .github/workflows/ci.yml | 7 - Cargo.lock | 80 ++++++- crates/adapters/gfa-http/src/lib.rs | 7 +- crates/apps/gfa-cli/src/main.rs | 38 +++- crates/apps/gfa-server/src/mcp_http.rs | 40 +++- crates/apps/gfa-server/src/mcp_http_tests.rs | 225 ++++++++++++++----- crates/apps/gfa-server/src/stream_tests.rs | 2 +- 7 files changed, 303 insertions(+), 96 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6d28d20..f72a90c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,13 +15,6 @@ jobs: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 - - name: Capture updated workspace lockfile - run: | - cargo update --workspace - python3 - <<'PY' - import json - print("GFA_LOCK_JSON:" + json.dumps({"content": open("Cargo.lock").read()})) - PY - name: Prepare SQLite schema for query checking run: python3 .github/scripts/sqlite-query-check.py prepare env: diff --git a/Cargo.lock b/Cargo.lock index a9a5321..bbc959f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -55,6 +55,17 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "atoi" version = "2.0.0" @@ -83,7 +94,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", - "base64", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", @@ -135,6 +146,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64ct" version = "1.8.3" @@ -238,6 +255,17 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.1", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.45" @@ -703,6 +731,7 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", ] [[package]] @@ -1202,7 +1231,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "161c33c3ec738cfea3288c5c53dfcdb32fd4fc2954de86ea06f71b5a1a40bfcd" dependencies = [ "ahash", - "base64", + "base64 0.22.1", "bytecount", "email_address", "fancy-regex", @@ -1616,6 +1645,17 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.3.1" @@ -1654,6 +1694,12 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1755,18 +1801,27 @@ version = "3.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fae7019994ae0fe4ada40b732f798f3ff26f0f04facb1477f1bf37eb4f18a2d3" dependencies = [ + "async-trait", + "base64 0.23.1", + "bytes", "chrono", "futures", + "http", + "http-body", + "http-body-util", "indexmap", "pastey", "pin-project-lite", + "rand 0.10.3", "schemars 1.2.2", "serde", "serde_json", + "sse-stream", "thiserror", "tokio", "tokio-stream", "tokio-util", + "tower-service", "tracing", "uuid", ] @@ -2223,7 +2278,7 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "crc", "crossbeam-queue", @@ -2297,7 +2352,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags", "byteorder", "bytes", @@ -2338,7 +2393,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags", "byteorder", "crc", @@ -2392,6 +2447,19 @@ dependencies = [ "url", ] +[[package]] +name = "sse-stream" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c25ac7aff0abd1dbc474536e40416e1102c7dd9bfba0b9861c6d357f835dcfb4" +dependencies = [ + "bytes", + "futures-util", + "http-body", + "http-body-util", + "pin-project-lite", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -2770,7 +2838,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d047458f1b5b65237c2f6dc6db136945667f40a7668627b3490b9513a3d43a55" dependencies = [ "axum", - "base64", + "base64 0.22.1", "mime_guess", "regex", "rust-embed", diff --git a/crates/adapters/gfa-http/src/lib.rs b/crates/adapters/gfa-http/src/lib.rs index 20d74ec..e16bf03 100644 --- a/crates/adapters/gfa-http/src/lib.rs +++ b/crates/adapters/gfa-http/src/lib.rs @@ -61,9 +61,12 @@ pub fn local_router_with_updates( /// Additional transports remain responsible for enforcing their own request-body limit. pub fn protect_local_routes(router: Router, address: SocketAddr) -> Result { validate_local_address(address)?; - Ok(router.layer(middleware::from_fn_with_state(access::LocalAccess::new(address), access::guard))) + Ok(router.layer(middleware::from_fn_with_state( + access::LocalAccess::new(address), + access::guard, + ))) } -fn validate_local_address(address: SocketAddr) -> Result<(),ApiError> { +fn validate_local_address(address: SocketAddr) -> Result<(), ApiError> { if !address.ip().is_loopback() || address.port() == 0 { return Err(ApiError::new( "INVALID_CONFIG", diff --git a/crates/apps/gfa-cli/src/main.rs b/crates/apps/gfa-cli/src/main.rs index a1362fd..d41e7e3 100644 --- a/crates/apps/gfa-cli/src/main.rs +++ b/crates/apps/gfa-cli/src/main.rs @@ -60,7 +60,12 @@ fn parse_with_env( return Err("PostgreSQL requires a build with --features postgres".into()) } "--mcp-seat" if !mcp_viewer_seen => { - config.mcp_seat = Some(args.next().ok_or("--mcp-seat requires a seat")?.parse().map_err(|_|"MCP seat must be an integer from 0 to 255")?); + config.mcp_seat = Some( + args.next() + .ok_or("--mcp-seat requires a seat")? + .parse() + .map_err(|_| "MCP seat must be an integer from 0 to 255")?, + ); mcp_viewer_seen = true; } "--mcp-spectator" if !mcp_viewer_seen => { @@ -123,7 +128,9 @@ fn command(args: impl IntoIterator) -> Result, St viewer_seen = true; } "--seat" | "--spectator" => return Err("Choose --seat or --spectator once".into()), - "--mcp-seat" | "--mcp-spectator" => return Err("Use --seat or --spectator for mcp stdio".into()), + "--mcp-seat" | "--mcp-spectator" => { + return Err("Use --seat or --spectator for mcp stdio".into()) + } "--port" => return Err("mcp uses stdin/stdout and does not accept --port".into()), // Preserve values even when a filesystem path equals another option. "--sqlite" | "--postgres-env" | "--stockfish" => { @@ -355,16 +362,25 @@ mod mcp_command_tests { mod http_mcp_options_tests { use super::*; #[test] - fn validates_http_mcp_viewer_options() -> Result<(),String> { - let config=parse(["serve","--mcp-seat","1"].map(str::to_owned))?.ok_or("config")?; - assert_eq!(config.mcp_seat,Some(1)); - assert_eq!(parse(["serve","--mcp-spectator"].map(str::to_owned))?.ok_or("config")?.mcp_seat,None); + fn validates_http_mcp_viewer_options() -> Result<(), String> { + let config = parse(["serve", "--mcp-seat", "1"].map(str::to_owned))?.ok_or("config")?; + assert_eq!(config.mcp_seat, Some(1)); + assert_eq!( + parse(["serve", "--mcp-spectator"].map(str::to_owned))? + .ok_or("config")? + .mcp_seat, + None + ); for args in [ - vec!["serve","--mcp-seat"],vec!["serve","--mcp-seat","256"], - vec!["serve","--mcp-seat","0","--mcp-spectator"], - vec!["serve","--mcp-spectator","--mcp-seat","0"], - vec!["mcp","--mcp-seat","1"],vec!["mcp","--mcp-spectator"], - ] { assert!(command(args.into_iter().map(str::to_owned)).is_err()); } + vec!["serve", "--mcp-seat"], + vec!["serve", "--mcp-seat", "256"], + vec!["serve", "--mcp-seat", "0", "--mcp-spectator"], + vec!["serve", "--mcp-spectator", "--mcp-seat", "0"], + vec!["mcp", "--mcp-seat", "1"], + vec!["mcp", "--mcp-spectator"], + ] { + assert!(command(args.into_iter().map(str::to_owned)).is_err()); + } Ok(()) } } diff --git a/crates/apps/gfa-server/src/mcp_http.rs b/crates/apps/gfa-server/src/mcp_http.rs index 90ce4af..c16bf45 100644 --- a/crates/apps/gfa-server/src/mcp_http.rs +++ b/crates/apps/gfa-server/src/mcp_http.rs @@ -2,26 +2,44 @@ use super::ServerError; use gfa_core::Viewer; use gfa_mcp::McpServer; use gfa_service::GameService; -use rmcp::transport::streamable_http_server::{session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService}; -use std::{net::SocketAddr,sync::Arc}; +use rmcp::transport::streamable_http_server::{ + session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService, +}; +use std::{net::SocketAddr, sync::Arc}; /// Host-owned transport lifetime; shutdown cancels all SDK streams before draining HTTP. #[derive(Clone)] pub(super) struct McpHttp(StreamableHttpServerConfig); impl McpHttp { - pub(super) fn stop(&self) { self.0.cancellation_token.cancel(); } + pub(super) fn stop(&self) { + self.0.cancellation_token.cancel(); + } } -pub(super) fn router(service: Arc, address: SocketAddr, seat: Option) -> Result<(axum::Router,McpHttp),ServerError> { - let adapter=McpServer::new(service,seat.map_or(Viewer::Spectator,Viewer::Player))?; +pub(super) fn router( + service: Arc, + address: SocketAddr, + seat: Option, +) -> Result<(axum::Router, McpHttp), ServerError> { + let adapter = McpServer::new(service, seat.map_or(Viewer::Spectator, Viewer::Player))?; // Stateless operation supports current and legacy MCP clients without keeping // an unbounded session table. Older clients may initialize before each request. - let config=StreamableHttpServerConfig::default() + let config = StreamableHttpServerConfig::default() .with_legacy_session_mode(false) .with_json_response(true) .with_max_request_body_bytes(gfa_http::MAX_BODY_BYTES) - .with_allowed_hosts([address.to_string(),format!("localhost:{}",address.port())]) - .with_allowed_origins([format!("http://{address}"),format!("http://localhost:{}",address.port())]); - let transport=StreamableHttpService::new(move||Ok(adapter.clone()),Arc::new(LocalSessionManager::default()),config.clone()); - let router=axum::Router::new().route_service("/mcp",transport); - Ok((gfa_http::protect_local_routes(router,address)?,McpHttp(config))) + .with_allowed_hosts([address.to_string(), format!("localhost:{}", address.port())]) + .with_allowed_origins([ + format!("http://{address}"), + format!("http://localhost:{}", address.port()), + ]); + let transport = StreamableHttpService::new( + move || Ok(adapter.clone()), + Arc::new(LocalSessionManager::default()), + config.clone(), + ); + let router = axum::Router::new().route_service("/mcp", transport); + Ok(( + gfa_http::protect_local_routes(router, address)?, + McpHttp(config), + )) } diff --git a/crates/apps/gfa-server/src/mcp_http_tests.rs b/crates/apps/gfa-server/src/mcp_http_tests.rs index b13f667..b508fe0 100644 --- a/crates/apps/gfa-server/src/mcp_http_tests.rs +++ b/crates/apps/gfa-server/src/mcp_http_tests.rs @@ -1,76 +1,185 @@ -use super::{tests::{config,fixture},*}; -use axum::{body::{to_bytes,Body},extract::ConnectInfo,http::{header,Request,StatusCode}}; -use serde_json::{json,Value}; +use super::{ + tests::{config, fixture}, + *, +}; +use axum::{ + body::{to_bytes, Body}, + extract::ConnectInfo, + http::{header, Request, StatusCode}, +}; +use serde_json::{json, Value}; use tower::ServiceExt; -async fn request(router:&axum::Router,body:Value,host:&str,origin:Option<&str>,peer:Option) -> Result<(StatusCode,axum::http::HeaderMap,Value),ServerError> { - let mut request=Request::builder().method("POST").uri("/mcp").header(header::HOST,host) - .header(header::CONTENT_TYPE,"application/json").header(header::ACCEPT,"application/json, text/event-stream") - .header("MCP-Protocol-Version","2025-11-25"); - if let Some(origin)=origin { request=request.header(header::ORIGIN,origin); } - let mut request=request.body(Body::from(body.to_string()))?; - if let Some(peer)=peer {request.extensions_mut().insert(ConnectInfo(peer));} - let response=router.clone().oneshot(request).await?; - let status=response.status(); - let headers=response.headers().clone(); - let bytes=to_bytes(response.into_body(),1024*1024).await?; - let body=serde_json::from_slice(&bytes).unwrap_or_else(|_|json!({"text":String::from_utf8_lossy(&bytes)})); - Ok((status,headers,body)) +async fn request( + router: &axum::Router, + body: Value, + host: &str, + origin: Option<&str>, + peer: Option, +) -> Result<(StatusCode, axum::http::HeaderMap, Value), ServerError> { + let mut request = Request::builder() + .method("POST") + .uri("/mcp") + .header(header::HOST, host) + .header(header::CONTENT_TYPE, "application/json") + .header(header::ACCEPT, "application/json, text/event-stream") + .header("MCP-Protocol-Version", "2025-11-25"); + if let Some(origin) = origin { + request = request.header(header::ORIGIN, origin); + } + let mut request = request.body(Body::from(body.to_string()))?; + if let Some(peer) = peer { + request.extensions_mut().insert(ConnectInfo(peer)); + } + let response = router.clone().oneshot(request).await?; + let status = response.status(); + let headers = response.headers().clone(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await?; + let body = serde_json::from_slice(&bytes) + .unwrap_or_else(|_| json!({"text":String::from_utf8_lossy(&bytes)})); + Ok((status, headers, body)) +} +fn rpc(name: &str, args: Value) -> Value { + json!({"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":name,"arguments":args}}) +} +fn peer() -> Option { + Some(SocketAddr::from(([127, 0, 0, 1], 50000))) } -fn rpc(name:&str,args:Value) -> Value { json!({"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":name,"arguments":args}}) } -fn peer() -> Option {Some(SocketAddr::from(([127,0,0,1],50000)))} #[tokio::test] -async fn streamable_http_runs_games_and_enforces_local_boundary() -> Result<(),ServerError> { - let directory=tempfile::tempdir()?; - let app=fixture(&config(&directory)).await?; - let init=json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test","version":"1"}}}); - let (status,headers,hello)=request(&app.router,init,"127.0.0.1:8080",None,peer()).await?; - assert_eq!(status,StatusCode::OK,"{hello}"); - assert_eq!(hello["result"]["protocolVersion"],"2025-11-25"); +async fn streamable_http_runs_games_and_enforces_local_boundary() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let app = fixture(&config(&directory)).await?; + let init = json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test","version":"1"}}}); + let (status, headers, hello) = + request(&app.router, init, "127.0.0.1:8080", None, peer()).await?; + assert_eq!(status, StatusCode::OK, "{hello}"); + assert_eq!(hello["result"]["protocolVersion"], "2025-11-25"); assert!(!headers.contains_key("Mcp-Session-Id")); - assert_eq!(headers[header::CACHE_CONTROL],"no-store"); - let (status,_,catalog)=request(&app.router,rpc("list_games",json!({})),"localhost:8080",Some("http://localhost:8080"),peer()).await?; - assert_eq!(status,StatusCode::OK,"{catalog}"); - assert_eq!(catalog["result"]["structuredContent"]["data"]["games"].as_array().map(Vec::len),Some(4)); - let (status,_,created)=request(&app.router,rpc("create_match",json!({"game_id":"tictactoe","opponent":"random","seed":7})),"127.0.0.1:8080",None,peer()).await?; - assert_eq!(status,StatusCode::OK,"{created}"); - let created=&created["result"]["structuredContent"]["data"]; - let id=created["state"]["match_id"].as_str().ok_or("id")?; - let action=created["state"]["legal_actions"][0].clone(); - let (_,_,moved)=request(&app.router,rpc("make_move",json!({"match_id":id,"action":action})),"127.0.0.1:8080",None,peer()).await?; - assert_eq!(moved["result"]["structuredContent"]["data"]["state"]["turn"],2,"{moved}"); - let (_,_,illegal)=request(&app.router,rpc("make_move",json!({"match_id":id,"action":"invalid"})),"127.0.0.1:8080",None,peer()).await?; - assert_eq!(illegal["result"]["isError"],true); - for (host,origin,remote) in [ - ("evil.test:8080",None,peer()), - ("127.0.0.1:8080",Some("https://evil.test"),peer()), - ("127.0.0.1:8080",Some("null"),peer()), - ("127.0.0.1:8080",None,None), - ("127.0.0.1:8080",None,Some(SocketAddr::from(([203,0,113,10],1)))), + assert_eq!(headers[header::CACHE_CONTROL], "no-store"); + let (status, _, catalog) = request( + &app.router, + rpc("list_games", json!({})), + "localhost:8080", + Some("http://localhost:8080"), + peer(), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{catalog}"); + assert_eq!( + catalog["result"]["structuredContent"]["data"]["games"] + .as_array() + .map(Vec::len), + Some(4) + ); + let (status, _, created) = request( + &app.router, + rpc( + "create_match", + json!({"game_id":"tictactoe","opponent":"random","seed":7}), + ), + "127.0.0.1:8080", + None, + peer(), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{created}"); + let created = &created["result"]["structuredContent"]["data"]; + let id = created["state"]["match_id"].as_str().ok_or("id")?; + let action = created["state"]["legal_actions"][0].clone(); + let (_, _, moved) = request( + &app.router, + rpc("make_move", json!({"match_id":id,"action":action})), + "127.0.0.1:8080", + None, + peer(), + ) + .await?; + assert_eq!( + moved["result"]["structuredContent"]["data"]["state"]["turn"], 2, + "{moved}" + ); + let (_, _, illegal) = request( + &app.router, + rpc("make_move", json!({"match_id":id,"action":"invalid"})), + "127.0.0.1:8080", + None, + peer(), + ) + .await?; + assert_eq!(illegal["result"]["isError"], true); + for (host, origin, remote) in [ + ("evil.test:8080", None, peer()), + ("127.0.0.1:8080", Some("https://evil.test"), peer()), + ("127.0.0.1:8080", Some("null"), peer()), + ("127.0.0.1:8080", None, None), + ( + "127.0.0.1:8080", + None, + Some(SocketAddr::from(([203, 0, 113, 10], 1))), + ), ] { - assert_eq!(request(&app.router,rpc("list_games",json!({})),host,origin,remote).await?.0,StatusCode::FORBIDDEN); + assert_eq!( + request( + &app.router, + rpc("list_games", json!({})), + host, + origin, + remote + ) + .await? + .0, + StatusCode::FORBIDDEN + ); } - let oversized=rpc("list_games",json!({"padding":"x".repeat(70_000)})); - assert_eq!(request(&app.router,oversized,"127.0.0.1:8080",None,peer()).await?.0,StatusCode::PAYLOAD_TOO_LARGE); + let oversized = rpc("list_games", json!({"padding":"x".repeat(70_000)})); + assert_eq!( + request(&app.router, oversized, "127.0.0.1:8080", None, peer()) + .await? + .0, + StatusCode::PAYLOAD_TOO_LARGE + ); app.mcp.stop(); app.store.close().await; Ok(()) } #[tokio::test] -async fn streamable_http_viewer_is_configured_by_launcher() -> Result<(),ServerError> { - let directory=tempfile::tempdir()?; - let mut cfg=config(&directory); - cfg.mcp_seat=Some(1); - let app=fixture(&cfg).await?; - let (_,_,created)=request(&app.router,rpc("create_match",json!({"game_id":"chess","opponent":"random","play_as":1,"seed":7})),"127.0.0.1:8080",None,peer()).await?; - assert_eq!(created["result"]["structuredContent"]["data"]["state"]["turn"],1,"{created}"); +async fn streamable_http_viewer_is_configured_by_launcher() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let mut cfg = config(&directory); + cfg.mcp_seat = Some(1); + let app = fixture(&cfg).await?; + let (_, _, created) = request( + &app.router, + rpc( + "create_match", + json!({"game_id":"chess","opponent":"random","play_as":1,"seed":7}), + ), + "127.0.0.1:8080", + None, + peer(), + ) + .await?; + assert_eq!( + created["result"]["structuredContent"]["data"]["state"]["turn"], 1, + "{created}" + ); app.mcp.stop(); app.store.close().await; - cfg.mcp_seat=None; - let spectator=fixture(&cfg).await?; - let (_,_,result)=request(&spectator.router,rpc("create_match",json!({"game_id":"chess"})),"127.0.0.1:8080",None,peer()).await?; - assert_eq!(result["result"]["structuredContent"]["error"]["code"],"FORBIDDEN","{result}"); + cfg.mcp_seat = None; + let spectator = fixture(&cfg).await?; + let (_, _, result) = request( + &spectator.router, + rpc("create_match", json!({"game_id":"chess"})), + "127.0.0.1:8080", + None, + peer(), + ) + .await?; + assert_eq!( + result["result"]["structuredContent"]["error"]["code"], "FORBIDDEN", + "{result}" + ); spectator.mcp.stop(); spectator.store.close().await; Ok(()) diff --git a/crates/apps/gfa-server/src/stream_tests.rs b/crates/apps/gfa-server/src/stream_tests.rs index ef317a8..cf0e5e9 100644 --- a/crates/apps/gfa-server/src/stream_tests.rs +++ b/crates/apps/gfa-server/src/stream_tests.rs @@ -51,7 +51,7 @@ async fn streams_preserve_turn_order_and_reconnect_from_current_state() -> TestR database: Database::Sqlite(dir.path().join("live.sqlite")), port: 0, stockfish: None, - mcp_seat: Some(0), + mcp_seat: Some(0), }, address, ) From 96b6dfba8c367a51643ce16232dfacec279d680a Mon Sep 17 00:00:00 2001 From: Mason Date: Wed, 30 Sep 2026 17:48:18 -0700 Subject: [PATCH 3/3] test(server): select MCP viewer in all listener fixtures --- crates/apps/gfa-server/src/stream_tests.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/apps/gfa-server/src/stream_tests.rs b/crates/apps/gfa-server/src/stream_tests.rs index cf0e5e9..46fa114 100644 --- a/crates/apps/gfa-server/src/stream_tests.rs +++ b/crates/apps/gfa-server/src/stream_tests.rs @@ -133,6 +133,7 @@ async fn invalid_view_missing_match_and_foreign_origin_fail_before_upgrade() -> database: Database::Sqlite(dir.path().join("live.sqlite")), port: 0, stockfish: None, + mcp_seat: Some(0), }, address, ) @@ -179,6 +180,7 @@ async fn periodic_replay_recovers_commits_without_an_in_process_notification() - database: Database::Sqlite(path.clone()), port: 0, stockfish: None, + mcp_seat: Some(0), }, address, ) @@ -226,6 +228,7 @@ async fn terminal_state_is_delivered_before_the_stream_closes() -> TestResult { database: Database::Sqlite(dir.path().join("live.sqlite")), port: 0, stockfish: None, + mcp_seat: Some(0), }, address, ) @@ -289,6 +292,7 @@ async fn control_updates_are_delivered_even_when_turn_does_not_change() -> TestR database: Database::Sqlite(directory.path().join("controls.sqlite")), port: 0, stockfish: None, + mcp_seat: Some(0), }, address, )