Feat/session options egress

README.md

@@ -70,6 +70,7 @@ asyncio.run(main())
 - 🤖 **Audio-passthrough** - Send TTS generated audio input and receive rendered synchronized audio/video avatar
 - 🗣️ **Direct text-to-speech** - Send text directly to TTS for immediate speech output (bypasses LLM processing)
 - 🎤 **Real-time user audio input** - Send raw audio samples (e.g. from microphone) to Anam for processing (turnkey solution: STT → LLM → TTS → Avatar)
+- 📤 **Direct egress** - (experimental) Publish the avatar's synchronised audio + video directly to a 3rd party video network provider. (currently supported providers: Daily)
 - 📡 **Async iterator API** - Clean, Pythonic async/await patterns for continuous stream of audio/video frames
 - 🎯 **Event-driven API** - Simple decorator-based event handlers for discrete events
 - 📝 **Fully typed** - Complete type hints for IDE support
@@ -99,6 +100,51 @@ async with client.connect(session_options=session_options) as session:
 
 Currently, only `"high"` or `"auto"` are supported `video_quality` values.
 
+## Direct Egress (Daily)
+
+> [!WARNING]
+> Direct Egress is experimental and only supported for Cara-4 avatars.
+> The transport and signalling path will change in upcoming alpha releases,
+> including our backend support. Expect breaking changes between alphas.
+
+Instead of consuming avatar frames over the SDK's WebRTC connection, Anam can publish the avatar's synchronised audio + video directly to a 3rd party real-time media network layer (e.g. WebRTC). The SDK's connection stays open for signalling; media goes straight from Anam to your channel/room/SFU/etc. Supported 3rd party networks: Daily.
+
+```python
+from anam import (
+    AnamClient,
+    EgressDailyOptions,
+    EgressOptions,
+    PersonaConfig,
+    SessionOptions,
+)
+
+client = AnamClient(
+    api_key="your-api-key",
+    persona_config=PersonaConfig(
+        avatar_id="your-avatar-id",
+        enable_audio_passthrough=True,
+    ),
+)
+
+session_options = SessionOptions(
+    egress=EgressOptions(
+        mode="daily",
+        daily=EgressDailyOptions(
+            room_url="https://your-domain.daily.co/your-room",
+            token="meeting-token-minted-by-you",  # optional for public rooms
+            user_name="anam-avatar",              # optional
+        ),
+    ),
+)
+
+async with client.connect(session_options=session_options) as session:
+    # The avatar is now publishing into your Daily room.
+    # Drive it via the normal SDK surface — e.g. send TTS audio:
+    await session.wait_until_closed()
+```
+
+**Daily tokens.** Mint meeting tokens through your own Daily app (Daily REST API or a server you control); the SDK never does this for you. A Daily meeting token is bound to the room it was minted for, so the `token` you pass must be minted for the same `room_url` — passing a token minted for a different room will fail at join time with a 403. For public rooms (no token required) you can omit `token` entirely.
+
 ## API Reference
 
 ### AnamClient

src/anam/__init__.py

@@ -59,6 +59,8 @@ async def consume_audio():
     AnamEvent,
     ClientOptions,
     ConnectionClosedCode,
+    EgressDailyOptions,
+    EgressOptions,
     Message,
     MessageRole,
     MessageStreamEvent,
@@ -80,6 +82,8 @@ async def consume_audio():
     "AudioFrame",
     "ClientOptions",
     "ConnectionClosedCode",
+    "EgressDailyOptions",
+    "EgressOptions",
     "Message",
     "MessageRole",
     "MessageStreamEvent",

src/anam/types.py

@@ -117,17 +117,68 @@ def to_dict(self) -> dict[str, Any]:
         return {"enableSessionReplay": self.enable_session_replay}
 
 
+@dataclass
+class EgressDailyOptions:
+    """Daily-specific egress credentials.
+
+    Args:
+        room_url: Daily room URL the avatar joins as a publisher.
+        token: Optional Daily meeting token. Omit for public rooms.
+        user_name: Optional display name for the avatar participant (default: "anam-avatar").
+    """
+
+    room_url: str
+    token: str | None = None
+    user_name: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        result: dict[str, Any] = {"roomUrl": self.room_url}
+        if self.token:
+            result["token"] = self.token
+        if self.user_name:
+            result["userName"] = self.user_name
+        return result
+
+
+@dataclass
+class EgressOptions:
+    """Direct egress to a third-party real-time transport.
+
+    When set, the avatar's audio + video is published into the named
+    provider's room. The WebRTC peer connection still exists for signalling
+    (interrupts, status messages).
+
+    Args:
+        mode: Egress provider. Only ``"daily"`` is currently supported.
+        daily: Required when ``mode`` is ``"daily"``.
+    """
+
+    mode: Literal["daily"]
+    daily: EgressDailyOptions
+
+    def __post_init__(self) -> None:
+        if self.mode == "daily" and self.daily is None:
+            raise ValueError('EgressOptions(mode="daily") requires a daily=... block')
+
+    def to_dict(self) -> dict[str, Any]:
+        if self.mode == "daily":
+            return {"mode": "daily", "daily": self.daily.to_dict()}
+        raise ValueError(f"Unsupported egress mode: {self.mode!r}")
+
+
 @dataclass
 class SessionOptions:
     """Configuration for an Anam session.
 
     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".
+        egress: Optional direct egress to a third-party transport (e.g. Daily). See :class:`EgressOptions`.
     """
 
     enable_session_replay: bool = True
     video_quality: Literal["high", "auto"] = "high"
+    egress: EgressOptions | None = None
 
     def __post_init__(self) -> None:
         self._session_replay = SessionReplayOptions(
@@ -137,9 +188,12 @@ def __post_init__(self) -> None:
             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
+        result: dict[str, Any] = {
+            "sessionReplay": self._session_replay.to_dict(),
+            "videoQuality": self.video_quality,
+        }
+        if self.egress is not None:
+            result["egress"] = self.egress.to_dict()
         return result