ai-stream-utils weekly downloads statistics

AI SDK: Filter and transform UI messages while streaming to the client

This library provides composable filter and transformation utilities for UI message streams created by streamText() in the AI SDK.

Why?

The AI SDK UI message stream created by toUIMessageStream() streams all parts (text, tools, reasoning, etc.) to the client by default. However, you may want to:

• Filter: Tool calls like database searches often contain large amounts of data or sensitive information that should not be streamed to the client • Transform: Modify text or tool outputs while they are streamed to the client • Observe: Log stream lifecycle events, update states, or run side-effects without modifying the stream

This library provides type-safe, composable utilities for all these use cases.

Installation

[!NOTE] Version compatibility:

- Use ai-stream-utils@2.x for AI SDK v6 - Use ai-stream-utils@3.x for AI SDK v7

Usage

The pipe function provides a composable pipeline API for filtering, transforming, and observing UI message streams. Multiple operators can be chained together, and type guards automatically narrow chunk and part types, thus enabling type-safe stream transformations with autocomplete.

.filter()

Filter chunks by returning true to keep or false to exclude. The predicate may be async.

Type Guards

Generic type guards provide a simpler API for common filtering patterns:

• includeChunks("text-delta") or includeChunks(["text-delta", "text-end"]): Include only specific chunk types • excludeChunks("text-delta") or excludeChunks(["text-delta", "text-end"]): Exclude only specific chunk types • includeParts("text") or includeParts(["text", "reasoning"]): Include only specific part types • excludeParts("reasoning") or excludeParts(["reasoning", "tool-database"]): Exclude only specific part types

Filtering tools is the most common use case and the tool-filter type guards provide a convenient API for filtering tool chunks by tool name:

• excludeTools() or excludeTools("weather") or excludeTools(["weather", "database"]): Exclude all tools or specific tools by name • includeTools() or includeTools("weather") or includeTools(["weather", "database"]): Include all tools or specific tools by name

[!NOTE] The tool-filter type guards only affect tool chunks. Non-tool chunks will pass through.

Examples

Exclude tool calls from the client.

[!NOTE] excludeTools() and includeTools() filters tool chunks on the server before streaming to the client. This affects all tool types including:

- Server-side tools with execute functions - Client-side tools without execute functions - Tools that require human approval via needsApproval

Excluded tools will not appear in the client's message parts, so users won't see tool call UI or be able to approve/reject filtered tools.

.map()

Transform chunks by returning a chunk, an array of chunks, or null to exclude. The callback may be async, which lets a chunk be rewritten from an awaited result.

Examples

Convert text to uppercase.

Helpers

transformProviderMetadata() changes providerMetadata on the chunks that carry it (text-*, reasoning-*, reasoning-file, tool-input-*, tool-output-available, tool-output-error, tool-approval-response, source-*, file, custom). Every other chunk passes through untouched, so you don't have to narrow by chunk type yourself.

The callback receives the current metadata (may be undefined) and returns one of three things:

• an object to set/replace the metadata (merge by spreading metadata) • undefined to leave the chunk unchanged • null to remove the providerMetadata field entirely (the chunk still passes through)

.on()

Observe chunks without modifying the stream. The callback is invoked for matching chunks. Both the predicate and the callback may be async.

Type Guards

Type guard provides a type-safe way to observe specific chunk types:

• chunkType("text-delta") or chunkType(["start", "finish"]): Observe specific chunk types • partType("text") or partType(["text", "reasoning"]): Observe chunks belonging to specific part types • toolCall() or toolCall({ tool: "weather" }) or toolCall({ state: "output-available" }): Observe tool state transitions

[!NOTE] The partType type guard still operates on chunks. That means partType("text") will match any text chunks such as text-start, text-delta, and text-end.

The toolCall() type guard matches tool chunks representing state transitions (not streaming events):

• input-available: Tool input fully parsed • approval-requested: Tool awaiting user approval • output-available: Tool execution completed • output-error: Tool execution failed • output-denied: User denied approval

Examples

Log stream lifecycle events:

Observe tool state transitions for a specific tool:

Observe all tool calls:

.toStream()

Convert the pipeline back to a AsyncIterableStream> that can be returned to the client or consumed.

Chaining and Type Narrowing

Multiple operators can be chained together. After filtering with type guards, chunk and part types are narrowed automatically.

Control Chunks

Control chunks always pass through regardless of filter/transform settings:

• start: Stream start marker • finish: Stream finish marker • abort: Stream abort marker • message-metadata: Message metadata updates • error: Error messages

Stream Utilities

Helper functions for consuming streams and converting between streams, arrays, and async iterables.

consumeUIMessageStream

Consumes a UI message stream by fully reading it and returns the final assembled message. Useful for server-side processing without streaming to the client.

createAsyncIterableStream

Adds async iterator protocol to a ReadableStream, enabling for await...of loops.

convertArrayToStream

Converts an array to a ReadableStream that emits each element.

convertAsyncIterableToStream

Converts an async iterable (e.g., async generator) to a ReadableStream.

convertAsyncIterableToArray

Collects all values from an async iterable into an array.…

Description from the npm listing.