Streaming Events

from orbiter import Agent, run

agent = Agent(name="assistant", model="gpt-4o", instructions="You are helpful.")

# Basic streaming — TextEvent and ToolCallEvent only
async for event in run.stream(agent, "What is 2 + 2?"):
    if event.type == "text":
        print(event.text, end="", flush=True)

# Detailed streaming — all event types
async for event in run.stream(agent, "What is 2 + 2?", detailed=True):
    match event.type:
        case "text":
            print(event.text, end="", flush=True)
        case "step":
            print(f"\n[Step {event.step_number} {event.status}]")
        case "tool_call":
            print(f"\n[Calling tool: {event.tool_name}]")
        case "tool_result":
            print(f"\n[Tool result: {event.result[:100]}]")
        case "error":
            print(f"\n[Error: {event.error}]")

StreamEvent = (
    TextEvent | ToolCallEvent | StepEvent | ToolResultEvent
    | ReasoningEvent | ErrorEvent | StatusEvent | UsageEvent
)

async for event in run.stream(agent, "Hello"):
    if event.type == "text":
        print(event.text, end="", flush=True)

if event.type == "tool_call":
    print(f"Calling tool: {event.tool_name} (id: {event.tool_call_id})")

if event.type == "step":
    if event.status == "started":
        print(f"Step {event.step_number} starting...")
    else:
        duration = event.completed_at - event.started_at
        print(f"Step {event.step_number} completed in {duration:.2f}s")

if event.type == "tool_result":
    status = "OK" if event.success else "FAILED"
    print(f"[{status}] {event.tool_name}: {event.result[:100]}")
    if event.error:
        print(f"  Error: {event.error}")

if event.type == "reasoning":
    print(f"[Thinking] {event.text}")

if event.type == "error":
    print(f"[ERROR] {event.error_type}: {event.error}")
    if event.step_number:
        print(f"  Occurred at step {event.step_number}")

if event.type == "status":
    print(f"[{event.status.upper()}] {event.agent_name}: {event.message}")

if event.type == "usage":
    u = event.usage
    print(f"Tokens: {u.input_tokens} in / {u.output_tokens} out / {u.total_tokens} total")
    print(f"  Model: {event.model}, Step: {event.step_number}")

StatusEvent(status="starting")           # Once at start
├── StepEvent(status="started")          # Per step
├── TextEvent(text="...")                # Text chunks from LLM
├── UsageEvent(...)                      # After LLM stream completes
├── ToolCallEvent(...)                   # Per tool call (if any)
├── ToolResultEvent(...)                 # Per tool result (if detailed)
├── StepEvent(status="completed")        # Step done
├── [next step repeats...]
StatusEvent(status="completed")          # Once at end

ErrorEvent(...)                          # Always emitted
StatusEvent(status="error")              # Only when detailed=True

# Only receive text and tool results
async for event in run.stream(
    agent, "Search for Python docs",
    detailed=True,
    event_types={"text", "tool_result"},
):
    match event.type:
        case "text":
            print(event.text, end="")
        case "tool_result":
            print(f"\n[Tool: {event.tool_name}] {event.result[:200]}")

# Only receive status updates
async for event in run.stream(
    agent, "Do the task",
    detailed=True,
    event_types={"status", "error"},
):
    print(f"[{event.type}] {getattr(event, 'status', event.error)}")

from orbiter import Agent, Swarm, run

researcher = Agent(name="researcher", model="gpt-4o", instructions="Research topics.")
writer = Agent(name="writer", model="gpt-4o", instructions="Write articles.")

swarm = Swarm(
    agents=[researcher, writer],
    flow="researcher >> writer",
    mode="workflow",
)

async for event in run.stream(swarm, "Write about Python", detailed=True):
    match event.type:
        case "status":
            print(f"\n--- [{event.agent_name}] {event.message} ---")
        case "text":
            print(event.text, end="", flush=True)
        case "step":
            print(f"\n[{event.agent_name}] Step {event.step_number} {event.status}")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from orbiter import Agent, run

app = FastAPI()
agent = Agent(name="assistant", model="gpt-4o", instructions="You are helpful.")

@app.get("/chat")
async def chat(message: str):
    async def event_generator():
        async for event in run.stream(agent, message, detailed=True):
            data = event.model_dump_json()
            yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )

@app.get("/chat")
async def chat(message: str):
    async def event_generator():
        async for event in run.stream(agent, message, detailed=True):
            data = event.model_dump_json()
            yield f"event: {event.type}\ndata: {data}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )

from django.http import StreamingHttpResponse
from orbiter import Agent, run
import asyncio

agent = Agent(name="assistant", model="gpt-4o", instructions="You are helpful.")

async def chat_view(request):
    message = request.GET.get("message", "")

    async def event_generator():
        async for event in run.stream(agent, message, detailed=True):
            data = event.model_dump_json()
            yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingHttpResponse(
        event_generator(),
        content_type="text/event-stream",
    )

const source = new EventSource("/chat?message=Hello");

source.addEventListener("text", (e) => {
  const event = JSON.parse(e.data);
  document.getElementById("output").textContent += event.text;
});

source.addEventListener("tool_call", (e) => {
  const event = JSON.parse(e.data);
  console.log(`Calling tool: ${event.tool_name}`);
});

source.addEventListener("step", (e) => {
  const event = JSON.parse(e.data);
  console.log(`Step ${event.step_number}: ${event.status}`);
});

source.addEventListener("error", (e) => {
  const event = JSON.parse(e.data);
  console.error(`Error: ${event.error}`);
  source.close();
});

source.addEventListener("status", (e) => {
  const event = JSON.parse(e.data);
  if (event.status === "completed") {
    source.close();
  }
});

async function streamChat(message) {
  const response = await fetch(`/chat?message=${encodeURIComponent(message)}`);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n\n");
    buffer = lines.pop(); // Keep incomplete chunk

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const data = line.slice(6);
        if (data === "[DONE]") return;
        const event = JSON.parse(data);
        handleEvent(event);
      }
    }
  }
}

function handleEvent(event) {
  switch (event.type) {
    case "text":
      process.stdout.write(event.text);
      break;
    case "tool_call":
      console.log(`\nTool: ${event.tool_name}`);
      break;
    case "tool_result":
      console.log(`Result: ${event.result.slice(0, 100)}`);
      break;
    case "status":
      console.log(`\n[${event.status}] ${event.message}`);
      break;
    case "error":
      console.error(`Error: ${event.error}`);
      break;
  }
}

event = TextEvent(text="Hello", agent_name="assistant")

# To dict
event.model_dump()
# {'type': 'text', 'text': 'Hello', 'agent_name': 'assistant'}

# To JSON string
event.model_dump_json()
# '{"type":"text","text":"Hello","agent_name":"assistant"}'

import json
from orbiter.types import (
    TextEvent, ToolCallEvent, StepEvent, ToolResultEvent,
    ReasoningEvent, ErrorEvent, StatusEvent, UsageEvent,
)

EVENT_TYPES = {
    "text": TextEvent,
    "tool_call": ToolCallEvent,
    "step": StepEvent,
    "tool_result": ToolResultEvent,
    "reasoning": ReasoningEvent,
    "error": ErrorEvent,
    "status": StatusEvent,
    "usage": UsageEvent,
}

def deserialize_event(data: dict) -> StreamEvent:
    event_cls = EVENT_TYPES[data["type"]]
    return event_cls(**data)

# Example
raw = json.loads('{"type":"text","text":"Hello","agent_name":"assistant"}')
event = deserialize_event(raw)  # TextEvent(type='text', text='Hello', agent_name='assistant')

from orbiter.distributed import distributed

# Submit task and stream events
handle = await distributed(agent, "Hello", redis_url="redis://localhost", detailed=True)

# Live streaming via Redis Pub/Sub
async for event in handle.stream():
    match event.type:
        case "text":
            print(event.text, end="")
        case "status" if event.status == "completed":
            print("\nDone!")

# Or wait for result
result = await handle.result()