Add downstream subscriptions/listen admission

[gandhipratik203]

Summary

• Carry request user + virtual host context into RMCP.

• Advertise virtual-host-specific capabilities via server/discover.

• Accept only supported/routable subscriptions/listen filters.

• Register active downstream subscription sinks and clean them up on close.

• Test modern/legacy subscription method gating.

• Closes [CF-DATAPLANE] Downstream subscriptions/listen: context-carried admission and sink registry IBM/mcp-context-forge#6117.

• Parent tracker: [Major-3] Notification rework: subscriptions/listen IBM/mcp-context-forge#5681.

Explanatory Diagrams

MCP client
   |
   | subscriptions/listen
   | "I want tools/resources change notifications"
   v
Dataplane gateway
   |
   | 1. identifies user + virtual host
   | 2. checks what this virtual host can expose
   | 3. drops unsupported or unroutable subscriptions
   v
RMCP acknowledgment
   |
   | "accepted: toolsListChanged, resourcesListChanged"
   v
Dataplane gateway
   |
   | stores SubscriptionSink in registry
   | key = user + virtual host + subscription + notification kind
   v
Long-lived listen stream
   |
   | stays open until client disconnects/cancels
   v
Drop guard cleanup
   |
   | removes that client's registered sinks

Scope boundary for this PR

This PR
=======

MCP client
   |
   | subscriptions/listen
   | asks to receive selected notification types
   v
Gateway
   |
   | - validates the requested notification filter
   | - keeps the accepted SubscriptionSink alive
   | - unregisters it when the listen stream closes
   v
Downstream subscription registry


Later PRs
=========

Backend MCP servers
   |
   | list_changed / resources/updated notifications
   v
Gateway
   |
   | - matches backend notifications to downstream subscriptions
   | - deduplicates repeated list_changed notifications
   | - forwards accepted notifications to subscribed clients
   v
MCP client

This PR adds the downstream subscription admission and registry path. It lets modern clients open subscriptions/listen streams, narrows each request to what the selected virtual host can expose, stores accepted sinks for future notification relay, and cleans them up when the stream closes.

Changed files layout

contextforge-data-plane-lib/
|
|-- src/lib.rs
|   `-- creates the shared subscription registry
|
|-- src/layers/
|   |-- request_context.rs
|   |   `-- remembers the user and virtual host for this request
|   |
|   `-- virtual_host_config.rs
|       `-- finds the virtual host from the request path
|
|-- src/gateway/
|   |-- mcp_service.rs
|   |   `-- handles server/discover and subscriptions/listen
|   |
|   `-- downstream_subscriptions.rs
|       `-- stores active client subscriptions
|
`-- tests/gateway_modern_subscriptions.rs
    `-- tests the new modern subscription flow

Runtime component flow

+---------------------------------------------------------------+
|                    contextforge-data-plane                    |
|                                                               |
|  +----------------------------+                               |
|  | Gateway middleware         |                               |
|  |                            |                               |
|  | - validates JWT            |                               |
|  | - loads UserConfig         |                               |
|  | - selects virtual host     |                               |
|  +-------------+--------------+                               |
|                |                                              |
|                | principal + virtual host                     |
|                v                                              |
|  +----------------------------+                               |
|  | RMCP service factory       |                               |
|  |                            |                               |
|  | builds McpService for this |                               |
|  | request context            |                               |
|  +-------------+--------------+                               |
|                |                                              |
|                v                                              |
|  +----------------------------+                               |
|  | McpService                 |                               |
|  |                            |                               |
|  | - server/discover          |                               |
|  | - subscriptions/listen     |                               |
|  +-------------+--------------+                               |
|                |                                              |
|                | accepted SubscriptionSink                    |
|                v                                              |
|  +----------------------------+                               |
|  | DownstreamSubscription     |                               |
|  | Registry                   |                               |
|  |                            |                               |
|  | stores active client sinks |                               |
|  | cleans up on stream close  |                               |
|  +----------------------------+                               |
|                                                               |
+---------------------------------------------------------------+

Notes

• Backend notification relay is intentionally left to follow-up work in #5346, #5615, and #6118.
• The downstream subscription registry uses std::sync::Mutex intentionally: registration and cleanup are short HashMap operations, no await happens while the lock is held, and cleanup runs from the guard Drop path where an async mutex cannot be awaited.

Validation

• cargo +1.96 test -p contextforge-data-plane-lib gateway:: --lib
• cargo +1.96 test -p contextforge-data-plane-lib --test gateway_modern_subscriptions -- --nocapture
• cargo +1.96 test -p contextforge-data-plane-lib
• git diff --check

E2E Verification

How to run

This E2E test is intentionally ignored by default because it starts a temporary redis-server and the real contextforge-data-plane binary.

cargo +1.96 test -p contextforge-data-plane --test modern_subscriptions_e2e -- --ignored --nocapture

The test pauses before the two MCP calls. Press Enter at each prompt:

Press Enter to run server/discover...
Press Enter to run subscriptions/listen...

E2E test script

// Copyright 2026
// SPDX-License-Identifier: Apache-2.0

use std::{
    collections::HashMap,
    fs,
    io::{self, Write},
    net::TcpStream as StdTcpStream,
    path::PathBuf,
    process::{Child, Command, Stdio},
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use contextforge_data_plane_apis::{
    User,
    user_store::{BackendMCPGateway, Transport, UserConfig, VirtualHost},
};
use jsonwebtoken::{Algorithm, EncodingKey, Header, encode};
use redis::aio::ConnectionManagerConfig;
use rmcp::{
    ErrorData, RoleServer, ServerHandler,
    model::{
        Implementation, InitializeRequestParams, InitializeResult, ProtocolVersion, ServerCapabilities, ServerInfo,
    },
    service::RequestContext,
    transport::{
        StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager,
    },
};
use serde_json::{Value, json};
use tokio::net::TcpListener;

const TEST_USER_ID: &str = "11111111-1111-1111-1111-111111111111";
const TEST_USER_EMAIL: &str = "admin@example.com";
const TEST_VIRTUAL_HOST_ID: &str = "vh-modern-subscriptions-e2e";
const TEST_TOKEN_TTL_SECS: u64 = 60 * 60;

#[derive(Clone)]
struct NotificationBackend;

impl ServerHandler for NotificationBackend {
    fn get_info(&self) -> ServerInfo {
        ServerInfo::new(subscription_capabilities())
            .with_server_info(Implementation::new("modern-subscriptions-e2e-backend", "0.1.0"))
            .with_protocol_version(ProtocolVersion::V_2026_07_28)
    }

    async fn initialize(
        &self,
        _request: InitializeRequestParams,
        _cx: RequestContext<RoleServer>,
    ) -> Result<InitializeResult, ErrorData> {
        Ok(InitializeResult::new(subscription_capabilities())
            .with_server_info(Implementation::new("modern-subscriptions-e2e-backend", "0.1.0")))
    }
}

struct ChildProcess {
    name: &'static str,
    child: Child,
    temp_dir: Option<PathBuf>,
    port: Option<u16>,
}

impl ChildProcess {
    fn new(name: &'static str, child: Child) -> Self {
        Self { name, child, temp_dir: None, port: None }
    }

    fn with_temp_dir(mut self, temp_dir: PathBuf) -> Self {
        self.temp_dir = Some(temp_dir);
        self
    }

    fn with_port(mut self, port: u16) -> Self {
        self.port = Some(port);
        self
    }

    fn port(&self) -> u16 {
        self.port.expect("child process records a port")
    }
}

impl Drop for ChildProcess {
    fn drop(&mut self) {
        if self.child.try_wait().ok().flatten().is_none() {
            let _ = self.child.kill();
        }
        let _ = self.child.wait();
        if let Some(temp_dir) = &self.temp_dir {
            let _ = fs::remove_dir_all(temp_dir);
        }
    }
}

struct RunningBackend {
    url: String,
    handle: tokio::task::JoinHandle<()>,
}

impl Drop for RunningBackend {
    fn drop(&mut self) {
        self.handle.abort();
    }
}

struct E2eEnvironment {
    gateway_url: String,
    _backend: RunningBackend,
    _redis: ChildProcess,
    _gateway: ChildProcess,
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "spawns redis-server and the contextforge-data-plane binary"]
async fn binary_e2e_modern_discover_and_listen_acknowledge_vhost_filter() {
    let env = start_environment().await;
    println!(
        "E2E setup complete: gateway={} backend={} redis_port={} virtual_host={TEST_VIRTUAL_HOST_ID}",
        env.gateway_url,
        env._backend.url,
        env._redis.port(),
    );

    let client = authenticated_client();

    wait_for_enter("Press Enter to run server/discover...");
    println!("E2E: sending server/discover");
    let discover = post_mcp(
        &client,
        &env.gateway_url,
        "server/discover",
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "server/discover",
            "params": {
                "_meta": request_meta()
            }
        }),
    )
    .await;

    assert_eq!(Some("complete"), discover.pointer("/result/resultType").and_then(Value::as_str));
    assert_eq!(Some(true), discover.pointer("/result/capabilities/tools/listChanged").and_then(Value::as_bool));
    assert_eq!(Some(true), discover.pointer("/result/capabilities/prompts/listChanged").and_then(Value::as_bool));
    assert_eq!(Some(true), discover.pointer("/result/capabilities/resources/subscribe").and_then(Value::as_bool));
    assert_eq!(Some(true), discover.pointer("/result/capabilities/resources/listChanged").and_then(Value::as_bool));
    println!(
        "E2E: server/discover capabilities tools.listChanged={} prompts.listChanged={} resources.subscribe={} resources.listChanged={}",
        bool_at(&discover, "/result/capabilities/tools/listChanged"),
        bool_at(&discover, "/result/capabilities/prompts/listChanged"),
        bool_at(&discover, "/result/capabilities/resources/subscribe"),
        bool_at(&discover, "/result/capabilities/resources/listChanged"),
    );

    wait_for_enter("Press Enter to run subscriptions/listen...");
    println!("E2E: sending subscriptions/listen with one unroutable resource URI");
    let acknowledged = post_streaming_mcp_until(
        &client,
        &env.gateway_url,
        "subscriptions/listen",
        "notifications/subscriptions/acknowledged",
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "subscriptions/listen",
            "params": {
                "notifications": {
                    "toolsListChanged": true,
                    "resourcesListChanged": true,
                    "resourceSubscriptions": ["unroutable://resource"]
                },
                "_meta": request_meta()
            }
        }),
    )
    .await;

    assert_eq!(Some("notifications/subscriptions/acknowledged"), acknowledged.get("method").and_then(Value::as_str));
    assert_eq!(
        Some(2),
        acknowledged.pointer("/params/_meta/io.modelcontextprotocol~1subscriptionId").and_then(Value::as_i64)
    );
    assert_eq!(Some(true), acknowledged.pointer("/params/notifications/toolsListChanged").and_then(Value::as_bool));
    assert_eq!(Some(true), acknowledged.pointer("/params/notifications/resourcesListChanged").and_then(Value::as_bool));
    assert!(
        acknowledged.pointer("/params/notifications/resourceSubscriptions").is_none(),
        "unroutable resource URI must be dropped from the acknowledged filter: {acknowledged}"
    );
    println!(
        "E2E: subscriptions/listen ack method={} subscription_id={} resourceSubscriptions_present={}",
        acknowledged.get("method").and_then(Value::as_str).unwrap_or("<missing>"),
        acknowledged.pointer("/params/_meta/io.modelcontextprotocol~1subscriptionId").and_then(Value::as_i64).unwrap_or(-1),
        acknowledged.pointer("/params/notifications/resourceSubscriptions").is_some(),
    );
}

async fn start_environment() -> E2eEnvironment {
    let backend = start_backend().await;

    let redis = start_redis().await;

    write_redis_config(redis.port(), &backend).await;

    let gateway_port = openport::pick_random_unused_port().expect("gateway port");
    let mut gateway = start_gateway_process(gateway_port, redis.port());
    wait_for_port(gateway_port, &mut gateway).await;

    E2eEnvironment {
        gateway_url: format!("http://127.0.0.1:{gateway_port}/contextforge-rs/servers/{TEST_VIRTUAL_HOST_ID}/mcp"),
        _backend: backend,
        _redis: redis,
        _gateway: gateway,
    }
}

async fn start_backend() -> RunningBackend {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("backend binds");
    let port = listener.local_addr().expect("backend address").port();
    let service = StreamableHttpService::new(
        || Ok(NotificationBackend),
        LocalSessionManager::default().into(),
        StreamableHttpServerConfig::default(),
    );
    let router = axum::Router::new().route_service("/mcp", service);
    let handle = tokio::spawn(async move {
        axum::serve(listener, router).await.expect("backend serves");
    });
    RunningBackend { url: format!("http://127.0.0.1:{port}/mcp"), handle }
}

async fn start_redis() -> ChildProcess {
    let port = openport::pick_random_unused_port().expect("redis port");
    let temp_dir = std::env::temp_dir()
        .join(format!("contextforge-data-plane-modern-subscriptions-redis-{}-{port}", std::process::id()));
    fs::create_dir_all(&temp_dir).expect("redis temp dir is created");

    let child = Command::new("redis-server")
        .args(["--port", &port.to_string(), "--save", "", "--appendonly", "no", "--dir"])
        .arg(temp_dir.to_str().expect("redis temp dir is UTF-8"))
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("redis-server starts; install redis-server to run this ignored E2E test");
    let mut redis = ChildProcess::new("redis-server", child).with_temp_dir(temp_dir).with_port(port);
    wait_for_redis(port, &mut redis).await;
    redis
}

fn start_gateway_process(gateway_port: u16, redis_port: u16) -> ChildProcess {
    let binary = env!("CARGO_BIN_EXE_contextforge-data-plane");
    let child = Command::new(binary)
        .current_dir(env!("CARGO_MANIFEST_DIR"))
        .args([
            "--address",
            &format!("127.0.0.1:{gateway_port}"),
            "--redis-port",
            &redis_port.to_string(),
            "--redis-address",
            "127.0.0.1",
            "--token-verification-public-key",
            "../../assets/jwt.key.pub",
            "--number-of-cpus",
            "1",
            "--redis-mode",
            "plain-text",
            "--upstream-connection-mode",
            "plain-text-or-tls",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("contextforge-data-plane binary starts");
    ChildProcess::new("contextforge-data-plane", child)
}

async fn write_redis_config(redis_port: u16, backend: &RunningBackend) {
    let client = redis::Client::open(format!("redis://127.0.0.1:{redis_port}/")).expect("redis client opens");
    let mut connection = client
        .get_connection_manager_with_config(ConnectionManagerConfig::default())
        .await
        .expect("redis connection opens");
    redis::cmd("FLUSHDB").query_async::<String>(&mut connection).await.expect("redis flush succeeds");

    let key = rmp_serde::encode::to_vec(&User::new(TEST_USER_ID)).expect("user key encodes");
    let config = UserConfig {
        virtual_hosts: HashMap::from([(
            TEST_VIRTUAL_HOST_ID.to_owned(),
            VirtualHost {
                backends: HashMap::from([
                    ("gateway-one".to_owned(), backend_config("gateway-one", &backend.url)),
                    ("gateway-two".to_owned(), backend_config("gateway-two", &backend.url)),
                ]),
            },
        )]),
    };
    let encoded = rmp_serde::encode::to_vec(&config).expect("user config encodes");
    redis::cmd("SET")
        .arg(key)
        .arg(encoded)
        .query_async::<String>(&mut connection)
        .await
        .expect("user config is written");
}

fn backend_config(name: &str, url: &str) -> BackendMCPGateway {
    BackendMCPGateway {
        name: name.to_owned(),
        url: url.parse().expect("backend URL parses"),
        transport: Transport::StreamableHttp,
        passthrough_headers: Vec::new(),
        add_headers: HashMap::new(),
        remove_headers: Vec::new(),
        allowed_tool_names: Vec::new(),
        tool_name_aliases: HashMap::new(),
        allowed_resource_names: Vec::new(),
        allowed_prompt_names: Vec::new(),
    }
}

async fn wait_for_redis(port: u16, child: &mut ChildProcess) {
    let client = redis::Client::open(format!("redis://127.0.0.1:{port}/")).expect("redis client opens");
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        assert_child_running(child);
        if let Ok(mut connection) = client.get_connection_manager_with_config(ConnectionManagerConfig::default()).await
            && redis::cmd("PING").query_async::<String>(&mut connection).await.is_ok()
        {
            return;
        }
        assert!(Instant::now() < deadline, "redis-server did not start on port {port}");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

async fn wait_for_port(port: u16, child: &mut ChildProcess) {
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        assert_child_running(child);
        if StdTcpStream::connect(("127.0.0.1", port)).is_ok() {
            return;
        }
        assert!(Instant::now() < deadline, "{} did not start on port {port}", child.name);
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

fn assert_child_running(child: &mut ChildProcess) {
    match child.child.try_wait() {
        Ok(None) => {},
        Ok(Some(status)) => panic!("{} exited before the E2E test completed: {status}", child.name),
        Err(error) => panic!("failed to inspect {} process: {error}", child.name),
    }
}

fn authenticated_client() -> reqwest::Client {
    reqwest::Client::builder()
        .default_headers({
            let mut headers = http::HeaderMap::new();
            headers.insert(
                http::header::AUTHORIZATION,
                http::HeaderValue::from_str(&format!("Bearer {}", token(TEST_USER_ID))).expect("auth header is valid"),
            );
            headers
        })
        .build()
        .expect("client builds")
}

async fn post_mcp(client: &reqwest::Client, gateway_url: &str, method: &str, body: Value) -> Value {
    println!("E2E request: POST {method}");
    let response = client
        .post(gateway_url)
        .header(http::header::CONTENT_TYPE, "application/json")
        .header(http::header::ACCEPT, "application/json, text/event-stream")
        .header("MCP-Protocol-Version", "2026-07-28")
        .header("Mcp-Method", method)
        .json(&body)
        .send()
        .await
        .expect("MCP request sends");
    let status = response.status();
    println!("E2E response: {method} status={status}");
    let body = response.text().await.expect("MCP response body reads");
    assert!(status.is_success(), "expected success response, got {status}: {body}");
    first_response_value(&body)
}

async fn post_streaming_mcp_until(
    client: &reqwest::Client,
    gateway_url: &str,
    method: &str,
    expected: &str,
    body: Value,
) -> Value {
    println!("E2E request: POST {method}, waiting for {expected}");
    let mut response = client
        .post(gateway_url)
        .header(http::header::CONTENT_TYPE, "application/json")
        .header(http::header::ACCEPT, "application/json, text/event-stream")
        .header("MCP-Protocol-Version", "2026-07-28")
        .header("Mcp-Method", method)
        .json(&body)
        .send()
        .await
        .expect("streaming MCP request sends");
    let status = response.status();
    println!("E2E response: {method} status={status}");
    assert!(status.is_success(), "expected success response, got {status}");

    let body = tokio::time::timeout(Duration::from_secs(2), async {
        let mut body = String::new();
        loop {
            if body.contains(expected) {
                return body;
            }
            let chunk = response.chunk().await.expect("stream chunk reads").expect("stream stays open until ack");
            body.push_str(&String::from_utf8_lossy(&chunk));
        }
    })
    .await
    .expect("acknowledgment arrives before timeout");

    println!("E2E stream: received {expected}");
    first_response_value(&body)
}

fn bool_at(value: &Value, pointer: &str) -> bool {
    value.pointer(pointer).and_then(Value::as_bool).unwrap_or(false)
}

fn wait_for_enter(message: &str) {
    print!("\n{message} ");
    io::stdout().flush().expect("stdout flushes");
    let mut line = String::new();
    io::stdin().read_line(&mut line).expect("stdin reads");
}

fn first_response_value(body: &str) -> Value {
    response_values(body).into_iter().next().expect("response contains JSON-RPC data")
}

fn response_values(body: &str) -> Vec<Value> {
    let values = body
        .lines()
        .filter_map(|line| line.strip_prefix("data:"))
        .map(str::trim)
        .filter(|data| !data.is_empty())
        .map(|data| serde_json::from_str(data).expect("SSE data is JSON"))
        .collect::<Vec<_>>();
    if !values.is_empty() {
        return values;
    }

    vec![serde_json::from_str(body.trim()).expect("response body is JSON")]
}

fn request_meta() -> Value {
    json!({
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {
            "name": "curl",
            "version": "0.1.0"
        },
        "io.modelcontextprotocol/clientCapabilities": {}
    })
}

fn subscription_capabilities() -> ServerCapabilities {
    ServerCapabilities::builder()
        .enable_prompts()
        .enable_prompts_list_changed()
        .enable_resources()
        .enable_resources_subscribe()
        .enable_resources_list_changed()
        .enable_tools()
        .enable_tool_list_changed()
        .build()
}

fn token(user_id: &str) -> String {
    let key = EncodingKey::from_rsa_pem(&fs::read("../../assets/jwt.key").expect("jwt key")).expect("encoding key");
    let mut header = Header::new(Algorithm::RS256);
    header.kid = Some("test".to_owned());
    let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("system clock").as_secs();
    let claims = json!({
        "iss": "mcpgateway",
        "sub": user_id,
        "aud": "mcpgateway-api",
        "exp": now + TEST_TOKEN_TTL_SECS,
        "iat": now,
        "jti": "test-token",
        "token_use": "api",
        "teams": ["team_awesome"],
        "user": {
            "email": TEST_USER_EMAIL,
            "full_name": "API Token User",
            "is_admin": true,
            "auth_provider": "api_token"
        },
        "scopes": {
            "server_id": "my_id",
            "permissions": ["tools.read", "servers.use"],
            "ip_restrictions": ["192.169.1.0/24"],
            "time_restrictions": null
        },
    });
    encode(&header, &claims, &key).expect("jwt token")
}

Expected result

E2E setup complete: gateway=http://127.0.0.1:.../contextforge-rs/servers/vh-modern-subscriptions-e2e/mcp backend=http://127.0.0.1:.../mcp redis_port=... virtual_host=vh-modern-subscriptions-e2e

Press Enter to run server/discover...
E2E: sending server/discover
E2E request: POST server/discover
E2E response: server/discover status=200 OK
E2E: server/discover capabilities tools.listChanged=true prompts.listChanged=true resources.subscribe=true resources.listChanged=true

Press Enter to run subscriptions/listen...
E2E: sending subscriptions/listen with one unroutable resource URI
E2E request: POST subscriptions/listen, waiting for notifications/subscriptions/acknowledged
E2E response: subscriptions/listen status=200 OK
E2E stream: received notifications/subscriptions/acknowledged
E2E: subscriptions/listen ack method=notifications/subscriptions/acknowledged subscription_id=2 resourceSubscriptions_present=false

test result: ok. 1 passed

This confirms server/discover reports the expected virtual-host capabilities, subscriptions/listen is accepted and acknowledged, and the unroutable resourceSubscriptions entry is not accepted.

Manual Verification

Manual test steps

Start Redis and sample backends:

docker compose -f docker/docker-compose-local.yaml up -d

Run the gateway with local helper routes enabled:

cargo +1.96 run --features contextforge-data-plane-lib/with_tools \
  --bin contextforge-data-plane -- \
  --address 0.0.0.0:8001 \
  --redis-port 6379 \
  --redis-address 127.0.0.1 \
  --token-verification-public-key assets/jwt.key.pub \
  --token-verification-private-key assets/jwt.key \
  --number-of-cpus 1 \
  --redis-mode=plain-text \
  --upstream-connection-mode=plain-text-or-tls

Mint a local token and seed a two-backend virtual host:

USER_ID=11111111-1111-1111-1111-111111111111
USER_EMAIL=admin@example.com

TOKEN=$(curl --silent --show-error \
  "http://127.0.0.1:8001/contextforge-rs/admin/tokens/${USER_ID}?email=${USER_EMAIL}")

curl --silent --show-error --request POST \
  --url "http://127.0.0.1:8001/contextforge-rs/admin/userconfigs/${USER_ID}" \
  --header "content-type: application/json" \
  --data '{
    "virtual_hosts": {
      "c0ffee00f001f00lf00ldeadbeefdead": {
        "backends": {
          "gateway-one": {
            "name": "gateway-one",
            "url": "http://127.0.0.1:5555/mcp",
            "transport": "STREAMABLEHTTP",
            "passthrough_headers": [],
            "allowed_tool_names": [],
            "allowed_resource_names": [],
            "allowed_prompt_names": []
          },
          "gateway-two": {
            "name": "gateway-two",
            "url": "http://127.0.0.1:5556/mcp",
            "transport": "STREAMABLEHTTP",
            "passthrough_headers": [],
            "allowed_tool_names": [],
            "allowed_resource_names": [],
            "allowed_prompt_names": []
          }
        }
      }
    }
  }'

Run server/discover:

ENDPOINT="http://127.0.0.1:8001/contextforge-rs/servers/c0ffee00f001f00lf00ldeadbeefdead/mcp"

curl -i --show-error \
  --url "${ENDPOINT}" \
  --header "authorization: Bearer ${TOKEN}" \
  --header "content-type: application/json" \
  --header "accept: application/json, text/event-stream" \
  --header "mcp-protocol-version: 2026-07-28" \
  --header "mcp-method: server/discover" \
  --data '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "server/discover",
    "params": {
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {"name": "curl", "version": "0.1.0"},
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }'

Run subscriptions/listen with one unroutable resource URI:

curl --no-buffer --max-time 5 --show-error \
  --url "${ENDPOINT}" \
  --header "authorization: Bearer ${TOKEN}" \
  --header "content-type: application/json" \
  --header "accept: application/json, text/event-stream" \
  --header "mcp-protocol-version: 2026-07-28" \
  --header "mcp-method: subscriptions/listen" \
  --data '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "subscriptions/listen",
    "params": {
      "notifications": {
        "toolsListChanged": true,
        "resourcesListChanged": true,
        "resourceSubscriptions": ["unroutable://resource"]
      },
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {"name": "curl", "version": "0.1.0"},
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }'

Manual test results

server/discover returned HTTP/1.1 200 OK with text/event-stream and vhost-derived notification capabilities:

{
  "capabilities": {
    "completions": {},
    "prompts": {"listChanged": true},
    "resources": {"subscribe": true, "listChanged": true},
    "tools": {"listChanged": true}
  }
}

subscriptions/listen returned notifications/subscriptions/acknowledged:

{
  "method": "notifications/subscriptions/acknowledged",
  "params": {
    "_meta": {"io.modelcontextprotocol/subscriptionId": 2},
    "notifications": {
      "toolsListChanged": true,
      "resourcesListChanged": true
    }
  }
}

The unroutable resourceSubscriptions entry was dropped from the acknowledged filter. The curl timeout after 5 seconds is expected because subscriptions/listen is a long-lived stream.

[lucarlig]

Requesting changes because the dataplane downstream contract is modern MCP 2026-07-28 only. This change still advertises and admits legacy protocol behavior, leaves RMCP legacy session routing enabled, and advertises subscriptions before any notification delivery path exists.

[lucarlig]

The protocol allowlist issue is fixed and its thread is resolved. Two P1 blockers remain: the downstream endpoint still exposes RMCP legacy session/fallback routing, and it advertises subscription capabilities before any notification delivery path exists.

[gandhipratik203]

Updated the checked-in wiki docs under _context/wiki to match this PR scope: downstream subscriptions/listen admission + sink registry, with notification relay delivery still tracked separately.

[lucarlig]

LGTM