Streaming

The SDK provides comprehensive streaming support for real-time response handling.

Basic Streaming

Platform Mode

const stream = aw.streamComplete({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Write a story' }]
});

for await (const chunk of stream) {
  if (chunk.type === 'text_delta') {
    process.stdout.write(chunk.text);
  }
}

Direct Provider Mode

import { createProvider } from '@agentic-work/sdk/providers';

const provider = createProvider({
  type: 'openai',
  apiKey: process.env.OPENAI_API_KEY
});

const stream = provider.stream({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Explain quantum computing' }]
});

for await (const chunk of stream) {
  // Handle different chunk types
}

Stream Chunk Types

Text Delta

Text content from the model:

interface TextDelta {
  type: 'text_delta';
  text: string;
}

Tool Call Delta

Partial tool call information (streamed incrementally):

interface ToolCallDelta {
  type: 'tool_call_delta';
  toolCall: {
    index: number;
    id?: string;
    name?: string;
    arguments?: string;  // JSON string, accumulated
  };
}

Stream Done

Completion signal:

interface StreamDone {
  type: 'done';
  finishReason?: 'stop' | 'length' | 'tool_calls' | 'content_filter';
  usage?: {
    promptTokens: number;
    completionTokens: number;
    totalTokens: number;
  };
}

Stream Error

Error during streaming:

interface StreamError {
  type: 'error';
  error: string;
}

Complete Handler Example

let fullText = '';
const toolCalls: Map<number, any> = new Map();

for await (const chunk of stream) {
  switch (chunk.type) {
    case 'text_delta':
      fullText += chunk.text;
      process.stdout.write(chunk.text);
      break;

    case 'tool_call_delta':
      const existing = toolCalls.get(chunk.toolCall.index) || {};
      toolCalls.set(chunk.toolCall.index, {
        ...existing,
        id: chunk.toolCall.id ?? existing.id,
        name: chunk.toolCall.name ?? existing.name,
        arguments: (existing.arguments || '') + (chunk.toolCall.arguments || '')
      });
      break;

    case 'done':
      console.log('\n--- Stream Complete ---');
      console.log('Finish reason:', chunk.finishReason);
      if (chunk.usage) {
        console.log('Tokens:', chunk.usage.totalTokens);
      }
      break;

    case 'error':
      console.error('Stream error:', chunk.error);
      break;
  }
}

// Process accumulated tool calls
for (const [index, call] of toolCalls) {
  console.log(`Tool ${index}:`, call.name, JSON.parse(call.arguments));
}

Agent Streaming

Agents stream text output while handling tools internally:

const agent = aw.createAgent({
  model: 'gpt-4o',
  tools: aw.getMCPTools()
});

// Text is yielded as it's generated
// Tool execution happens automatically between yields
for await (const text of agent.run({
  prompt: 'Create a Python script that fetches weather data'
})) {
  process.stdout.write(text);
}

Streaming with Callbacks

For more granular control over agent execution:

const agent = aw.createAgent({
  model: 'gpt-4o',
  tools: aw.getMCPTools(),

  onToolCall: (toolCall) => {
    console.log(`\n[Calling ${toolCall.name}...]`);
  },

  onToolResult: (result) => {
    if (result.isError) {
      console.log(`[Tool error: ${result.content}]`);
    } else {
      console.log(`[Tool complete]`);
    }
  },

  onText: (text) => {
    // Called for each text chunk
    process.stdout.write(text);
  }
});

// The run() generator also yields text
for await (const text of agent.run({ prompt: 'Help me debug this code' })) {
  // Text is also available here
}

Cancellation

Cancel a streaming request:

const controller = new AbortController();

// Cancel after 5 seconds
setTimeout(() => controller.abort(), 5000);

try {
  const stream = aw.streamComplete({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: 'Write a very long story' }]
  });

  for await (const chunk of stream) {
    if (controller.signal.aborted) {
      break;
    }
    if (chunk.type === 'text_delta') {
      process.stdout.write(chunk.text);
    }
  }
} catch (error) {
  if (error.name === 'AbortError') {
    console.log('\nStream cancelled');
  }
}

Rich Stream Events (CLI/UI)

For building interactive UIs:

type StreamEventType =
  | 'text'
  | 'thinking'
  | 'tool_start'
  | 'tool_progress'
  | 'tool_complete'
  | 'tool_error'
  | 'usage'
  | 'done'
  | 'error';

interface StreamEvent {
  type: StreamEventType;
  text?: string;
  tool?: {
    id: string;
    name: string;
    args: Record<string, unknown>;
    status: 'pending' | 'running' | 'success' | 'error';
    output?: string;
    error?: string;
    duration?: number;
  };
  usage?: TokenUsage;
  error?: string;
}

Best Practices

1. Always handle all chunk types - Don't assume only text will be streamed

2. Accumulate tool call arguments - They arrive in pieces

3. Check for errors - Handle error chunks gracefully

4. Use done signal - Don't assume stream ends with text

5. Implement cancellation - Allow users to abort long streams

6. Track usage - Token counts are only available in done chunk