# `ConduitMcp.Subscriptions`
[🔗](https://github.com/nyo16/conduit_mcp/blob/v0.11.0/lib/conduit_mcp/subscriptions.ex#L1)

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

| Topic | Filter field | Notification |
|---|---|---|
| `:tools` | `toolsListChanged` | `notifications/tools/list_changed` |
| `:prompts` | `promptsListChanged` | `notifications/prompts/list_changed` |
| `:resources` | `resourcesListChanged` | `notifications/resources/list_changed` |
| `{:resource, uri}` | `resourceSubscriptions` | `notifications/resources/updated` |
| `{:task, owner, task_id}` | `taskIds` | `notifications/tasks` |

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.

# `topic`

```elixir
@type topic() ::
  :tools
  | :prompts
  | :resources
  | {:resource, String.t()}
  | {:task, String.t() | nil, String.t()}
```

What a stream subscribes to; see the Topics table.

# `child_spec`

```elixir
@spec child_spec(term()) :: Supervisor.child_spec()
```

The child spec of the registry, started by `ConduitMcp.Application`.

# `max_queue`

```elixir
@spec max_queue() :: pos_integer()
```

The mailbox bound of a stream: `publish/2` skips a stream holding this many
messages or more. `config :conduit_mcp, :listen_max_queue` (a positive
integer), default 1000.

# `notify_prompts_changed`

```elixir
@spec notify_prompts_changed() :: :ok
```

Sends `notifications/prompts/list_changed` to every local stream subscribed
to `promptsListChanged`. See `ConduitMcp.notify_prompts_changed/0`.

# `notify_resource_updated`

```elixir
@spec notify_resource_updated(String.t()) :: :ok
```

Sends `notifications/resources/updated` for `uri` to every local stream
whose `resourceSubscriptions` named exactly `uri`. See
`ConduitMcp.notify_resource_updated/1`.

# `notify_resources_changed`

```elixir
@spec notify_resources_changed() :: :ok
```

Sends `notifications/resources/list_changed` to every local stream
subscribed to `resourcesListChanged`. See
`ConduitMcp.notify_resources_changed/0`.

# `notify_tools_changed`

```elixir
@spec notify_tools_changed() :: :ok
```

Sends `notifications/tools/list_changed` to every local stream subscribed
to `toolsListChanged`. See `ConduitMcp.notify_tools_changed/0`.

# `publish`

```elixir
@spec publish(term(), map()) :: :ok
```

Delivers `notification` — a complete JSON-RPC notification map, such as a
`ConduitMcp.Notifications` builder returns — to every local stream
subscribed to `topic`.

The stream adds `_meta["io.modelcontextprotocol/subscriptionId"]` itself.
Delivery is a message send per stream: this returns once every stream has
been sent the notification, not once each has written it.

# `subscribers`

```elixir
@spec subscribers(term()) :: [pid()]
```

The pids of the local streams subscribed to `topic`. For introspection and
tests; the order is unspecified.

# `task_topic`

```elixir
@spec task_topic(String.t() | nil, String.t()) :: topic()
```

The topic of `task_id`'s changes for streams of the principal `owner`
(`ConduitMcp.Principal.id/1`; `nil` when unauthenticated).

---

*Consult [api-reference.md](api-reference.md) for complete listing*