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

Cleans up when a streaming request is closed from the connection side. Optional.

This callback fires only when a streaming request is torn down by something other than the handler's own return value — practically, when the client disconnects or the network fails (the transport-level cancellation signal on this binding). It is not called when the handler ends the request itself by returning {:result, …}, {:stop, …}, or {:error, …} from handle_request/3 or handle_message/3; those finish the request directly, with no cleanup hook. (A handler that closes its own stream with GenMCP.Mux.Channel.close/1 is the one server-side exception that still runs this callback.)

It receives the request's GenMCP.Mux.Channel.t/0, now marked closed, and the last state. It is the place to release resources the handler acquired while streaming, such as unsubscribing from a Phoenix.PubSub topic. The return value is ignored.

The callback is optional. Define it only when a streaming handler holds resources that must be released on disconnect:

@impl true
def handle_close(_channel, state) do
  Phoenix.PubSub.unsubscribe(MyApp.PubSub, state.topic)
end

@callback handle_message(message :: term(), GenMCP.Mux.Channel.t(), state()) ::
  {:stream, state()}
  | {:result, result()}
  | {:result, result(), stop_reason :: term()}
  | {:stop, reason :: term()}
  | {:error, reason :: term()}

Handles a process message while a request is streaming.

This callback runs only after handle_request/3 returned {:stream, state}. Once a request is streaming, the worker forwards every Erlang message it receives to this callback, so a handler that spawns tasks, subscribes to a Phoenix.PubSub topic, or monitors another process receives those messages here. It is passed the raw message, the request's GenMCP.Mux.Channel.t/0, and the current state.

Return one of:

• {:stream, state} keeps the stream open and waits for the next message, carrying the updated state.
• {:result, result} ends the stream with the request's final result.
• {:result, result, stop_reason} ends the stream with result, then stops the worker process with stop_reason instead of the default {:shutdown, :reply}. The client sees the same final result; stop_reason only sets the worker's exit reason. Use :normal, :shutdown, or {:shutdown, term} for a clean exit.
• {:error, reason} ends the stream with a JSON-RPC error.
• {:stop, reason} ends the stream with no further result, for example when a process the handler was listening to exits normally. Use :normal, :shutdown, or {:shutdown, term} for a clean exit.

Send intermediate progress and log notifications through the channel with GenMCP.Mux.Channel.send_progress/4 and GenMCP.Mux.Channel.send_log/4 before returning.

Examples

A handler waiting on a background job. The job sends its own status messages to the worker process; an update keeps the stream open and reports progress to the client, and the completion message ends the request:

@impl true
def handle_message({:job_update, job_id, done, total}, channel, %{job_id: job_id} = state) do
  GenMCP.Mux.Channel.send_progress(channel, done, total)
  {:stream, state}
end

def handle_message({:job_finished, job_id, output}, _channel, %{job_id: job_id}) do
  {:result, MCP.call_tool_result(text: output)}
end

@callback handle_notification(notification(), GenMCP.Mux.Channel.t(), state()) :: :ok

Observes a client notification. Returns :ok.

Each client notification arrives as its own HTTP POST that is answered with 202 Accepted and never streams, so there is nothing to return beyond :ok, and no state is carried forward. The notification struct, its own GenMCP.Mux.Channel.t/0, and the state from init/1 are passed in. The notification types are listed in notification/0.

The channel is the notification's own per-request context, a sibling of any in-flight request rather than a handle to it. Read channel.meta (client info, capabilities, authorization assigns) to decide whether to trust the sender. A notification cannot reach or cancel another request: on this transport, cancellation is signalled by the client closing the connection, which the framework already turns into handle_close/2.

The default behaviour, and a fine implementation when there is nothing to observe, is to accept and ignore:

@impl true
def handle_notification(_notification, _channel, _state) do
  :ok
end

@callback handle_request(request(), GenMCP.Mux.Channel.t(), state()) ::
  {:result, result()}
  | {:result, result(), stop_reason :: term()}
  | {:error, reason :: term()}
  | {:stream, state()}

Handles one MCP request and either answers it or upgrades it to a stream.

This is the primary callback. It receives the decoded request struct, the request's GenMCP.Mux.Channel.t/0, and the state from init/1. Match on the request struct to route the call. The request types are listed in request/0: tools/list, tools/call, the resources/* and prompts/* requests, and the server/discover capability snapshot.

Return one of:

• {:result, result} answers the request and ends it. Build result with the helpers in GenMCP.MCP.V2607, for example GenMCP.MCP.V2607.list_tools_result/2 or GenMCP.MCP.V2607.call_tool_result/1.
• {:result, result, stop_reason} answers the request with result, then stops the worker process with stop_reason instead of the default {:shutdown, :reply}. The client sees the same response; stop_reason only sets the worker's exit reason, observable to whatever monitors or supervises it (and in telemetry). Use :normal, :shutdown, or {:shutdown, term} for a clean exit.
• {:error, reason} ends the request with a JSON-RPC error.
• {:stream, state} holds the response open as an SSE stream and routes every later message to handle_message/3 with the returned state.

A handler that answers with {:result, result} after a long stretch of work can open the stream while it works by calling GenMCP.Mux.Channel.start_stream/1 on its channel, which gets the response its periodic keepalives.

Examples

Answer tools/call for one known tool and reject the rest:

@impl true
def handle_request(%MCP.CallToolRequest{params: %{name: "ping"}}, _channel, _state) do
  {:result, MCP.call_tool_result(text: "pong")}
end

def handle_request(%MCP.CallToolRequest{}, _channel, _state) do
  {:error, :method_not_found}
end

@callback init(init_arg :: term()) :: {:ok, state()} | {:stop, term()}

Builds the per-request state before any handler runs.

init/1 is called once for every incoming request or notification, on a fresh worker, before the matching handler. Keep it cheap: the stateless core runs it on the hot path of each message, not once per session.

The argument is the server configuration. When the server is wired as a bare module (server: MyServer), init/1 receives the leftover transport options as a keyword list. When wired as {MyServer, arg}, it receives arg unchanged.

Return {:ok, state} to proceed, where state is threaded into the handler, or {:stop, reason} to abort the request before it is handled.