Align Streamable HTTP transport with the 2025-06-18/2025-11-25 spec

CHANGELOG.md

@@ -1,3 +1,26 @@
+## Unreleased
+
+Streamable HTTP transport fixes for the protocol versions before 2026-07-28:
+
+- New: `Phantom.Plug` option `validate_session: true` answers `404 Not Found`
+  to a request whose `mcp-session-id` the server did not issue, or that was
+  terminated with `DELETE`, so the client initializes again. It is off by
+  default because Phantom only keeps sessions in memory; apps that persist
+  sessions keep answering `{:not_found, message}` from `connect/2`.
+- Reject requests with an unsupported `MCP-Protocol-Version` header with
+  `400 Bad Request`. Requests without the header are still accepted.
+- Respond `200 OK` instead of `202 Accepted` when opening a GET SSE stream.
+- Respond `406 Not Acceptable` to GET requests whose `Accept` header excludes
+  `text/event-stream`. A missing `Accept`, `text/*`, or `*/*` is allowed.
+- Accept a client's JSON-RPC error response (for example, to
+  `elicitation/create`) with `202 Accepted`. The waiting `Session.elicit/3`
+  returns `:error`; it used to wait until it timed out. Over stdio too.
+- Respond `400 Bad Request` to an invalid JSON-RPC response instead of
+  `202 Accepted`.
+- Allow `DELETE` in CORS `access-control-allow-methods`.
+- Stop using JSON-RPC ids as SSE event ids, since SSE event ids must be unique
+  within a session.
+
 ## 0.6.0 (2026-10-04)
 
 - New: `Phantom.Test` for unit-testing MCP routers without an HTTP transport.

lib/phantom/plug.ex

@@ -3,6 +3,7 @@ defmodule Phantom.Plug do
     pubsub: nil,
     origins: ["http://localhost:4000"],
     validate_origin: true,
+    validate_session: false,
     hosts: :all,
     session_timeout: :timer.seconds(30),
     max_request_size: 1_048_576
@@ -145,6 +146,7 @@ defmodule Phantom.Plug do
           router: module(),
           origins: [String.t()] | :all | mfa(),
           validate_origin: boolean(),
+          validate_session: boolean(),
           hosts: [String.t()] | :all | mfa(),
           session_timeout: pos_integer(),
           max_request_size: pos_integer()
@@ -160,6 +162,12 @@ defmodule Phantom.Plug do
   - `:validate_origin` - Whether to validate the Origin header (default: true). Requests
     without one are allowed: only browsers send it, and only browsers can be used for DNS
     rebinding.
+  - `:validate_session` - Whether to answer `404 Not Found` to a request whose
+    `mcp-session-id` this server did not issue in an `initialize` response, or that was
+    terminated with `DELETE`, so the client initializes again (default: false). Phantom
+    keeps that record in memory, so a restart forgets every session. Apps that persist
+    sessions across restarts should leave this off and answer `{:not_found, message}` from
+    `c:Phantom.Router.connect/2` instead.
   - `:hosts` - List of allowed hosts (from the `Host` header, without the port), `:all`, or an
     MFA called with the host prepended to its arguments (default: `:all`). A server bound to
     localhost should set `hosts: ["localhost", "127.0.0.1", "[::1]"]` to reject DNS
@@ -210,7 +218,7 @@ defmodule Phantom.Plug do
       )
 
     try do
-      case router.connect(session, conn) do
+      case session |> router.connect(conn) |> validate_session(conn, opts) do
         {:ok, session} ->
           session = inherit_session_meta(session)
 
@@ -267,6 +275,41 @@ defmodule Phantom.Plug do
     end
   end
 
+  # A session id the server did not issue (or terminated) is answered with
+  # 404 so the client initializes again. `initialize` itself starts a session,
+  # and requests without a session id are not checked.
+  defp validate_session({:ok, session} = result, conn, %{validate_session: true} = opts) do
+    session_id = get_req_header(conn, "mcp-session-id") |> List.first()
+
+    cond do
+      is_nil(session_id) or conn.private.phantom.modern -> result
+      conn.body_params["method"] == "initialize" -> result
+      known_session?(session.id, opts.pubsub) -> result
+      true -> {:not_found, "Session not found"}
+    end
+  end
+
+  defp validate_session(result, _conn, _opts), do: result
+
+  # Without the table (an escript), sessions cannot be told apart. With a
+  # pubsub, another node's `initialize` may not have replicated here yet.
+  defp known_session?(session_id, pubsub, retries \\ 5) do
+    cond do
+      not Phantom.SessionMeta.available?() ->
+        true
+
+      Phantom.SessionMeta.get(session_id) ->
+        true
+
+      is_nil(pubsub) or retries == 0 ->
+        false
+
+      true ->
+        Process.sleep(20)
+        known_session?(session_id, pubsub, retries - 1)
+    end
+  end
+
   defp validate_request(conn, opts) do
     cond do
       not valid_host?(conn.host, opts[:hosts]) ->
@@ -289,6 +332,22 @@ defmodule Phantom.Plug do
         |> put_status(405)
         |> request_error(Request.not_found("Method not allowed"))
 
+      not supported_protocol_version?(conn) ->
+        conn
+        |> put_status(400)
+        |> request_error(
+          Request.invalid("Unsupported protocol version")
+          |> Map.put(:data, %{
+            supported: Request.supported_protocols(),
+            requested: mcp_header(conn, "mcp-protocol-version")
+          })
+        )
+
+      conn.method == "GET" and not accepts_event_stream?(conn) ->
+        conn
+        |> put_status(406)
+        |> request_error(Request.invalid("Accept header must include text/event-stream"))
+
       conn.method not in ~w[DELETE GET OPTIONS] and map_size(conn.body_params) == 0 ->
         conn
         |> put_status(400)
@@ -304,6 +363,36 @@ defmodule Phantom.Plug do
     end
   end
 
+  defp client_response?(params),
+    do:
+      (is_map_key(params, "result") or is_map_key(params, "error")) and
+        not is_map_key(params, "method")
+
+  # Clients before 2025-06-18 do not send the header, so its absence is allowed.
+  # `validate_protocol_request/1` checks the version of a 2026-07-28 request.
+  defp supported_protocol_version?(%Plug.Conn{private: %{phantom: %{modern: true}}}), do: true
+
+  defp supported_protocol_version?(conn) do
+    case mcp_header(conn, "mcp-protocol-version") do
+      nil -> true
+      version -> version in Request.supported_protocols()
+    end
+  end
+
+  # A missing Accept header means the client accepts any media type.
+  defp accepts_event_stream?(conn) do
+    case get_req_header(conn, "accept") do
+      [] ->
+        true
+
+      accept ->
+        accept
+        |> Enum.flat_map(&String.split(&1, ","))
+        |> Enum.map(&(&1 |> String.split(";") |> hd() |> String.trim() |> String.downcase()))
+        |> Enum.any?(&(&1 in ~w[text/event-stream text/* */*]))
+    end
+  end
+
   defp validate_protocol_request(%Plug.Conn{halted: true} = conn), do: conn
 
   defp validate_protocol_request(%Plug.Conn{method: "POST"} = conn) do
@@ -319,7 +408,7 @@ defmodule Phantom.Plug do
       not is_map(conn.body_params) ->
         protocol_error(conn, nil, Request.invalid())
 
-      is_map_key(conn.body_params, "result") and not is_map_key(conn.body_params, "method") ->
+      client_response?(conn.body_params) ->
         protocol_error(conn, conn.body_params["id"], Request.invalid())
 
       is_nil(version) ->
@@ -383,7 +472,7 @@ defmodule Phantom.Plug do
           |> put_resp_header("cache-control", "no-cache, no-transform")
           |> put_resp_content_type("text/event-stream")
           |> put_resp_header("x-accel-buffering", "no")
-          |> send_chunked(202)
+          |> send_chunked(200)
           |> stream_loop(opts)
       end
     else
@@ -395,21 +484,22 @@ defmodule Phantom.Plug do
 
   # JSON-RPC response POST (e.g. elicitation response) → 202 Accepted per MCP spec §4
   defp dispatch(%Plug.Conn{body_params: params, method: "POST"} = conn, _opts)
-       when is_map(params) and is_map_key(params, "result") and
+       when is_map(params) and
+              (is_map_key(params, "result") or is_map_key(params, "error")) and
               not is_map_key(params, "method") do
     session = conn.private.phantom.session
 
     case Request.build(params) do
       {:ok, request} ->
         session.router.dispatch_method([request.method, request.params, request, session])
 
-      {:error, _} ->
-        :ok
-    end
+        conn
+        |> maybe_put_session_header(params, session.id)
+        |> send_resp(202, "")
 
-    conn
-    |> maybe_put_session_header(params, session.id)
-    |> send_resp(202, "")
+      {:error, _request} ->
+        protocol_error(conn, nil, Request.invalid())
+    end
   end
 
   # JSON-RPC notification POST (method, no id) → 202 Accepted per MCP spec §4
@@ -891,7 +981,7 @@ defmodule Phantom.Plug do
     )
     |> put_resp_header("access-control-allow-origin", origin || "*")
     |> put_resp_header("access-control-allow-credentials", "true")
-    |> put_resp_header("access-control-allow-methods", "GET, POST, OPTIONS")
+    |> put_resp_header("access-control-allow-methods", "GET, POST, DELETE, OPTIONS")
     |> put_resp_header(
       "access-control-allow-headers",
       allowed_headers
@@ -921,9 +1011,8 @@ defmodule Phantom.Plug do
 
   defp stream_fun(%{conn: %{state: :sent}} = state, _id, _event, _payload), do: state
 
-  defp stream_fun(%{conn: %{halted: false} = conn} = state, id, event, payload) do
-    id = if Session.stateless?(state.session), do: nil, else: id
-    conn = send_sse_event(conn, id, event, payload)
+  defp stream_fun(%{conn: %{halted: false} = conn} = state, _id, event, payload) do
+    conn = send_sse_event(conn, event, payload)
     put_in(state.conn, conn)
   end
 
@@ -993,7 +1082,10 @@ defmodule Phantom.Plug do
     end
   end
 
-  defp send_sse_event(conn, _id, "comment", _data) do
+  # SSE event ids must be unique within a session, so JSON-RPC ids (chosen by
+  # the client, and free to repeat) are not used. Streams are not resumable,
+  # so events carry no id.
+  defp send_sse_event(conn, "comment", _data) do
     case chunk(conn, ": keepalive\n\n") do
       {:ok, conn} ->
         conn
@@ -1004,9 +1096,8 @@ defmodule Phantom.Plug do
     end
   end
 
-  defp send_sse_event(conn, id, _event_type, nil) do
-    id = if id, do: ["id: #{id}\n"], else: []
-    data = id ++ ["event: message\n", "data: \"\"\n\n"]
+  defp send_sse_event(conn, _event_type, nil) do
+    data = ["event: message\n", "data: \"\"\n\n"]
 
     case chunk(conn, data) do
       {:ok, conn} ->
@@ -1018,13 +1109,12 @@ defmodule Phantom.Plug do
     end
   end
 
-  defp send_sse_event(conn, id, event_type, %{} = data) do
-    send_sse_event(conn, id, event_type, JSON.encode!(data))
+  defp send_sse_event(conn, event_type, %{} = data) do
+    send_sse_event(conn, event_type, JSON.encode!(data))
   end
 
-  defp send_sse_event(conn, id, event_type, data) when is_binary(data) do
-    id = if id, do: ["id: #{id}\n"], else: []
-    data = id ++ ["event: #{event_type}\n", "data: #{data}\n\n"]
+  defp send_sse_event(conn, event_type, data) when is_binary(data) do
+    data = ["event: #{event_type}\n", "data: #{data}\n\n"]
 
     case chunk(conn, data) do
       {:ok, conn} ->

lib/phantom/request.ex

@@ -1,13 +1,14 @@
 defmodule Phantom.Request do
   @moduledoc "Standard requests and responses for the MCP protocol"
-  defstruct [:id, :type, :method, :params, :response, :spec, meta: %{}]
+  defstruct [:id, :type, :method, :params, :response, :error, :spec, meta: %{}]
 
   @opaque t :: %__MODULE__{
             id: String.t(),
             type: String.t(),
             method: String.t(),
             params: map(),
             response: map(),
+            error: map() | nil,
             spec: Phantom.ResourceTemplate.t() | Phantom.Tool.t() | Phantom.Prompt.t(),
             meta: map()
           }
@@ -190,6 +191,15 @@ defmodule Phantom.Request do
      )}
   end
 
+  def build(%{"jsonrpc" => "2.0", "error" => error} = response)
+      when is_map(error) do
+    {:ok,
+     struct!(__MODULE__,
+       error: error,
+       id: response["id"]
+     )}
+  end
+
   def build(request) when is_map(request) do
     id = Map.get(request, "id")
     {:error, struct!(__MODULE__, id: id, response: error(id, invalid()))}

lib/phantom/router.ex

@@ -611,19 +611,13 @@ defmodule Phantom.Router do
 
       def dispatch_method(_method, _params, %{id: request_id, response: %{} = response}, session)
           when is_binary(request_id) do
-        require Logger
-
-        case Phantom.Router.await_request_meta(request_id) do
-          %{type: :elicitation, reply_ref: ref, reply_pid: pid} when is_pid(pid) ->
-            Logger.debug("Routing elicitation response #{request_id} to #{inspect(pid)}")
-
-            send(pid, {:phantom_elicitation_response, ref, response})
-            Phantom.Tracker.untrack_request(request_id)
-
-          _ ->
-            Logger.debug("No tracked handler for response #{request_id}")
-        end
+        Phantom.Router.route_client_response(request_id, response)
+        {:reply, nil, session}
+      end
 
+      def dispatch_method(_method, _params, %{id: request_id, error: %{} = error}, session)
+          when is_binary(request_id) do
+        Phantom.Router.route_client_response(request_id, {:error, error})
         {:reply, nil, session}
       end
 
@@ -1416,6 +1410,24 @@ defmodule Phantom.Router do
     end)
   end
 
+  @doc false
+  # Deliver the client's response to a server-initiated request (a result
+  # map, or `{:error, error}`) to the process waiting on it.
+  def route_client_response(request_id, response) do
+    require Logger
+
+    case await_request_meta(request_id) do
+      %{type: :elicitation, reply_ref: ref, reply_pid: pid} when is_pid(pid) ->
+        Logger.debug("Routing elicitation response #{request_id} to #{inspect(pid)}")
+
+        send(pid, {:phantom_elicitation_response, ref, response})
+        Phantom.Tracker.untrack_request(request_id)
+
+      _ ->
+        Logger.debug("No tracked handler for response #{request_id}")
+    end
+  end
+
   @doc false
   # Wait for a tracked request to become visible via Tracker replication.
   # In a distributed setup, the request may be tracked on one node but