Support additional session request headers

src/anam/_api.py

@@ -64,6 +64,7 @@ async def start_session(
         url = f"{self._api_url}/engine/session"
         authorization = self._session_token or self._api_key
         headers = {
+            **(self._options.request_headers or {}),
             "Content-Type": "application/json",
             "Authorization": f"Bearer {authorization}",
         }

src/anam/types.py

@@ -261,6 +261,9 @@ class ClientOptions:
         ice_servers: Custom ICE servers for WebRTC (optional).
         client_label: Custom label for session tracking (optional).
             Defaults to 'python-sdk' if not specified.
+        request_headers: Additional HTTP headers to include when starting a
+            session (optional). SDK-owned headers such as ``Authorization``
+            and ``Content-Type`` cannot be overridden.
         environment: Engine routing overrides for non-production targets
             (optional), e.g. ``{"podName": ..., "engineVersion": ...}`` to pin
             the session to a specific engine pod / devspace / preview. Sent
@@ -272,6 +275,7 @@ class ClientOptions:
     api_version: str = "v1"
     ice_servers: list[dict[str, Any]] | None = None
     client_label: str | None = None
+    request_headers: dict[str, str] | None = None
     environment: dict[str, str] | None = None
 
 

tests/test_client.py

@@ -116,6 +116,9 @@ def test_init_with_options(self) -> None:
     def test_environment_defaults_to_none(self) -> None:
         assert ClientOptions().environment is None
 
+    def test_request_headers_default_to_none(self) -> None:
+        assert ClientOptions().request_headers is None
+
 
 class TestCoreApiClientSessionBody:
     """The session request body matches the selected authentication mode."""
@@ -203,6 +206,34 @@ async def test_pre_minted_session_token_uses_snapshot_only(
         assert captured["headers"]["Authorization"] == f"Bearer {token}"
         assert captured["body"] == {"clientMetadata": CLIENT_METADATA}
 
+    @pytest.mark.asyncio
+    async def test_additional_request_headers_cannot_override_sdk_headers(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from anam._api import CoreApiClient
+
+        captured: dict[str, Any] = {}
+        monkeypatch.setattr("aiohttp.ClientSession", self._fake_session(captured))
+
+        token = "header.payload.signature"
+        client = CoreApiClient(
+            session_token=token,
+            options=ClientOptions(
+                request_headers={
+                    "x-vercel-protection-bypass": "preview-secret",
+                    "Authorization": "Bearer attacker-controlled",
+                    "Content-Type": "text/plain",
+                }
+            ),
+        )
+        await client.start_session(None, SessionOptions())
+
+        assert captured["headers"] == {
+            "x-vercel-protection-bypass": "preview-secret",
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+        }
+
 
 class TestAnamClientEvents:
     """Tests for event handling."""