# Streaming and Live Functions
> An app function can do more than return one result at the end:
**URL:** https://inference.sh/docs/extend/streaming
**Last updated:** 2026-11-07
---
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.
```python Python
from typing import AsyncGenerator
class App(BaseApp):
async def run(self, input_data: AppInput) -> AsyncGenerator[AppOutput, None]:
text = ""
async for piece in produce(input_data):
text += piece
yield AppOutput(text=text) # a snapshot of the output so far
yield AppOutput(text=text, output_meta=OutputMeta( # the result
inputs=[TextMeta(tokens=count_in)], outputs=[TextMeta(tokens=count_out)],
))
```
```javascript Node.js
export class App {
async *run(input) {
let text = "";
for await (const piece of produce(input)) {
text += piece;
yield { text }; // a snapshot of the output so far
}
yield { text, output_meta: { ... } }; // the result
}
}
```
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](/docs/api/sdk/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.
```python Python
from typing import Literal, Union
from pydantic import BaseModel, Field
from inferencesh import BaseAppInput, BaseAppOutput, PCM16, Stream
class UserText(BaseModel):
type: Literal["text"] = "text"
text: str
class Interrupt(BaseModel):
type: Literal["interrupt"] = "interrupt"
class TalkInput(BaseAppInput):
audio: Stream[PCM16(24000)] = Field(description="Microphone audio") # live media
events: Stream[Union[UserText, Interrupt]] = Field(description="Messages") # live typed items
voice: str = "eve" # ordinary
class TalkOutput(BaseAppOutput):
audio: Stream[PCM16(24000)] = Field(description="The assistant's voice")
transcript: str = ""
```
```javascript Node.js
import { z } from "zod";
import { createStreamSchema, pcm16 } from "@inferencesh/app";
const UserText = z.object({ type: z.literal("text"), text: z.string() });
const Interrupt = z.object({ type: z.literal("interrupt") });
export const TalkInput = z.object({
audio: createStreamSchema(z, pcm16(z, 24000)).describe("Microphone audio"), // live media
events: createStreamSchema(z, z.union([UserText, Interrupt])), // live typed items
voice: z.string().default("eve"), // ordinary
});
export const TalkOutput = z.object({
audio: createStreamSchema(z, pcm16(z, 24000)).describe("The assistant's voice"),
transcript: z.string().default(""),
});
```
In the JSON schema a live field is an array marked `"format": "stream"`, the way a file field is marked `"format": "file"`:
```json
"audio": {"type": "array", "format": "stream",
"items": {"type": "string", "format": "binary",
"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:
| Frame | Means |
|---|---|
| binary | one item of the binary live field (e.g. 20 ms of PCM audio) |
| `{"events": {"type": "text", "text": "hi"}}` | one item of a JSON live field |
| `{"voice": "ara"}` | a new value for an ordinary field, while the task runs |
| `{"$error": {"field": ..., "message": ...}}` | app → caller: the app refused a frame, or has something to report; the stream goes on |
| `{"$clear": "audio"}` | app → caller: drop what you have buffered of this live output field |
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.
```python Python
from inferencesh import Live, Socket
class App(BaseApp):
async def talk(self, input_data: TalkInput, socket: Socket) -> TalkOutput:
live = Live(socket, input_data, TalkOutput)
await live.send(transcript="") # speak first: the caller knows you are there
async for update in live: # ends when the caller closes
if update.field == "audio":
await live.send(audio=process(update.value)) # bytes out, a binary frame
elif update.field == "events":
handle(update.value) # a UserText or an Interrupt
# an ordinary field (voice) is already set on input_data
return TalkOutput(transcript=..., output_meta=OutputMeta(
inputs=[AudioMeta(seconds=seconds)], outputs=[],
))
```
```javascript Node.js
import { Live } from "@inferencesh/app";
export class App {
async talk(input, socket) {
const live = new Live(socket, input, TalkInput, TalkOutput);
await live.send({ transcript: "" }); // speak first: the caller knows you are there
for await (const update of live) { // ends when the caller closes
if (update.field === "audio") {
await live.send({ audio: process(update.value) }); // Buffer out, a binary frame
} else if (update.field === "events") {
handle(update.value); // a UserText or an Interrupt
}
}
return { transcript: "...", output_meta: { inputs: [{ type: "audio", seconds }], outputs: [] } };
}
}
```
- **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 is kept as the output of the cancelled task and billed (see [Returning what you have](/docs/extend/cancellation#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:
```python Python
await live.clear("audio") # sends {"$clear": "audio"}
```
```javascript Node.js
await 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](/docs/extend/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
```bash
belt 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:
```python Python
from inferencesh import async_inference
client = async_inference(api_key="...")
task, session = await client.live({
"app": "xai/grok-voice", "function": "talk",
"input": {"voice": "eve"},
})
await session.send(mic_frame) # bytes: an item of the binary live field
await session.send({"events": {"type": "text", "text": "hi"}}) # dict: a JSON frame
async for frame in session: # bytes, dict, or str for plain text
... # control frames go to on_clear= / on_error= when set
await session.close() # the function returns; the task completes
```
```javascript JavaScript
import { inference } from "@inferencesh/sdk";
const client = inference({ apiKey: "..." });
const { task, session } = await client.live(
{ app: "xai/grok-voice", function: "talk", input: { voice: "eve" } },
{
onState: (state) => console.log(state), // connecting → waiting → live → ended
onBinary: (pcm) => play(pcm),
onPatch: (patch) => console.log(patch), // {"user_text": "..."}
onClear: (field) => player.flush(), // the app cut an answer short
onError: (field, message) => console.warn(field, message),
onText: (text) => console.log(text), // text the app sent as text, not a patch
},
);
session.sendBinary(micFrame);
session.sendPatch({ events: { type: "text", text: "hi" } });
session.close();
```
```swift Swift
import InferenceSDK
let client = InferenceClient(apiKey: "...")
let (task, session) = try await client.live(ApiAppRunRequest(
app: "xai/grok-voice", input: ["voice": "eve"], function: "talk"
))
Task {
for await event in session.events { // ends when the session does
switch event {
case .state(let state): print(state) // connecting → waiting → live → ended
case .binary(let pcm): player.play(pcm)
case .patch(let patch): print(patch) // ["user_text": "..."]
case .clear: player.flush() // the app cut an answer short
case .error(let field, let message): print(field ?? "-", message)
case .text(let text): print(text) // text the app sent as text, not a patch
}
}
}
session.sendBinary(micFrame)
session.sendPatch(["events": ["type": "text", "text": "hi"]])
session.close()
```
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:
```javascript
const session = await client.sockets.open(task, { onUpdate: ({ field, value }) => show(field, value) }, {
inputSchema: fn.input_schema,
outputSchema: fn.output_schema,
});
session.sendField("audio", micFrame); // binary: the input's binary live field
session.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`.