Streaming

Real-time progress updates via Server-Sent Events.

Basic Streaming

1from inferencesh import inference, TaskStatus23client = inference(api_key="inf_your_key")45for update in client.tasks.run({6    "app": "infsh/flux",7    "input": {"prompt": "A sunset"}8}, stream=True):9    print(f"Status: {TaskStatus(update['status']).name}")10    11    if update.get("logs"):12        print(f"Logs: {update['logs']}")13    14    if update["status"] == TaskStatus.COMPLETED:15        print(f"Output: {update['output']}")

Stream Existing Task

1# Start task without waiting2task = client.tasks.run(params, wait=False)34# Stream updates later5with client.tasks.stream(task["id"]) as stream:6    for update in stream:7        print(f"Status: {update['status']}")8        if update["status"] == TaskStatus.COMPLETED:9            break1011# Access final result12print(f"Final: {stream.result}")

Partial updates

The API may send incremental task updates as a {data, fields} envelope instead of a full task object. fields lists which keys changed in data.

1# TaskStream iterators receive the inner task object — no manual unwrapping2for update in client.tasks.run(params, stream=True):3    # Same shape as a GET /tasks/:id response4    print(update["status"], update.get("output"))

The Python SDK unwraps {data, fields} lines automatically before yielding. Unlike the JavaScript onPartialUpdate callback, Python iterators do not expose the fields list — compare successive updates if you need to detect which keys changed.

Stream completion and GET reconciliation (Python)

TaskStream and AsyncTaskStream (from client.tasks.run(..., stream=True) and client.tasks.stream()) reconcile against GET /tasks/:id when the NDJSON stream ends without a terminal COMPLETED, FAILED, or CANCELLED event. This covers connection drops, heartbeat-only idle periods (default 30 seconds), and cases where the task finished server-side but the stream never delivered the final update.

1with client.tasks.run(params, stream=True) as stream:2    for update in stream:3        print(update["status"])45# Populated after reconciliation if the stream missed the terminal event6print(stream.result["output"])

Sync and async clients share this behavior.

client.tasks.wait_for_completion() uses the same stream-and-reconcile path internally. If the NDJSON stream ends without a terminal event, the client performs a final GET /tasks/:id before returning or raising.

Pass an optional timeout (seconds) to cap how long the call waits. When the deadline is reached, the client still attempts that final GET — a task that finished while the stream was idle is returned even if the timeout has elapsed. If the task is still non-terminal after the GET, TimeoutError is raised. Failed or cancelled tasks raise RuntimeError immediately from the stream or from GET reconciliation.

1try:2    result = client.tasks.wait_for_completion(task_id, timeout=300)3except TimeoutError:4    print("Task still running after 5 minutes")5except RuntimeError as e:6    print(f"Task failed or cancelled: {e}")

The async client accepts the same timeout keyword: await client.tasks.wait_for_completion(task_id, timeout=300).

Combined Callbacks (JavaScript)

1await client.run(params, {2  onUpdate: (update) => {3    console.log(`Status: ${update.status}`);4  },5  onPartialUpdate: (update, fields) => {6    if (fields.includes('logs')) {7      console.log('New logs:', update.logs);8    }9  }10});

Async Streaming (Python)

1from inferencesh import async_inference, TaskStatus23async def main():4    client = async_inference(api_key="inf_your_key")56    # Stream while running a new task7    async for update in await client.tasks.run(params, stream=True):8        print(f"Status: {update['status']}")910    # Or stream an existing task11    async with client.tasks.stream(task_id) as stream:12        async for update in stream:13            if update["status"] == TaskStatus.COMPLETED:14                print(update["output"])15                break

Reconnection Options

1# Tune SSE performance2client = inference(3    api_key="inf_your_key",4    sse_chunk_size=8192,5    sse_mode="iter_lines"6)78# Or via environment variables9# export INFERENCE_SSE_READ_BYTES=819210# export INFERENCE_SSE_MODE=iter_lines1112# Per-stream reconnect tuning (also on client.tasks.stream())13with client.tasks.run(14    params,15    stream=True,16    auto_reconnect=True,17    max_reconnects=5,18    reconnect_delay_ms=1000,19) as stream:20    for update in stream:21        print(update["status"])

client.run() streams task progress over NDJSON (StreamableManager) by default — it does not accept autoReconnect. Use StreamManager (see below) when you need EventSource reconnect options.

SSE EventSource streams (JavaScript)

Namespace .stream() helpers return a browser EventSource for SSE endpoints — for example client.tasks.stream(taskId), client.chats.stream(chatId), client.flows.stream(flowId), and client.flowRuns.stream(flowRunId).

Failed initial SSE handshakes (HTTP 401/403, OTP, expired tokens) are routed through the client-level onError handler, the same as REST requests. Return the result of retry() from your handler to reconnect after refreshing credentials:

1import { inference, InferenceError } from '@inferencesh/sdk';23const client = inference({4  apiKey: 'inf_your_key',5  onError: async (error, retry) => {6    if (error instanceof InferenceError && error.statusCode === 401) {7      await refreshToken();8      return retry();9    }10    throw error;11  },12});1314const source = await client.tasks.stream('task_abc123');

For lower-level control, StreamManager from @inferencesh/sdk wraps EventSource with autoReconnect, maxReconnects, and reconnectDelayMs for drops after the stream is open:

1import { createHttpClient, StreamManager } from '@inferencesh/sdk';23const http = createHttpClient({ apiKey: 'inf_your_key' });4const manager = new StreamManager({5  createEventSource: () => http.createEventSource('/tasks/task_abc123/stream'),6  autoReconnect: true,7  maxReconnects: 5,8  reconnectDelayMs: 1000,9  onData: (update) => console.log(update.status),10});11manager.start();

→ Error Handling — InferenceError shape and RFC 9457 parsing

Status Constants

1from inferencesh import TaskStatus23TaskStatus.RECEIVED    # 1 - Request received4TaskStatus.QUEUED      # 2 - Waiting in queue5TaskStatus.DISPATCHED  # 3 - Worker assigned (legacy alias: SCHEDULED)6TaskStatus.PREPARING   # 4 - Preparing7TaskStatus.SERVING     # 5 - Serving8TaskStatus.SETTING_UP  # 6 - Setting up9TaskStatus.RUNNING     # 7 - Executing10TaskStatus.CANCELLING  # 8 - Cancelling11TaskStatus.UPLOADING   # 9 - Uploading results12TaskStatus.COMPLETED   # 10 - Done13TaskStatus.FAILED      # 11 - Error14TaskStatus.CANCELLED   # 12 - Cancelled