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"}}

Give each kind of item in a JSON live field a constant that names it: a type literal as above, or any field pydantic's discriminator= points at. Callers label the kinds by it (alternative_label / alternativeLabel), and the app page fills it in for you.

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:

Keys that start with $ are reserved for control frames; no field name can start with one. Apps on SDKs before inferencesh 0.10.1 / @inferencesh/app 0.1.16 send errors as {"error": ...}; the SDK sessions still read those as errors unless the output has a field named error.

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        ))

• Tell the caller what went wrong without ending the stream. await live.error("the provider rejected the voice") sends a $error frame; Live sends one itself for every frame that does not fit the input.
• 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. Cancelling the task ends the loop the same way: what the function returns within the grace period completes the task (see Returning what you have).
• Audio is not continuous. The app page stops sending microphone audio a few seconds after the last sound and resumes when the level rises, so count time from the clock, not from samples received, and don't wait for frames to arrive during silence.
• 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.

Cutting an answer short

A caller plays audio as it arrives, and a provider usually sends it faster than real time, so seconds of an answer can be queued in the caller's player. When the user talks over the answer, stop sending it and tell the caller to drop what it queued:

1await live.clear("audio")      # sends {"$clear": "audio"}

clear takes a live output field. The app page flushes its player when it receives one; SDK callers handle it with on_clear / onClear (below).

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.
• Pass barge-in on. When the provider reports that the user started speaking (input_audio_buffer.speech_started on OpenAI-compatible APIs), call live.clear for the audio field if any of the answer went out.
• End idle sessions yourself. Providers bill every minute a session is open, and silence keeps it open. xai/grok-voice takes an idle_minutes input and ends the session when nobody has spoken or typed for that long.
• 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, dict, or str for plain text12    ...                                                        # control frames go to on_clear= / on_error= when set13await 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. While still waiting, the JavaScript and Swift LiveSession redial automatically when the relay drops the WebSocket early (up to five attempts, each with a fresh credential from POST /sockets/{id}/access). After a page reload or an app relaunch, call client.sockets.open(taskId) to obtain a new credential and dial again. The Python socket needs the async extra: pip install inferencesh[async].

A task whose caller never connects keeps waiting for it, up to 15 minutes. When a session ends before it reached the app (close code 1006: the dial failed or the network dropped), dial again with client.sockets.open(taskId) or cancel the task.

Given the function's schemas, a session maps frames to fields the way Live does on the app side. Pass the output schema and each frame arrives as a field update (on_update / onUpdate, LiveUpdate(field, value) from the Python iterator, session.updates(for: event) in Swift); pass the input schema and send_field / sendField sends a binary frame or a JSON one as the field requires:

1const session = await client.sockets.open(task, { onUpdate: ({ field, value }) => show(field, value) }, {2  inputSchema: fn.input_schema,3  outputSchema: fn.output_schema,4});5session.sendField("audio", micFrame);   // binary: the input's binary live field6session.sendField("voice", "ara");      // JSON: an ordinary field

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.