Feat/lemonslice plugin stream

agents-core/vision_agents/core/agents/inference/transcribing_flow.py

@@ -538,6 +538,7 @@ async def process_tts(
                         if remainder:
                             async for chunk in self._tts.send_iter(remainder):
                                 await tts_output.send(chunk)
+                        await tts_output.send(TTSOutputEnd())
                     elif item.delta:
                         # Chunk: accumulate and emit on sentence boundaries.
                         text = self._tts_tokenizer.update(item.text)
@@ -551,17 +552,26 @@ async def process_tts(
                     if isinstance(item, TTSInput) and not item.delta:
                         async for chunk in self._tts.send_iter(item.text):
                             await tts_output.send(chunk)
+                        await tts_output.send(TTSOutputEnd())
 
     async def write_audio_output(
         self,
         tts_output: Stream[TTSOutputChunk | TTSOutputEnd],
         audio_output: AudioOutputStream,
     ):
+        # A streaming TTS speaks one reply as several segments, and each segment
+        # ends with a chunk where final=True.
+        # TTSOutputEnd is the only signal that the complete reply was synthesized,
+        # and it triggers AudioOutputChunk.final for downstream consumers.
         speaking = False
         async for item in tts_output:
             with log_exceptions(logger, "Error while processing TTS output"):
                 if isinstance(item, TTSOutputEnd):
                     if speaking:
+                        # An interrupt discards the audio already queued, so
+                        # there is no complete reply to mark.
+                        if not item.interrupted:
+                            await audio_output.send(AudioOutputChunk(final=True))
                         self.events.send(
                             AgentTurnEndedEvent(interrupted=item.interrupted)
                         )
@@ -570,9 +580,4 @@ async def write_audio_output(
                 if not speaking:
                     self.events.send(AgentTurnStartedEvent())
                     speaking = True
-                await audio_output.send(
-                    AudioOutputChunk(data=item.data, final=item.final)
-                )
-                if item.final:
-                    self.events.send(AgentTurnEndedEvent())
-                    speaking = False
+                await audio_output.send(AudioOutputChunk(data=item.data))

plugins/lemonslice/README.md

@@ -64,10 +64,9 @@ LEMONSLICE_AGENT_ID=your_agent_id
 # Or, instead of LEMONSLICE_AGENT_ID:
 # LEMONSLICE_AGENT_IMAGE_URL=https://example.com/avatar.png
 
-# LemonSlice uses LiveKit as a transport for audio and video
-LIVEKIT_URL=wss://your-livekit-server.com
-LIVEKIT_API_KEY=your_livekit_api_key
-LIVEKIT_API_SECRET=your_livekit_api_secret
+# LemonSlice uses Stream as the transport for audio and video
+STREAM_API_KEY=your_stream_api_key
+STREAM_API_SECRET=your_stream_api_secret
 ```
 
 ### Avatar Options
@@ -79,26 +78,61 @@ lemonslice.Avatar(
     agent_prompt=None,  # Prompt to influence avatar expressions/movements
     api_key=None,  # Optional: override LEMONSLICE_API_KEY env var
     idle_timeout=None,  # Session timeout in seconds
-    livekit_url=None,  # Optional: override LIVEKIT_URL env var
-    livekit_api_key=None,  # Optional: override LIVEKIT_API_KEY env var
-    livekit_api_secret=None,  # Optional: override LIVEKIT_API_SECRET env var
-    width=1920,  # Output video width in pixels
-    height=1080,  # Output video height in pixels
+    stream_api_key=None,  # Optional: override STREAM_API_KEY env var
+    stream_api_secret=None,  # Optional: override STREAM_API_SECRET env var
+    width=1280,  # Output video width in pixels
+    height=720,  # Output video height in pixels
+    fps=30,  # Output video frame rate
+    buffer_seconds=1.0,  # Max video buffer depth in seconds
+    avatar_join_timeout=30.0,  # Seconds to wait for the avatar to join the bridge call
+    lemonslice_properties=None,  # Extra fields merged into the LemonSlice session request
 )
 ```
 
 ## How It Works
 
-1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the LiveKit room as a participant
-2. **Audio Forwarding**: TTS audio is captured and sent to LemonSlice via the room
+1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the Stream call as a participant. `agent.join()` blocks until the avatar is on the call, so no audio is sent before it can receive it — if the avatar does not show up within `avatar_join_timeout`, the connection is torn down and the error is raised
+2. **Audio Forwarding**: TTS audio is captured and sent to LemonSlice via the Stream call
 3. **Avatar Generation**: LemonSlice generates synchronized avatar video and audio
 4. **Video Streaming**: Avatar video is streamed to call participants via GetStream Edge
 
+## Custom Stream Call Type (recommended)
+
+The plugin runs its own internal Stream call as a bridge between your process and the LemonSlice avatar service — this is separate from the user-facing call the agent joins. Only two users ever need to be on the bridge call: the plugin user and the avatar user. We recommend passing a custom `stream_call_type` whose permissions allow **only those two** to join, so no other token-holder in your app can accidentally enter the bridge.
+
+Reference docs:
+- [Built-in call types](https://getstream.io/video/docs/api/call_types/builtin/)
+- [Managing call types](https://getstream.io/video/docs/api/call_types/manage/)
+- [Permissions & capabilities](https://getstream.io/video/docs/api/call_types/permissions/)
+
+The plugin attaches both users to the bridge call as members with `role="call_member"`. Configure your custom call type so the `call_member` role has exactly the capabilities the plugin needs — and no other role has `join-call`:
+
+```python
+client.video.create_call_type(
+    name="lemonslice_bridge",
+    grants={
+        # plugin + avatar — everything they need to bridge audio/video
+        "call_member": ["join-call", "read-call", "send-audio", "send-video"],
+        # everyone else — denied
+        "user": [],
+        "admin": [],
+        "host": [],
+        "moderator": [],
+    },
+)
+
+lemonslice.Avatar(
+    agent_id="your-avatar-id",
+    stream_call_type="lemonslice_bridge",
+)
+```
+
+If you stick with the default `"default"` call type the plugin still works, but the bridge call uses the same broad permissions as any default Stream call.
+
 ## Requirements
 
 - Python 3.10+
 - LemonSlice API key (get one at [lemonslice.com](https://lemonslice.com))
-- LiveKit server (cloud or self-hosted)
 - GetStream account for video calls
 - TTS provider (Cartesia, ElevenLabs, etc.) or Realtime LLM
 

plugins/lemonslice/example/lemonslice_avatar_example.py

@@ -6,9 +6,6 @@
 Required environment variables:
     LEMONSLICE_API_KEY
     LEMONSLICE_AGENT_ID  (or LEMONSLICE_AGENT_IMAGE_URL)
-    LIVEKIT_URL
-    LIVEKIT_API_KEY
-    LIVEKIT_API_SECRET
     STREAM_API_KEY
     STREAM_API_SECRET
 """

plugins/lemonslice/pyproject.toml

@@ -12,8 +12,6 @@ requires-python = ">=3.10"
 license = "MIT"
 dependencies = [
     "vision-agents",
-    "livekit>=1.1.2,<2",
-    "livekit-api>=1.1.0,<2",
     "httpx>=0.28.1,<1",
 ]
 

plugins/lemonslice/tests/test_lemonslice_plugin.py

@@ -1,20 +1,57 @@
+import asyncio
+import json
+
+import httpx
+import numpy as np
 import pytest
-from vision_agents.core.agents.inference import AudioOutputStream
+from getstream import AsyncStream
+from getstream.video.rtc.track_util import AudioFormat, PcmData
+from vision_agents.core.agents.inference import AudioOutputFlush, AudioOutputStream
+from vision_agents.core.utils.utils import cancel_and_wait
 from vision_agents.core.utils.video_track import QueuedVideoTrack
 from vision_agents.plugins.lemonslice.lemonslice_avatar import LemonSliceAvatar
+from vision_agents.plugins.lemonslice.track import AvatarInputTrack
 
 
 def _make_avatar(**overrides) -> LemonSliceAvatar:
     default_kwargs = {
         "agent_id": "test-agent",
-        "api_key": "ls-test-key",
-        "livekit_url": "wss://test.livekit.cloud",
-        "livekit_api_key": "devkey",
-        "livekit_api_secret": "devsecret",
+        "api_key": "lemonslice-key",
+        "stream_api_key": "key",
+        "stream_api_secret": "secret",
     }
     return LemonSliceAvatar(**{**default_kwargs, **overrides})
 
 
+@pytest.fixture
+def session_requests() -> list[httpx.Request]:
+    return []
+
+
+@pytest.fixture
+def session_transport(session_requests: list[httpx.Request]) -> httpx.MockTransport:
+    def handler(request: httpx.Request) -> httpx.Response:
+        session_requests.append(request)
+        return httpx.Response(200, json={"session_id": "session-1"})
+
+    return httpx.MockTransport(handler)
+
+
+@pytest.fixture
+def call_events() -> list[dict]:
+    return []
+
+
+@pytest.fixture
+def call_event_transport(call_events: list[dict]) -> httpx.MockTransport:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/event"):
+            call_events.append(json.loads(request.content)["custom"])
+        return httpx.Response(200, json={"duration": "0ms"})
+
+    return httpx.MockTransport(handler)
+
+
 class TestLemonSliceAvatar:
     async def test_init_with_agent_image_url_instead_of_id(self):
         avatar = _make_avatar(
@@ -32,20 +69,13 @@ async def test_init_missing_api_key_raises(self, monkeypatch: pytest.MonkeyPatch
         with pytest.raises(ValueError, match="API key required"):
             _make_avatar(api_key=None)
 
-    async def test_init_missing_livekit_url_raises(
+    async def test_init_missing_stream_secret_raises(
         self, monkeypatch: pytest.MonkeyPatch
     ):
-        monkeypatch.delenv("LIVEKIT_URL", raising=False)
-        with pytest.raises(ValueError, match="LiveKit URL required"):
-            _make_avatar(livekit_url=None)
-
-    async def test_init_missing_livekit_secret_raises(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        monkeypatch.delenv("LIVEKIT_API_KEY", raising=False)
-        monkeypatch.delenv("LIVEKIT_API_SECRET", raising=False)
-        with pytest.raises(ValueError, match="LiveKit API key and secret required"):
-            _make_avatar(livekit_api_key=None, livekit_api_secret=None)
+        monkeypatch.delenv("STREAM_API_KEY", raising=False)
+        monkeypatch.delenv("STREAM_API_SECRET", raising=False)
+        with pytest.raises(ValueError, match="Stream API key and secret required"):
+            _make_avatar(stream_api_key=None, stream_api_secret=None)
 
     async def test_video_output(self):
         avatar = _make_avatar(width=640, height=480)
@@ -65,3 +95,102 @@ async def test_init_odd_height_raises(self):
     async def test_audio_output(self):
         avatar = _make_avatar()
         assert isinstance(avatar.audio_output(), AudioOutputStream)
+
+    async def test_extra_params_are_sent_in_the_session_request(
+        self,
+        session_transport: httpx.MockTransport,
+        session_requests: list[httpx.Request],
+    ):
+        avatar = _make_avatar(
+            lemonslice_properties={"voice_id": "nova", "metadata": {"tier": "pro"}}
+        )
+        avatar._client._http_client = httpx.AsyncClient(transport=session_transport)
+
+        await avatar._client.create_session(
+            call_id="call-1", call_type="default", token="token", api_key="stream-key"
+        )
+
+        payload = json.loads(session_requests[0].content)
+        assert payload["voice_id"] == "nova"
+        assert payload["metadata"] == {"tier": "pro"}
+
+    async def test_extra_params_do_not_override_transport_fields(
+        self,
+        session_transport: httpx.MockTransport,
+        session_requests: list[httpx.Request],
+    ):
+        avatar = _make_avatar(
+            lemonslice_properties={"transport_type": "websocket", "properties": {}}
+        )
+        avatar._client._http_client = httpx.AsyncClient(transport=session_transport)
+
+        await avatar._client.create_session(
+            call_id="call-1", call_type="default", token="token", api_key="stream-key"
+        )
+
+        payload = json.loads(session_requests[0].content)
+        assert payload["transport_type"] == "stream"
+        assert payload["properties"]["call_id"] == "call-1"
+
+    async def test_end_utterance_and_interrupt_events_respect_audio_boundaries(
+        self, call_events: list[dict], call_event_transport: httpx.MockTransport
+    ):
+        avatar = _make_avatar()
+        manager = avatar._rtc_manager
+        manager._client = AsyncStream(
+            api_key="key", api_secret="secret", transport=call_event_transport
+        )
+        manager._call = manager._client.video.call("default", "call-1")
+        manager._input_track = AvatarInputTrack(sample_rate=16000, channels=1)
+        manager._connected = True
+
+        for _ in range(3):
+            await manager.send_audio(
+                PcmData(
+                    samples=np.zeros(480, dtype=np.int16),
+                    sample_rate=24000,
+                    format=AudioFormat.S16,
+                    channels=1,
+                )
+            )
+
+        await manager.flush()
+
+        emitted_samples = 0
+        while True:
+            try:
+                emitted_samples += (
+                    await asyncio.wait_for(manager._input_track.recv(), timeout=0.05)
+                ).samples
+            except TimeoutError:
+                break
+
+        assert call_events[0] == {
+            "type": "lemonslice.end_utterance",
+            "pts": emitted_samples,
+            "event_id": 1,
+        }
+
+        # An interruption discards the buffered audio, so announcing an
+        # end-of-utterance PTS that covers it points the avatar at audio that
+        # never arrives.
+        await manager._input_track.write(
+            PcmData(
+                samples=np.zeros(16000, dtype=np.int16),
+                sample_rate=16000,
+                format=AudioFormat.S16,
+                channels=1,
+            )
+        )
+
+        stream = AudioOutputStream()
+        avatar.attach_audio_input(stream)
+        task = asyncio.create_task(avatar._process_audio_input())
+        stream.send_nowait(AudioOutputFlush())
+        await asyncio.sleep(0.05)
+        await cancel_and_wait(task)
+
+        assert [event["type"] for event in call_events] == [
+            "lemonslice.end_utterance",
+            "lemonslice.interrupt",
+        ]