---
title: AI SDK
description: Use AI SDK's streamText directly inside durable workflows for lower-level control over model calls and tool execution.
type: guide
summary: Use streamText() inside a workflow for full control over model options, stop conditions, and output schemas — while tools remain durable steps.
related:
  - /docs/ai
  - /docs/ai/chat-session-modeling
  - /docs/ai/defining-tools
  - /docs/ai/resumable-streams
  - /docs/api-reference/workflow-ai/durable-agent
---

[AI SDK](https://ai-sdk.dev/) is Vercel's framework-agnostic TypeScript toolkit for building AI-powered apps and agents — unified provider access, streaming, tool calling, structured output, and UI hooks. Workflow SDK complements it by making those calls durable: the model request, the tool loop, and the multi-turn conversation all survive restarts and timeouts.

For the full AI SDK reference (providers, `streamText`, `generateObject`, `useChat`, tool calling, etc.) see the [AI SDK docs](https://ai-sdk.dev/docs). This page covers the Workflow-specific integration points.

<Callout type="info">
For most agent use cases, prefer [`DurableAgent`](/cookbook/agent-patterns/durable-agent) which wraps [`streamText`](https://ai-sdk.dev/docs/reference/ai-sdk-core/stream-text) and manages the tool loop automatically. This page covers using `streamText()` directly when you need lower-level control.
</Callout>

## When to use streamText directly

Use [`streamText()`](https://ai-sdk.dev/docs/reference/ai-sdk-core/stream-text) instead of `DurableAgent` when you need:

* **Custom stop conditions** — [`stopWhen`](https://ai-sdk.dev/docs/ai-sdk-core/agents#stop-conditions), [`prepareStep`](https://ai-sdk.dev/docs/ai-sdk-core/agents#prepare-step), or [`onStepFinish`](https://ai-sdk.dev/docs/reference/ai-sdk-core/stream-text#on-step-finish) callbacks
* **Structured output** — [`Output.object()`](https://ai-sdk.dev/docs/ai-sdk-core/generating-structured-data) or `Output.array()` alongside tool calling
* **Step-level callbacks** — `onStepFinish` for logging, metrics, or branching logic
* **Provider options** — per-step model switching, reasoning budgets, or custom [provider options](https://ai-sdk.dev/docs/ai-sdk-core/provider-options)

## Multi-turn pattern

One workflow run = one full conversation. The workflow suspends between turns on a hook and resumes when the next user message arrives. Conversation state, tool history, and intermediate computation all live inside the run.

<Callout type="info">
Because the conversation is one workflow run, it stays on the deployment that started it. If each turn should run on the latest deployment while preserving selected state or streams, see [Versioning](/docs/foundations/versioning) for the child-run continuation pattern.
</Callout>

<Tabs items={['Workflow', 'API Route', 'Client']}>

<Tab value="Workflow">

```typescript title="workflows/support.ts" lineNumbers
import { streamText, stepCountIs } from "ai";
import { defineHook, getWritable, getWorkflowMetadata } from "workflow";
import type { ModelMessage, UIMessageChunk } from "ai";
import { z } from "zod";

const MAX_TURNS = 20;

export const turnHook = defineHook({ // [!code highlight]
  schema: z.object({ message: z.string() }),
});

async function lookupOrder({ orderId }: { orderId: string }) {
  "use step";
  const res = await fetch(`https://api.store.com/orders/${orderId}`);
  return res.json();
}

async function processRefund({ orderId, reason }: { orderId: string; reason: string }) {
  "use step";
  const res = await fetch("https://api.store.com/refunds", {
    method: "POST",
    body: JSON.stringify({ orderId, reason }),
  });
  return res.json();
}

const TOOLS = {
  lookupOrder: {
    description: "Look up an order by ID",
    inputSchema: z.object({ orderId: z.string() }),
    execute: lookupOrder,
  },
  processRefund: {
    description: "Process a refund",
    inputSchema: z.object({ orderId: z.string(), reason: z.string() }),
    execute: processRefund,
  },
};

// Per-turn step — streams one agent response to the durable writable // [!code highlight]
async function runTurn(messages: ModelMessage[]) {
  "use step";

  const result = streamText({
    model: "anthropic/claude-haiku-4.5",
    system: "You are a customer support agent.",
    messages,
    tools: TOOLS,
    stopWhen: stepCountIs(8),
  });

  const writable = getWritable<UIMessageChunk>();
  // preventClose keeps the durable writable open so the next turn can  // write to it. Each turn still emits its own start + finish chunks.
  await result.toUIMessageStream().pipeTo(writable, { preventClose: true }); // [!code highlight]

  const response = await result.response;
  return { responseMessages: response.messages };
}

export async function supportWorkflow(initialMessages: ModelMessage[]) {
  "use workflow";

  const { workflowRunId } = getWorkflowMetadata();
  // Create the hook once, outside the loop — same token = HookConflictError // [!code highlight]
  const hook = turnHook.create({ token: workflowRunId }); // [!code highlight]
  let allMessages = initialMessages;

  for (let turn = 0; turn < MAX_TURNS; turn++) {
    const { responseMessages } = await runTurn(allMessages);
    allMessages = [...allMessages, ...responseMessages];

    const { message } = await hook; // [!code highlight] suspend until next user message
    if (message === "/done") break;

    allMessages = [...allMessages, { role: "user", content: message }];
  }

  return { turns: MAX_TURNS };
}
```

</Tab>

<Tab value="API Route">

One endpoint handles first turn, follow-ups, and the `/done` exit. The client sends `runId` in the body to distinguish first vs follow-up.

```typescript title="app/api/support/route.ts" lineNumbers
import type { UIMessage, UIMessageChunk } from "ai";
import { convertToModelMessages, createUIMessageStreamResponse } from "ai";
import { start, getRun } from "workflow/api";
import { supportWorkflow, turnHook } from "@/workflows/support";

// Pump the durable stream until this turn's `finish` chunk, then close  // the HTTP response. The source reader is released (not cancelled) so the
// workflow's durable stream keeps flowing for the next turn.
function sliceUntilFinish( // [!code highlight]
  source: ReadableStream<UIMessageChunk>
): ReadableStream<UIMessageChunk> {
  return new ReadableStream<UIMessageChunk>({
    async start(controller) {
      const reader = source.getReader();
      try {
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          controller.enqueue(value);
          if (value.type === "finish") break; // [!code highlight]
        }
        controller.close();
      } catch (e) {
        controller.error(e);
      } finally {
        reader.releaseLock();
      }
    },
  });
}

// `/done` exits the workflow without emitting chunks. Return a synthetic
// start+finish so useChat's lifecycle terminates cleanly.
function emptyTurnStream(): ReadableStream<UIMessageChunk> {
  return new ReadableStream<UIMessageChunk>({
    start(controller) {
      controller.enqueue({ type: "start", messageId: crypto.randomUUID() });
      controller.enqueue({ type: "finish" });
      controller.close();
    },
  });
}

export async function POST(req: Request) {
  const { messages, runId }: { messages: UIMessage[]; runId?: string } =
    await req.json();
  const modelMessages = await convertToModelMessages(messages);

  // Follow-up turn: resume hook, return stream starting AFTER the last turn // [!code highlight]
  if (runId) {
    try {
      const run = getRun(runId);

      // Snapshot tail before resuming so our slice only contains this turn // [!code highlight]
      const probe = run.getReadable();
      const tailIndex = await probe.getTailIndex();
      await probe.cancel();

      const lastUser = modelMessages.filter((m) => m.role === "user").at(-1);
      const text =
        typeof lastUser?.content === "string"
          ? lastUser.content
          : Array.isArray(lastUser?.content)
            ? lastUser.content
                .filter((p): p is { type: "text"; text: string } =>
                  "type" in p && p.type === "text"
                )
                .map((p) => p.text)
                .join("")
            : "";

      await turnHook.resume(runId, { message: text }); // [!code highlight]

      if (text === "/done") {
        return createUIMessageStreamResponse({
          stream: emptyTurnStream(),
          headers: { "x-workflow-run-id": runId },
        });
      }

      const stream = sliceUntilFinish(
        run.getReadable({ startIndex: tailIndex + 1 }) // [!code highlight]
      );

      return createUIMessageStreamResponse({
        stream,
        headers: { "x-workflow-run-id": runId },
      });
    } catch (e: unknown) {
      const msg = e instanceof Error ? e.message.toLowerCase() : "";
      if (!msg.includes("not found") && !msg.includes("expired")) throw e;
      // Stale runId — fall through to start fresh
    }
  }

  // First turn: start a new workflow // [!code highlight]
  const run = await start(supportWorkflow, [modelMessages]);
  const stream = sliceUntilFinish(run.readable);

  return createUIMessageStreamResponse({
    stream,
    headers: { "x-workflow-run-id": run.runId },
  });
}
```

</Tab>

<Tab value="Client">

Store the `runId` in a ref and pass it in the body of every follow-up. `WorkflowChatTransport` forwards it for you.

```tsx title="components/support-chat.tsx" lineNumbers
"use client";

import { useChat } from "@ai-sdk/react";
import { WorkflowChatTransport } from "@workflow/ai";
import { useMemo, useRef, useState } from "react";

export function SupportChat() {
  const [input, setInput] = useState("");
  const runIdRef = useRef<string | null>(null); // [!code highlight]

  const transport = useMemo(
    () =>
      new WorkflowChatTransport({
        api: "/api/support",
        prepareSendMessagesRequest: ({ messages, body }) => ({
          body: { ...body, messages, runId: runIdRef.current }, // [!code highlight]
        }),
        onChatSendMessage: (response) => {
          const id = response.headers.get("x-workflow-run-id");
          if (id) runIdRef.current = id; // [!code highlight]
        },
      }),
    []
  );

  const { messages, sendMessage, status } = useChat({ transport });
  const busy = status === "streaming" || status === "submitted";

  return (
    <form
      onSubmit={(e) => {
        e.preventDefault();
        if (busy || !input.trim()) return;
        sendMessage({ text: input });
        setInput("");
      }}
    >
      {messages.map((m) => (
        <div key={m.id}>{m.role}: {m.parts.map((p) => p.type === "text" ? p.text : "").join("")}</div>
      ))}
      <input value={input} onChange={(e) => setInput(e.target.value)} disabled={busy} />
    </form>
  );
}
```

</Tab>

</Tabs>

## How it works

1. **One workflow = one conversation.** The workflow loops on a hook, keeping `allMessages`, tool history, and state alive across turns.
2. **Hook is created once.** `turnHook.create({ token: workflowRunId })` outside the loop — calling it twice with the same token throws `HookConflictError`.
3. **`preventClose: true`** on `pipeTo` keeps the durable writable open so the next turn can write to it.
4. **`sliceUntilFinish`** in the API reads chunks until `type === "finish"`, then closes the HTTP response. The source reader is released — not cancelled — so the workflow stream keeps flowing.
5. **`startIndex: tailIndex + 1`** gives each follow-up response only the new chunks, avoiding replay of previous turns.
6. **`/done`** resumes the hook so the workflow exits cleanly, then returns a synthetic `start` + `finish` so `useChat` transitions out of "streaming".

## Pitfalls

Non-obvious correctness details worth knowing before adapting this pattern.

### Snapshot `tailIndex` *before* resuming the hook

{/* @skip-typecheck - fragment referencing variables from the surrounding multi-turn pattern */}
```typescript
const tailIndex = await probe.getTailIndex(); // [!code highlight] FIRST
await probe.cancel();
await turnHook.resume(runId, { message: text }); // [!code highlight] THEN
const stream = run.getReadable({ startIndex: tailIndex + 1 });
```

Reversing the order races the workflow: by the time you read `tailIndex`, the next turn has already written its `start` chunk, and your `startIndex + 1` skips past it.

### Don't call `writable.close()` inside a workflow function

I/O operations like closing streams must happen inside a `"use step"` function. Calling `writable.close()` directly in the workflow body throws `Not supported in workflow functions`. When the workflow returns, the runtime closes the underlying writable for you.

### Don't use `TransformStream.terminate()` to slice the stream

A `TransformStream` with `controller.terminate()` on the `finish` chunk seems like the obvious fit for `sliceUntilFinish`, but throws `Invalid state: TransformStream has been terminated` when late-arriving chunks hit the transform callback. Manual pumping through a custom `ReadableStream` (as shown above) sidesteps the problem entirely.

### Release the source reader, don't cancel it

In `sliceUntilFinish`, use `reader.releaseLock()` in the `finally` block rather than `source.cancel()`. Cancelling propagates upstream and closes the durable writable, breaking the next turn. Releasing the lock just detaches our reader; the durable stream keeps flowing.

### Handle stale `runId` gracefully

Clients can send a `runId` from a long-gone workflow (localStorage, back button, server restart). Wrap the follow-up path in a try/catch for `not found` / `expired` and fall through to the first-turn code path to start a fresh workflow.

## streamText vs DurableAgent

| | `streamText()` | `DurableAgent` |
|---|---|---|
| **Tool loop** | AI SDK handles via `stopWhen` | DurableAgent handles internally |
| **LLM call durability** | Re-executes on replay | Each LLM call is a durable step |
| **Stop conditions** | `stopWhen`, `prepareStep` | `prepareStep` only |
| **Structured output** | `Output.object()`, `Output.array()` | Not available |
| **Step callbacks** | `onStepFinish`, `onChunk` | Not available |
| **Setup** | Manual stream piping | Automatic |

Use `DurableAgent` for most agent use cases. Use `streamText` when you need the additional control.

## Key APIs

**AI SDK** ([docs](https://ai-sdk.dev/docs))

* [`streamText()`](https://ai-sdk.dev/docs/reference/ai-sdk-core/stream-text) — core streaming function; `toUIMessageStream()` pipes into the durable writable
* [`tool()` / tool calling](https://ai-sdk.dev/docs/ai-sdk-core/tools-and-tool-calling) — tools wrap `"use step"` functions so each tool call is replayed from the log, not re-executed
* [`stepCountIs()` / `stopWhen`](https://ai-sdk.dev/docs/ai-sdk-core/agents#stop-conditions) — bound the agent loop inside each turn
* [`convertToModelMessages()`](https://ai-sdk.dev/docs/reference/ai-sdk-ui/convert-to-model-messages) / [`createUIMessageStreamResponse()`](https://ai-sdk.dev/docs/reference/ai-sdk-ui/create-ui-message-stream-response) — UI ↔ model message conversion at the API boundary
* [`useChat()`](https://ai-sdk.dev/docs/reference/ai-sdk-ui/use-chat) — React hook that consumes the UI message stream on the client

**Workflow SDK**

* [`"use step"`](/docs/api-reference/workflow/use-step) — makes tool executions durable
* [`defineHook()`](/docs/api-reference/workflow/define-hook) — suspension point for follow-up messages
* [`getWritable()`](/docs/api-reference/workflow/get-writable) — resumable stream output
* [`getRun()`](/docs/api-reference/workflow-api/get-run) — `run.getReadable({ startIndex })` for slicing per-turn streams
* [`WorkflowChatTransport`](/docs/api-reference/workflow-ai/workflow-chat-transport) — passes `runId` between turns
