Skip to content

Conversation

@mattapperson
Copy link
Collaborator

Summary

  • Adds real-time streaming of generator tool preliminary results as they yield, rather than batch-emitting after tool execution completes
  • Introduces ToolEventBroadcaster class for push-based multi-consumer event delivery following the ReusableReadableStream pattern
  • Updates getToolStream() and getFullResponsesStream() to deliver events in real-time

Motivation

Similar to Vercel AI SDK v6's streaming preliminary tool outputs feature. This enables streaming UI updates during long-running tool operations - users can see progress as it happens rather than waiting for completion.

Changes

  • New: src/lib/tool-event-broadcaster.ts - Push-based broadcaster with buffering and multiple consumer support
  • Modified: src/lib/model-result.ts - Added executeToolsWithBroadcast() method, updated stream methods
  • Export: ToolEventBroadcaster exported for advanced use cases

Usage

const progressTool = tool({
  name: 'progress_task',
  inputSchema: z.object({ task: z.string() }),
  eventSchema: z.object({ progress: z.number() }),
  outputSchema: z.object({ completed: z.boolean() }),
  execute: async function* (params) {
    yield { progress: 25 };  // Streamed immediately!
    await doWork();
    yield { progress: 50 };  // Streamed immediately!
    yield { completed: true };
  },
});

// Consume real-time events
for await (const event of response.getToolStream()) {
  if (event.type === 'preliminary_result') {
    console.log(`Progress: ${event.result.progress}%`);
  }
}

Test plan

  • Unit tests for ToolEventBroadcaster (11 tests covering single/multiple consumers, async waiting, error handling)
  • E2E tests for real-time streaming via getToolStream() and getFullResponsesStream()
  • All existing tests pass

Generator tools can now stream preliminary results in real-time as they yield,
rather than batch-emitting them after all tool execution completes.

Changes:
- Add ToolEventBroadcaster class for push-based multi-consumer event delivery
- Update executeToolRound to pass onPreliminaryResult callback to executeTool
- Add executeToolsWithBroadcast method for real-time streaming flows
- Update getToolStream and getFullResponsesStream for real-time delivery
- Export ToolEventBroadcaster for advanced use cases

This enables streaming UI updates during long-running tool operations, similar
to Vercel AI SDK's streaming preliminary tool outputs feature.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants