Streaming (agentvisor.stream)

Emit stream chunks during agent execution for real-time delivery via the /threads/{id}/runs/{runId}/stream SSE endpoint. The guest runtime injects the authoritative thread_id and run_id before forwarding to the host — agent code never sets them directly.

Outside of a run (for example, during schema extraction or a standalone script with no run_id in the environment), emit_chunk() silently does nothing rather than raising.

Usage

from agentvisor import stream

# Emit an LLM token as it streams
for token in llm.stream("Hello"):
    stream.emit_token(token)

# Emit a chat message
stream.emit_message("assistant", "I'm working on that...")

# Emit a state update
stream.emit_state({"step": 2, "progress": 0.5})

# Emit arbitrary application-defined data
stream.emit_chunk({"event": "tool_started", "tool": "search"}, chunk_type="custom")

API Reference

Functions

emit_chunk()

def emit_chunk(data: dict[str, Any], chunk_type: str = "custom") -> None

The base function every convenience wrapper below calls. data must be JSON-serializable. Common chunk_type values: "token", "state", "message", "custom".

emit_token()

def emit_token(token: str, **kwargs: Any) -> None

Equivalent to emit_chunk({"token": token}, chunk_type="token").

emit_message()

def emit_message(role: str, content: str, **kwargs: Any) -> None

Equivalent to emit_chunk({"role": role, "content": content}, chunk_type="message").

emit_state()

def emit_state(state: dict[str, Any], **kwargs: Any) -> None

Equivalent to emit_chunk(state, chunk_type="state").

Error Handling

Like other SDK operations, stream calls are wrapped with fire-and-forget retry semantics — see Fire-and-Forget Operations in the Retry reference. A transient failure is retried and then silently suppressed rather than raised, so a dropped stream chunk never crashes agent execution; a permanent error (e.g. INVALID_ARGUMENT) is still raised.