feat(rtc): add LocalParticipant.send_bytes (#858)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: 1egoman

livekit-rtc/livekit/rtc/participant.py

@@ -974,6 +974,56 @@ async def send_file(
 
         return _byte_stream_info_from_proto(cb.send_file.info)
 
+    async def send_bytes(
+        self,
+        data: bytes,
+        *,
+        topic: str = "",
+        name: str | None = None,
+        mime_type: str | None = None,
+        destination_identities: Optional[List[str]] = None,
+        attributes: Optional[Dict[str, str]] = None,
+        stream_id: str | None = None,
+        compress: bool = True,
+    ) -> ByteStreamInfo:
+        """
+        Sends an in-memory buffer as a complete byte stream in one call.
+        Unlike stream_bytes(), the whole payload is known up front, so short payloads can be
+        sent as a single packet and larger ones compressed, when every recipient supports it.
+        `name` defaults to "unknown" and `mime_type` to "application/octet-stream".
+        """
+        req = proto_ffi.FfiRequest()
+        req.send_bytes.local_participant_handle = self._ffi_handle.handle
+        req.send_bytes.bytes = data
+        options = req.send_bytes.options
+        options.topic = topic
+        if name is not None:
+            options.name = name
+        if mime_type is not None:
+            options.mime_type = mime_type
+        if attributes:
+            options.attributes.update(attributes)
+        if destination_identities:
+            options.destination_identities.extend(destination_identities)
+        if stream_id is not None:
+            options.id = stream_id
+        options.compress = compress
+        options.sender_identity = self.identity
+
+        queue = FfiClient.instance.queue.subscribe()
+        try:
+            resp = FfiClient.instance.request(req)
+            cb: proto_ffi.FfiEvent = await queue.wait_for(
+                lambda e: e.send_bytes.async_id == resp.send_bytes.async_id
+            )
+        finally:
+            FfiClient.instance.queue.unsubscribe(queue)
+
+        if cb.send_bytes.HasField("error"):
+            raise StreamError(cb.send_bytes.error.description)
+
+        return _byte_stream_info_from_proto(cb.send_bytes.info)
+
     async def publish_data_track(
         self,
         *,

tests/rtc/test_e2e_data_streams.py

@@ -192,6 +192,40 @@ async def read() -> None:
         await asyncio.gather(receiver.disconnect(), sender.disconnect())
 
 
+@pytest.mark.asyncio
+@skip_if_no_credentials()  # type: ignore[untyped-decorator]
+async def test_send_bytes_small() -> None:
+    """Small in-memory buffer via send_bytes: info assertions + round trip."""
+    receiver, sender = await connect_rooms(unique_room_name("ds-send-bytes-small"))
+    try:
+        bytes_to_send = b"hello world"
+        received: asyncio.Future[bytes] = asyncio.get_event_loop().create_future()
+
+        def handler(reader: rtc.ByteStreamReader, participant_identity: str) -> None:
+            assert participant_identity == "sender"
+
+            async def read() -> None:
+                chunks = [chunk async for chunk in reader]
+                received.set_result(b"".join(chunks))
+
+            asyncio.create_task(read())
+
+        receiver.register_byte_stream_handler("some-topic", handler)
+
+        info = await sender.local_participant.send_bytes(
+            bytes_to_send, topic="some-topic", name="hello.bin"
+        )
+        assert info.stream_id
+        assert info.topic == "some-topic"
+        assert info.name == "hello.bin"
+        assert info.size == len(bytes_to_send)
+        assert info.mime_type == "application/octet-stream"
+
+        assert await asyncio.wait_for(received, timeout=5.0) == bytes_to_send
+    finally:
+        await asyncio.gather(receiver.disconnect(), sender.disconnect())
+
+
 @pytest.mark.asyncio
 @skip_if_no_credentials()  # type: ignore[untyped-decorator]
 async def test_send_text_large_compressible() -> None: