@callback handle_close(GenMCP.Mux.Channel.t(), state(), arg()) :: term()

Cleans up when the stream is closed from the connection side. Optional.

The Suite-level mirror of GenMCP.handle_close/2, invoked when the subscription stream is torn down from the outside while this handler owns it — the client disconnecting or the network failing — rather than by the handler's own {:stop, reason} (which ends the stream with a SubscriptionsListenResult and does not run this callback). The channel is already :closed, so nothing more can be sent and the return value is ignored. Use it only for side effects: unsubscribing the event source joined in subscribe/3, stopping helper processes, and so on.

If the handler does not implement it, the worker stops immediately with no cleanup.

Examples

@impl true
def handle_close(_channel, _state, _arg) do
  Phoenix.PubSub.unsubscribe(MyApp.PubSub, "tools")
  :ok
end

@callback handle_message(msg :: term(), GenMCP.Mux.Channel.t(), state(), arg()) ::
  {:stream, state()} | {:stop, reason :: term()}

Turns one process message received while the stream is open into notifications.

Invoked for each message the worker process receives during the subscription, typically a delivery from the event source joined in subscribe/3. Send the matching change notifications with GenMCP.Mux.Channel.send_notification/2 (which stamps each one with the io.modelcontextprotocol/subscriptionId that ties it to this stream), then return to continue or stop.

• {:stream, state} - keep the stream open with updated state.
• {:stop, reason} - tear the subscription down gracefully. The Suite ends the stream by sending a GenMCP.MCP.V2607.SubscriptionsListenResult.t/0 (resultType: "complete", stamped with this stream's io.modelcontextprotocol/subscriptionId). The reason is split two ways: it is not surfaced to the client — the spec has no way to signal a failed teardown, so every stop yields the same graceful complete result — but it is forwarded to the OTP layer as the worker's exit reason. Use :normal, :shutdown, or {:shutdown, term} to keep that a clean exit.

Examples

Forward an application broadcast as a tool-list-changed notification, then keep the stream open:

@impl true
def handle_message(:tools_changed, channel, state, _arg) do
  Channel.send_notification(channel, %MCP.ToolListChangedNotification{})
  {:stream, state}
end

@callback subscribe(
  filter :: GenMCP.MCP.V2607.SubscriptionFilter.t(),
  GenMCP.Mux.Channel.t(),
  arg()
) ::
  {:stream, state()}
  | {:stream, honored :: GenMCP.MCP.V2607.SubscriptionFilter.t(), state()}
  | {:stop, reason :: term()}

Accepts or rejects a subscriptions/listen request and sets up the event source.

Runs in the connection-scoped worker process that owns the stream. Subscribe the worker to the application's event source here (Phoenix.PubSub, a GenStage consumer, ...), and use the channel's read-only meta to authorize. The filter is the GenMCP.MCP.V2607.SubscriptionFilter.t/0 the client requested, naming the notification types it wants.

The Suite sends the acknowledged message when this callback accepts, and the spec requires that message to be the first one carrying this subscription's id. Keep this callback to setup and authorization, and emit every notification from handle_message/4, once the stream is acknowledged and open. A handler that wants to push an initial notification right away can send/2 itself a message here and emit from handle_message/4 when it arrives.

Return one of:

• {:stream, state} - accept as requested. The Suite opens the stream and sends the acknowledged message reporting the full requested filter. state is the handler-private value carried to handle_message/4.
• {:stream, honored, state} - accept but downgrade to honored, a filter that narrows (never widens) the request, for auth or capability reasons. The Suite reports honored in the acknowledged message.
• {:stop, reason} - reject. No stream opens and no ack is sent; the Suite returns a normal error response instead.

Enforcing the filter, that is not emitting a type the client did not request, is the handler's job: the Suite does not drop notifications at send time.

Examples

Accept only when the client asked for the type this handler serves, joining the app's pub/sub and starting from empty state:

@impl true
def subscribe(%MCP.SubscriptionFilter{toolsListChanged: true}, _channel, _arg) do
  Phoenix.PubSub.subscribe(MyApp.PubSub, "tools")
  {:stream, %{}}
end

Downgrade a broad request to only the part an unauthenticated caller may see:

@impl true
def subscribe(filter, channel, _arg) do
  honored =
    case channel.meta do
      %{current_user: %{role: :admin}} -> filter
      _ -> %MCP.SubscriptionFilter{toolsListChanged: filter.toolsListChanged}
    end

  {:stream, honored, %{}}
end

@callback subscription_capabilities(GenMCP.Mux.Channel.t(), arg()) ::
  subscription_capabilities()

Declares which subscription notification types this handler can emit. Optional.

The Suite calls this while building server/discover and folds the types declared true into the advertised capabilities, so a conformant client knows it may request them on subscriptions/listen. Return a subscription_capabilities/0 map; the recognized keys are:

• :tools_list_changed - notifications/tools/list_changed.
• :prompts_list_changed - notifications/prompts/list_changed.
• :resources_list_changed - notifications/resources/list_changed.
• :resources_updated - notifications/resources/updated.

Keys that are absent or false advertise nothing. If the handler does not implement this callback, the Suite advertises no subscription capability, so a conformant client will not subscribe.

Examples

@impl true
def subscription_capabilities(_channel, _arg) do
  %{tools_list_changed: true, resources_updated: true}
end