The node-local registry behind subscriptions/listen (protocol 2026-07-28), and the dispatch of change notifications to it.

A modern client opens a long-lived notification stream by POSTing subscriptions/listen with a notifications filter (ConduitMcp.Subscriptions.Filter). The request process serving that stream registers itself here under one topic per notification type it agreed to deliver, then relays whatever is published to those topics as Server-Sent Events until the client or the server closes the stream. The server pushes with the public API on ConduitMcp:

ConduitMcp.notify_tools_changed()
ConduitMcp.notify_prompts_changed()
ConduitMcp.notify_resources_changed()
ConduitMcp.notify_resource_updated("file:///project/config.json")

Topics

A resource topic matches its URI exactly: notify_resource_updated/1 for "file:///a/b" does not reach a stream subscribed to "file:///a".

A task topic carries the task's owner (ConduitMcp.Principal.id/1, nil when unauthenticated), and a stream subscribes under its own principal's id, so a task's changes reach only its owner's streams (SV-1). They are published by ConduitMcp.Ext.Tasks.ListenNotifier, the default notifier of ConduitMcp.Ext.Tasks.

Topics are shared by every Server and Endpoint in the node: the notify_* functions take no server module, so a notification reaches the matching streams of every endpoint. When endpoints serve different audiences, gate each resource with scope or the endpoint's handle_subscribe_resource/2 (both apply to resourceSubscriptions), and don't put anything an audience must not learn in a URI another endpoint lets it subscribe to.

Node-local

The registry is a Registry started under ConduitMcp.Supervisor (duplicate keys), so a notification reaches only the streams served by the node that published it. On a cluster, broadcast the change on your own PubSub and call the ConduitMcp.notify_* function on every node:

# wherever the change happens
Phoenix.PubSub.broadcast(MyApp.PubSub, "mcp", :tools_changed)

# in a process subscribed to "mcp" on every node
def handle_info(:tools_changed, state) do
  ConduitMcp.notify_tools_changed()
  {:noreply, state}
end

Lifecycle and bounds

Registry entries belong to the stream's process: a stream that ends unregisters them, and the registry drops the entries of a process that dies (it is linked to every registrant), so a killed stream leaks nothing. The same holds for its entry in ConduitMcp.Subscriptions.Admission, which monitors every admitted stream.

Each stream is bounded by the transport (see ConduitMcp.Transport.StreamableHTTP): at most :max_listens_per_principal live streams per principal (default 4), the shared :max_streams cap (503 when full), :max_stream_lifetime, the bearer token's exp, and :listen_idle_timeout. A filter may name at most 1 000 resource URIs (ConduitMcp.Subscriptions.Filter).

A stream's mailbox is bounded too. A stream writes each notification to its socket synchronously, so a client that stops reading parks it in the send (up to the socket's send timeout) while notifications keep arriving. publish/2 skips any stream already holding max_queue/0 messages (config :conduit_mcp, :listen_max_queue, default 1 000): that stream misses the notification, every other stream still gets it, and [:conduit_mcp, :subscription, :drop] is emitted. The stream itself is left to end through its own bounds (a failed write, the lifetime, idle timeout or supersession); a client that has missed notifications should re-read what it cares about after reconnecting.

The per-principal cap is never answered 429: a listen that would exceed it is admitted and the principal's oldest stream is superseded - it ends with notifications/cancelled for its listen id, then the graceful-closure result, then closes (telemetry subscription.stop reason :superseded). A client that went away is only noticed on the stream's next write, so a client opening one listen per run (e.g. the MCP Inspector CLI) would otherwise be locked out by its own dead streams. Admission and eviction are one step of ConduitMcp.Subscriptions.Admission, a node-local process that answers admissions in arrival order, so even a burst of concurrent listens never holds more than the cap live at once; superseded streams still writing their teardown are not counted (each also holds a :max_streams slot until it closes). One principal still never holds more than the cap (SV-10), and another principal's streams are never evicted.