feat: propagating headers on requests (#214)

## Problem

Currently we are not propagating / accepting headers for StreamableHttp
transport

## Solution

Implementation of header optional param

## Rationale

Will be handled only for StreamableHTTP for now.

Author: yoschihirokaimoto

lib/hermes/client/base.ex

@@ -15,6 +15,7 @@ defmodule Hermes.Client.Base do
   alias Hermes.MCP.Response
   alias Hermes.Protocol
   alias Hermes.Telemetry
+  alias Hermes.Transport.StreamableHTTP
 
   require Message
 
@@ -81,7 +82,7 @@ defmodule Hermes.Client.Base do
              Hermes.Transport.STDIO
              | Hermes.Transport.SSE
              | Hermes.Transport.WebSocket
-             | Hermes.Transport.StreamableHTTP}
+             | StreamableHTTP}
             | {:name, GenServer.server()}
           )
 
@@ -325,6 +326,7 @@ defmodule Hermes.Client.Base do
     * `:progress` - Progress tracking options
       * `:token` - A unique token to track progress (string or integer)
       * `:callback` - A function to call when progress updates are received
+    * `:headers` - HTTP headers to send with this request (map, only for StreamableHTTP transport)
   """
   @spec call_tool(t, String.t(), map() | nil, keyword) ::
           {:ok, Response.t()} | {:error, Error.t()}
@@ -337,7 +339,8 @@ defmodule Hermes.Client.Base do
         method: "tools/call",
         params: params,
         progress_opts: Keyword.get(opts, :progress),
-        timeout: Keyword.get(opts, :timeout)
+        timeout: Keyword.get(opts, :timeout),
+        headers: Keyword.get(opts, :headers)
       })
 
     buffer_timeout = operation.timeout + to_timeout(second: 1)
@@ -429,7 +432,7 @@ defmodule Hermes.Client.Base do
       ref = %{"type" => "ref/prompt", "name" => "code_review"}
       argument = %{"name" => "language", "value" => "py"}
       {:ok, response} = Hermes.Client.complete(client, ref, argument)
-      
+
       # Access the completion values
       values = get_in(Response.unwrap(response), ["completion", "values"])
   """
@@ -706,14 +709,14 @@ defmodule Hermes.Client.Base do
 
       MyClient.register_sampling_callback(fn params ->
         messages = params["messages"]
-        
+
         # Show UI for user approval
         case MyUI.approve_sampling(messages) do
           {:approved, edited_messages} ->
             # Call LLM with approved/edited messages
             response = MyLLM.generate(edited_messages, params["modelPreferences"])
             {:ok, response}
-            
+
           :rejected ->
             {:error, "User rejected sampling request"}
         end
@@ -799,7 +802,7 @@ defmodule Hermes.Client.Base do
          {request_id, updated_state} =
            State.add_request_from_operation(state, operation, from),
          {:ok, request_data} <- encode_request(method, params_with_token, request_id),
-         :ok <- send_to_transport(state.transport, request_data) do
+         :ok <- send_to_transport(state.transport, request_data, operation.headers) do
       Telemetry.execute(
         Telemetry.event_client_request(),
         %{system_time: System.system_time()},
@@ -1466,6 +1469,18 @@ defmodule Hermes.Client.Base do
     send_notification(state, "notifications/cancelled", params)
   end
 
+  defp send_to_transport(transport, data, headers) when not is_nil(headers) do
+    opts = [headers: headers]
+
+    with {:error, reason} <- transport.layer.send_message(transport.name, data, opts) do
+      {:error, Error.transport(:send_failure, %{original_reason: reason})}
+    end
+  end
+
+  defp send_to_transport(transport, data, _headers) do
+    send_to_transport(transport, data)
+  end
+
   defp send_to_transport(transport, data) do
     with {:error, reason} <- transport.layer.send_message(transport.name, data) do
       {:error, Error.transport(:send_failure, %{original_reason: reason})}

lib/hermes/client/operation.ex

@@ -20,14 +20,16 @@ defmodule Hermes.Client.Operation do
           method: String.t(),
           params: map(),
           progress_opts: progress_options() | nil,
-          timeout: pos_integer()
+          timeout: pos_integer(),
+          headers: map() | nil
         }
 
   defstruct [
     :method,
     params: %{},
     progress_opts: [],
-    timeout: @default_timeout
+    timeout: @default_timeout,
+    headers: nil
   ]
 
   @doc """
@@ -40,19 +42,22 @@ defmodule Hermes.Client.Operation do
       * `:params` - The parameters to send to the server (required)
       * `:progress_opts` - Progress tracking options (optional)
       * `:timeout` - The timeout for this operation in milliseconds (optional, defaults to 30s)
+      * `:headers` - HTTP headers to send with this request (optional, only for StreamableHTTP transport)
   """
   @spec new(%{
           required(:method) => String.t(),
           optional(:params) => map(),
           optional(:progress_opts) => progress_options() | nil,
-          optional(:timeout) => pos_integer()
+          optional(:timeout) => pos_integer(),
+          optional(:headers) => map()
         }) :: t()
   def new(%{method: method} = attrs) do
     %__MODULE__{
       method: method,
       params: Map.get(attrs, :params) || %{},
       progress_opts: Map.get(attrs, :progress_opts),
-      timeout: Map.get(attrs, :timeout) || @default_timeout
+      timeout: Map.get(attrs, :timeout) || @default_timeout,
+      headers: Map.get(attrs, :headers)
     }
   end
 end

lib/hermes/transport/streamable_http.ex

@@ -81,7 +81,7 @@ defmodule Hermes.Transport.StreamableHTTP do
     name: {{:custom, &Hermes.genserver_name/1}, {:default, __MODULE__}},
     client: {:required, Hermes.get_schema(:process_name)},
     base_url: {:required, {:string, {:transform, &URI.new!/1}}},
-    mcp_path: {:string, {:default, "/mcp"}},
+    mcp_path: {:string, {:default, "/mcp/"}},
     headers: {:map, {:default, %{}}},
     transport_opts: {:any, {:default, []}},
     http_options: {:any, {:default, []}},
@@ -96,8 +96,8 @@ defmodule Hermes.Transport.StreamableHTTP do
   end
 
   @impl Transport
-  def send_message(pid \\ __MODULE__, message) when is_binary(message) do
-    GenServer.call(pid, {:send, message})
+  def send_message(pid \\ __MODULE__, message, opts \\ []) when is_binary(message) do
+    GenServer.call(pid, {:send, message, opts})
   end
 
   @impl Transport
@@ -136,31 +136,36 @@ defmodule Hermes.Transport.StreamableHTTP do
   end
 
   @impl GenServer
-  def handle_call({:send, message}, from, state) do
+  def handle_call({:send, message, opts}, from, state) do
     emit_telemetry(:send, state, %{message_size: byte_size(message)})
 
-    Logging.transport_event("sending_http_request", %{
+    headers = Keyword.get(opts, :headers)
+    log_event = if headers, do: "sending_http_request_with_headers", else: "sending_http_request"
+
+    log_data = %{
       url: URI.to_string(state.mcp_url),
       size: byte_size(message)
-    })
+    }
+
+    log_data = if headers, do: Map.put(log_data, :additional_headers, Map.keys(headers)), else: log_data
+
+    Logging.transport_event(log_event, log_data)
 
     new_state = %{state | active_request: from}
 
-    case send_http_request(new_state, message) do
+    case send_http_request(new_state, message, headers) do
       {:ok, response} ->
         Logging.transport_event("got_http_response", %{status: response.status})
         handle_response(response, new_state)
 
       {:error, {:http_error, 404, _body}} when not is_nil(state.session_id) ->
         Logging.transport_event("session_expired", %{session_id: state.session_id})
-        GenServer.cast(state.client, :session_expired)
-        {:reply, {:error, :session_expired}, %{state | session_id: nil}}
+        {:reply, {:error, :session_expired}, %{new_state | session_id: nil}}
 
       {:error, reason} ->
         Logging.transport_event("http_request_error", %{reason: inspect(reason)}, level: :error)
-
         log_error(reason)
-        {:reply, {:error, reason}, state}
+        {:reply, {:error, reason}, %{new_state | active_request: nil}}
     end
   end
 
@@ -224,13 +229,31 @@ defmodule Hermes.Transport.StreamableHTTP do
 
   # Private functions
 
-  defp send_http_request(state, message) do
-    headers =
+  defp send_http_request(state, message, additional_headers) do
+    base_headers =
       state.headers
       |> Map.put("accept", "application/json, text/event-stream")
       |> Map.put("content-type", "application/json")
       |> put_session_header(state.session_id)
 
+    case additional_headers do
+      nil ->
+        do_send_http_request(state, message, base_headers)
+
+      headers when is_map(headers) ->
+        case validate_headers(headers) do
+          :ok ->
+            final_headers = Map.merge(base_headers, headers)
+            do_send_http_request(state, message, final_headers)
+
+          {:error, reason} ->
+            {:error, reason}
+        end
+    end
+  end
+
+  # Common HTTP request logic
+  defp do_send_http_request(state, message, headers) do
     options = [transport_opts: state.transport_opts] ++ state.http_options
     url = URI.to_string(state.mcp_url)
 
@@ -253,7 +276,6 @@ defmodule Hermes.Transport.StreamableHTTP do
 
       {:error, reason} = error ->
         Logging.transport_event("http_request_failed", %{reason: reason}, level: :error)
-
         error
     end
   end
@@ -324,6 +346,23 @@ defmodule Hermes.Transport.StreamableHTTP do
     Logging.transport_event("unknown_sse_event", event, level: :warning)
   end
 
+  # Header validation - fail fast with clear error messages
+  defp validate_headers(headers) when is_map(headers) do
+    case find_invalid_header(headers) do
+      nil ->
+        :ok
+
+      {key, value} ->
+        {:error, "Invalid header value for '#{key}': HTTP headers must be strings, got #{inspect(value)}"}
+    end
+  end
+
+  defp find_invalid_header(headers) do
+    Enum.find(headers, fn {_key, value} ->
+      not is_binary(value) and not is_nil(value)
+    end)
+  end
+
   defp put_session_header(headers, nil), do: headers
 
   defp put_session_header(headers, session_id), do: Map.put(headers, "mcp-session-id", session_id)

test/hermes/transport/streamable_http_test.exs

@@ -394,6 +394,59 @@ defmodule Hermes.Transport.StreamableHTTPTest do
       StreamableHTTP.shutdown(transport)
       StubClient.clear_messages()
     end
+
+    test "sends per-request headers via opts parameter", %{bypass: bypass} do
+      server_url = "http://localhost:#{bypass.port}"
+      {:ok, stub_client} = StubClient.start_link()
+
+      Bypass.expect(bypass, "POST", "/mcp", fn conn ->
+        # Verify transport-level header is present
+        assert "Bearer base-token" ==
+                 conn |> Plug.Conn.get_req_header("authorization") |> List.first()
+
+        # Verify per-request header is present
+        assert "request-123" ==
+                 conn |> Plug.Conn.get_req_header("x-request-id") |> List.first()
+
+        # Verify per-request header overrides transport-level header
+        assert "request-specific" ==
+                 conn |> Plug.Conn.get_req_header("x-client-version") |> List.first()
+
+        conn = Plug.Conn.put_resp_header(conn, "content-type", "application/json")
+        Plug.Conn.resp(conn, 200, ~s|{"jsonrpc":"2.0","id":"1","result":{}}|)
+      end)
+
+      {:ok, transport} =
+        StreamableHTTP.start_link(
+          client: stub_client,
+          base_url: server_url,
+          mcp_path: "/mcp",
+          headers: %{
+            "authorization" => "Bearer base-token",
+            "x-client-version" => "1.0.0"
+          },
+          transport_opts: @test_http_opts
+        )
+
+      Process.sleep(100)
+
+      {:ok, ping_message} =
+        Message.encode_request(%{"method" => "ping", "params" => %{}}, "1")
+
+      # Test new opts-based header functionality
+      per_request_headers = %{
+        "x-request-id" => "request-123",
+        # This should override transport header
+        "x-client-version" => "request-specific"
+      }
+
+      assert :ok = StreamableHTTP.send_message(transport, ping_message, headers: per_request_headers)
+
+      Process.sleep(100)
+
+      StreamableHTTP.shutdown(transport)
+      StubClient.clear_messages()
+    end
   end
 
   describe "error handling" do