Stream object does not have direct access to all WebRTC connections when using Gradio UI

[marcusvaltonen]

Hi,

I was struggling to find a good title, so let me explain with a minimal working example:

from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from fastrtc import ReplyOnPause, Stream
import gradio as gr
import numpy as np


def echo(audio: tuple[int, np.ndarray]):
    yield audio

stream = Stream(
    handler=ReplyOnPause(echo),
    modality="audio",
    mode="send-receive",
)

app = FastAPI()
stream.mount(app)

@app.get("/")
async def _():
    return RedirectResponse(url="/ui")

@app.get("/connections")
async def _():
    return list(stream.connections.keys())

app = gr.mount_gradio_app(app, stream.ui, path="/ui")

import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)

Here we set up a dummy stream and add basic API endpoints, in particular /connections which should return all WebRTC connections. The Streams UI components are mounted to /ui and going from root redirects you here.

Now, the problem is that if I connect through the gradio interface, and then call my /connections endpoint, it will return empty. This design isn't great.

Why is this happening?
I had to dig a little and realized it is because the WebRTC component is itself a Stream object and accessible through stream.webrtc_component. This means that I can modify the code above to

@app.get("/connections")
async def _():
    all_connections = list(stream.connections.keys())
    if stream.webrtc_component:
        all_connections.extend(list(stream.webrtc_component.connections.keys()))
    return all_connections

which does indeed return all webrtc_ids regardless of how they were created (directly from an API call or through the Gradio UI).

Why this is a problem?
This is of course a simple issue with a simple fix, but it is hard to understand why it happens. Furthermore, since most API calls rely on webrtc_ids, e.g., stream.stream_outputs(webrtc_id) that is called in many demo applications with additional_outputs configured, this type of approach would have to be implemented for all endpoints. It will just result in boilerplate code that is bloating your code.

Possible solutions
Some sort of way to handle this internally in the stream class and expose the WebRTC streams of the stream.webrtc_component directly.

I could prepare an MR for this, should you agree with the above. Or perhaps this is not the intended way of usage? I do like it myself...

[marcusvaltonen]

I linked a more scaled down version that should do the same thing in #405.

[marcusvaltonen]

I'd like to thank @DeveloperViraj for helping out; however, the current MR only fixes the problem raised above, not the underlying problem. Consider now that I want to eavesdrop on an output stream (all security-related aspects put aside!): I call my /connections endpoint and I get the active connections and pick one of them, say webrtc_id=foo. I then call my /outputs endpoint as in most demos:

@app.get("/outputs")
def _(webrtc_id: str):
    async def output_stream():
        import json

        async for output in stream.output_stream(webrtc_id):
            s = json.dumps(output.args[0])
            yield f"event: output\ndata: {s}\n\n"

    return StreamingResponse(output_stream(), media_type="text/event-stream")

This would still not work, since stream.connections doesn't have webrtc_id=foo. This work around would still be necessary:

@app.get("/outputs")
def _(webrtc_id: str):

    if webrtc_id in stream.connections:
        stream_object = stream
    elif webrtc_id in stream.webrtc_component.connections:
        stream_object = stream.webrtc_component
    else:
        <probably rasie an HTTPException>

    async def output_stream():
        import json

        async for output in stream_object.output_stream(webrtc_id):
            s = json.dumps(output.args[0])
            yield f"event: output\ndata: {s}\n\n"

    return StreamingResponse(output_stream(), media_type="text/event-stream")

Note: There are other issues with eavesdropping on a Gradio UI component as it is consuming itself - but this is aside from this discussion topic.

Possible mitigations
One way would be to move connections -> _connections internally and have

@property
def connections(self) -> :
    """Return all connections."""
       connections = self.connections
        if self.webrtc_component is not None:
            connections += self.webrtc_component.connections

Alternatively, one could create a WebRTCConnections class that could handle more advanced logic later on, including e.g., security aspects. This way one could keep self.connections.append() if this class handles it internally.

[DeveloperViraj]

Thanks for the clarification, @marcusvaltonen
That makes a lot of sense, exposing the connections via a property or a WebRTCConnections handler definitely feels cleaner than the workaround in /outputs.

The @Property pattern:

@Property
def connections(self):

seems like a nice, backward-compatible way to handle it.

Happy to give that a try if you and @freddyaboulton think it fits with the overall FastRTC direction.