WorkflowChatTransport

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 which assumes the full response arrives in a single HTTP request, WorkflowChatTransport 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.

'use client';

import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow/client';

export default function Chat() {
  const { messages, sendMessage } = useChat({
    transport: new WorkflowChatTransport({
      api: '/api/chat',
      maxConsecutiveErrors: 5,
    }),
  });

  // ... render chat UI
}

Import

import { WorkflowChatTransport } from "@ai-sdk/workflow/client"

Constructor

Parameters

Methods

sendMessages()

Sends messages to the chat endpoint via POST and returns a streaming response. 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.

const stream = await transport.sendMessages({
  chatId: 'chat-123',
  trigger: 'submit-message',
  messages: [...],
  abortSignal: controller.signal,
});

Returns

Returns a Promise<ReadableStream<UIMessageChunk>> that includes chunks from both the initial POST response and any automatic reconnection.

reconnectToStream()

Reconnects to an existing chat stream that was previously interrupted. Useful for resuming after a page refresh or when the client needs to re-establish a connection.

const stream = await transport.reconnectToStream({
  chatId: 'chat-123',
  startIndex: -50, // Optional: fetch last 50 chunks
});

Returns

Returns a Promise<ReadableStream<UIMessageChunk> | null>.

How Reconnection Works

The transport follows this 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.

Negative Start Index

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).

Negative indexes require a durable server stream whose stored objects are already UIMessageChunk objects. The raw WorkflowAgent conversion shown below supports non-negative indexes only.

Server Requirements

For WorkflowChatTransport to work, your server must provide two endpoints:

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

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
• For negative startIndex, resolve to the tail and include x-workflow-stream-tail-index response header

See the WorkflowAgent guide for complete endpoint examples.

Examples

Basic Usage with useChat

'use client';

import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow/client';
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>
  );
}

With Callbacks

'use client';

import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow/client';
import { useMemo } from 'react';

export default function Chat() {
  const transport = useMemo(
    () =>
      new WorkflowChatTransport({
        api: '/api/chat',
        maxConsecutiveErrors: 5,
        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
}

Server-Side Endpoints (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,
    },
  });
}

import { createModelCallToUIChunkTransform } from '@ai-sdk/workflow';
import { createUIMessageStreamResponse } from 'ai';
import type { NextRequest } from 'next/server';
import { getRun } from 'workflow/api';

export async function GET(
  request: NextRequest,
  { params }: { params: Promise<{ runId: string }> },
) {
  const { runId } = await params;
  const startIndex = Number(
    new URL(request.url).searchParams.get('startIndex') ?? '0',
  );
  if (!Number.isSafeInteger(startIndex) || startIndex < 0) {
    return Response.json(
      { error: 'startIndex must be a non-negative safe integer' },
      { status: 400 },
    );
  }

  const run = await getRun(runId);
  const readable = run
    .getReadable({ startIndex: 0 })
    .pipeThrough(
      createModelCallToUIChunkTransform({ uiStartIndex: startIndex }),
    );

  return createUIMessageStreamResponse({
    stream: readable,
    headers: {
      'x-workflow-run-id': runId,
    },
  });
}

This WorkflowAgent endpoint replays raw ModelCallStreamPart objects from index 0, then applies the transport's non-negative cursor after converting them to UIMessageChunk objects. Negative start indexes require a durable stream whose stored objects are already UIMessageChunk objects.