Add transcription example

AHGIJMKLKKZNPJKQR · AHGIJMKLKKZNPJKQR · commit 7bcd2aaf17a0 · 2025-08-22T01:27:36.000+02:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -8,9 +8,9 @@ jobs:
     strategy: &python-matrix
       matrix:
         python-version:
-          - "3.10"
           - "3.11"
           - "3.12"
+          - "3.13"
     name: static-checks
     steps:
       - name: Checkout code
@@ -28,7 +28,7 @@ jobs:
           enable-cache: true
 
       - name: Install project dependencies
-        run: uv sync --locked --all-extras --dev
+        run: uv sync --locked --all-extras --dev --all-packages
 
       - name: Format
         run: uv run format_check
diff --git a/examples/transcription/.python-version b/examples/transcription/.python-version
@@ -0,0 +1 @@
+3.12
diff --git a/examples/transcription/README.md b/examples/transcription/README.md
@@ -0,0 +1,36 @@
+# Transcription demo
+
+This directory contains a demo app, which uses [Fishjam](https://fishjam.io) and [Gemini Live API](https://ai.google.dev/gemini-api/docs/live)
+for real-time transcription of ongoing calls.
+
+The application contains an HTTP server written in [FastAPI](https://fastapi.tiangolo.com/)
+and uses [uv](https://docs.astral.sh/uv/) for dependency management.
+
+## Running
+
+Make sure to [install uv](https://docs.astral.sh/uv/getting-started/installation/) if you don't have it already.
+
+Once you have `uv` installed, fetch the dependencies with
+
+```bash
+uv sync --all-packages
+```
+
+To run the app, you will need 3 environment variables:
+
+- `FISHJAM_ID`: Your Fishjam ID, which you can get on the [Fishjam website](https://fishjam.io/app)
+- `FISHJAM_MANAGEMENT_TOKEN`: Your Fishjam managemen token, which you can get on the [Fishjam website](https://fishjam.io/app)
+- `GEMINI_API_KEY`: An API key for the Gemini API. You can generate one on the [Gemini website](https://aistudio.google.com/app/apikey).
+
+Once you have these variables, you can run the demo with
+
+```bash
+FISHJAM_ID=<your-fishjam-id> \
+FISHJAM_MANAGEMENT_TOKEN=<your-management-token> \
+GEMINI_API_KEY=<your-api-token> \
+uv run fastapi dev
+```
+
+Now, you can create peer tokens by going to <http://localhost:3000>.
+You can then use the [minimal-react](https://github.com/fishjam-cloud/web-client-sdk/tree/main/examples/react-client)
+demo app to connect as these peers and see your transcriptions live in the console!
diff --git a/examples/transcription/main.py b/examples/transcription/main.py
@@ -0,0 +1,40 @@
+from contextlib import asynccontextmanager
+from typing import Annotated
+
+from fastapi import Depends, FastAPI
+from transcription.notifier import make_notifier
+from transcription.room import RoomService, fishjam
+from transcription.worker import async_worker
+
+from fishjam import PeerOptions, SubscribeOptions
+
+_room_service: RoomService | None = None
+
+
+def get_room_service():
+    if not _room_service:
+        raise RuntimeError("Application skipped lifespan events!")
+    return _room_service
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    async with async_worker() as worker:
+        global _room_service
+        _room_service = RoomService(worker)
+        notifier = make_notifier(_room_service)
+        worker.run_in_background(notifier.connect())
+
+        yield
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.get("/")
+def get_peer(room_service: Annotated[RoomService, Depends(get_room_service)]):
+    _peer, token = fishjam.create_peer(
+        room_service.get_room().id,
+        PeerOptions(subscribe=SubscribeOptions()),
+    )
+    return token
diff --git a/examples/transcription/pyproject.toml b/examples/transcription/pyproject.toml
@@ -0,0 +1,14 @@
+[project]
+name = "transcription"
+version = "0.1.0"
+description = "Fishjam transcription demo"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi[standard]==0.116.0",
+    "fishjam-server-sdk",
+    "google-genai>=1.31.0",
+]
+
+[tool.uv.sources]
+fishjam-server-sdk = { workspace = true }
diff --git a/examples/transcription/transcription/__init__.py b/examples/transcription/transcription/__init__.py
diff --git a/examples/transcription/transcription/agent.py b/examples/transcription/transcription/agent.py
@@ -0,0 +1,55 @@
+import asyncio
+
+from fishjam.agent import Agent, AgentResponseTrackData
+from transcription.worker import BackgroundWorker
+
+from .transcription import TranscriptionSession
+
+
+class TranscriptionAgent:
+    def __init__(self, room_id: str, agent: Agent, worker: BackgroundWorker):
+        self._room_id = room_id
+        self._agent = agent
+        self._sessions: dict[str, TranscriptionSession] = {}
+        self._disconnect = asyncio.Event()
+        self._worker = worker
+
+        @agent.on_track_data
+        def _(track_data: AgentResponseTrackData):
+            if track_data.peer_id not in self._sessions:
+                return
+            self._sessions[track_data.peer_id].transcribe(track_data.data)
+
+    async def _start(self):
+        async with self._agent:
+            print(f"Agent connected to room {self._room_id}")
+            await self._disconnect.wait()
+        self._disconnect.clear()
+        print(f"Agent disconnected from room {self._room_id}")
+
+    def _stop(self):
+        self._disconnect.set()
+
+    def _handle_transcription(self, peer_id: str, text: str):
+        print(f"Peer {peer_id} in room {self._room_id} said: {text}")
+
+    def on_peer_enter(self, peer_id: str):
+        if peer_id in self._sessions:
+            return
+
+        if len(self._sessions) == 0:
+            self._worker.run_in_background(self._start())
+
+        session = TranscriptionSession(lambda t: self._handle_transcription(peer_id, t))
+        self._sessions[peer_id] = session
+        self._worker.run_in_background(session.start(peer_id))
+
+    def on_peer_leave(self, peer_id: str):
+        if peer_id not in self._sessions:
+            return
+
+        self._sessions[peer_id].end()
+        self._sessions.pop(peer_id)
+
+        if len(self._sessions) == 0:
+            self._stop()
diff --git a/examples/transcription/transcription/config.py b/examples/transcription/transcription/config.py
@@ -0,0 +1,12 @@
+import os
+
+from google.genai.types import AudioTranscriptionConfig, LiveConnectConfig, Modality
+
+FISHJAM_ID = os.environ["FISHJAM_ID"]
+FISHJAM_TOKEN = os.environ["FISHJAM_MANAGEMENT_TOKEN"]
+FISHJAM_URL = os.getenv("FISHJAM_URL")
+TRANSCRIPTION_MODEL = "gemini-live-2.5-flash-preview"
+TRANSCRIPTION_CONFIG = LiveConnectConfig(
+    response_modalities=[Modality.TEXT],
+    input_audio_transcription=AudioTranscriptionConfig(),
+)
diff --git a/examples/transcription/transcription/notifier.py b/examples/transcription/transcription/notifier.py
@@ -0,0 +1,29 @@
+from fishjam import FishjamNotifier
+from fishjam.events import ServerMessagePeerConnected, ServerMessagePeerDisconnected
+from fishjam.events.allowed_notifications import AllowedNotification
+
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .room import RoomService
+
+
+def make_notifier(room_service: RoomService):
+    notifier = FishjamNotifier(FISHJAM_ID, FISHJAM_TOKEN, fishjam_url=FISHJAM_URL)
+
+    @notifier.on_server_notification
+    def _(notification: AllowedNotification):
+        match notification:
+            case ServerMessagePeerConnected(peer_id=peer_id, room_id=room_id):
+                handle_peer_connected(peer_id, room_id)
+
+            case ServerMessagePeerDisconnected(peer_id=peer_id, room_id=room_id):
+                handle_peer_disconnected(peer_id, room_id)
+
+    def handle_peer_connected(peer_id: str, room_id: str):
+        if room_id == room_service.room.id:
+            room_service.agent.on_peer_enter(peer_id)
+
+    def handle_peer_disconnected(peer_id: str, room_id: str):
+        if room_id == room_service.room.id:
+            room_service.agent.on_peer_leave(peer_id)
+
+    return notifier
diff --git a/examples/transcription/transcription/room.py b/examples/transcription/transcription/room.py
@@ -0,0 +1,39 @@
+from fishjam import FishjamClient, Room
+from fishjam.errors import NotFoundError
+from transcription.worker import BackgroundWorker
+
+from .agent import TranscriptionAgent
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+
+fishjam = FishjamClient(
+    FISHJAM_ID,
+    FISHJAM_TOKEN,
+    fishjam_url=FISHJAM_URL,
+)
+
+
+class RoomService:
+    def __init__(self, worker: BackgroundWorker):
+        self._worker = worker
+        self._create_room()
+
+    def get_room(self) -> Room:
+        try:
+            self.room = fishjam.get_room(self.room.id)
+        except NotFoundError:
+            self._create_room()
+        return self.room
+
+    def _create_room(self):
+        self.room = fishjam.create_room()
+        self._create_agent()
+
+    def _create_agent(self):
+        self.agent = TranscriptionAgent(
+            self.room.id,
+            fishjam.create_agent(self.room.id),
+            self._worker,
+        )
+
+    def get_agent(self):
+        return self.agent
diff --git a/examples/transcription/transcription/transcription.py b/examples/transcription/transcription/transcription.py
@@ -0,0 +1,63 @@
+from asyncio import Event, Queue, TaskGroup
+from typing import Callable
+
+from google import genai
+from google.genai.live import AsyncSession
+from google.genai.types import Blob
+
+from .config import TRANSCRIPTION_CONFIG, TRANSCRIPTION_MODEL
+
+
+class TranscriptionSession:
+    def __init__(self, on_text: Callable[[str], None]):
+        self._gemini = genai.Client()
+        self._audio_queue = Queue[bytes]()
+        self._end_event = Event()
+        self._model = TRANSCRIPTION_MODEL
+        self._on_text = on_text
+
+    async def start(self, peer_id: str):
+        async with self._gemini.aio.live.connect(
+            model=self._model,
+            config=TRANSCRIPTION_CONFIG,
+        ) as session:
+            async with TaskGroup() as tg:
+                send_task = tg.create_task(self._send_loop(session))
+                recv_task = tg.create_task(self._recv_loop(session))
+
+                print(f"Started transcription session for peer {peer_id}")
+
+                await self._end_event.wait()
+
+                send_task.cancel()
+                recv_task.cancel()
+            print(f"Stopped transcription session for peer {peer_id}")
+
+    def transcribe(self, audio: bytes):
+        self._audio_queue.put_nowait(audio)
+
+    def end(self):
+        self._end_event.set()
+
+    async def _send_loop(self, session: AsyncSession):
+        while True:
+            audio_frame = await self._audio_queue.get()
+            await session.send_realtime_input(
+                audio=Blob(data=audio_frame, mime_type="audio/pcm;rate=16000")
+            )
+
+    async def _recv_loop(self, session: AsyncSession):
+        while True:
+            acc = ""
+            async for res in session.receive():
+                if (
+                    (content := res.server_content)
+                    and (transcription := content.input_transcription)
+                    and (text := transcription.text)
+                ):
+                    acc += text
+
+            if acc:
+                self._on_text(acc)
+
+            return session
diff --git a/examples/transcription/transcription/worker.py b/examples/transcription/transcription/worker.py
@@ -0,0 +1,30 @@
+from asyncio import Task, TaskGroup
+from contextlib import asynccontextmanager
+from typing import Any, Coroutine
+
+
+class BackgroundWorker:
+    def __init__(self, tg: TaskGroup) -> None:
+        self._tg = tg
+        self._tasks: set[Task[None]] = set()
+
+    def run_in_background(self, coro: Coroutine[Any, Any, None]):
+        task = self._tg.create_task(coro)
+        task.add_done_callback(self._remove_task)
+        self._tasks.add(task)
+
+    def _remove_task(self, task: Task[None]):
+        self._tasks.discard(task)
+
+    def cleanup(self):
+        for task in self._tasks:
+            task.cancel()
+        self._tasks = set()
+
+
+@asynccontextmanager
+async def async_worker():
+    async with TaskGroup() as tg:
+        worker = BackgroundWorker(tg)
+        yield worker
+        worker.cleanup()
diff --git a/fishjam/__init__.py b/fishjam/__init__.py
@@ -21,13 +21,15 @@
     PeerOptions,
     Room,
     RoomOptions,
+    SubscribeOptions,
 )
 
 __all__ = [
     "FishjamClient",
     "FishjamNotifier",
     "receive_binary",
     "PeerOptions",
+    "SubscribeOptions",
     "RoomOptions",
     "Room",
     "Peer",
diff --git a/fishjam/_ws_notifier.py b/fishjam/_ws_notifier.py
diff --git a/fishjam/agent/agent.py b/fishjam/agent/agent.py
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/tests/Dockerfile b/tests/Dockerfile
diff --git a/uv.lock b/uv.lock
