From 81ee0a37ebfeb9dc2c3d02543a3913788dc29504 Mon Sep 17 00:00:00 2001 From: satoren Date: Mon, 5 Oct 2026 09:35:39 +0900 Subject: [PATCH] Align Streamable HTTP transport with the 2025-06-18/2025-11-25 spec For the protocol versions before 2026-07-28: - Add `validate_session: true` to answer 404 to session ids the server did not issue or that were terminated. Off by default, since apps that persist sessions answer 404 from connect/2. - Reject an unsupported MCP-Protocol-Version header with 400. - Open GET SSE streams with 200 instead of 202. - Answer 406 to a GET whose Accept header excludes text/event-stream. - Accept a client's JSON-RPC error response with 202 and fail the waiting elicitation with :error (HTTP and stdio). - Answer 400 to an invalid JSON-RPC response instead of 202. - Allow DELETE in CORS allow-methods. - Stop using JSON-RPC ids as SSE event ids. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 23 ++++ lib/phantom/plug.ex | 136 ++++++++++++++++---- lib/phantom/request.ex | 12 +- lib/phantom/router.ex | 36 ++++-- lib/phantom/session.ex | 15 ++- lib/phantom/session_meta.ex | 3 + test/phantom/plug_test.exs | 205 +++++++++++++++++++++++++++++- test/phantom/sse_headers_test.exs | 70 +++++++++- test/phantom/stdio_test.exs | 30 +++++ 9 files changed, 485 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9934f6d..e209ca0 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/lib/phantom/plug.ex b/lib/phantom/plug.ex index 2b859c9..d29e554 100644 --- a/lib/phantom/plug.ex +++ b/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,7 +484,8 @@ 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 @@ -403,13 +493,13 @@ defmodule Phantom.Plug 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} -> diff --git a/lib/phantom/request.ex b/lib/phantom/request.ex index 844d422..e390838 100644 --- a/lib/phantom/request.ex +++ b/lib/phantom/request.ex @@ -1,6 +1,6 @@ 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(), @@ -8,6 +8,7 @@ defmodule Phantom.Request do 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()))} diff --git a/lib/phantom/router.ex b/lib/phantom/router.ex index b2e711b..db606bd 100644 --- a/lib/phantom/router.ex +++ b/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 diff --git a/lib/phantom/session.ex b/lib/phantom/session.ex index 27627d4..6f52a91 100644 --- a/lib/phantom/session.ex +++ b/lib/phantom/session.ex @@ -1066,7 +1066,7 @@ defmodule Phantom.Session do {:noreply, state} {%{from: from, request_id: request_id}, state} -> - GenServer.reply(from, {:ok, response}) + GenServer.reply(from, elicitation_reply(response)) state = forget_pending_elicit(state, request_id) {:noreply, state |> set_activity() |> schedule_inactivity()} end @@ -1332,7 +1332,7 @@ defmodule Phantom.Session do Enum.reduce(requests, {[], []}, fn raw_request, {resp_acc, other_acc} -> if elicit_response?(raw_request, pending) do - {[{raw_request["id"], raw_request["result"]} | resp_acc], other_acc} + {[{raw_request["id"], elicit_response(raw_request)} | resp_acc], other_acc} else {resp_acc, [raw_request | other_acc]} end @@ -1342,8 +1342,14 @@ defmodule Phantom.Session do defp elicit_response?(%{"id" => id, "result" => result}, pending) when is_map(result), do: Map.has_key?(pending, id) + defp elicit_response?(%{"id" => id, "error" => error}, pending) when is_map(error), + do: Map.has_key?(pending, id) + defp elicit_response?(_, _), do: false + defp elicit_response(%{"result" => result}) when is_map(result), do: result + defp elicit_response(%{"error" => error}), do: {:error, error} + defp route_local_elicit_response({request_id, response}, state) do {ref, state} = pop_in(state, [Access.key(:pending_elicit_ids, %{}), request_id]) @@ -1352,6 +1358,11 @@ defmodule Phantom.Session do state end + # The client answers `elicitation/create` with a result, or with a JSON-RPC + # error when it could not handle the request. + defp elicitation_reply({:error, _error}), do: :error + defp elicitation_reply(response), do: {:ok, response} + defp forget_pending_elicit(state, request_id) do Map.update(state, :pending_elicit_ids, %{}, &Map.delete(&1, request_id)) end diff --git a/lib/phantom/session_meta.ex b/lib/phantom/session_meta.ex index 84f5c28..7a670cb 100644 --- a/lib/phantom/session_meta.ex +++ b/lib/phantom/session_meta.ex @@ -37,6 +37,9 @@ defmodule Phantom.SessionMeta do end end + @doc "Whether session metadata is kept on this node." + def available?, do: table?() + @doc "Remove a session's metadata from every node." def delete(pubsub, session_id) do if table?(), do: :ets.delete(@table, session_id) diff --git a/test/phantom/plug_test.exs b/test/phantom/plug_test.exs index 67bc542..426ef49 100644 --- a/test/phantom/plug_test.exs +++ b/test/phantom/plug_test.exs @@ -40,7 +40,11 @@ defmodule Phantom.PlugTest do assert conn.status == 204 assert get_resp_header(conn, "access-control-allow-origin") == ["http://localhost:4000"] - assert get_resp_header(conn, "access-control-allow-methods") == ["GET, POST, OPTIONS"] + + assert get_resp_header(conn, "access-control-allow-methods") == [ + "GET, POST, DELETE, OPTIONS" + ] + assert get_resp_header(conn, "access-control-allow-credentials") == ["true"] allowed_headers = @@ -97,7 +101,10 @@ defmodule Phantom.PlugTest do assert get_resp_header(conn, "access-control-allow-origin") == ["http://localhost:4000"] assert get_resp_header(conn, "access-control-allow-credentials") == ["true"] - assert get_resp_header(conn, "access-control-allow-methods") == ["GET, POST, OPTIONS"] + + assert get_resp_header(conn, "access-control-allow-methods") == [ + "GET, POST, DELETE, OPTIONS" + ] allowed_headers = String.split( @@ -412,9 +419,177 @@ defmodule Phantom.PlugTest do assert_receive {:response, nil, "closed", "finished"} end + + test "returns 400 when a response is not valid JSON-RPC" do + :post + |> conn("/mcp", %{jsonrpc: "2.0", id: "abc", result: "not a map"}) + |> put_req_header("content-type", "application/json") + |> call() + + assert_receive {:conn, conn} + assert conn.status == 400 + + error = JSON.decode!(conn.resp_body) + assert error["error"]["code"] == -32600 + end + end + + describe "MCP-Protocol-Version header" do + test "accepts a supported protocol version" do + request_ping(before_call: &put_req_header(&1, "mcp-protocol-version", "2025-06-18")) + + assert_receive {:conn, conn} + assert conn.status == 200 + assert_receive {:response, 1, "message", %{result: %{}}} + end + + test "accepts requests without the header" do + request_ping() + + assert_receive {:conn, conn} + assert conn.status == 200 + end + + test "rejects an unsupported protocol version with 400" do + request_ping(before_call: &put_req_header(&1, "mcp-protocol-version", "1999-01-01")) + + assert_receive {:conn, conn} + assert conn.status == 400 + + error = JSON.decode!(conn.resp_body) + assert error["error"]["code"] == -32600 + assert error["error"]["message"] == "Unsupported protocol version" + assert error["error"]["data"]["requested"] == "1999-01-01" + assert "2025-11-25" in error["error"]["data"]["supported"] + refute_receive {:response, 1, _, _} + end + + test "rejects an unsupported protocol version on GET" do + :get + |> conn("/mcp") + |> put_req_header("accept", "text/event-stream") + |> put_req_header("mcp-protocol-version", "1999-01-01") + |> call() + + assert_receive {:conn, conn} + assert conn.status == 400 + end + end + + describe "session ids with validate_session: true" do + defp initialize_session(session_id) do + :post + |> conn("/mcp", %{ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: %{ + protocolVersion: "2025-11-25", + capabilities: %{}, + clientInfo: %{name: "Client", version: "1.0"} + } + }) + |> put_req_header("content-type", "application/json") + |> call(%{session_id: session_id}) + + assert_receive {:response, 1, "message", %{result: _}}, 500 + assert_receive {:conn, %{status: 200}}, 500 + end + + test "a session the server issued is accepted", context do + session_id = to_string(context.test) + initialize_session(session_id) + + request_ping(session_id: session_id, validate_session: true) + + assert_receive {:conn, conn} + assert conn.status == 200 + end + + test "a POST with an unknown session id returns 404", context do + request_ping(session_id: to_string(context.test), validate_session: true) + + assert_receive {:conn, conn}, 500 + assert conn.status == 404 + + error = JSON.decode!(conn.resp_body) + assert error["error"]["message"] == "Session not found" + refute_receive {:response, 1, _, _} + end + + test "a GET with an unknown session id returns 404", context do + request_sse_stream(session_id: to_string(context.test), validate_session: true) + + assert_receive {:conn, conn}, 500 + assert conn.status == 404 + end + + test "a session terminated with DELETE returns 404", context do + session_id = to_string(context.test) + initialize_session(session_id) + + :delete + |> conn("/mcp") + |> call(session_id: session_id) + + assert_receive {:conn, %{status: status}} + assert status in [200, 204] + + request_ping(session_id: session_id, validate_session: true) + + assert_receive {:conn, conn}, 500 + assert conn.status == 404 + end + + test "an initialize request is not checked", context do + :post + |> conn("/mcp", %{ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: %{ + protocolVersion: "2025-11-25", + capabilities: %{}, + clientInfo: %{name: "Client", version: "1.0"} + } + }) + |> put_req_header("content-type", "application/json") + |> call(%{session_id: to_string(context.test), validate_session: true}) + + assert_receive {:conn, conn} + assert conn.status == 200 + end + + test "an unknown session id is accepted by default", context do + request_ping(session_id: to_string(context.test)) + + assert_receive {:conn, conn} + assert conn.status == 200 + end end describe "SSE handling" do + test "GET request without text/event-stream in Accept returns 406" do + :get + |> conn("/mcp") + |> put_req_header("accept", "application/json") + |> call() + + assert_receive {:conn, conn} + assert conn.status == 406 + error = JSON.decode!(conn.resp_body) + assert error["error"]["code"] == -32600 + end + + test "GET request accepting any media type opens a stream" do + :get + |> conn("/mcp") + |> put_req_header("accept", "*/*") + |> call() + + assert_sse_connected() + end + test "GET request returns error" do :get |> conn("/mcp") @@ -795,6 +970,32 @@ defmodule Phantom.PlugTest do assert %{result: %{isError: true}} = response end + test "an error response to an elicitation is accepted and fails the elicitation", context do + session_id = to_string(context.test) + initialize_with_elicitation(session_id) + + request_tool("elicit_tool", %{}, session_id: session_id, id: 45) + + assert_receive {:response, elicit_id, "message", %{"method" => "elicitation/create"}}, + 500 + + :post + |> conn("/mcp", %{ + jsonrpc: "2.0", + id: elicit_id, + error: %{code: -32603, message: "Client failed to show the form"} + }) + |> put_req_header("content-type", "application/json") + |> call(%{session_id: session_id}) + + assert_receive {:conn, %{status: 202}}, 500 + + assert_response(45, response) + + assert %{result: %{isError: true, content: [%{text: "Elicitation failed"}]}} = + response + end + test "elicitation without client capability returns fallback", context do session_id = to_string(context.test) diff --git a/test/phantom/sse_headers_test.exs b/test/phantom/sse_headers_test.exs index b9ddda0..8201301 100644 --- a/test/phantom/sse_headers_test.exs +++ b/test/phantom/sse_headers_test.exs @@ -65,6 +65,36 @@ defmodule Phantom.SSEHeadersTest do end end + test "GET SSE response returns 200 OK" do + {status, _headers} = + response_status_and_headers( + "GET /mcp HTTP/1.1\r\n" <> + "Host: localhost:#{@port}\r\n" <> + "Accept: application/json, text/event-stream\r\n" <> + "\r\n" + ) + + assert status == 200 + end + + test "POST SSE events do not reuse JSON-RPC ids as event ids" do + json = JSON.encode!(%{jsonrpc: "2.0", id: 1, method: "ping"}) + + body = + response_body( + "POST /mcp HTTP/1.1\r\n" <> + "Host: localhost:#{@port}\r\n" <> + "Content-Type: application/json\r\n" <> + "Accept: application/json, text/event-stream\r\n" <> + "Content-Length: #{byte_size(json)}\r\n" <> + "\r\n" <> + json + ) + + assert body =~ ~s|"result":{}| + refute body =~ ~r/^id: 1$/m + end + test "GET SSE response omits connection-specific headers" do headers = response_headers( @@ -84,18 +114,48 @@ defmodule Phantom.SSEHeadersTest do end defp response_headers(request) do + {_status, headers} = response_status_and_headers(request) + headers + end + + defp response_status_and_headers(request) do {:ok, socket} = :gen_tcp.connect(~c"localhost", @port, [:binary, active: false]) :ok = :gen_tcp.send(socket, request) response = recv_until_headers(socket, "") :gen_tcp.close(socket) [headers, _body] = String.split(response, "\r\n\r\n", parts: 2) - [_status_line | header_lines] = String.split(headers, "\r\n") + [status_line | header_lines] = String.split(headers, "\r\n") + [_version, status | _] = String.split(status_line, " ") + + headers = + Map.new(header_lines, fn line -> + [name, value] = String.split(line, ":", parts: 2) + {String.downcase(name), String.trim(value)} + end) - Map.new(header_lines, fn line -> - [name, value] = String.split(line, ":", parts: 2) - {String.downcase(name), String.trim(value)} - end) + {String.to_integer(status), headers} + end + + defp response_body(request) do + {:ok, socket} = :gen_tcp.connect(~c"localhost", @port, [:binary, active: false]) + :ok = :gen_tcp.send(socket, request) + response = recv_until_closed(socket, "") + :gen_tcp.close(socket) + + [_headers, body] = String.split(response, "\r\n\r\n", parts: 2) + body + end + + defp recv_until_closed(socket, acc) do + if String.contains?(acc, "event: closed") do + acc + else + case :gen_tcp.recv(socket, 0, 1000) do + {:ok, data} -> recv_until_closed(socket, acc <> data) + {:error, _} -> acc + end + end end defp recv_until_headers(socket, acc) do diff --git a/test/phantom/stdio_test.exs b/test/phantom/stdio_test.exs index 7aaa9f1..41c6091 100644 --- a/test/phantom/stdio_test.exs +++ b/test/phantom/stdio_test.exs @@ -683,6 +683,36 @@ defmodule Phantom.StdioTest do text = get_in(result, ["result", "content", Access.at(0), "text"]) assert %{"hello" => "async my name is Stdio Alice"} = JSON.decode!(text) end + + test "an error response to an elicitation fails the elicitation" do + ctx = start_stdio() + initialize_with_elicitation(ctx) + + send_request(ctx, %{ + jsonrpc: "2.0", + id: 43, + method: "tools/call", + params: %{name: "async_elicit_tool", arguments: %{}} + }) + + elicit = + find_in_output(ctx, fn r -> r["method"] == "elicitation/create" end, 2_000) + + assert elicit, "expected elicitation/create on output" + + send_request(ctx, %{ + jsonrpc: "2.0", + id: elicit["id"], + error: %{code: -32603, message: "Client failed to show the form"} + }) + + result = + find_in_output(ctx, fn r -> r["id"] == 43 and is_map(r["result"]) end, 2_000) + + assert result, "expected tool result for id=43" + assert result["result"]["isError"] == true + assert get_in(result, ["result", "content", Access.at(0), "text"]) == "Elicitation failed" + end end describe "duplicate request dedup" do