From 0952c316685d8912eb56c1c603ac336f17474976 Mon Sep 17 00:00:00 2001 From: Zoey Pessanha Date: Tue, 7 Oct 2025 14:28:15 -0300 Subject: [PATCH 1/3] fix: correctly handle timeouts and keepalive --- .gitignore | 5 + .../server/transport/streamable_http.ex | 161 +++++++++++------- .../server/transport/streamable_http/plug.ex | 136 ++++++--------- .../streamable_http/request_params.ex | 34 ++++ lib/anubis/sse/streaming.ex | 7 + priv/dev/upcase/.formatter.exs | 2 +- priv/dev/upcase/lib/upcase/router.ex | 4 +- priv/dev/upcase/lib/upcase/server.ex | 28 ++- priv/dev/upcase/mix.lock | 2 +- .../server/transport/streamable_http_test.exs | 12 +- 10 files changed, 237 insertions(+), 154 deletions(-) create mode 100644 lib/anubis/server/transport/streamable_http/request_params.ex diff --git a/.gitignore b/.gitignore index 3f59f569..14b0a9cd 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,8 @@ result # Claude /.claude/ /rag/ + +# LSP +/.lexical/ +/.expert/ +/.elixir-tools/ diff --git a/lib/anubis/server/transport/streamable_http.ex b/lib/anubis/server/transport/streamable_http.ex index bf65b24f..ba7c3007 100644 --- a/lib/anubis/server/transport/streamable_http.ex +++ b/lib/anubis/server/transport/streamable_http.ex @@ -51,7 +51,10 @@ defmodule Anubis.Server.Transport.StreamableHTTP do import Peri + alias Anubis.MCP.Error + alias Anubis.MCP.ID alias Anubis.MCP.Message + alias Anubis.Server.Transport.StreamableHTTP.RequestParams alias Anubis.Telemetry alias Anubis.Transport.Behaviour, as: Transport @@ -64,21 +67,17 @@ defmodule Anubis.Server.Transport.StreamableHTTP do - `:server` - The server process (required) - `:name` - Name for registering the GenServer (required) - - `:call_timeout` - Timeout for internal GenServer calls in milliseconds (default: 10 minutes) """ @type option :: {:server, GenServer.server()} | {:name, GenServer.name()} - | {:call_timeout, pos_integer()} | GenServer.option() defschema(:parse_options, [ {:server, {:required, Anubis.get_schema(:process_name)}}, {:name, {:required, {:custom, &Anubis.genserver_name/1}}}, {:registry, {:atom, {:default, Anubis.Server.Registry}}}, - {:request_timeout, {:integer, {:default, to_timeout(second: 30)}}}, - {:task_supervisor, {:required, {:custom, &Anubis.genserver_name/1}}}, - {:call_timeout, {:integer, {:default, to_timeout(second: 30)}}} + {:task_supervisor, {:required, {:custom, &Anubis.genserver_name/1}}} ]) @doc """ @@ -108,10 +107,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do * `{:error, reason}` otherwise """ @impl Transport - @spec send_message(GenServer.server(), binary(), keyword()) :: :ok | {:error, term()} - def send_message(transport, message, opts \\ []) when is_binary(message) do - timeout = Keyword.get(opts, :call_timeout, 5000) - GenServer.call(transport, {:send_message, message}, timeout) + @spec send_message(GenServer.server(), binary()) :: :ok | {:error, term()} + def send_message(transport, message) when is_binary(message) do + GenServer.call(transport, {:send_message, message}, 5000) end @doc """ @@ -137,11 +135,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do Called by the Plug when establishing an SSE connection. The calling process becomes the SSE handler for the session. """ - @spec register_sse_handler(GenServer.server(), String.t(), keyword()) :: - :ok | {:error, term()} - def register_sse_handler(transport, session_id, opts \\ []) do - timeout = Keyword.get(opts, :call_timeout, 5000) - GenServer.call(transport, {:register_sse_handler, session_id, self()}, timeout) + @spec register_sse_handler(GenServer.server(), String.t()) :: :ok | {:error, term()} + def register_sse_handler(transport, session_id) do + GenServer.call(transport, {:register_sse_handler, session_id, self()}, 5000) end @doc """ @@ -159,11 +155,10 @@ defmodule Anubis.Server.Transport.StreamableHTTP do Called by the Plug when a message is received via HTTP POST. """ - @spec handle_message(GenServer.server(), String.t(), map() | list(map), map(), keyword()) :: - {:ok, binary() | nil} | {:error, term()} - def handle_message(transport, session_id, message, context, opts \\ []) do - timeout = Keyword.get(opts, :call_timeout, 5000) - GenServer.call(transport, {:handle_message, session_id, message, context}, timeout) + @spec handle_message(RequestParams.t()) :: {:ok, binary() | nil} | {:error, term()} + def handle_message(%RequestParams{transport: transport} = params) do + timeout = params.timeout + 1_000 + GenServer.call(transport, {:handle_message, params}, timeout) end @doc """ @@ -172,16 +167,11 @@ defmodule Anubis.Server.Transport.StreamableHTTP do This allows the Plug to know whether to stream the response via SSE or return it as a regular HTTP response. """ - @spec handle_message_for_sse(GenServer.server(), String.t(), map(), map(), keyword()) :: + @spec handle_message_for_sse(RequestParams.t()) :: {:ok, binary()} | {:sse, binary()} | {:error, term()} - def handle_message_for_sse(transport, session_id, message, context, opts \\ []) do - timeout = Keyword.get(opts, :call_timeout, 5000) - - GenServer.call( - transport, - {:handle_message_for_sse, session_id, message, context}, - timeout - ) + def handle_message_for_sse(%RequestParams{transport: transport} = params) do + timeout = params.timeout + 1_000 + GenServer.call(transport, {:handle_message_for_sse, params}, timeout) end @doc """ @@ -190,10 +180,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do Returns the pid of the process handling SSE for this session, or nil if no SSE connection exists. """ - @spec get_sse_handler(GenServer.server(), String.t(), keyword()) :: pid() | nil - def get_sse_handler(transport, session_id, opts \\ []) do - timeout = Keyword.get(opts, :call_timeout, 5000) - GenServer.call(transport, {:get_sse_handler, session_id}, timeout) + @spec get_sse_handler(GenServer.server(), String.t()) :: pid() | nil + def get_sse_handler(transport, session_id) do + GenServer.call(transport, {:get_sse_handler, session_id}) end @doc """ @@ -216,14 +205,19 @@ defmodule Anubis.Server.Transport.StreamableHTTP do state = %{ server: server, registry: opts.registry, - request_timeout: opts.request_timeout, - call_timeout: opts.call_timeout, task_supervisor: opts.task_supervisor, # Map of session_id => {pid, monitor_ref} sse_handlers: %{}, - active_tasks: %{} + active_tasks: %{}, + # keepalive + keepalive_interval: opts[:keepalive_interval] || 5_000, + keepalive_enabled: opts[:keepalive] || true } + if should_keepalive?(state) do + schedule_keepalive(state.keepalive_interval) + end + Logger.metadata(mcp_transport: :streamable_http, mcp_server: server) Logging.transport_event("starting", %{ @@ -255,12 +249,12 @@ defmodule Anubis.Server.Transport.StreamableHTTP do end @impl GenServer - def handle_call({:handle_message, session_id, message, context}, from, state) when is_map(message) do + def handle_call({:handle_message, %{message: message} = params}, from, state) when is_map(message) do + %{session_id: session_id, context: context, timeout: timeout} = params server = state.registry.whereis_server(state.server) - timeout = state.request_timeout cond do - Message.is_notification(message) -> + Message.is_notification(params.message) -> GenServer.cast(server, {:notification, message, session_id, context}) {:reply, {:ok, nil}, state} @@ -270,14 +264,17 @@ defmodule Anubis.Server.Transport.StreamableHTTP do true -> task = - Task.Supervisor.async(state.task_supervisor, fn -> - forward_request_to_server(server, message, session_id, context, timeout) + Task.Supervisor.async_nolink(state.task_supervisor, fn -> + forward_request_to_server(server, params) end) + task_timeout_ref = Process.send_after(self(), {:task_timeout, task.ref}, timeout) + task_info = %{ type: :handle_message, session_id: session_id, - from: from + from: from, + task_timeout: task_timeout_ref } {:noreply, put_in(state.active_tasks[task.ref], task_info)} @@ -285,9 +282,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do end @impl GenServer - def handle_call({:handle_message_for_sse, session_id, message, context}, from, state) when is_map(message) do + def handle_call({:handle_message_for_sse, %{message: message} = params}, from, state) when is_map(message) do + %{session_id: session_id, context: context, timeout: timeout} = params server = state.registry.whereis_server(state.server) - timeout = state.request_timeout if Message.is_notification(message) do GenServer.cast(server, {:notification, message, session_id, context}) @@ -296,22 +293,18 @@ defmodule Anubis.Server.Transport.StreamableHTTP do sse_handler? = Map.has_key?(state.sse_handlers, session_id) task = - Task.Supervisor.async(state.task_supervisor, fn -> - forward_request_to_server( - server, - message, - session_id, - context, - timeout, - sse_handler? - ) + Task.Supervisor.async_nolink(state.task_supervisor, fn -> + forward_request_to_server(server, params, sse_handler?) end) + task_timeout_ref = Process.send_after(self(), {:task_timeout, task.ref}, timeout) + task_info = %{ type: :handle_message_for_sse, session_id: session_id, from: from, - has_sse_handler: sse_handler? + has_sse_handler: sse_handler?, + task_timeout: task_timeout_ref } {:noreply, put_in(state.active_tasks[task.ref], task_info)} @@ -352,10 +345,10 @@ defmodule Anubis.Server.Transport.StreamableHTTP do {:reply, :ok, state} end - defp forward_request_to_server(server, message, session_id, context, timeout, has_sse_handler \\ false) do - msg = {:request, message, session_id, context} + defp forward_request_to_server(server, params, has_sse_handler \\ false) do + msg = {:request, params.message, params.session_id, params.context} - case GenServer.call(server, msg, timeout) do + case GenServer.call(server, msg, params.timeout) do {:ok, response} when has_sse_handler -> {:sse, response} @@ -365,7 +358,7 @@ defmodule Anubis.Server.Transport.StreamableHTTP do {:error, reason} -> Logging.transport_event( "server_error", - %{reason: reason, session_id: session_id}, + %{reason: reason, session_id: params.session_id}, level: :error ) @@ -411,9 +404,40 @@ defmodule Anubis.Server.Transport.StreamableHTTP do # Handle successful task completion @impl GenServer + def handle_info({:task_timeout, ref}, %{active_tasks: active_tasks} = state) when is_map_key(active_tasks, ref) do + {task_info, active_tasks} = Map.pop(active_tasks, ref) + + timeout_error = + Error.protocol(:internal_error, %{ + message: "Request timeout - tool execution exceeded limit", + session_id: task_info.session_id + }) + + {:ok, error_json} = Error.to_json_rpc(timeout_error, ID.generate_error_id()) + + GenServer.reply(task_info.from, {:error, error_json}) + + Logging.transport_event( + "task_timeout", + %{ + session_id: task_info.session_id + }, + level: :warning + ) + + {:noreply, %{state | active_tasks: active_tasks}} + end + + def handle_info({:task_timeout, _ref}, state), do: {:noreply, state} + def handle_info({ref, result}, %{active_tasks: active_tasks} = state) when is_reference(ref) and is_map_key(active_tasks, ref) do {task_info, active_tasks} = Map.pop(active_tasks, ref) + + if Map.has_key?(task_info, :timeout_ref) do + Process.cancel_timer(task_info.task_timeout) + end + GenServer.reply(task_info.from, result) Process.demonitor(ref, [:flush]) @@ -455,7 +479,20 @@ defmodule Anubis.Server.Transport.StreamableHTTP do {:noreply, %{state | sse_handlers: sse_handlers}} end - def handle_info(_msg, state) do + def handle_info(:send_keepalive, state) do + for {_session_id, {pid, _ref}} <- state.sse_handlers do + send(pid, :sse_keepalive) + end + + if should_keepalive?(state) do + schedule_keepalive(state.keepalive_interval) + end + + {:noreply, state} + end + + def handle_info(msg, state) do + Logging.transport_event("unknown handle_info message", msg, level: :warning) {:noreply, state} end @@ -469,4 +506,12 @@ defmodule Anubis.Server.Transport.StreamableHTTP do :ok end + + defp schedule_keepalive(interval) do + Process.send_after(self(), :send_keepalive, interval) + end + + defp should_keepalive?(state) do + state.keepalive_enabled and not Enum.empty?(state.sse_handlers) + end end diff --git a/lib/anubis/server/transport/streamable_http/plug.ex b/lib/anubis/server/transport/streamable_http/plug.ex index d5802d69..3387249a 100644 --- a/lib/anubis/server/transport/streamable_http/plug.ex +++ b/lib/anubis/server/transport/streamable_http/plug.ex @@ -34,7 +34,7 @@ if Code.ensure_loaded?(Plug) do - `:server` - The server process name (required) - `:session_header` - Custom header name for session ID (default: "mcp-session-id") - - `:timeout` - Request timeout in milliseconds (default: 30000) + - `:request_timeout` - Request timeout in milliseconds (default: 30000) - `:registry` - The registry to use. See `Anubis.Server.Registry.Adapter` for more information (default: Elixir's Registry implementation) ## Security Features @@ -64,6 +64,7 @@ if Code.ensure_loaded?(Plug) do alias Anubis.MCP.ID alias Anubis.MCP.Message alias Anubis.Server.Transport.StreamableHTTP + alias Anubis.Server.Transport.StreamableHTTP.RequestParams alias Anubis.SSE.Streaming alias Plug.Conn.Unfetched @@ -80,10 +81,9 @@ if Code.ensure_loaded?(Plug) do registry = Keyword.get(opts, :registry, Anubis.Server.Registry) transport = registry.transport(server, :streamable_http) session_header = Keyword.get(opts, :session_header, @default_session_header) - timeout = Keyword.get(opts, :timeout, @default_timeout) - call_timeout = Keyword.get(opts, :call_timeout, @default_timeout) + request_timeout = Keyword.get(opts, :request_timeout, @default_timeout) - %{transport: transport, session_header: session_header, timeout: timeout, call_timeout: call_timeout} + %{transport: transport, session_header: session_header, timeout: request_timeout} end @impl Plug @@ -98,13 +98,13 @@ if Code.ensure_loaded?(Plug) do # GET request handler - establishes SSE connection - defp handle_get(conn, %{transport: transport, session_header: session_header, call_timeout: call_timeout}) do + defp handle_get(conn, %{transport: transport, session_header: session_header} = opts) do if wants_sse?(conn) do session_id = get_or_create_session_id(conn, session_header) - case StreamableHTTP.register_sse_handler(transport, session_id, call_timeout: call_timeout) do + case StreamableHTTP.register_sse_handler(transport, session_id) do :ok -> - start_sse_streaming(conn, transport, session_id, session_header) + start_sse_streaming(conn, Map.put(opts, :session_id, session_id)) {:error, reason} -> Logging.transport_event("sse_registration_failed", %{reason: reason}, level: :error) @@ -130,7 +130,17 @@ if Code.ensure_loaded?(Plug) do session_id: session_id }) - process_message(message, conn, transport, session_id, context, session_header, opts.call_timeout) + process_message( + conn, + RequestParams.new( + message: message, + transport: transport, + session_id: session_id, + context: context, + session_header: session_header, + timeout: opts.timeout + ) + ) else {:error, :invalid_accept_header} -> send_error( @@ -157,22 +167,13 @@ if Code.ensure_loaded?(Plug) do end end - defp process_message(message, conn, transport, session_id, context, session_header, call_timeout) - when is_map(message) do + defp process_message(conn, %{message: message} = params) when is_map(message) do if Message.is_request(message) do - handle_request_with_possible_sse( - conn, - transport, - session_id, - message, - context, - session_header, - call_timeout - ) + handle_request_with_possible_sse(conn, params) else # Notification - transport - |> StreamableHTTP.handle_message(session_id, message, context, call_timeout: call_timeout) + params + |> StreamableHTTP.handle_message() |> format_notification_response(conn) end end @@ -215,91 +216,54 @@ if Code.ensure_loaded?(Plug) do # Handle requests that might need SSE streaming - defp handle_request_with_possible_sse(conn, transport, session_id, body, context, session_header, call_timeout) do + defp handle_request_with_possible_sse(conn, params) do if wants_sse?(conn) do - handle_sse_request( - conn, - transport, - session_id, - body, - context, - session_header, - call_timeout - ) + handle_sse_request(conn, params) else - handle_json_request( - conn, - transport, - session_id, - body, - context, - session_header, - call_timeout - ) + handle_json_request(conn, params) end end - defp handle_sse_request(conn, transport, session_id, body, context, session_header, call_timeout) do - case StreamableHTTP.handle_message_for_sse( - transport, - session_id, - body, - context, - call_timeout: call_timeout - ) do + defp handle_sse_request(conn, params) do + case StreamableHTTP.handle_message_for_sse(params) do {:sse, response} -> - route_sse_response( - conn, - transport, - session_id, - response, - body, - context, - session_header, - call_timeout - ) + route_sse_response(conn, response, params) {:ok, response} -> conn |> put_resp_content_type("application/json") - |> maybe_add_session_header(session_header, session_id) + |> maybe_add_session_header(params.session_header, params.session_id) |> send_resp(200, response) {:error, error} -> - handle_request_error(conn, error, body) + handle_request_error(conn, error, params.message) end end - defp handle_json_request(conn, transport, session_id, body, context, session_header, call_timeout) do - case StreamableHTTP.handle_message(transport, session_id, body, context, call_timeout: call_timeout) do + defp handle_json_request(conn, params) do + case StreamableHTTP.handle_message(params) do {:ok, response} -> conn |> put_resp_content_type("application/json") - |> maybe_add_session_header(session_header, session_id) + |> maybe_add_session_header(params.session_header, params.session_id) |> send_resp(200, response) {:error, error} -> - handle_request_error(conn, error, body) + handle_request_error(conn, error, params.message) end end - defp route_sse_response(conn, transport, session_id, response, body, context, session_header, call_timeout) do - if handler_pid = StreamableHTTP.get_sse_handler(transport, session_id, call_timeout: call_timeout) do + defp route_sse_response(conn, response, params) do + %{transport: transport, session_id: session_id} = params + + if handler_pid = StreamableHTTP.get_sse_handler(transport, session_id) do send(handler_pid, {:sse_message, response}) conn |> put_resp_content_type("application/json") |> send_resp(202, "{}") else - establish_sse_for_request( - conn, - transport, - session_id, - body, - context, - session_header, - call_timeout - ) + establish_sse_for_request(conn, params) end end @@ -317,11 +281,13 @@ if Code.ensure_loaded?(Plug) do ) end - defp establish_sse_for_request(conn, transport, session_id, body, context, session_header, call_timeout) do - case StreamableHTTP.register_sse_handler(transport, session_id, call_timeout: call_timeout) do + defp establish_sse_for_request(conn, params) do + %{transport: transport, session_id: session_id} = params + + case StreamableHTTP.register_sse_handler(transport, session_id) do :ok -> - start_background_request(transport, session_id, body, context, call_timeout) - start_sse_streaming(conn, transport, session_id, session_header) + start_background_request(params) + start_sse_streaming(conn, params) {:error, reason} -> Logging.transport_event("sse_registration_failed", %{reason: reason}, level: :error) @@ -329,16 +295,16 @@ if Code.ensure_loaded?(Plug) do send_jsonrpc_error( conn, Error.protocol(:internal_error, %{reason: reason}), - extract_request_id(body) + extract_request_id(params.message) ) end end - defp start_background_request(transport, session_id, body, context, call_timeout) do + defp start_background_request(params) do self_pid = self() Task.start(fn -> - case StreamableHTTP.handle_message(transport, session_id, body, context, call_timeout: call_timeout) do + case StreamableHTTP.handle_message(params) do {:ok, response} when is_binary(response) -> send(self_pid, {:sse_message, response}) @@ -352,9 +318,11 @@ if Code.ensure_loaded?(Plug) do end) end - defp start_sse_streaming(conn, transport, session_id, session_header) do + defp start_sse_streaming(conn, params) do + %{transport: transport, session_id: session_id} = params + conn - |> put_resp_header(session_header, session_id) + |> put_resp_header(params.session_header, session_id) |> Streaming.prepare_connection() |> Streaming.start(transport, session_id, on_close: fn -> diff --git a/lib/anubis/server/transport/streamable_http/request_params.ex b/lib/anubis/server/transport/streamable_http/request_params.ex new file mode 100644 index 00000000..d2adc9c8 --- /dev/null +++ b/lib/anubis/server/transport/streamable_http/request_params.ex @@ -0,0 +1,34 @@ +defmodule Anubis.Server.Transport.StreamableHTTP.RequestParams do + @moduledoc false + + @type t :: %__MODULE__{ + transport: GenServer.server(), + session_id: String.t() | nil, + session_header: String.t(), + timeout: pos_integer(), + context: map() | nil, + message: map() | binary() | nil + } + + @enforce_keys [:transport, :session_header, :timeout] + defstruct [ + :transport, + :session_id, + :session_header, + :timeout, + :context, + :message + ] + + @spec new(keyword()) :: t() + def new(opts) do + %__MODULE__{ + transport: Keyword.fetch!(opts, :transport), + session_id: Keyword.get(opts, :session_id), + session_header: Keyword.fetch!(opts, :session_header), + timeout: Keyword.fetch!(opts, :timeout), + context: Keyword.get(opts, :context), + message: Keyword.get(opts, :message) + } + end +end diff --git a/lib/anubis/sse/streaming.ex b/lib/anubis/sse/streaming.ex index b0afb6df..8a29e0e8 100644 --- a/lib/anubis/sse/streaming.ex +++ b/lib/anubis/sse/streaming.ex @@ -78,6 +78,9 @@ if Code.ensure_loaded?(Plug) do defp loop(conn, transport, session_id, event_counter) do receive do + :sse_keepalive -> + loop(keep_alive(conn), transport, session_id, event_counter + 1) + {:sse_message, message} when is_binary(message) -> case send_event(conn, message, event_counter) do {:ok, conn} -> @@ -117,5 +120,9 @@ if Code.ensure_loaded?(Plug) do loop(conn, transport, session_id, event_counter) end end + + defp keep_alive(conn) do + Plug.Conn.chunk(conn, ": keepalive\n\n") + end end end diff --git a/priv/dev/upcase/.formatter.exs b/priv/dev/upcase/.formatter.exs index bef567b5..15237520 100644 --- a/priv/dev/upcase/.formatter.exs +++ b/priv/dev/upcase/.formatter.exs @@ -1,5 +1,5 @@ # Used by "mix format" [ inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"], - import_deps: [:plug, :hermes_mcp] + import_deps: [:plug, :anubis_mcp] ] diff --git a/priv/dev/upcase/lib/upcase/router.ex b/priv/dev/upcase/lib/upcase/router.ex index 97186d10..0d6293fe 100644 --- a/priv/dev/upcase/lib/upcase/router.ex +++ b/priv/dev/upcase/lib/upcase/router.ex @@ -7,7 +7,9 @@ defmodule Upcase.Router do plug :match plug :dispatch - forward "/mcp", to: StreamableHTTP.Plug, init_opts: [server: Upcase.Server] + forward "/mcp", + to: StreamableHTTP.Plug, + init_opts: [server: Upcase.Server, request_timeout: 10_000] match _ do send_resp(conn, 404, "not found") diff --git a/priv/dev/upcase/lib/upcase/server.ex b/priv/dev/upcase/lib/upcase/server.ex index 0645ba0c..15f784c6 100644 --- a/priv/dev/upcase/lib/upcase/server.ex +++ b/priv/dev/upcase/lib/upcase/server.ex @@ -3,7 +3,9 @@ defmodule Upcase.Server do A simple MCP server that upcases input text. """ - use Anubis.Server + use Anubis.Server, capabilities: [:tools, :resources, :prompts] + + alias Anubis.Server.Response require Logger @@ -12,11 +14,6 @@ defmodule Upcase.Server do %{"name" => "Upcase MCP Server", "version" => "1.0.0"} end - @impl true - def server_capabilities do - %{"tools" => %{}} - end - @impl true def supported_protocol_versions do ["2025-03-26", "2024-10-07", "2024-05-11"] @@ -30,8 +27,15 @@ defmodule Upcase.Server do @impl true def init(client_info, frame) do Logger.info("We had the client_info: #{inspect(client_info)}") - schedule_hello() - {:ok, assign(frame, counter: 0) |> put_pagination_limit(1)} + # schedule_hello() + + {:ok, + assign(frame, counter: 0) + |> put_pagination_limit(10) + |> register_tool("timeout", + description: "tests the server timeout", + input_schema: %{interval: {:required, :integer}} + )} end @impl true @@ -42,6 +46,14 @@ defmodule Upcase.Server do {:noreply, frame} end + @impl true + def handle_tool_call("timeout", %{interval: interval}, frame) do + IO.puts("sleeping...") + Process.sleep(interval) + IO.puts("slept!") + {:reply, Response.text(Response.tool(), "slept for #{interval}"), frame} + end + defp schedule_hello do Process.send_after(self(), :hello, 1_650) end diff --git a/priv/dev/upcase/mix.lock b/priv/dev/upcase/mix.lock index 65f989ac..0b40ec03 100644 --- a/priv/dev/upcase/mix.lock +++ b/priv/dev/upcase/mix.lock @@ -6,7 +6,7 @@ "mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"}, "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, - "peri": {:hex, :peri, "0.6.1", "6a90ca728a27aef8fef37ce307444255d20364b0c8f8d39e52499d8d825cb514", [:mix], [{:ecto, "~> 3.12", [hex: :ecto, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:stream_data, "~> 1.1", [hex: :stream_data, repo: "hexpm", optional: true]}], "hexpm", "e20ffc659967baf9c4f28799fe7302b656d6662a8b3db7646fdafd017e192743"}, + "peri": {:hex, :peri, "0.6.0", "0758aa037f862f7a3aa0823cb82195916f61a8071f6eaabcff02103558e61a70", [:mix], [{:ecto, "~> 3.12", [hex: :ecto, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:stream_data, "~> 1.1", [hex: :stream_data, repo: "hexpm", optional: true]}], "hexpm", "b27f118f3317fbc357c4a04b3f3c98561efdd8865edd4ec0e24fd936c7ff36c8"}, "plug": {:hex, :plug, "1.18.1", "5067f26f7745b7e31bc3368bc1a2b818b9779faa959b49c934c17730efc911cf", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "57a57db70df2b422b564437d2d33cf8d33cd16339c1edb190cd11b1a3a546cc2"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"}, diff --git a/test/anubis/server/transport/streamable_http_test.exs b/test/anubis/server/transport/streamable_http_test.exs index a1d07d5b..16601a95 100644 --- a/test/anubis/server/transport/streamable_http_test.exs +++ b/test/anubis/server/transport/streamable_http_test.exs @@ -4,6 +4,7 @@ defmodule Anubis.Server.Transport.StreamableHTTPTest do import ExUnit.CaptureLog alias Anubis.Server.Transport.StreamableHTTP + alias Anubis.Server.Transport.StreamableHTTP.RequestParams setup :with_default_registry @@ -60,7 +61,16 @@ defmodule Anubis.Server.Transport.StreamableHTTPTest do assert :ok = StreamableHTTP.register_sse_handler(transport, session_id) message = build_request("ping", %{}) - StreamableHTTP.handle_message_for_sse(transport, session_id, message, %{}) + params = %RequestParams{ + transport: transport, + session_id: session_id, + message: message, + context: %{}, + session_header: nil, + timeout: 5 + } + + StreamableHTTP.handle_message_for_sse(params) # Clean up to avoid logs after test ends capture_log(fn -> From 5060fab0f822da47a7b38fa06f4d668e5884b08e Mon Sep 17 00:00:00 2001 From: Zoey Pessanha Date: Tue, 7 Oct 2025 14:37:06 -0300 Subject: [PATCH 2/3] chore: correctly handle keepalive errors --- lib/anubis/sse/streaming.ex | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/lib/anubis/sse/streaming.ex b/lib/anubis/sse/streaming.ex index 8a29e0e8..1467365a 100644 --- a/lib/anubis/sse/streaming.ex +++ b/lib/anubis/sse/streaming.ex @@ -79,7 +79,15 @@ if Code.ensure_loaded?(Plug) do defp loop(conn, transport, session_id, event_counter) do receive do :sse_keepalive -> - loop(keep_alive(conn), transport, session_id, event_counter + 1) + case keep_alive(conn) do + {:ok, conn} -> + loop(conn, transport, session_id, event_counter + 1) + + {:error, reason} -> + Logging.transport_event("sse_keepalive_failed", %{session_id: session_id, reason: reason}, level: :error) + + conn + end {:sse_message, message} when is_binary(message) -> case send_event(conn, message, event_counter) do From 40f243fce78a3af7d410719402fc1113e9227595 Mon Sep 17 00:00:00 2001 From: Zoey Pessanha Date: Tue, 7 Oct 2025 16:43:15 -0300 Subject: [PATCH 3/3] chore: pr suggestions --- lib/anubis/server/transport/streamable_http.ex | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/lib/anubis/server/transport/streamable_http.ex b/lib/anubis/server/transport/streamable_http.ex index ba7c3007..73078815 100644 --- a/lib/anubis/server/transport/streamable_http.ex +++ b/lib/anubis/server/transport/streamable_http.ex @@ -77,7 +77,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do {:server, {:required, Anubis.get_schema(:process_name)}}, {:name, {:required, {:custom, &Anubis.genserver_name/1}}}, {:registry, {:atom, {:default, Anubis.Server.Registry}}}, - {:task_supervisor, {:required, {:custom, &Anubis.genserver_name/1}}} + {:task_supervisor, {:required, {:custom, &Anubis.genserver_name/1}}}, + {:keepalive, {:boolean, {:default, true}}}, + {:keepalive_interval, {:integer, {:default, 5_000}}} ]) @doc """ @@ -210,8 +212,8 @@ defmodule Anubis.Server.Transport.StreamableHTTP do sse_handlers: %{}, active_tasks: %{}, # keepalive - keepalive_interval: opts[:keepalive_interval] || 5_000, - keepalive_enabled: opts[:keepalive] || true + keepalive_interval: opts.keepalive_interval, + keepalive_enabled: opts.keepalive } if should_keepalive?(state) do @@ -274,7 +276,8 @@ defmodule Anubis.Server.Transport.StreamableHTTP do type: :handle_message, session_id: session_id, from: from, - task_timeout: task_timeout_ref + task_timeout: task_timeout_ref, + task: task } {:noreply, put_in(state.active_tasks[task.ref], task_info)} @@ -304,7 +307,8 @@ defmodule Anubis.Server.Transport.StreamableHTTP do session_id: session_id, from: from, has_sse_handler: sse_handler?, - task_timeout: task_timeout_ref + task_timeout: task_timeout_ref, + task: task } {:noreply, put_in(state.active_tasks[task.ref], task_info)} @@ -416,6 +420,7 @@ defmodule Anubis.Server.Transport.StreamableHTTP do {:ok, error_json} = Error.to_json_rpc(timeout_error, ID.generate_error_id()) GenServer.reply(task_info.from, {:error, error_json}) + if task = task_info.task, do: Task.shutdown(task, :brutal_kill) Logging.transport_event( "task_timeout",