Streaming and Live Functions

An app function can do more than return one result at the end:

• Streaming output: the function yields partial results as it works, and callers see each one as a task update. Use it for anything that produces results incrementally: LLM text, transcripts, progress.
• Live (stream) functions: the function holds a two-way socket with its caller for the life of the task. Audio, video frames and messages flow both ways while it runs. Use it for voice agents, live dictation, anything interactive.

Both are ordinary functions of your app with ordinary input and output schemas. Nothing else changes: deploys, secrets, pricing and output_meta work the same.

Streaming output

Make the function an async generator. Every yield replaces the task's output; the last yield is the result.

1from typing import AsyncGenerator23class App(BaseApp):4    async def run(self, input_data: AppInput) -> AsyncGenerator[AppOutput, None]:5        text = ""6        async for piece in produce(input_data):7            text += piece8            yield AppOutput(text=text)                      # a snapshot of the output so far9        yield AppOutput(text=text, output_meta=OutputMeta(  # the result10            inputs=[TextMeta(tokens=count_in)], outputs=[TextMeta(tokens=count_out)],11        ))

Rules:

• Yield the whole output so far, not the new part. Each yield replaces the previous one.
• Put output_meta on the last yield. The last yield is the result, which is what gets billed.
• Yield at least once. A generator that ends without yielding fails with function 'run' yielded no output.
• Keep snapshots small. Each one resends the whole output, and intermediate ones can be coalesced (they are sent at most every 100 ms). Put detail that only matters at the end, like word timestamps, in the last yield.
• Don't block the event loop. Pull from a blocking model generator with await asyncio.to_thread(next, it, None).

Callers receive the snapshots as task updates: stream=True in the Python SDK, onUpdate in JavaScript, or GET /tasks/{id}/stream over SSE. See Streaming.

Live functions

A live function takes a socket after its input. The platform opens the socket when the task starts and pairs it with the caller's end through a relay, so the app and the caller exchange frames directly for as long as the task runs.

Declaring what the socket carries

What travels over the socket is declared in the function's input and output schemas, the same way the request body is. Some fields are live: their values arrive over the socket while the task runs instead of in the request body.

1from typing import Literal, Union2from pydantic import BaseModel, Field3from inferencesh import BaseAppInput, BaseAppOutput, PCM16, Stream45class UserText(BaseModel):6    type: Literal["text"] = "text"7    text: str89class Interrupt(BaseModel):10    type: Literal["interrupt"] = "interrupt"1112class TalkInput(BaseAppInput):13    audio: Stream[PCM16(24000)] = Field(description="Microphone audio")       # live media14    events: Stream[Union[UserText, Interrupt]] = Field(description="Messages") # live typed items15    voice: str = "eve"                                                           # ordinary1617class TalkOutput(BaseAppOutput):18    audio: Stream[PCM16(24000)] = Field(description="The assistant's voice")19    transcript: str = ""

In the JSON schema a live field is an array marked "format": "stream", the way a file field is marked "format": "file":

1"audio": {"type": "array", "format": "stream",2          "items": {"type": "string", "format": "binary",3                    "contentMediaType": "audio/pcm;format=s16le;rate=24000;channels=1"}}

Live fields are never required, and the request body carries only the ordinary fields. A schema has at most one binary live field per direction, because a binary frame carries no field name.

On the wire:

Writing the function

Live reads and writes the socket through your models: incoming frames become typed updates, ordinary fields are applied to the input as they change, and send turns output fields into frames.

1from inferencesh import Live, Socket23class App(BaseApp):4    async def talk(self, input_data: TalkInput, socket: Socket) -> TalkOutput:5        live = Live(socket, input_data, TalkOutput)6        await live.send(transcript="")             # speak first: the caller knows you are there78        async for update in live:                  # ends when the caller closes9            if update.field == "audio":10                await live.send(audio=process(update.value))   # bytes out, a binary frame11            elif update.field == "events":12                handle(update.value)               # a UserText or an Interrupt13            # an ordinary field (voice) is already set on input_data1415        return TalkOutput(transcript=..., output_meta=OutputMeta(16            inputs=[AudioMeta(seconds=seconds)], outputs=[],17        ))

• The function name is yours. Any function that takes a socket is a live function; set default_function in inf.yml if it is the main one.
• Send a frame first. Until the app's first frame the caller is waiting (the worker may still be starting), so send one as soon as you are ready.
• The caller closing ends the loop. Returning ends the stream for the caller. Either way, what the function returns is the task's result and is billed through its output_meta.
• Slow readers lose old audio, not new. If the app reads more slowly than binary frames arrive, the kernel keeps the newest 256 and drops the oldest (socket.dropped counts them). JSON frames are never dropped.

A live function can also be an async generator. Its yields are task updates, exactly as in streaming output, while the socket stays the low-latency channel. infsh/live-transcribe does this: the transcript goes over the socket as you speak, and each settled phrase is also a snapshot of the task's output.

Wrapping a realtime API

Relaying a provider's realtime WebSocket (xAI Grok Voice, OpenAI realtime) is the common case. Open the provider's socket in the function, pump both directions concurrently, and translate at the edges. Keep the audio as raw PCM end to end where the provider allows it. xai/grok-voice is a complete example.

• Bill the session, not the call. Report what the provider bills in output_meta, for example session seconds as an audio input, and price it per minute. See Usage-Based Pricing.
• A provider closing the session is not a failure. When it ends a session it had accepted (an idle timeout, for instance), return the conversation so far with its output_meta. A failed task bills nothing, but the provider still bills you.

Testing locally

1belt app test --stream --function talk --stream-addr 127.0.0.1:8765

This serves a browser console with a microphone, playback and a frame log, plus a WebSocket at /ws for scripted clients. Each connection is a fresh call; yields print as [progress] and the return value as [result].

Browsers only allow the microphone on HTTPS or localhost. If the dev server runs on another machine, tunnel it (ssh -L 8765:127.0.0.1:8765 ...) and open http://localhost:8765.

Calling a live function

The run response of a live function carries its socket (task.socket: where to dial and a short-lived credential). The SDKs start the task and dial it in one call:

1from inferencesh import async_inference23client = async_inference(api_key="...")4task, session = await client.live({5    "app": "xai/grok-voice", "function": "talk",6    "input": {"voice": "eve"},7})89await session.send(mic_frame)                                  # bytes: an item of the binary live field10await session.send({"events": {"type": "text", "text": "hi"}}) # dict: a JSON frame11async for frame in session:                                    # bytes or dict, as the app sends them12    ...13await session.close()                                          # the function returns; the task completes

The session is waiting until the app's first frame, which includes a cold start, and gives up if the task ends first. After a page reload, client.sockets.open(taskId, handlers) asks for a fresh credential and dials again. The Python socket needs the async extra: pip install inferencesh[async].

Callers can read a function's schema to build a UI: split_live_schema / splitLiveSchema separate the ordinary fields (a form) from the live ones (controls), and pcm_format / pcmFormat reads a PCM field's sample rate. The app page on inference.sh does exactly this: a PCM input field becomes a microphone, a JSON live field a message sender, and a PCM output field is played back.

Over plain HTTP, the socket is POST /apps/run (the response's socket), then a WebSocket to socket.url?access_token=socket.token.