Fishjam Python Server SDK

160class FishjamClient(Client):
161    """Allows for managing rooms."""
162
163    def __init__(
164        self,
165        fishjam_id: str,
166        management_token: str,
167    ):
168        """Create a FishjamClient instance.
169
170        Does not contact the Fishjam backend — use :meth:`create_and_verify`
171        or :meth:`check_credentials` to verify credentials live.
172
173        Args:
174            fishjam_id: The unique identifier for the Fishjam instance.
175            management_token: The token used for authenticating management operations.
176        """
177        super().__init__(fishjam_id=fishjam_id, management_token=management_token)
178
179    @classmethod
180    def create_and_verify(
181        cls, *, fishjam_id: str, management_token: str
182    ) -> "FishjamClient":
183        """Construct a FishjamClient and verify its credentials against the backend.
184
185        Args:
186            fishjam_id: The unique identifier for the Fishjam instance.
187            management_token: The token used for authenticating management operations.
188
189        Returns:
190            FishjamClient: A client whose credentials have been verified.
191
192        Raises:
193            InvalidFishjamCredentialsError: If the token is rejected.
194        """
195        client = cls(fishjam_id=fishjam_id, management_token=management_token)
196        client.check_credentials()
197        return client
198
199    def check_credentials(self) -> None:
200        """Verify the management token via a single ``/validate`` call.
201
202        Raises:
203            InvalidFishjamCredentialsError: If the token is rejected.
204        """
205        response = credentials_validate_credentials.sync_detailed(client=self.client)
206        self._handle_deprecation_header(response.headers)
207
208        if response.status_code == HTTPStatus.NOT_FOUND:
209            raise InvalidFishjamCredentialsError("Invalid Fishjam credentials")
210
211    def create_peer(
212        self,
213        room_id: str,
214        options: PeerOptions | None = None,
215    ) -> tuple[Peer, str]:
216        """Creates a peer in the room.
217
218        Args:
219            room_id: The ID of the room where the peer will be created.
220            options: Configuration options for the peer. Defaults to None.
221
222        Returns:
223            A tuple containing:
224                - Peer: The created peer object.
225                - str: The peer token needed to authenticate to Fishjam.
226        """
227        options = options or PeerOptions()
228
229        peer_metadata = self.__parse_peer_metadata(options.metadata)
230        peer_options = PeerOptionsWebRTC(
231            metadata=peer_metadata,
232            subscribe_mode=SubscribeMode(options.subscribe_mode),
233        )
234        body = PeerConfig(type_=PeerType.WEBRTC, options=peer_options)
235
236        resp = cast(
237            PeerDetailsResponse,
238            self._request(room_add_peer, room_id=room_id, body=body),
239        )
240
241        return (resp.data.peer, resp.data.token)
242
243    def create_agent(self, room_id: str, options: AgentOptions | None = None):
244        """Creates an agent in the room.
245
246        Args:
247            room_id: The ID of the room where the agent will be created.
248            options: Configuration options for the agent. Defaults to None.
249
250        Returns:
251            Agent: The created agent instance initialized with peer ID, room ID, token,
252                and Fishjam URL.
253        """
254        options = options or AgentOptions()
255        body = PeerConfig(
256            type_=PeerType.AGENT,
257            options=PeerOptionsAgent(
258                output=AgentOutput(
259                    audio_format=AudioFormat(options.output.audio_format),
260                    audio_sample_rate=AudioSampleRate(options.output.audio_sample_rate),
261                ),
262                subscribe_mode=SubscribeMode(options.subscribe_mode),
263            ),
264        )
265
266        resp = cast(
267            PeerDetailsResponse,
268            self._request(room_add_peer, room_id=room_id, body=body),
269        )
270
271        return Agent(resp.data.peer.id, room_id, resp.data.token, self._fishjam_url)
272
273    def create_vapi_agent(
274        self,
275        room_id: str,
276        options: PeerOptionsVapi,
277    ) -> Peer:
278        """Creates a vapi agent in the room.
279
280        Args:
281            room_id: The ID of the room where the vapi agent will be created.
282            options: Configuration options for the vapi peer.
283
284        Returns:
285            - Peer: The created peer object.
286        """
287        body = PeerConfig(type_=PeerType.VAPI, options=options)
288
289        resp = cast(
290            PeerDetailsResponse,
291            self._request(room_add_peer, room_id=room_id, body=body),
292        )
293
294        return resp.data.peer
295
296    def create_room(self, options: RoomOptions | None = None) -> Room:
297        """Creates a new room.
298
299        Args:
300            options: Configuration options for the room. Defaults to None.
301
302        Returns:
303            Room: The created Room object.
304        """
305        options = options or RoomOptions()
306
307        if options.video_codec is None:
308            codec = UNSET
309        else:
310            codec = VideoCodec(options.video_codec)
311
312        config = RoomConfig(
313            max_peers=options.max_peers,
314            video_codec=codec,
315            webhook_url=options.webhook_url,
316            room_type=RoomType(options.room_type),
317            public=options.public,
318            batch_webhook_notifications=options.batch_webhook_notifications,
319        )
320
321        room = cast(
322            RoomCreateDetailsResponse, self._request(room_create_room, body=config)
323        ).data.room
324
325        return Room(config=room.config, id=room.id, peers=room.peers)
326
327    def get_all_rooms(self) -> list[Room]:
328        """Returns list of all rooms.
329
330        Returns:
331            list[Room]: A list of all available Room objects.
332        """
333        rooms = cast(RoomsListingResponse, self._request(room_get_all_rooms)).data
334
335        return [
336            Room(config=room.config, id=room.id, peers=room.peers) for room in rooms
337        ]
338
339    def get_room(self, room_id: str) -> Room:
340        """Returns room with the given id.
341
342        Args:
343            room_id: The ID of the room to retrieve.
344
345        Returns:
346            Room: The Room object corresponding to the given ID.
347        """
348        room = cast(
349            RoomDetailsResponse, self._request(room_get_room, room_id=room_id)
350        ).data
351
352        return Room(config=room.config, id=room.id, peers=room.peers)
353
354    def delete_peer(self, room_id: str, peer_id: str) -> None:
355        """Deletes a peer from a room.
356
357        Args:
358            room_id: The ID of the room the peer belongs to.
359            peer_id: The ID of the peer to delete.
360        """
361        return self._request(room_delete_peer, id=peer_id, room_id=room_id)
362
363    def delete_room(self, room_id: str) -> None:
364        """Deletes a room.
365
366        Args:
367            room_id: The ID of the room to delete.
368        """
369        return self._request(room_delete_room, room_id=room_id)
370
371    def refresh_peer_token(self, room_id: str, peer_id: str) -> str:
372        """Refreshes a peer token.
373
374        Args:
375            room_id: The ID of the room.
376            peer_id: The ID of the peer whose token needs refreshing.
377
378        Returns:
379            str: The new peer token.
380        """
381        response = cast(
382            PeerRefreshTokenResponse,
383            self._request(room_refresh_token, id=peer_id, room_id=room_id),
384        )
385
386        return response.data.token
387
388    def create_livestream_viewer_token(self, room_id: str) -> str:
389        """Generates a viewer token for livestream rooms.
390
391        Args:
392            room_id: The ID of the livestream room.
393
394        Returns:
395            str: The generated viewer token.
396        """
397        response = cast(
398            ViewerToken, self._request(viewer_generate_viewer_token, room_id=room_id)
399        )
400
401        return response.token
402
403    def create_livestream_streamer_token(self, room_id: str) -> str:
404        """Generates a streamer token for livestream rooms.
405
406        Args:
407            room_id: The ID of the livestream room.
408
409        Returns:
410            str: The generated streamer token.
411        """
412        response = cast(
413            StreamerToken,
414            self._request(streamer_generate_streamer_token, room_id=room_id),
415        )
416
417        return response.token
418
419    def create_moq_access(
420        self,
421        publish_path: str | None = None,
422        subscribe_path: str | None = None,
423    ) -> MoqAccess:
424        """Generates MoQ relay connection details.
425
426        Args:
427            publish_path: Path the access grants publish access to.
428            subscribe_path: Path the access grants subscribe access to.
429
430        Returns:
431            MoqAccess: The relay connection details, containing the
432            ``connection_url`` (with the JWT embedded as a ``?jwt=`` query
433            parameter) and the ``token`` itself.
434        """
435        config = MoqAccessConfig(
436            publish_path=publish_path, subscribe_path=subscribe_path
437        )
438        response = cast(
439            MoqAccess,
440            self._request(moq_create_access, body=config),
441        )
442
443        return response
444
445    def subscribe_peer(self, room_id: str, peer_id: str, target_peer_id: str):
446        """Subscribes a peer to all tracks of another peer.
447
448        Args:
449            room_id: The ID of the room.
450            peer_id: The ID of the subscribing peer.
451            target_peer_id: The ID of the peer to subscribe to.
452        """
453        self._request(
454            room_subscribe_peer,
455            room_id=room_id,
456            id=peer_id,
457            peer_id=target_peer_id,
458        )
459
460    def subscribe_tracks(self, room_id: str, peer_id: str, track_ids: list[str]):
461        """Subscribes a peer to specific tracks of another peer.
462
463        Args:
464            room_id: The ID of the room.
465            peer_id: The ID of the subscribing peer.
466            track_ids: A list of track IDs to subscribe to.
467        """
468        self._request(
469            room_subscribe_tracks,
470            room_id=room_id,
471            id=peer_id,
472            body=SubscribeTracksBody(track_ids=track_ids),
473        )
474
475    def __parse_peer_metadata(self, metadata: dict | None) -> WebRTCMetadata:
476        peer_metadata = WebRTCMetadata()
477
478        if not metadata:
479            return peer_metadata
480
481        for key, value in metadata.items():
482            peer_metadata.additional_properties[key] = value
483
484        return peer_metadata

 33class FishjamNotifier:
 34    """Allows for receiving WebSocket messages from Fishjam."""
 35
 36    def __init__(
 37        self,
 38        fishjam_id: str,
 39        management_token: str,
 40    ):
 41        """Create a FishjamNotifier instance with an ID and management token."""
 42        websocket_url = get_fishjam_url(fishjam_id).replace("http", "ws")
 43        self._fishjam_url = f"{websocket_url}/socket/server/websocket"
 44        self._management_token: str = management_token
 45        self._websocket: client.ClientConnection | None = None
 46        self._ready: bool = False
 47
 48        self._ready_event: asyncio.Event | None = None
 49
 50        self._notification_handler: NotificationHandler | None = None
 51
 52    def on_server_notification(self, handler: NotificationHandler):
 53        """Decorator for defining a handler for Fishjam notifications.
 54
 55        Args:
 56            handler: The function to be registered as the notification handler.
 57
 58        Returns:
 59            NotificationHandler: The original handler function (unmodified).
 60        """
 61        self._notification_handler = handler
 62        return handler
 63
 64    async def connect(self):
 65        """Connects to Fishjam and listens for all incoming messages.
 66
 67        It runs until the connection isn't closed.
 68
 69        The incoming messages are handled by the functions defined using the
 70        `on_server_notification` decorator.
 71
 72        The handler have to be defined before calling `connect`,
 73        otherwise the messages won't be received.
 74        """
 75        async with client.connect(self._fishjam_url) as websocket:
 76            try:
 77                self._websocket = websocket
 78                await self._authenticate()
 79
 80                if self._notification_handler:
 81                    await self._subscribe_event(
 82                        event=ServerMessageEventType.EVENT_TYPE_SERVER_NOTIFICATION
 83                    )
 84
 85                self._ready = True
 86                if self._ready_event:
 87                    self._ready_event.set()
 88
 89                await self._receive_loop()
 90            finally:
 91                self._websocket = None
 92
 93    async def wait_ready(self) -> None:
 94        """Waits until the notifier is connected and authenticated to Fishjam.
 95
 96        If already connected, returns immediately.
 97        """
 98        if self._ready:
 99            return
100
101        if self._ready_event is None:
102            self._ready_event = asyncio.Event()
103
104        await self._ready_event.wait()
105
106    async def _authenticate(self):
107        if not self._websocket:
108            raise RuntimeError("Websocket is not connected")
109
110        msg = ServerMessage(
111            auth_request=ServerMessageAuthRequest(token=self._management_token)
112        )
113        await self._websocket.send(bytes(msg))
114
115        try:
116            message = await self._websocket.recv(decode=False)
117        except ConnectionClosed as exception:
118            if "invalid token" in str(exception):
119                raise RuntimeError("Invalid management token") from exception
120            raise
121
122        message = ServerMessage().parse(message)
123
124        _type, message = betterproto.which_one_of(message, "content")
125        assert isinstance(message, ServerMessageAuthenticated)
126
127    async def _receive_loop(self):
128        if not self._websocket:
129            raise RuntimeError("Websocket is not connected")
130        if not self._notification_handler:
131            raise RuntimeError("Notification handler is not defined")
132
133        while True:
134            message = await self._websocket.recv(decode=False)
135            message = ServerMessage().parse(message)
136            _which, message = betterproto.which_one_of(message, "content")
137
138            if isinstance(message, ALLOWED_NOTIFICATIONS):
139                res = self._notification_handler(message)
140                if inspect.isawaitable(res):
141                    await res
142
143    async def _subscribe_event(self, event: ServerMessageEventType):
144        if not self._websocket:
145            raise RuntimeError("Websocket is not connected")
146
147        request = ServerMessage(subscribe_request=ServerMessageSubscribeRequest(event))
148
149        await self._websocket.send(bytes(request))
150        message = cast(bytes, await self._websocket.recv())
151        message = ServerMessage().parse(message)
152        _which, message = betterproto.which_one_of(message, "content")
153        assert isinstance(message, ServerMessageSubscribeResponse)

45def decode_server_notifications(binary: bytes) -> List[AllowedNotification]:
46    """Decode a received protobuf payload into a list of notifications.
47
48    Handles both single notifications and batches transparently: a single
49    notification is returned as a one-element list, a batch is unpacked into
50    its members (in order), and anything unsupported yields an empty list.
51
52    The available notifications are listed in the `fishjam.events` module.
53
54    Args:
55        binary: The raw binary data received from the webhook.
56
57    Returns:
58        list[AllowedNotification]: The decoded notifications, in order. Empty
59            when the payload carries no supported notification.
60    """
61    message = ServerMessage().parse(binary)
62    _which, content = betterproto.which_one_of(message, "content")
63
64    if isinstance(content, ServerMessageNotificationBatch):
65        return _unpack_batch(content)
66
67    if isinstance(content, ALLOWED_NOTIFICATIONS):
68        return [content]
69
70    return []

 73def receive_binary(
 74    binary: bytes,
 75) -> Union[AllowedNotification, List[AllowedNotification], None]:
 76    """Transforms a received protobuf notification into a notification instance.
 77
 78    .. deprecated::
 79        Use `decode_server_notifications` instead, which always returns a list
 80        and handles batched payloads with a single, consistent return type.
 81
 82    The available notifications are listed in `fishjam.events` module.
 83
 84    Args:
 85        binary: The raw binary data received from the webhook.
 86
 87    Returns:
 88        AllowedNotification: A single notification when the payload carries one.
 89        list[AllowedNotification]: The unpacked notifications, in order, when the
 90            payload is a batch (webhook batching enabled).
 91        None: When the payload is not a supported notification.
 92    """
 93    warnings.warn(
 94        "receive_binary is deprecated; use decode_server_notifications instead.",
 95        DeprecationWarning,
 96        stacklevel=2,
 97    )
 98
 99    message = ServerMessage().parse(binary)
100    _which, content = betterproto.which_one_of(message, "content")
101
102    if isinstance(content, ServerMessageNotificationBatch):
103        return _unpack_batch(content)
104
105    if isinstance(content, ALLOWED_NOTIFICATIONS):
106        return content
107
108    return None

13@_attrs_define
14class PeerMetadata:
15    """Custom metadata set by the peer
16
17    Example:
18        {'name': 'FishjamUser'}
19
20    """
21
22    additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
23
24    def to_dict(self) -> dict[str, Any]:
25        field_dict: dict[str, Any] = {}
26        field_dict.update(self.additional_properties)
27
28        return field_dict
29
30    @classmethod
31    def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
32        d = dict(src_dict)
33        peer_metadata = cls()
34
35        peer_metadata.additional_properties = d
36        return peer_metadata
37
38    @property
39    def additional_keys(self) -> list[str]:
40        return list(self.additional_properties.keys())
41
42    def __getitem__(self, key: str) -> Any:
43        return self.additional_properties[key]
44
45    def __setitem__(self, key: str, value: Any) -> None:
46        self.additional_properties[key] = value
47
48    def __delitem__(self, key: str) -> None:
49        del self.additional_properties[key]
50
51    def __contains__(self, key: str) -> bool:
52        return key in self.additional_properties

118@dataclass
119class PeerOptions:
120    """Options specific to a WebRTC Peer.
121
122    Attributes:
123        metadata: Peer metadata.
124        subscribe_mode: Configuration of peer's subscribing policy.
125    """
126
127    metadata: dict[str, Any] | None = None
128    """Peer metadata"""
129    subscribe_mode: Literal["auto", "manual"] = "auto"
130    """Configuration of peer's subscribing policy"""

15@_attrs_define
16class PeerOptionsVapi:
17    """Options specific to the VAPI peer
18
19    Attributes:
20        api_key (str): VAPI API key
21        call_id (str): VAPI call ID
22        subscribe_mode (SubscribeMode | Unset): Configuration of peer's subscribing policy
23    """
24
25    api_key: str
26    call_id: str
27    subscribe_mode: SubscribeMode | Unset = UNSET
28
29    def to_dict(self) -> dict[str, Any]:
30        api_key = self.api_key
31
32        call_id = self.call_id
33
34        subscribe_mode: str | Unset = UNSET
35        if not isinstance(self.subscribe_mode, Unset):
36            subscribe_mode = self.subscribe_mode.value
37
38        field_dict: dict[str, Any] = {}
39
40        field_dict.update({
41            "apiKey": api_key,
42            "callId": call_id,
43        })
44        if subscribe_mode is not UNSET:
45            field_dict["subscribeMode"] = subscribe_mode
46
47        return field_dict
48
49    @classmethod
50    def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
51        d = dict(src_dict)
52        api_key = d.pop("apiKey")
53
54        call_id = d.pop("callId")
55
56        _subscribe_mode = d.pop("subscribeMode", UNSET)
57        subscribe_mode: SubscribeMode | Unset
58        if isinstance(_subscribe_mode, Unset):
59            subscribe_mode = UNSET
60        else:
61            subscribe_mode = SubscribeMode(_subscribe_mode)
62
63        peer_options_vapi = cls(
64            api_key=api_key,
65            call_id=call_id,
66            subscribe_mode=subscribe_mode,
67        )
68
69        return peer_options_vapi