diff --git a/_context/wiki/architecture.md b/_context/wiki/architecture.md index c9146b29..d2c81e38 100644 --- a/_context/wiki/architecture.md +++ b/_context/wiki/architecture.md @@ -1,8 +1,7 @@ # Architecture -> This page describes the **current Rust implementation**, including temporary -> backend fan-out and session behavior. The tentative configuration-driven end -> state is in +> This page describes the **current stateless Rust implementation**. The +> tentative configuration-driven end state is in > [ContextForge 2.0 Target Architecture and Roadmap](mcp-capability-allocation.md). ## Middleware Stack Order @@ -19,7 +18,6 @@ TCP/TLS listener -> mcp_header_limits_layer → MCP standard header budgets (431 when exceeded) -> virtual_host_id_layer → inserts VirtualHostId (400 on path mismatch) -> claims_layer → inserts ContextForgeClaims (401 on bad/missing JWT) - -> session_id_layer → inserts SessionId if present -> user_config_store_layer → inserts UserConfig (400 no config, 500 store error) -> virtual_host_config_layer → rejects unknown vhost (404 "Server not found") -> /servers/{virtual_host_name}/mcp RMCP service → validates Host, then dispatches MCP @@ -30,7 +28,7 @@ present Origin that is malformed or not allowlisted; requests without Origin continue. RMCP validates the optional Host allowlist at the MCP service boundary. See [Security](security.md#mcp-origin-and-host-validation). `mcp_header_limits_layer` rejects excessive MCP standard headers before JWT -validation, config lookup, session creation, backend fanout, or RMCP body +validation, config lookup, backend connection setup, or RMCP body parsing. MCP handlers read typed extensions — they never parse headers, paths, or Redis keys directly. @@ -39,20 +37,20 @@ MCP handlers read typed extensions — they never parse headers, paths, or Redis ```text downstream request - -> Origin validation → MCP header limits → virtual host extraction → JWT validation → session extraction + -> Origin validation → MCP header limits → virtual host extraction → JWT validation -> user config lookup → RMCP Host validation → MCP handler validation -> request plugin hooks - -> backend MCP call (concurrent via join_all for initialize/list) + -> per-request backend MCP call upstream response - -> response plugin hooks → merge/namespace/passthrough + -> response plugin hooks → passthrough -> metrics, tracing, logging → downstream response ``` ```mermaid flowchart TD bin["binary\nCLI · logging · runtime"] - lib["lib\nrouting · middleware\nsessions · transports"] + lib["lib\nrouting · middleware\ntransports"] apis["apis\nUserConfig · VirtualHost\nBackendMCPGateway"] cpex["cpex\nCPEX hook factories"] bin --> lib @@ -65,12 +63,12 @@ flowchart TD ```mermaid flowchart TD D(["downstream request"]) - A["virtual host · JWT\nsession extract"] + A["virtual host · JWT"] C["user config lookup\nMCP validate"] P1["request plugins\ntool_pre_invoke"] - B["backend MCP call\njoin_all for init/list"] + B["per-request backend MCP call"] P2["response plugins\ntool_post_invoke"] - M["merge · namespace\npassthrough"] + M["response passthrough"] T["metrics · tracing · logging"] U(["downstream response"]) D --> A --> C --> P1 --> B --> P2 --> M --> T --> U @@ -85,8 +83,7 @@ Order is invariant: auth/config before backend selection; request plugins before | --- | --- | | `common.rs` | CLI config shape, JWT claims, Redis config validation, `reqwest::Client` construction | | `layers/` | HTTP request extension extraction, request-bound validation | -| `gateway/` | MCP server behavior, initialize fanout, list merging, prefixed routing, backend service state | -| `gateway/session_store/` | Local and Redis user session storage | +| `gateway/` | Stateless MCP server behavior, identifier routing, and per-request backend calls | | `user_config_store/` | `UserConfigStore` trait, Redis-backed store | | `transports/` | Downstream TCP and TLS listener setup | | `tools.rs` | Local bootstrap helpers (`with_tools` feature only) | @@ -99,33 +96,25 @@ Order is invariant: auth/config before backend selection; request plugins before | JWT decoders | `ContextForgeDataPlaneAppState` | Process | | User config | `RedisUserConfigStore` (LRU + Redis) | Request-path consumed; control-plane authored | | Request identity / VirtualHostId | Request extensions | One HTTP request | -| Downstream session id | RMCP + `SessionId` extension | MCP session | -| Backend RMCP services (initialize, list ops) | `BackendTransports` map | Local process, per principal/backend/session | -| Backend RMCP services (call_tool) | Per-request connection | Single HTTP request | -| Local user session mapping | `LocalUserSessionStore` | Local LRU, 50k entries, 1 hour | +| Backend RMCP service | Targeted MCP handler | Single downstream request | | Plugin manager | `CpexRuntimeRegistry` | Process, reloadable | -> **Session rule:** backend MCP services are local process state. Sticky routing required for load-balanced deployments. +> **Stateless rule:** the dataplane never issues or requires `Mcp-Session-Id`; no request depends on process or runtime affinity. ## Executor Shapes | `--single-runtime` | Shape | | --- | --- | -| `true` (default) | One multi-thread Tokio runtime, `--number-of-cpus` workers. All connections share one `BackendTransports`. | -| `false` | One OS thread per CPU, each with its own current-thread Tokio runtime and own `BackendTransports`. `SO_REUSEPORT` spreads connections — no session affinity. **Stateful MCP sessions need `--single-runtime true`**. | +| `true` (default) | One multi-thread Tokio runtime with `--number-of-cpus` workers. | +| `false` | One OS thread per CPU, each with its own current-thread Tokio runtime. `SO_REUSEPORT` spreads connections without affecting stateless MCP correctness. | In multi-runtime mode, the first thread initializes the optional CPEX plugin runtime before the others start; the current-thread builders are tuned with a global queue interval of `1024` and `4` I/O events per tick. -> **Multi-runtime consequence:** each runtime thread builds its own `BackendTransports` map and user-session store. Backend session state is per-runtime-thread, and `SO_REUSEPORT` gives no connection affinity — later requests in a streamable HTTP session can land on a thread that does not own the session. Treat single-runtime as the only mode supporting stateful MCP sessions today. - ## Lock Design | State | Lock | Contention profile | | --- | --- | --- | -| `BackendTransports` map | `Arc>>` | Locked briefly on initialize insert, list-op borrow, and cleanup. Borrowing clones `Arc` handles so the lock is not held across backend calls. `call_tool` bypasses this map entirely. | -| Subscription set | `Arc>>` | Local `subscribe`/`unsubscribe` only. | | User config LRU cache | `Arc>` inside `RedisUserConfigStore` | One lock per config lookup on the hot path; misses add a Redis round trip. | -| User session LRU cache | Same pattern in `LocalUserSessionStore` | Initialize and delete paths. | | JWT decoders, upstream `reqwest::Client`, process `Config` | No lock — immutable after startup, shared by `Arc`/clone. | None. | Design rule: locks guard maps of handles, not I/O. Backend calls, Redis reads, and plugin hooks all run outside any gateway lock. @@ -138,12 +127,11 @@ The TCP listener binds with `reuseaddr`, `reuseport`, and keepalive, listens wit The binary sets `tikv_jemallocator` as the global allocator. jemalloc holds up better than the system allocator under the many small, short-lived allocations of per-request JSON and header processing. -## Fanout And Cancellation +## Request Lifecycle And Cancellation -- `initialize` opens one backend transport per configured backend concurrently (`futures::future::join_all`); a failed backend degrades that backend only. -- List methods fan out to all connected backends concurrently and merge. -- Targeted calls (except `call_tool`) resolve exactly one backend service handle from `BackendTransports`. -- `call_tool` creates a fresh per-request backend connection via `connect_backend_for_request`, runs pre/post plugin hooks, executes the call, then explicitly closes the connection before returning. +- Legacy `initialize` is answered locally as a best-effort stateless compatibility request. +- List methods and subscriptions are rejected locally and remain control-plane responsibilities. +- Each supported targeted call resolves one configured backend, creates a fresh connection via `connect_backend_for_request`, executes the call, then explicitly closes the connection. - `call_tool` watches the downstream cancellation token and forwards a cancel to the backend if the client gives up first; backend progress notifications are forwarded downstream while the call is in flight. ## Startup And Response Flow @@ -159,7 +147,6 @@ install rustls crypto provider -> Gateway::builder() .with_config(config) .with_user_config_store_type(UserConfigStoreType::Redis) - .with_session_manager(LocalSessionManager::default()) .with_plugin_runtime(...) .build() -> runtime.execute(gateway, plugin_registry) @@ -170,10 +157,9 @@ Response unwind order (Tower layers execute outside-in, so unwind is inside-out) ```text backend response -> response plugin hooks (call_tool only) - -> merge / namespace / pass through + -> pass through -> virtual_host_config_layer response side -> user_config_store_layer response side - -> session_id_layer response side ← on DELETE success: remove session + backend transports -> claims_layer response side -> virtual_host_id_layer response side -> CORS, mcp_origin_layer, TraceLayer, HttpMetricsLayer @@ -196,10 +182,10 @@ Flow checkpoints — each must exist before the next dependency runs: The current code implements MCP behavior, but the gateway shell is broader: ```text -auth → config lookup → transport setup → plugin runtime → telemetry → session strategy +auth → config lookup → transport setup → plugin runtime → telemetry ``` -Keep protocol-neutral concerns (auth, config ingestion, TLS handling, plugin execution, telemetry, runtime shape, session strategy) reusable. Future A2A or model-provider routing should reuse the gateway shell without copying the MCP routing stack. MCP-specific behavior must remain isolated to the current MCP modules. +Keep protocol-neutral concerns (auth, config ingestion, TLS handling, plugin execution, telemetry, runtime shape) reusable. Future A2A or model-provider routing should reuse the gateway shell without copying the MCP routing stack. MCP-specific behavior must remain isolated to the current MCP modules. ## Transport Security Split @@ -236,7 +222,7 @@ Changing a load-bearing choice requires updating more than one file: | --- | --- | | Downstream MCP version | Coordinate with the ContextForge control plane and built-in dataplane; update the `2026-07-28`/`2025-11-25` compatibility matrix, protocol tests, examples, and front-door routing. The ContextForge built-in dataplane handles both stateful and stateless traffic; the ContextForge external dataplane handles both supported Streamable HTTP versions statelessly. | | Backend namespace / prefix contract | Update merge logic, split logic, tests, docs, and control-plane integration if client-facing surface moves. | -| Session state moves external | Update `SessionManager`, cleanup behavior, load-balancing docs, and failure-mode tests. | +| MCP lifecycle changes | Preserve stateless request independence and update the four-version compatibility matrix and failure-mode tests. | | Config transport changes | Keep `UserConfigStore` as the boundary; update adapter tests. | | Plugin hook surface expands | Document ordering, failure, timeout, cancellation, streaming, and telemetry before landing. | | New protocol joins the gateway | Keep shared shell protocol-neutral; isolate new protocol-specific routing. | diff --git a/_context/wiki/deployment.md b/_context/wiki/deployment.md index 876b3026..18c61a92 100644 --- a/_context/wiki/deployment.md +++ b/_context/wiki/deployment.md @@ -1,8 +1,7 @@ # Deployment -> This page describes **current deployment requirements**, including session -> affinity. The tentative target removes live aggregate fan-out and durable -> upstream-session dependence; see +> This page describes **current deployment requirements** for the stateless +> external dataplane; see > [ContextForge 2.0 Target Architecture and Roadmap](mcp-capability-allocation.md). ## Checklist @@ -11,7 +10,7 @@ 2. JWT verification key/secret matches the control plane's signing material; clients use control-plane API tokens whose `sub` matches the published user-config key. 3. Redis reachable; TLS/mTLS across trust zones; write access restricted to the control plane; `DATAPLANE_PUBLISHER=true` on the control plane. 4. Upstream connection mode matches backend URL schemes. -5. One replica per `Mcp-session-id` (single replica or sticky routing). +5. Scale replicas without MCP session affinity; every request is independent. 6. `with_tools` feature **disabled** in the production build. 7. Telemetry export pointed at the collector. 8. System limits raised: `nofile 65535`, TCP tuning (`tcp_fin_timeout=15`, widened local port range). @@ -28,12 +27,12 @@ Reference `docker/nginx.conf` split: - Other MCP routes, including stateful and legacy/SSE compatibility routes → ContextForge built-in dataplane. - Upstream retries on `error timeout http_502/503/504`: 2 tries, 10-second window. Non-idempotent MCP `POST` bodies are not re-sent after they reached an upstream — only connection-stage failures retry. -## Session Affinity And Failover +## Stateless Failover -Backend MCP sessions are **local process state** — see [routing.md](routing.md). +The external dataplane neither issues nor requires `Mcp-Session-Id` and retains no backend transport between requests — see [routing.md](routing.md). -- >1 replica requires sticky routing by `Mcp-session-id`. The reference nginx config does not provide this; safe shapes today are a single replica or a front door with stickiness. -- On restart or failover, all sessions are lost. Design clients to treat session-not-found as "reinitialize", not "retry". +- Replicas do not require sticky routing. +- Restart or failover loses no MCP session state; an interrupted request may be retried only when its method is safe to retry. ## Redis Availability diff --git a/_context/wiki/routing.md b/_context/wiki/routing.md index 1013311f..8f9f260f 100644 --- a/_context/wiki/routing.md +++ b/_context/wiki/routing.md @@ -1,7 +1,6 @@ # MCP Routing Semantics -> This page describes the **current transitional routing behavior**. Its live -> upstream fan-out and durable-session assumptions are not the Phase 3 target. +> This page describes the **current stateless routing behavior**. > See [ContextForge 2.0 Target Architecture and Roadmap](mcp-capability-allocation.md) > for the proposed ownership boundary and migration. @@ -22,16 +21,16 @@ Single-backend virtual hosts: identifiers pass through **unchanged**. `BackendMCPGateway.tool_name_aliases` maps `{downstream_alias: upstream_original}`. Aliases take precedence over prefix fallback. They are advertised and routed exactly as published (case, dots, underscores preserved). -## List Operations (fan-out) +## List Operations -All four list methods fan out to all connected backends concurrently and merge results: +All four list methods are rejected locally and remain ContextForge built-in dataplane responsibilities: ```text list_tools / list_resources / list_prompts / list_resource_templates - → all connected backends → merged sorted output + → INVALID_REQUEST: use the ContextForge built-in dataplane ``` -Failed/unavailable backends are logged and skipped. Single-backend: identifiers unchanged. Multi-backend: prefixed with backend map key. +Aliases and legacy prefixes remain usable for targeted calls published by the control plane. ## Routed Operations (single backend) @@ -46,96 +45,39 @@ gateway-oneincrement → rejected (no - separator) Methods using the same conditional routing: `read_resource`, `subscribe`, `unsubscribe`, `get_prompt`, `complete`. -## Federated Pagination +## Stateless Request Lifecycle -The gateway wraps per-backend cursors inside its own opaque token (JSON, treated as opaque by MCP clients). First request: all backends queried. Resume: cursor decoded, exhausted backends skipped. New cursor emitted when any backend has more pages. - -**Known limitation:** if backend set changes between pages, removed backend's cursor is silently dropped. - -## Session State (local process) - -Backend RMCP services are stored in `BackendTransports` keyed by: -```text -principal (claims.sub) + backend_name (map key) + downstream_session_id -``` - -This is **local process state only**. Implications: -- After `initialize`, later requests must reach the same process. -- Sticky routing required for load-balanced deployments. -- Gateway restart → all sessions lost → clients must re-run `initialize`. -- Multi-runtime mode (`--single-runtime false`): each runtime thread has its own `BackendTransports` with no cross-thread affinity. - -**Exception: `call_tool` uses per-request backend lifecycle.** Each tool call creates a fresh backend connection, executes the call with plugin hooks, then closes the connection. This bypasses `BackendTransports` entirely and does not require session affinity for tool calls specifically (though other MCP methods still do). +The dataplane uses RMCP's `NeverSessionManager` and disables legacy session mode. It never issues or requires `Mcp-Session-Id`, and it retains no backend transport between downstream requests. Legacy `initialize` remains a local compatibility response only; each later request independently resolves configuration and opens its own backend connection. ```mermaid sequenceDiagram participant C as MCP Client participant GW as Gateway (RMCP) - participant BT as BackendTransports
(local process state) - participant LU as LocalUserSessionStore
(LRU 50k / 1h) participant BA as Backend A - participant BB as Backend B - - C->>GW: POST initialize (Mcp-Session-Id: S) - GW->>BA: initialize (concurrent) - GW->>BB: initialize (concurrent) - BA-->>GW: InitializeResult - BB-->>GW: InitializeResult - GW->>BT: store RunningService keyed by sub+backend+S - GW->>LU: store session entry for sub+S - GW-->>C: merged InitializeResult - - C->>GW: POST call_tool (Mcp-Session-Id: S) - GW->>BT: lookup sub+backend+S → Arc - BT-->>GW: RunningService handle - GW->>BA: call_tool (routed by name prefix) - BA-->>GW: ToolResult - GW-->>C: ToolResult - C->>GW: DELETE (Mcp-Session-Id: S) - GW->>GW: RMCP handles DELETE - GW->>LU: remove sub+S entry - GW->>BT: remove all sub+*+S entries - GW-->>C: 200 OK -``` - -## Capability Merge - -On `initialize`, the gateway builds one downstream `InitializeResult` — not a passthrough of any one backend. The source of truth is each backend's `InitializeResult`; the gateway reads `peer_info().capabilities` from each running service and stores them with the backend transport state. - -The merge rule (gateway-aware, not a raw union): -- Enable a top-level capability when ≥1 backend supports it **and** the gateway has a routing story for it. -- `resources.subscribe` preserved if any backend advertises it (the gateway routes subscribe/unsubscribe and forwards resource-update notifications). -- `listChanged` not yet advertised (gateway doesn't emit downstream list-changed notifications when upstream lists change). -- Single-backend passthrough is not a stable contract (`HashMap` iteration order). -- If no backend reports supported capabilities, returns `ServerCapabilities::default()`. - -**Do not** initialize the downstream capability from just one backend entry — the gateway fronts multiple backends, `HashMap` iteration is non-deterministic, and list methods already merge across all backends. + C->>GW: POST initialize (2025 compatibility) + GW-->>C: local InitializeResult, no session id -## Cleanup - -`DELETE` with `Mcp-session-id`: -```text -→ RMCP handles request -→ on success: remove LocalUserSessionStore entry + BackendTransports entries for principal+session + C->>GW: POST tools/call (no session id) + GW->>GW: resolve configured backend + tool + GW->>BA: server/discover + GW->>BA: tools/call + BA-->>GW: ToolResult + GW->>BA: close connection + GW-->>C: ToolResult ``` -If RMCP rejects the delete, local state is untouched. ## MCP Method Quick Reference | Method | Group | Behavior | | --- | --- | --- | -| `initialize` | Session | Concurrent fanout to all backends; failure of one backend is non-fatal (stored with no service). Returns merged capability set. Requires `DownstreamSessionId`, `UserConfig`, `VirtualHostId`, `ContextForgeClaims`. | -| `list_tools` | List | Fan-out all connected backends → merged sorted result. Cursor-based pagination across backends. | -| `list_resources` | List | Same as list_tools. | -| `list_prompts` | List | Same as list_tools. | -| `list_resource_templates` | List | Same — both name and URI template get prefixed for multi-backend. | -| `call_tool` | Targeted | **Per-request backend lifecycle:** creates fresh connection via `connect_backend_for_request`, runs pre-hook, executes call, runs post-hook, closes connection. Resolves alias → single/multi-backend fallback. Forwards downstream cancellation to backend. Tracks backend progress tokens: RMCP assigns a new token per backend request; the gateway maps each backend token to the downstream token. Request enqueue and mapping publication are serialized against progress lookup so an immediate backend notification cannot overtake registration. When the notification matches an in-flight token, the gateway restores the downstream token and forwards it to the client. Does not use session-backed `BackendTransports`. | -| `read_resource` | Targeted | Single-backend: URI unchanged. Multi-backend: strips prefix. | -| `subscribe` / `unsubscribe` | Targeted | Same resource-URI routing; forwards/stops resource-update notifications. | -| `get_prompt` | Targeted | Single-backend: name unchanged. Multi-backend: strips prefix. Runs pre/post prompt hooks around the backend call: the pre hook may rewrite arguments or deny, the post hook may rewrite or reject the rendered messages. | -| `complete` | Targeted | Routes on prompt name or resource URI inside `ref`. | +| `initialize` | Compatibility | For `2025-11-25`, returns a local stateless compatibility result with no session id; modern clients use `server/discover`. | +| `list_tools` / `list_resources` / `list_prompts` / `list_resource_templates` | List | Rejected locally; use the ContextForge built-in dataplane. | +| `call_tool` | Targeted | Resolves the upstream name, creates a per-request backend connection, runs pre/post hooks, executes, and closes. Forwards cancellation and request-scoped progress. | +| `read_resource` | Targeted | Resolves alias/prefix, creates a per-request backend connection, executes, and closes. | +| `subscribe` / `unsubscribe` | Unsupported | Rejected locally; use the ContextForge built-in dataplane. | +| `get_prompt` | Targeted | Resolves alias/prefix and runs pre/post prompt hooks around a per-request backend call. | +| `complete` | Unsupported | Rejected locally; use the ContextForge built-in dataplane. | | `ping` | Local | Returns success; no backend fanout. | -| `DELETE` | Session | RMCP handles first; on success `session_id_layer` removes local session + backend transports. | diff --git a/_context/wiki/testing.md b/_context/wiki/testing.md index 607b9e63..96842625 100644 --- a/_context/wiki/testing.md +++ b/_context/wiki/testing.md @@ -30,7 +30,7 @@ Protocol-sensitive tests and fixtures must cover MCP `2026-07-28` and `2025-11-2 | `gateway_list_tools.rs` | List fanout, prefixing, and merged output. | | `gateway_prompts.rs` | Prompt listing and prefixed `get_prompt` routing. | | `gateway_resource_templates.rs` | Template fanout with prefixed names and URI templates, plus `read_resource` round-trips. | -| `gateway_plugins.rs` | CPEX pre/post tool hooks around `call_tool` and stream events, and prompt hooks around `get_prompt`. | +| `gateway_plugins.rs` | Stateless legacy request behavior, CPEX hooks, and request-scoped progress. | These run in `cargo nextest run` with no Docker dependencies. diff --git a/crates/contextforge-data-plane-lib/src/lib.rs b/crates/contextforge-data-plane-lib/src/lib.rs index a5d72eae..cfa22b3f 100644 --- a/crates/contextforge-data-plane-lib/src/lib.rs +++ b/crates/contextforge-data-plane-lib/src/lib.rs @@ -8,7 +8,7 @@ use http::uri::Authority; use jsonwebtoken::DecodingKey; use rmcp::transport::{ StreamableHttpServerConfig, - streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService}, + streamable_http_server::{session::never::NeverSessionManager, tower::StreamableHttpService}, }; mod common; mod const_values; @@ -59,7 +59,6 @@ pub enum UserConfigStoreType { #[builder(field_defaults(setter(prefix = "with_")))] pub struct Gateway { config: Config, - session_manager: Arc, user_config_store_type: UserConfigStoreType, #[builder(default)] plugin_runtime: Option, @@ -97,7 +96,7 @@ impl Gateway { } async fn build_app(self) -> Result { - let Gateway { config, session_manager, user_config_store_type, plugin_runtime } = self; + let Gateway { config, user_config_store_type, plugin_runtime } = self; let user_config_store = match user_config_store_type { UserConfigStoreType::Redis => Arc::new(get_config_store(&config).await?), UserConfigStoreType::Test(store) => store, @@ -113,19 +112,20 @@ impl Gateway { .disable_allowed_origins() } else { StreamableHttpServerConfig::default().disable_allowed_hosts().disable_allowed_origins() - }; + } + .with_legacy_session_mode(false); let reqwest_backend_client = reqwest::Client::try_from(&config)?; // Create streamable HTTP service - let mcp_service: StreamableHttpService = StreamableHttpService::new( + let mcp_service: StreamableHttpService = StreamableHttpService::new( move || { Ok(McpService::builder() .with_http_client(reqwest_backend_client.clone()) .with_plugin_runtime(plugin_runtime.clone()) .build()) }, - session_manager, + Arc::new(NeverSessionManager::default()), streamable_config, ); @@ -169,7 +169,7 @@ impl Gateway { .layer(middleware::from_fn_with_state(mcp_standard_header_limits, mcp_header_limits_layer)) .layer(cors_layer) // mcp_origin_layer is the outermost wrapper: fires before JWT auth, - // session creation, and backend fan-out. + // MCP dispatch, and backend connection setup. .layer(middleware::from_fn_with_state(config.clone(), mcp_origin_layer)); #[cfg(feature = "with_tools")] @@ -199,7 +199,6 @@ mod tests { use axum::body::Body; use contextforge_data_plane_apis::{User, user_store::UserConfig}; use http::{Request, StatusCode}; - use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; use tower::ServiceExt; use crate::{ @@ -226,7 +225,6 @@ mod tests { let config = Config { mcp_standard_header_max_count: 1, ..Config::default() }; let app = Gateway::builder() .with_config(config) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(UnusedConfigStore))) .build() .build_app() diff --git a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs index c3c48e01..dd660cdd 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs @@ -68,7 +68,6 @@ async fn start_gateway(config: Config, virtual_host_id: &str, user_config: UserC let gateway = Gateway::builder() .with_config(config) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(store))) .build(); diff --git a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs index 51560f6a..ea419fdf 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs @@ -100,19 +100,15 @@ fn raw_mcp_request( client: &reqwest::Client, gateway: &RunningGateway, user: &str, - session_id: Option<&str>, body: &Value, ) -> reqwest::RequestBuilder { - let mut request = client + client .post(gateway.gateway_url()) .bearer_auth(token(user)) .header(http::header::CONTENT_TYPE, "application/json") .header(http::header::ACCEPT, "application/json, text/event-stream") - .json(body); - if let Some(session_id) = session_id { - request = request.header("Mcp-Session-Id", session_id).header("MCP-Protocol-Version", "2025-11-25"); - } - request + .header("MCP-Protocol-Version", "2025-11-25") + .json(body) } fn raw_tool_call(tool_name: &str, request_id: i64, progress_token: &str) -> Value { @@ -200,12 +196,11 @@ fn assert_raw_progress_stream(body: &str, response_id: i64, progress_token: &str assert_eq!(Some("completed 4 packages"), result.pointer("/result/content/0/text").and_then(Value::as_str)); } -async fn start_raw_mcp_session(client: &reqwest::Client, gateway: &RunningGateway, user: &str) -> String { +async fn initialize_raw_stateless_client(client: &reqwest::Client, gateway: &RunningGateway, user: &str) { let initialize = raw_mcp_request( client, gateway, user, - None, &serde_json::json!({ "method": "initialize", "params": { @@ -221,20 +216,13 @@ async fn start_raw_mcp_session(client: &reqwest::Client, gateway: &RunningGatewa .await .expect("initialize request is sent"); assert!(initialize.status().is_success(), "initialize failed: {initialize:?}"); - let session_id = initialize - .headers() - .get("mcp-session-id") - .expect("initialize response has MCP session id") - .to_str() - .expect("MCP session id is valid") - .to_owned(); + assert!(initialize.headers().get("mcp-session-id").is_none(), "initialize must remain stateless"); let _initialize_body = initialize.text().await.expect("initialize body is read"); let initialized = raw_mcp_request( client, gateway, user, - Some(&session_id), &serde_json::json!({ "method": "notifications/initialized", "jsonrpc": "2.0" }), ) .send() @@ -242,8 +230,6 @@ async fn start_raw_mcp_session(client: &reqwest::Client, gateway: &RunningGatewa .expect("initialized notification is sent"); assert!(initialized.status().is_success(), "initialized notification failed: {initialized:?}"); let _initialized_body = initialized.text().await.expect("initialized body is read"); - - session_id } async fn read_concurrent_raw_progress_streams( @@ -306,26 +292,14 @@ async fn concurrent_progress_calls_forward_each_token_without_plugins() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn raw_streamable_http_concurrent_progress_calls_complete_without_plugins() { +async fn legacy_raw_concurrent_progress_calls_complete_without_sessions() { let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; let client = reqwest::Client::new(); - let session_id = start_raw_mcp_session(&client, &gateway, TEST_USER_ID).await; + initialize_raw_stateless_client(&client, &gateway, TEST_USER_ID).await; let tool_name = "progress_sum"; - let first = raw_mcp_request( - &client, - &gateway, - TEST_USER_ID, - Some(&session_id), - &raw_tool_call(tool_name, 2, "downstream-first"), - ); - let second = raw_mcp_request( - &client, - &gateway, - TEST_USER_ID, - Some(&session_id), - &raw_tool_call(tool_name, 3, "downstream-second"), - ); + let first = raw_mcp_request(&client, &gateway, TEST_USER_ID, &raw_tool_call(tool_name, 2, "downstream-first")); + let second = raw_mcp_request(&client, &gateway, TEST_USER_ID, &raw_tool_call(tool_name, 3, "downstream-second")); let (first_body, second_body) = read_concurrent_raw_progress_streams(first, second).await; assert_raw_progress_stream(&first_body, 2, "downstream-first"); diff --git a/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs b/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs index f88d8afc..6f9f1bee 100644 --- a/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs +++ b/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs @@ -102,7 +102,6 @@ pub(crate) async fn create_gateway_with_four_counters(user: &str, config: Config let gateway = Gateway::builder() .with_config(config.clone()) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(mocked_user_config_store))) .build(); @@ -180,7 +179,6 @@ pub(crate) async fn create_tls_gateway_with_four_tls_counters( let gateway = Gateway::builder() .with_config(config.clone()) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(mocked_user_config_store))) .build(); diff --git a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs b/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs index f0c262e0..e6f4f16c 100644 --- a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs +++ b/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs @@ -359,7 +359,6 @@ async fn start_gateway_with_state( runtime_plugins_enabled: Some(runtime_plugins_enabled), ..Default::default() }) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(user_store))) .with_plugin_runtime(runtime_plugins_enabled.then(|| plugin_runtime.handle())) .build(); diff --git a/crates/contextforge-data-plane/Cargo.toml b/crates/contextforge-data-plane/Cargo.toml index 6de6816e..26d8e09d 100644 --- a/crates/contextforge-data-plane/Cargo.toml +++ b/crates/contextforge-data-plane/Cargo.toml @@ -30,7 +30,6 @@ opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", opentelemetry_sdk.workspace = true tonic = "0.14" num_cpus = "1.17.0" -rmcp.workspace = true tikv-jemallocator = "0.7.0" rustls.workspace = true @@ -46,6 +45,7 @@ jsonwebtoken.workspace = true openport.workspace = true redis.workspace = true reqwest.workspace = true +rmcp.workspace = true rmp-serde.workspace = true serde_json.workspace = true diff --git a/crates/contextforge-data-plane/src/main.rs b/crates/contextforge-data-plane/src/main.rs index 4dfa9fda..e2d07791 100644 --- a/crates/contextforge-data-plane/src/main.rs +++ b/crates/contextforge-data-plane/src/main.rs @@ -8,7 +8,6 @@ use std::sync::Arc; use clap::Parser; use contextforge_data_plane_cpex::CpexRuntimeRegistry; use contextforge_data_plane_lib::{Config, Gateway, RedisClient, RedisConfig, UserConfigStoreType}; -use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; use rustls::crypto; use tikv_jemallocator::Jemalloc; use tracing::info; @@ -35,7 +34,6 @@ fn main() -> Result<(), Box> { let gateway = Gateway::builder() .with_config(config) .with_user_config_store_type(UserConfigStoreType::Redis) - .with_session_manager(Arc::new(LocalSessionManager::default())) .with_plugin_runtime(plugin_runtime.clone()) .build();