Feat: pin high video quality

README.md

@@ -75,6 +75,30 @@ asyncio.run(main())
 - 📝 **Fully typed** - Complete type hints for IDE support
 - 🔒 **Server-side ready** - Designed for server-side Python applications (e.g. for backend pipelines)
 
+## Video Quality Notes (Server-to-Server)
+
+The Python SDK is primarily intended for server-side use with a high-capacity network connection. In this setup, adaptive bitrate (ABR) is often not required. By default, `SessionOptions` uses `video_quality="high"` which disables ABR and pins the video quality to the highest available rendition.
+
+
+Omitting `session_options` or setting `video_quality="high"` achieves this: 
+```python
+from anam import SessionOptions
+session_options = SessionOptions(video_quality="high")
+async with client.connect(session_options=session_options) as session:
+```
+
+This sends `sessionOptions.videoQuality="high"` to the API and pins the video bitrate for the session to the highest available bitrate.
+
+If you want to use ABR, set `video_quality="auto"` instead:
+
+```python
+from anam import SessionOptions
+session_options = SessionOptions(video_quality="auto")
+async with client.connect(session_options=session_options) as session:
+```
+
+Currently, only `"high"` or `"auto"` are supported `video_quality` values.
+
 ## API Reference
 
 ### AnamClient

examples/persona_interactive_video.py

@@ -237,6 +237,7 @@ async def consume_audio_frames(session) -> None:
         except Exception as e:
             logger.error(f"Error consuming audio frames: {e}")
 
+    # connect defaults high quality video rendition and disables ABR
     async with client.connect() as session:
         print(f"Session: {session.session_id}")
         print("Type 'q' in CLI to quit")

examples/user_audio_from_wav.py

@@ -61,6 +61,7 @@
 if missing:
     raise EnvironmentError(f"Missing required environment variables: {', '.join(missing)}")
 
+
 def _build_persona_config() -> PersonaConfig:
     """Build a persona configuration from environment variables."""
     avatar_id = os.environ.get("ANAM_AVATAR_ID", "").strip().strip('"')
@@ -76,11 +77,14 @@ def _build_persona_config() -> PersonaConfig:
         enable_audio_passthrough=False,
     )
 
+
 def _format_ids(correlation_ids: list[str | None]) -> str:
     """Format correlation IDs for log output."""
     if not correlation_ids:
         return "none"
-    return ", ".join("None" if correlation_id is None else correlation_id for correlation_id in correlation_ids)
+    return ", ".join(
+        "None" if correlation_id is None else correlation_id for correlation_id in correlation_ids
+    )
 
 
 def _compact_text(text: str) -> str:
@@ -141,9 +145,7 @@ async def _stream_wav_file_realtime(
         total_frames = wav_file.getnframes()
 
         if sample_width != 2:
-            raise ValueError(
-                f"Expected 16-bit PCM WAV input (sample width 2), got {sample_width}"
-            )
+            raise ValueError(f"Expected 16-bit PCM WAV input (sample width 2), got {sample_width}")
         if num_channels not in (1, 2):
             raise ValueError(f"Expected mono or stereo WAV input, got {num_channels} channels")
 
@@ -258,7 +260,9 @@ async def on_message_stream_event(event: MessageStreamEvent) -> None:
         role = "user" if event.role == MessageRole.USER else "assistant"
         role_emoji = "👤" if event.role == MessageRole.USER else "🤖"
         status_emoji = "✗" if event.interrupted else "✓"
-        await log(f"{role_emoji} {role} [{event.correlation_id}] ({status_emoji} {status}): {message}")
+        await log(
+            f"{role_emoji} {role} [{event.correlation_id}] ({status_emoji} {status}): {message}"
+        )
 
         if event.role == MessageRole.USER:
             _append_unique(transcript_ids, event.correlation_id)
@@ -285,7 +289,9 @@ async def on_message_stream_event(event: MessageStreamEvent) -> None:
         await asyncio.sleep(2.0)
 
         wav_duration, sample_rate, num_channels = await _stream_wav_file_realtime(session, wav_path)
-        await log(f"📤 Finished sending {wav_duration:.2f}s of WAV input. Sending trailing silence...")
+        await log(
+            f"📤 Finished sending {wav_duration:.2f}s of WAV input. Sending trailing silence..."
+        )
 
         silence_task = asyncio.create_task(
             _send_silence_until_cancelled(

src/anam/_api.py

@@ -46,7 +46,7 @@ async def get_session_token(
 
         Args:
             persona_config: The persona configuration to use.
-            session_options: Session options (optional).
+            session_options: Session options.
         Returns:
             The session token string.
 
@@ -107,7 +107,7 @@ async def start_session(
 
         Args:
             persona_config: The persona configuration.
-            session_options: Additional session options (optional).
+            session_options: Additional session options.
 
         Returns:
             SessionInfo with connection details.

src/anam/client.py

@@ -190,7 +190,9 @@ async def _emit(self, event: AnamEvent, *args: Any, **kwargs: Any) -> None:
             except Exception as e:
                 logger.error("Error in event callback for %s: %s", event.value, e)
 
-    def connect(self) -> "_SessionContextManager":
+    def connect(
+        self, session_options: SessionOptions = SessionOptions()
+    ) -> "_SessionContextManager":
         """Connect to Anam and start streaming.
 
         Returns:
@@ -216,13 +218,13 @@ def connect(self) -> "_SessionContextManager":
                 await session.close()
             ```
         """
-        return _SessionContextManager(self)
+        return _SessionContextManager(self, session_options)
 
     async def connect_async(self, session_options: SessionOptions = SessionOptions()) -> "Session":
         """Connect to Anam and start streaming (without context manager).
 
         Args:
-            session_options: Session options (default: SessionOptions(enable_session_replay=True)).
+            session_options: Session options (default: SessionOptions(enable_session_replay=True, video_quality="high")).
 
         Returns:
             A Session object for interacting with the avatar.
@@ -448,13 +450,14 @@ def get_persona_config(self) -> PersonaConfig | None:
 class _SessionContextManager:
     """Async context manager for AnamClient.connect()."""
 
-    def __init__(self, client: AnamClient):
+    def __init__(self, client: AnamClient, session_options: SessionOptions):
         self._client = client
         self._session: Session | None = None
+        self._session_options = session_options
 
     async def __aenter__(self) -> "Session":
         """Enter the context and connect."""
-        self._session = await self._client.connect_async()
+        self._session = await self._client.connect_async(self._session_options)
         return self._session
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:

src/anam/types.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any
+from typing import Any, Literal
 
 
 class AnamEvent(str, Enum):
@@ -123,18 +123,23 @@ class SessionOptions:
 
     Args:
         enable_session_replay: If True (default), session is recorded. Set False to disable.
+        video_quality: Video quality profile to pin the video quality. Supported values are "high" (default) and "auto".
     """
 
     enable_session_replay: bool = True
+    video_quality: Literal["high", "auto"] = "high"
 
     def __post_init__(self) -> None:
         self._session_replay = SessionReplayOptions(
             enable_session_replay=self.enable_session_replay
         )
+        if self.video_quality not in {"high", "auto"}:
+            raise ValueError('video_quality must be either "high" or "auto"')
 
     def to_dict(self) -> dict[str, Any]:
         result: dict[str, Any] = {}
         result["sessionReplay"] = self._session_replay.to_dict()
+        result["videoQuality"] = self.video_quality
         return result
 
 

tests/test_client.py

@@ -4,7 +4,15 @@
 
 import pytest
 
-from anam import AnamClient, AnamEvent, ClientOptions, MessageRole, MessageStreamEvent, PersonaConfig
+from anam import (
+    AnamClient,
+    AnamEvent,
+    ClientOptions,
+    MessageRole,
+    MessageStreamEvent,
+    PersonaConfig,
+    SessionOptions,
+)
 from anam.errors import ConfigurationError
 
 
@@ -210,3 +218,38 @@ def test_to_dict_full(self) -> None:
         assert result["languageCode"] == "en"
         assert result["llmId"] == "gpt-4"
         assert result["maxSessionLengthSeconds"] == 300
+
+
+class TestSessionOptions:
+    """Tests for SessionOptions serialization."""
+
+    def test_to_dict_defaults(self) -> None:
+        options = SessionOptions()
+        result = options.to_dict()
+
+        assert result == {
+            "sessionReplay": {"enableSessionReplay": True},
+            "videoQuality": "high",
+        }
+
+    def test_to_dict_with_video_quality_high(self) -> None:
+        options = SessionOptions(video_quality="high")
+        result = options.to_dict()
+
+        assert result == {
+            "sessionReplay": {"enableSessionReplay": True},
+            "videoQuality": "high",
+        }
+
+    def test_to_dict_with_video_quality_auto(self) -> None:
+        options = SessionOptions(video_quality="auto")
+        result = options.to_dict()
+
+        assert result == {
+            "sessionReplay": {"enableSessionReplay": True},
+            "videoQuality": "auto",
+        }
+
+    def test_invalid_video_quality_raises_value_error(self) -> None:
+        with pytest.raises(ValueError, match='video_quality must be either "high" or "auto"'):
+            SessionOptions(video_quality="medium")  # type: ignore[arg-type]