[msureshkumar88]

Pull Request

🔗 Related Issue

Closes #6644

Part of #6571. Stacked on #7414 (#6573): this PR targets feat/6573-subscriptions-listen and shows only the #6644 work. I retarget it to main after #7414 merges.

---

📝 Summary

A 2026-07-28 client can now name resource URIs in resourceSubscriptions on a subscriptions/listen request. The gateway acknowledges the URIs the caller can see. When one of them changes on its modern upstream server, the client gets notifications/resources/updated with that uri, stamped with its subscription ID in _meta.

#7414 dropped resourceSubscriptions and hid resources.subscribe (its D9), because the gateway had no source of per-resource events. This PR adds that source and reverses D9.

---

✅ Acceptance criteria coverage

Criterion (#6644)	Delivered by
A modern client's listen request naming resource URIs in resourceSubscriptions is acknowledged	_adapt_subscriptions_listen resolves the URIs through ResourceService.resolve_subscription_targets() and puts the honored ones in the ack
When a subscribed resource changes on its modern upstream, the client receives notifications/resources/updated with that uri, tagged with the subscription ID in _meta	ResourceSubscriptionRelay holds an upstream subscriptions/listen for each honored URI and republishes each update on the worker's listen bus. The SDK ListenHandler delivers it only to streams that honored the URI, stamped with their subscription ID

---

🏗️ Architecture

flowchart LR
    C[Modern client] -->|"subscriptions/listen<br/>resourceSubscriptions"| A[_adapt_subscriptions_listen]
    A -->|resources.read gate| R["resolve_subscription_targets()<br/>visibility · virtual server · reachable upstream"]
    R -->|honored URIs| H[SDK ListenHandler]
    A -->|"watch(targets) for the stream lifetime"| Y[ResourceSubscriptionRelay<br/>one per worker]
    Y -->|"subscriptions/listen<br/>resourceSubscriptions=[uri]"| U[Modern upstream]
    U -->|notifications/resources/updated| Y
    Y -->|"UpstreamResourceUpdated(uri, gateway)"| B[Listen bus, same worker]
    B -->|"scoped to the stream's (gateway, uri) pairs"| H
    H -->|stamped resources/updated| C

Loading

1. Gate (mcpgateway/transports/streamablehttp_transport.py). URIs need resources.read, the same token-scope gate as resources/list and resources/read. Without it the URIs drop from the ack. A request with only URIs and no resources.read gets the same PermissionError as resources/read.

2. Resolve (ResourceService.resolve_subscription_targets). A URI is honored when an enabled resource with that URI:

   • passes _check_resource_access() for the caller's scoped identity (Layer 1);
   • belongs to the virtual server on /servers/{id}/mcp;
   • comes from an enabled upstream server that a standing stream reaches with its stored headers alone: not SSE (legacy-only), not OAuth (no per-user token on a standing stream), not query-parameter auth, and no private CA or mTLS client certificate.

   Unknown and hidden URIs drop from the ack without an error, so the ack reveals nothing that resources/read would not. One request considers at most 64 distinct URIs. The DB session closes before the stream opens.

3. Relay (mcpgateway/services/resource_subscription_relay.py, new). The stream holds relay.watch(targets) for its lifetime. The relay keeps one upstream Client.listen(resource_subscriptions=[uri]) per (gateway_id, uri) on this worker, shared by reference count. The last release cancels it.

   • Each upstream update publishes UpstreamResourceUpdated(uri, gateway_id) on the worker-local listen bus.
   • Two upstream servers can expose one URI to different teams. A stream that honored URIs is served by its own ListenHandler over scoped_listen_bus(), which passes a resource update only for the (gateway_id, uri) pairs that stream watches.
   • Each worker holds at most 256 distinct watches. A URI that would start one more drops from the ack.
   • A pre-2026 upstream, or an upstream ack that leaves the URI out, ends the relay for that target without reconnecting. A pre-2026 upstream is remembered for an hour, so its URIs drop from later acks and no stream probes it again.
   • A dropped upstream stream reconnects with the modern listener's jittered backoff and publishes one update, because the spec defines no replay.
   • Upstream auth headers come from gateway_listen_headers(), now shared with the modern listener.

4. Capabilities. 2026-07-28 capabilities (also server/discover) advertise resources.subscribe again. Legacy capabilities stay byte-identical.

5. Shutdown. The lifespan shuts the relay down with the other services.

---

🧭 Decisions and why

Decision	Why	Cost if wrong
The upstream stream lives on the worker that holds the downstream stream	Delivery needs no cross-worker channel and no Redis, and the watch ends exactly when the client goes away	Two workers that serve the same URI each hold one upstream stream
One upstream stream per (gateway_id, uri), not one per upstream server	The URI set never changes during a stream, and reference counting is simple	More upstream connections than a batched stream would need; the per-worker cap of 256 watches bounds them
Per-stream ListenHandler over a scoped bus for streams with URIs	The SDK matches resource updates on the URI string only, and the same URI can come from two upstreams in different teams	These streams sit outside the shared handler's 1024-stream cap; the per-principal and anonymous caps from #7414 still bound them
Honor only URIs from upstreams the relay can reach with stored headers	The ack "reflects the subset the server agreed to honor"	OAuth, query-parameter, private-CA and mTLS upstream resources cannot be subscribed yet
A pre-2026 upstream not yet probed is honored in the ack but sends no updates	The upstream era is known only after a connection, and probing before the ack would delay it. The SDK treats a subscription that never fires as valid. After the first probe its URIs drop from acks for an hour	The first stream on a legacy upstream URI gets an ack and no events
The relay publishes the subscribed URI	The upstream stream watches exactly one URI, and the SDK ListenHandler matches URIs exactly	A sub-resource update reaches the client as an update to the subscribed URI
Cap at the first 64 distinct requested URIs	Bounds the DB query and the upstream streams from client input	A client that lists more than 64 URIs gets fewer honored

---

🔒 Security

• Two layers. resources.read gates the request (Layer-1 token-scope ceiling, same helper as */list). Like */list, no per-method Layer-2 RBAC check runs; [FEATURE]: Add per-method Layer-2 RBAC check to streamable HTTP catalog reads and listen #7494 tracks it. _check_resource_access() and virtual-server membership filter each URI (Layer 1), from the server-side user context only. Nothing in params changes the identity.
• No cross-scope delivery. The listen bus is shared per worker. A stream that honored URIs listens through a bus scoped to its own (gateway_id, uri) pairs, so an update from another team's upstream that shares the URI never reaches it.
• No existence leak. A hidden URI and an unknown URI both drop from the ack silently.
• Upstream credentials are the gateway's stored credentials, as for the modern listener and direct-proxy reads. No client token goes upstream.
• Bounds. At most 64 URIs per request and 256 distinct upstream resource streams per worker. Upstream streams are shared per (gateway_id, uri), and the per-principal and anonymous stream caps from feat: serve MCP 2026-07-28 subscriptions/listen to modern clients (#6573) #7414 apply.
• Deny-path tests (unit and live): no resources.read → URIs dropped or PermissionError; another team's resource, a public-only token, a URI outside the virtual server, a disabled resource or gateway, SSE, OAuth, query-parameter, private-CA and mTLS upstreams → URI dropped; an update from another upstream with the same URI → not delivered.

---

🧪 Verification

Check	Command	Status
Pre-commit	make pre-commit	✅ pass
Unit tests (touched and related files)	uv run pytest tests/unit/mcpgateway/transports/test_streamablehttp_transport.py tests/unit/mcpgateway/services/test_resource_subscription_relay.py tests/unit/mcpgateway/services/test_resource_subscription_targets.py tests/unit/mcpgateway/services/test_modern_listener_service.py tests/unit/mcpgateway/services/test_notification_service.py tests/unit/mcpgateway/services/test_resource_service.py tests/unit/mcpgateway/test_main_extended.py tests/unit/mcpgateway/test_main.py -n 8	✅ 2343 passed, 1 skipped
Live gateway proofs	uv run pytest tests/live_gateway/mcp/test_subscriptions_listen.py	✅ 25 passed, cross-worker test skipped (needs the compose stack)
Doctests of touched modules	pytest --doctest-modules on the four touched modules, with the Makefile doctest secrets	✅ 49 passed
Lint, coverage, e2e	make ruff pylint interrogate, make coverage diff-cover, make test-e2e	Left to CI

Live proofs (real gateway and SDK MCPServer upstream subprocesses): the upstream gains a test://note resource and a touch tool that announces an update.

• test_listen_resource_subscription_receives_upstream_update: the ack lists test://note and not an unknown URI; touch on the upstream delivers a stamped notifications/resources/updated for test://note.
• test_listen_resource_subscription_outside_virtual_server_not_acked and test_listen_resource_subscription_needs_resources_read: deny paths.
• test_modern_discover_advertises_resources_subscribe.

Unit proofs: the relay against an in-memory SDK upstream (delivery, sharing, the watch cap, legacy upstream and its cooldown, URI not honored, reconnect, shutdown, the scoped bus); the resolver against a real SQLite session (14 cases); the adapter on the real 2026-07-28 wire, including an update from another upstream with the same URI.

Review: a fresh whole-branch review found no blockers. Its three functionally-impacting findings are fixed with tests: cross-upstream delivery for a shared URI, unbounded upstream streams, and upstreams the relay cannot reach. Its suggestion to remember pre-2026 upstreams is also in.

---

📁 Files

Area	Files
New	mcpgateway/services/resource_subscription_relay.py, tests/unit/mcpgateway/services/test_resource_subscription_relay.py, tests/unit/mcpgateway/services/test_resource_subscription_targets.py
Listen	mcpgateway/transports/streamablehttp_transport.py
Resolve	mcpgateway/services/resource_service.py, mcpgateway/services/modern_listener_service.py (gateway_listen_headers())
Lifespan	mcpgateway/main.py
Docs	docs/docs/architecture/get-stream-quickstart.md, CHANGELOG.md

---

📏 Reviewability

• This PR has one clear purpose
• The linked issue is not labeled triage
• Unrelated bugs or improvements are tracked in separate issues/PRs
• Tests are included with the code they validate
• If AI-assisted, I understand and can explain the generated changes

---

🏷️ Type of Change

• Bug fix
• Feature / Enhancement
• Documentation
• Refactor
• Chore (deps, CI, tooling)
• Other (describe below)

---

✅ Checklist

• Code formatted (make black isort pre-commit)
• Tests added/updated for changes
• Documentation updated (if applicable)
• No secrets or credentials committed

---

📓 Notes

Known limits

• OAuth, SSE, query-parameter, private-CA and mTLS upstream resources cannot be subscribed (dropped from the ack).
• An open watch keeps the upstream URL and credentials it started with. A credential rotation or gateway disable applies when the stream ends (at most MCP_LISTEN_MAX_LIFETIME_SECONDS).
• A pre-2026 upstream resource is acknowledged but sends no updates; legacy resources/subscribe relaying is a possible follow-up.
• Resources that the gateway hosts itself (no upstream) are not honored.

[gandhipratik203]

mcpgateway/transports/streamablehttp_transport.py:3974 | Access is checked only at stream start, so a caller who loses access (team removal, resource made private, disabled or deleted) keeps getting update notifications until the stream ends (up to MCP_LISTEN_MAX_LIFETIME_SECONDS) | Document it in the known limits, or re-check on a timer

[madhu-mohan-jaishankar]

LGTM