UI message streams

Stream UI messages to a chat client, as Effect Streams you can compose and return as an HTTP response.

The AI SDK’s UI message stream is what useChat reads from a server. ai-effect exposes it as an Effect Stream of the SDK’s chunks, so you can merge, transform and tap it with the usual Stream operators. Chunks are the SDK’s own; see its docs for their shape.

There are three ways to make one, depending on what you start from.

From streamText

The result of streamText has toUIMessageStream:

import { streamText } from 'ai-effect';
import { Console, Effect, Stream } from 'effect';

const program = Effect.gen(function* () {
	const result = yield* streamText({
		model: 'openai/gpt-5',
		prompt: 'Write a haiku about rivers.',
	});
	yield* Stream.runForEach(result.toUIMessageStream(), (chunk) =>
		Console.log(chunk.type),
	);
});

It takes the SDK’s options. messageMetadata and onError stay synchronous functions; the step and end callbacks are Effects, like callbacks on generateText.

Compose with createUIMessageStream

To write your own chunks, or to combine several sources into one response, use createUIMessageStream. Its execute is an Effect that gets a writer:

import { createUIMessageStream, streamText } from 'ai-effect';
import { Effect } from 'effect';

const stream = createUIMessageStream({
	execute: ({ writer }) =>
		Effect.gen(function* () {
			yield* writer.write({ type: 'start' });
			const result = yield* streamText({
				model: 'openai/gpt-5',
				prompt: 'Write a haiku about rivers.',
			});
			yield* writer.merge(result.toUIMessageStream());
		}),
	onEnd: () => Effect.log('stream finished'),
});

onStepEnd, onStepFinish, onEnd and onFinish are Effects, and execute can use any service available where you build the stream. The writer has:

The result is a Stream that fails with an AiError and requires Ai. Running it again calls execute again.

From an agent

createAgentUIStream runs a ToolLoopAgent over the UIMessages a client sends. It takes the SDK’s options plus agent and uiMessages, and the agent must come from ToolLoopAgent.make:

import { createAgentUIStream, ToolLoopAgent } from 'ai-effect';
import type { UIMessage } from 'ai';
import { Effect } from 'effect';

const chat = (uiMessages: Array<UIMessage>) =>
	Effect.gen(function* () {
		const agent = yield* ToolLoopAgent.make({ model: 'openai/gpt-5' });
		return yield* createAgentUIStream({ agent, uiMessages });
	});

Unlike the other two, it returns an Effect of the stream. Invalid uiMessages (an InvalidRequestError, which is not retryable) and a model that fails to start fail the outer Effect before any chunk exists, so a route can still answer with an error status.

The SDK’s createAgentUIStreamResponse does not accept an ai-effect agent. Use createAgentUIStream and the conversion below instead.

Return it as an HTTP response

ai-effect has no wrapper for the SDK’s response helpers. Convert the stream with Stream.toReadableStreamEffect and pass it to the SDK’s createUIMessageStreamResponse. The Ai layer has to outlive the request, so run the handler on a ManagedRuntime built once (why):

import { createUIMessageStreamResponse, type UIMessage } from 'ai';
import { Ai, createAgentUIStream, ToolLoopAgent } from 'ai-effect';
import { Effect, ManagedRuntime, Stream } from 'effect';

const runtime = ManagedRuntime.make(Ai.layer);

export const POST = (request: Request) =>
	Effect.gen(function* () {
		const body = yield* Effect.promise(
			() => request.json() as Promise<{ messages: Array<UIMessage> }>,
		);
		const agent = yield* ToolLoopAgent.make({ model: 'openai/gpt-5' });
		const stream = yield* createAgentUIStream({
			agent,
			uiMessages: body.messages,
		});
		const readable = yield* Stream.toReadableStreamEffect(stream);
		return createUIMessageStreamResponse({ stream: readable });
	}).pipe(
		Effect.catch((error) =>
			Effect.succeed(
				new Response(error.message, {
					status:
						error.reason._tag === 'ai-effect/InvalidRequestError' ? 400 : 500,
				}),
			),
		),
		runtime.runPromise,
	);

A stream from createUIMessageStream or toUIMessageStream converts the same way.

Cancellation

Interrupting the fiber that consumes a UI stream stops the work behind it. Stopping early does not:

Closing the Ai layer also stops merged streams and aborts model calls, and an abortSignal you pass to a call is honored as well. Dropping a stream you never ran aborts nothing.