AI TOOLKIT
Get started
Recipes/Next.js

Streaming with Custom Format

Build a custom format to stream LLM responses

3 min readnextstreamingtool useView source

Create a custom stream to control the streaming format and structure of tool calls instead of using the built-in AI TOOLKIT data stream format (toUIMessageStream()). fullStream (on StreamTextResult) gives you direct access to all model events. You can transform, filter, and structure these events into your own streaming format. This gives you the benefits of the AI TOOLKIT's unified provider interface without prescribing how you consume the stream. You can: - Define your own stream chunk format - Control how steps and tool calls are structured - Parse the stream manually on the client - Build custom UI from your stream data For complete control over both the streaming format and the execution loop, combine this pattern with a manual agent loop.

Run it locally

$ npm install ai

Implementation

### Server

Create a route handler that calls a model and then streams the responses in a custom format:

import { tools } from '@/ai/tools'; // your tools

import { stepCountIs, streamText } from 'ai-toolkit';

__PROVIDER_IMPORT__;

export type StreamEvent =

| { type: 'text'; text: string }

| { type: 'tool-call'; toolName: string; input: unknown }

| { type: 'tool-result'; toolName: string; result: unknown };

const encoder = new TextEncoder();

function formatEvent(event: StreamEvent): Uint8Array {

return encoder.encode('data: ' + JSON.stringify(event) + '\n\n');

}

export async function POST(request: Request) {

const { prompt } = await request.json();

const result = streamText({

prompt,

model: __MODEL__,

tools,

stopWhen: stepCountIs(5),

});

const transformStream = new TransformStream({

transform(chunk, controller) {

switch (chunk.type) {

case 'text-delta':

controller.enqueue(formatEvent({ type: 'text', text: chunk.text }));

break;

case 'tool-call':

controller.enqueue(

formatEvent({

type: 'tool-call',

toolName: chunk.toolName,

input: chunk.input,

}),

);

break;

case 'tool-result':

controller.enqueue(

formatEvent({

type: 'tool-result',

toolName: chunk.toolName,

result: chunk.output,

}),

);

break;

}

},

});

return new Response(result.fullStream.pipeThrough(transformStream), {

headers: { 'Content-Type': 'text/event-stream' },

});

}

The route uses streamText to process the prompt with tools. Each event (text, tool calls, tool results) is encoded as a Server-Sent Event with a data: prefix and sent to the client.

### Client

Create a simple interface that parses and displays the stream:

'use client';

import { useState } from 'react';

import { StreamEvent } from './api/stream/route';

export default function Home() {

const [prompt, setPrompt] = useState('');

const [events, setEvents] = useState<StreamEvent[]>([]);

const [isStreaming, setIsStreaming] = useState(false);

const handleSubmit = async () => {

setEvents([]);

setIsStreaming(true);

setPrompt('');

const response = await fetch('/api/stream', {

method: 'POST',

headers: { 'Content-Type': 'application/json' },

body: JSON.stringify({ prompt }),

});

const reader = response.body?.getReader();

const decoder = new TextDecoder();

if (reader) {

let buffer = '';

while (true) {

const { done, value } = await reader.read();

if (done) break;

buffer += decoder.decode(value, { stream: true });

const lines = buffer.split('\n');

buffer = lines.pop() || '';

for (const line of lines) {

if (line.trim()) {

const dataStr = line.replace(/^data: /, '');

const event = JSON.parse(dataStr) as StreamEvent;

setEvents(prev => [...prev, event]);

}

}

}

}

setIsStreaming(false);

};

return (

<div>

<input

value={prompt}

onChange={e => setPrompt(e.target.value)}

placeholder="Enter a prompt..."

/>

<button onClick={handleSubmit} disabled={isStreaming}>

{isStreaming ? 'Streaming...' : 'Send'}

</button>

<pre>{JSON.stringify(events, null, 2)}</pre>

</div>

);

}

app/api/stream/route.ts
tsx
  1. 1import { tools } from '@/ai/tools'; // your tools
  2. 2import { stepCountIs, streamText } from 'ai-toolkit';
  3. 3__PROVIDER_IMPORT__;
  4. 4export type StreamEvent =
  5. 5 | { type: 'text'; text: string }
  6. 6 | { type: 'tool-call'; toolName: string; input: unknown }
  7. 7 | { type: 'tool-result'; toolName: string; result: unknown };
  8. 8const encoder = new TextEncoder();
  9. 9function formatEvent(event: StreamEvent): Uint8Array {
  10. 10 return encoder.encode('data: ' + JSON.stringify(event) + '\n\n');
  11. 11}
  12. 12export async function POST(request: Request) {
  13. 13 const { prompt } = await request.json();
  14. 14 const result = streamText({
  15. 15 prompt,
  16. 16 model: __MODEL__,
  17. 17 tools,
  18. 18 stopWhen: stepCountIs(5),
  19. 19 });
  20. 20 const transformStream = new TransformStream({
  21. 21 transform(chunk, controller) {
  22. 22 switch (chunk.type) {
  23. 23 case 'text-delta':
  24. 24 controller.enqueue(formatEvent({ type: 'text', text: chunk.text }));
  25. 25 break;
  26. 26 case 'tool-call':
  27. 27 controller.enqueue(
  28. 28 formatEvent({
  29. 29 type: 'tool-call',
  30. 30 toolName: chunk.toolName,
  31. 31 input: chunk.input,
  32. 32 }),
  33. 33 );
  34. 34 break;
  35. 35 case 'tool-result':
  36. 36 controller.enqueue(
  37. 37 formatEvent({
  38. 38 type: 'tool-result',
  39. 39 toolName: chunk.toolName,
  40. 40 result: chunk.output,
  41. 41 }),
  42. 42 );
  43. 43 break;
  44. 44 }
  45. 45 },
  46. 46 });
  47. 47 return new Response(result.fullStream.pipeThrough(transformStream), {
  48. 48 headers: { 'Content-Type': 'text/event-stream' },
  49. 49 });
  50. 50}
app/page.tsx
tsx
  1. 1'use client';
  2. 2import { useState } from 'react';
  3. 3import { StreamEvent } from './api/stream/route';
  4. 4export default function Home() {
  5. 5 const [prompt, setPrompt] = useState('');
  6. 6 const [events, setEvents] = useState<StreamEvent[]>([]);
  7. 7 const [isStreaming, setIsStreaming] = useState(false);
  8. 8 const handleSubmit = async () => {
  9. 9 setEvents([]);
  10. 10 setIsStreaming(true);
  11. 11 setPrompt('');
  12. 12 const response = await fetch('/api/stream', {
  13. 13 method: 'POST',
  14. 14 headers: { 'Content-Type': 'application/json' },
  15. 15 body: JSON.stringify({ prompt }),
  16. 16 });
  17. 17 const reader = response.body?.getReader();
  18. 18 const decoder = new TextDecoder();
  19. 19 if (reader) {
  20. 20 let buffer = '';
  21. 21 while (true) {
  22. 22 const { done, value } = await reader.read();
  23. 23 if (done) break;
  24. 24 buffer += decoder.decode(value, { stream: true });
  25. 25 const lines = buffer.split('\n');
  26. 26 buffer = lines.pop() || '';
  27. 27 for (const line of lines) {
  28. 28 if (line.trim()) {
  29. 29 const dataStr = line.replace(/^data: /, '');
  30. 30 const event = JSON.parse(dataStr) as StreamEvent;
  31. 31 setEvents(prev => [...prev, event]);
  32. 32 }
  33. 33 }
  34. 34 }
  35. 35 }
  36. 36 setIsStreaming(false);
  37. 37 };
  38. 38 return (
  39. 39 <div>
  40. 40 <input
  41. 41 value={prompt}
  42. 42 onChange={e => setPrompt(e.target.value)}
  43. 43 placeholder="Enter a prompt..."
  44. 44 />
  45. 45 <button onClick={handleSubmit} disabled={isStreaming}>
  46. 46 {isStreaming ? 'Streaming...' : 'Send'}
  47. 47 </button>
  48. 48 <pre>{JSON.stringify(events, null, 2)}</pre>
  49. 49 </div>
  50. 50 );
  51. 51}

How it works

The client uses the Fetch API to stream responses from the server. Since the server sends Server-Sent Events (newline-delimited with data: prefix), the client:

1. Reads chunks from the stream using getReader()

2. Decodes the binary chunks to text

3. Splits by newlines to identify complete events

4. Removes the data: prefix and parses the JSON, then appends it to the events list

Events are rendered in order as they arrive, giving you a linear representation of the AI's response.

Browse all recipes