WorkflowChatTransport overview
WorkflowChatTransport is a ChatTransport implementation for useChat that enables automatic stream reconnection for workflow-based chat apps. It posts messages to a chat endpoint, extracts the x-workflow-run-id response header, and reconnects to a /{runId}/stream endpoint on interruption (network failures, page refreshes, function timeouts). Unlike DefaultChatTransport, it is designed for the Workflow SDK where the initial response stream may be interrupted by function timeouts. The transport automatically detects missing finish events and reconnects to resume from where the stream left off.
WorkflowChatTransport import
Import WorkflowChatTransport from @ai-sdk/workflow using: import { WorkflowChatTransport } from '@ai-sdk/workflow'
WorkflowChatTransport with callbacks and page refresh recovery example
'use client';
import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow';
import { useMemo } from 'react';
export default function Chat() {
const transport = useMemo(
() =>
new WorkflowChatTransport({
api: '/api/chat',
maxConsecutiveErrors: 5,
initialStartIndex: -50, // Resume from last 50 chunks on page refresh
onChatSendMessage: response => {
const runId = response.headers.get('x-workflow-run-id');
console.log('Workflow run started:', runId);
},
onChatEnd: ({ chatId, chunkIndex }) => {
console.log(`Chat ${chatId} complete, ${chunkIndex} chunks`);
},
}),
[],
);
const { messages, sendMessage } = useChat({ transport });
// ... render chat UI
}
This example demonstrates using WorkflowChatTransport with callbacks for extracting workflow run ID and handling chat completion, plus page refresh recovery using negative initialStartIndex.
WorkflowChatTransport constructor parameters
WorkflowChatTransport constructor accepts the following parameters: (1) api (string, optional): API endpoint for chat requests. The reconnection endpoint is derived from this as {api}/{runId}/stream. Default: '/api/chat'. (2) fetch (typeof fetch, optional): Custom fetch implementation to use for HTTP requests. Default: global fetch. (3) maxConsecutiveErrors (number, optional): Maximum number of consecutive errors allowed during reconnection attempts before giving up. Default: 3. (4) initialStartIndex (number, optional): Default chunk index to start from when reconnecting. Negative values read from the end of the stream (e.g., -50 fetches the last 50 chunks), useful for resuming after a page refresh without replaying the full conversation. Can be overridden per-call via reconnectToStream options. Default: 0. (5) onChatSendMessage ((response: Response, options: SendMessagesOptions) => void | Promise<void>, optional): Callback invoked after the initial POST request succeeds. Useful for inspecting response headers (e.g., extracting workflow run ID) or tracking chat history on the client side. (6) onChatEnd (({ chatId, chunkIndex }) => void | Promise<void>, optional): Callback invoked when the stream ends (receives a finish chunk). Receives the chat ID and total chunk count. Useful for cleanup or state updates. (7) prepareSendMessagesRequest (PrepareSendMessagesRequest, optional): Function to customize the POST request before sending. Can override the API endpoint, headers, credentials, and body. (8) prepareReconnectToStreamRequest (PrepareReconnectToStreamRequest, optional): Function to customize the reconnection GET request. Can override the API endpoint, headers, and credentials.
WorkflowChatTransport.sendMessages() method signature
The sendMessages() method sends messages to the chat endpoint via POST and returns a streaming response. It accepts parameters: (1) chatId (string): Unique identifier for the chat session. (2) trigger ('submit-message' | 'regenerate-message'): The type of message submission. (3) messageId (string | undefined): ID of the message to regenerate, or undefined for new messages. (4) messages (UIMessage[]): Array of UI messages representing the conversation history. (5) abortSignal (AbortSignal | undefined): Signal to abort the request. Propagated to both the initial POST and any reconnection GET requests. It returns Promise<ReadableStream<UIMessageChunk>> that includes chunks from both the initial POST response and any automatic reconnection. If the stream is interrupted (no finish event received), the transport automatically reconnects via GET to {api}/{runId}/stream?startIndex={chunkIndex} to resume from where it left off. The POST request includes the messages as JSON and expects the response to include an x-workflow-run-id header identifying the workflow run.
WorkflowChatTransport.reconnectToStream() method signature
The reconnectToStream() method reconnects to an existing chat stream that was previously interrupted. It accepts parameters: (1) chatId (string): The chat ID to reconnect to. Used to construct the reconnection URL. (2) abortSignal (AbortSignal | undefined): Signal to abort the reconnection request. (3) startIndex (number, optional): Override the start index for this reconnection. Negative values read from the end of the stream. When omitted, falls back to the constructor's initialStartIndex. It returns Promise<ReadableStream<UIMessageChunk> | null>.
WorkflowChatTransport reconnection flow
The transport follows this reconnection flow: (1) POST to {api} with messages. The response must include an x-workflow-run-id header. (2) Stream the SSE response, counting chunks as they arrive. (3) Detect interruption: If the stream closes without a finish event (e.g., function timeout, network error), the transport knows the response is incomplete. (4) Reconnect via GET to {api}/{runId}/stream?startIndex={chunkIndex} to resume from the last received chunk. (5) Retry: If the reconnection stream also interrupts, retry up to maxConsecutiveErrors times. (6) Complete: Once a finish event is received, call onChatEnd and close the stream.
WorkflowChatTransport negative start index handling
When initialStartIndex is negative (e.g., -50), the transport sends it as-is in the first reconnection request. The server should resolve this to an absolute position and return the x-workflow-stream-tail-index response header so the transport can compute the correct position for subsequent retries. If the header is missing or invalid, the transport falls back to replaying from the beginning (startIndex=0).
WorkflowChatTransport server endpoint requirements
For WorkflowChatTransport to work, the server must provide two endpoints: (1) POST {api} (e.g., /api/chat): Accept messages as JSON body, return an SSE stream of UIMessageChunk events, include an x-workflow-run-id response header. (2) GET {api}/{runId}/stream (e.g., /api/chat/{runId}/stream): Accept a startIndex query parameter, return the SSE stream starting from the given chunk index, and for negative startIndex, resolve to the tail and include x-workflow-stream-tail-index response header.
WorkflowChatTransport basic usage example
'use client';
import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow';
import { useMemo } from 'react';
export default function Chat() {
const transport = useMemo(
() => new WorkflowChatTransport({ api: '/api/chat' }),
[],
);
const { messages, sendMessage, status } = useChat({ transport });
return (
<div>
{messages.map(message => (
<div key={message.id}>
{message.role === 'user' ? 'User: ' : 'AI: '}
{message.parts.map((part, index) =>
part.type === 'text' ? <span key={index}>{part.text}</span> : null,
)}
</div>
))}
<button onClick={() => sendMessage({ text: 'Hello!' })}>Send</button>
</div>
);
}
This example shows basic usage of WorkflowChatTransport with useChat hook.
WorkflowChatTransport server POST endpoint example (Next.js)
import { createModelCallToUIChunkTransform } from '@ai-sdk/workflow';
import { createUIMessageStreamResponse, type UIMessage } from 'ai';
import { start } from 'workflow/api';
import { chat } from '@/workflow/agent-chat';
export async function POST(request: Request) {
const { messages }: { messages: UIMessage[] } = await request.json();
const run = await start(chat, [messages]);
return createUIMessageStreamResponse({
stream: run.readable.pipeThrough(createModelCallToUIChunkTransform()),
headers: {
'x-workflow-run-id': run.runId,
},
});
}
filename: app/api/chat/route.ts
This example shows the POST endpoint that accepts messages, starts a workflow, and returns an SSE stream with the x-workflow-run-id header.
Rate limiting with Upstash Redis and Ratelimit library
Rate limiting protects APIs from abuse by setting a maximum threshold on requests per timeframe. Upstash Redis and Upstash Ratelimit provide a straightforward implementation. Import Ratelimit from '@upstash/ratelimit' and Redis from '@upstash/redis'. Create a Ratelimit instance with redis: Redis.fromEnv() and limiter: Ratelimit.fixedWindow(5, '30s') to allow 5 requests per 30 seconds. Call ratelimit.limit(ip) with the client IP to check limits; it returns {success, remaining}. Return a 429 response if success is false.
Ratelimit.fixedWindow configuration
Ratelimit.fixedWindow(requests, timeframe) sets a fixed window rate limiting policy. The first parameter is the maximum number of requests allowed, and the second parameter is the timeframe duration as a string (e.g., '30s' for 30 seconds).
Rate limit check returns success and remaining fields
The ratelimit.limit(ip) method returns an object with two fields: success (boolean indicating if the request is allowed) and remaining (number of remaining requests in the current window).
Complete rate-limited streaming API endpoint example
This example shows a complete Next.js API route protecting a streaming text generation endpoint with rate limiting:
```tsx filename='app/api/generate/route.ts'
import {
createUIMessageStreamResponse,
streamText,
toUIMessageStream,
} from 'ai';
__PROVIDER_IMPORT__;
import { Ratelimit } from '@upstash/ratelimit';
import { Redis } from '@upstash/redis';
import { NextRequest } from 'next/server';
// Allow streaming responses up to 30 seconds
export const maxDuration = 30;
// Create Rate limit
const ratelimit = new Ratelimit({
redis: Redis.fromEnv(),
limiter: Ratelimit.fixedWindow(5, '30s'),
});
export async function POST(req: NextRequest) {
// call ratelimit with request ip
const ip = req.ip ?? 'ip';
const { success, remaining } = await ratelimit.limit(ip);
// block the request if unsuccessful
if (!success) {
return new Response('Ratelimited!', { status: 429 });
}
const { messages } = await req.json();
const result = streamText({
model: __MODEL__,
messages,
});
return createUIMessageStreamResponse({
stream: toUIMessageStream({ stream: result.stream }),
});
}
```
This pattern rate limits using client IP, extracts messages from the request, streams text responses, and returns a 429 status when rate limit is exceeded.
Rate limit by client IP in Next.js
Extract the client IP from the NextRequest object using req.ip, with a fallback value if it is not available (e.g., const ip = req.ip ?? 'ip'). Pass this IP to ratelimit.limit(ip) to enforce per-IP rate limits.