Docs/API Reference/Streaming

Streaming

Enable real-time streaming to receive responses as they're generated using Server-Sent Events (SSE).

#Basic Streaming

Set stream=True to enable streaming responses.

Python

from openai import OpenAI

client = OpenAI(
    api_key="your-alphagent-api-key",
    base_url="https://api.alphagent.co/v1"
)

stream = client.responses.create(
    model="alphagent-smart",
    input="Analyze TSLA fundamentals",
    stream=True
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        print("\nDone!")

TypeScript

import OpenAI from 'openai';

const client = new OpenAI({
  apiKey: process.env.ALPHAGENT_API_KEY,
  baseURL: 'https://api.alphagent.co/v1',
});

const stream = await client.responses.create({
  model: 'alphagent-smart',
  input: 'Analyze TSLA fundamentals',
  stream: true,
});

for await (const event of stream) {
  if (event.type === 'response.output_text.delta') {
    process.stdout.write(event.delta);
  }
}

#Event Types

All event types you may receive during streaming, organized by category:

Text Events

Event	Description
response.output_text.delta	Text chunk received
response.output_text.done	Text generation complete

Lifecycle Events

Event	Description
response.created	Response object created
response.output_item.added	New output item started (message, function_call, or function_call_output)
response.content_part.added	Content part started within message
response.content_part.done	Content part finished
response.output_item.done	Output item finished
response.completed	Response fully complete with metadata

Function Call Events

Event	Description
response.function_call_arguments.delta	Tool arguments chunk
response.function_call_arguments.done	Tool arguments complete

Verification Events

Event	Description
verification.started	Verification process started
verification.done	Verification complete with results

Error Events

Event	Description
error	Stream error occurred

#Function Call Lifecycle

When agents use tools, you'll receive events for the complete function call lifecycle. Click each step to see the event payload.

Python Example - Handling Function Calls

Python

tool_calls = {}  # track by call_id

for event in stream:
    if event.type == "response.output_item.added":
        item = event.item
        if item.get("type") == "function_call":
            # Tool call started
            call_id = item["call_id"]
            tool_calls[call_id] = {
                "name": item["name"],
                "arguments": "",
                "status": "in_progress"
            }
        elif item.get("type") == "function_call_output":
            # Tool result received
            call_id = item["call_id"]
            if call_id in tool_calls:
                tool_calls[call_id]["output"] = item["output"]
                tool_calls[call_id]["status"] = "completed"

    elif event.type == "response.function_call_arguments.delta":
        call_id = event.call_id
        if call_id in tool_calls:
            tool_calls[call_id]["arguments"] += event.delta

    elif event.type == "response.function_call_arguments.done":
        call_id = event.call_id
        if call_id in tool_calls:
            tool_calls[call_id]["arguments"] = event.arguments

#Code Execution Lifecycle

When agents execute code, you'll receive events for the complete execution lifecycle. Click each step to see the event payload.

Output Fields

Field	Type	Description
stdout	string	Standard output from execution
stderr	string	Standard error output
return_code	number	Exit code (0 = success)
files	array	Generated files with `file_id` and optional `filename`

Python Example - Handling Code Execution

Python

code_executions = {}  # track by item_id

for event in stream:
    if event.type == "response.code_execution.added":
        exec_id = event.item_id
        code_executions[exec_id] = {
            "name": event.name,
            "input": "",
            "status": "generating"
        }

    elif event.type == "response.code_execution.input_delta":
        exec_id = event.item_id
        if exec_id in code_executions:
            code_executions[exec_id]["input"] += event.delta

    elif event.type == "response.code_execution.started":
        exec_id = event.item_id
        if exec_id in code_executions:
            code_executions[exec_id]["status"] = "executing"

    elif event.type == "response.code_execution.output":
        exec_id = event.item_id
        if exec_id in code_executions:
            code_executions[exec_id]["status"] = "completed"
            code_executions[exec_id]["output"] = {
                "stdout": event.stdout,
                "stderr": event.stderr,
                "return_code": event.return_code,
                "files": event.files or []
            }

#File Access

Files created during code execution (charts, exports, etc.) are returned as file_id references. Use the /v1/files/{file_id} endpoint to download them.

File Reference Structure

JSON

{
  "files": [
    { "file_id": "file_abc123", "filename": "chart.png" },
    { "file_id": "file_def456", "filename": "data.csv" }
  ]
}

Downloading Files

cURL

curl https://api.alphagent.co/v1/files/file_abc123 \
  -H "Authorization: Bearer $ALPHAGENT_API_KEY" \
  --output chart.png

Python

import requests

def download_file(file_id: str, api_key: str) -> bytes:
    response = requests.get(
        f"https://api.alphagent.co/v1/files/{file_id}",
        headers={"Authorization": f"Bearer {api_key}"}
    )
    return response.content

# After receiving code execution output
for file_ref in output.get("files", []):
    content = download_file(file_ref["file_id"], api_key)
    filename = file_ref.get("filename", file_ref["file_id"])
    with open(filename, "wb") as f:
        f.write(content)

Displaying in HTML

HTML

<!-- Image files -->
<img src="https://api.alphagent.co/v1/files/file_abc123" />

<!-- PDF files -->
<embed src="https://api.alphagent.co/v1/files/file_def456" type="application/pdf" />

File Fields

Field	Type	Description
file_id	string	Unique identifier for downloading via `/v1/files/{file_id}`
filename	string (optional)	Original filename if available (e.g., "chart.png")

#SSE Wire Format

The raw Server-Sent Events format sent over the wire. Each event is prefixed with data: and separated by double newlines. The stream ends with [DONE].

Raw SSE

data: {"type":"response.created","response":{...}}

data: {"type":"response.output_text.delta","delta":"Hello"}

data: {"type":"response.completed","response":{...}}

data: [DONE]

#Common Pattern

For most use cases, you only need to handle response.output_text.delta for streaming text and response.completed for the final response with metadata.

Python

full_text = ""
final_response = None

for event in stream:
    if event.type == "response.output_text.delta":
        full_text += event.delta
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        final_response = event.response

# Access verification after streaming completes
if final_response and final_response.metadata:
    verification = final_response.metadata.get("verification")
    print("Confidence:", verification.get("confidence"))

Response Format

Verification