feat: forward engine routing (incl. namespace pools) via ClientOptions.environment (ENG-2612)

src/anam/_api.py

@@ -69,6 +69,10 @@ async def start_session(
             "sessionOptions": session_options.to_dict(),
             "clientMetadata": CLIENT_METADATA,
         }
+        # Engine routing overrides (e.g. pin to a specific pod / devspace /
+        # preview). Only sent when set; omitted entirely for production.
+        if self._options.environment:
+            body["environment"] = self._options.environment
 
         logger.debug("Starting session at %s (direct API-key auth)", url)
 

src/anam/types.py

@@ -261,12 +261,18 @@ 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.
+        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
+            verbatim as the ``environment`` field of the session request; the
+            backend ignores it for standard production routing.
     """
 
     api_base_url: str = "https://api.anam.ai"
     api_version: str = "v1"
     ice_servers: list[dict[str, Any]] | None = None
     client_label: str | None = None
+    environment: dict[str, str] | None = None
 
 
 @dataclass

tests/test_client.py

@@ -72,13 +72,84 @@ def test_init_with_options(self) -> None:
         """Test initialization with ClientOptions."""
         options = ClientOptions(
             api_base_url="https://custom.api.com",
+            environment={"podName": "anam-engine-abc", "engineVersion": "v5.13.2"},
         )
         client = AnamClient(
             api_key="test-key",
             persona_id="test-persona",
             options=options,
         )
         assert client._options.api_base_url == "https://custom.api.com"
+        assert client._options.environment == {
+            "podName": "anam-engine-abc",
+            "engineVersion": "v5.13.2",
+        }
+
+    def test_environment_defaults_to_none(self) -> None:
+        assert ClientOptions().environment is None
+
+
+class TestCoreApiClientSessionBody:
+    """The session request body carries engine routing overrides."""
+
+    @staticmethod
+    def _fake_session(captured: dict):
+        response_payload = {
+            "sessionId": "sess-1",
+            "engineHost": "engine.example",
+            "engineProtocol": "https",
+            "signallingEndpoint": "wss://engine.example/ws",
+        }
+
+        class _Resp:
+            status = 201
+
+            async def json(self) -> dict:
+                return response_payload
+
+            async def __aenter__(self) -> "_Resp":
+                return self
+
+            async def __aexit__(self, *_: object) -> bool:
+                return False
+
+        class _Session:
+            async def __aenter__(self) -> "_Session":
+                return self
+
+            async def __aexit__(self, *_: object) -> bool:
+                return False
+
+            def post(self, url: str, headers=None, json=None):  # noqa: A002
+                captured["body"] = json
+                return _Resp()
+
+        return _Session
+
+    @pytest.mark.asyncio
+    async def test_environment_is_sent_when_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        from anam._api import CoreApiClient
+
+        captured: dict[str, Any] = {}
+        monkeypatch.setattr("aiohttp.ClientSession", self._fake_session(captured))
+
+        overrides = {"podName": "anam-engine-abc", "engineVersion": "v5.13.2"}
+        client = CoreApiClient("key", ClientOptions(environment=overrides))
+        await client.start_session(PersonaConfig(persona_id="p"), SessionOptions())
+
+        assert captured["body"]["environment"] == overrides
+
+    @pytest.mark.asyncio
+    async def test_environment_is_omitted_when_unset(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        from anam._api import CoreApiClient
+
+        captured: dict[str, Any] = {}
+        monkeypatch.setattr("aiohttp.ClientSession", self._fake_session(captured))
+
+        client = CoreApiClient("key", ClientOptions())
+        await client.start_session(PersonaConfig(persona_id="p"), SessionOptions())
+
+        assert "environment" not in captured["body"]
 
 
 class TestAnamClientEvents: