feat(ai-ui-message-stream): client tools, structured output and reconnect for AI SDK routes

.changeset/ai-ui-message-stream-client.md

@@ -0,0 +1,12 @@
+---
+'@tanstack/ai-ui-message-stream': minor
+'@tanstack/ai-client': patch
+---
+
+TanStack `useChat` gets more from an AI SDK route through `fromChatTransport`:
+
+- Client tools go to the route in the request body. Read them on the route with `clientToolsFromAISDKBody(body, serverTools)`, so the route needs no copy.
+- With `outputSchema`, an `Output.object` answer is a typed `structured-output` part with partial objects.
+- A dropped stream reconnects with the backoff of the TanStack HTTP connections. Set it with the `reconnect` option.
+
+`@tanstack/ai-client` exports `createReconnectTracker` and its `ReconnectTracker` type.

docs/migration/tanstack-client-ai-sdk-server.md

@@ -19,6 +19,14 @@ Point TanStack AI `useChat` at a Vercel AI SDK route. Wrap an AI SDK `ChatTransp
 
 Use this page to move the client first. To move the server first, see [AI SDK useChat with a TanStack AI server](./ai-sdk-client-tanstack-server).
 
+Against your AI SDK route, TanStack `useChat` gives you:
+
+- Client persistence and resume after a reload, with no server code.
+- A [message queue](../chat/queueing) while a reply streams.
+- Client tools that the route reads from the request, with no copy on the route.
+- Typed structured output, with partial objects while it streams.
+- Typed approvals, TanStack devtools, and hooks for React, Solid, Vue, Svelte, Preact, Angular and more.
+
 > [!WARNING]
 > This bridge is a stop gap while you port. Some features change shape or are dropped on the way (see [AI SDK bridge compatibility](./ai-sdk-compatibility)). When the client works, port the route to `chat()` with the [migration guide](./migration-from-vercel-ai), then remove the bridge.
 
@@ -96,7 +104,38 @@ export async function POST(request: Request) {
 
 ## 4. Run client tools and approve tool calls
 
-Define the tools of the route on the client too, with the same names. Pass them to `useChat` and to `fromChatTransport`:
+Define client tools on the client only. The connection sends their names, descriptions and JSON Schemas to the route in the request body. Read them on the route with `clientToolsFromAISDKBody`, together with the server tools:
+
+```typescript
+import { openai } from '@ai-sdk/openai'
+import { clientToolsFromAISDKBody } from '@tanstack/ai-ui-message-stream'
+import { convertToModelMessages, streamText, tool } from 'ai'
+import { z } from 'zod'
+
+const getWeather = tool({
+  description: 'Get the weather for a city',
+  inputSchema: z.object({ city: z.string() }),
+  execute: async ({ city }) => ({ city, temperature: 72 }),
+})
+
+export async function POST(request: Request) {
+  const body = await request.json()
+
+  const result = streamText({
+    model: openai('gpt-6.1-sol'),
+    messages: await convertToModelMessages(body.messages),
+    tools: clientToolsFromAISDKBody(body, { getWeather }),
+  })
+
+  return result.toUIMessageStreamResponse()
+}
+```
+
+- A server tool wins when a client tool has the same name.
+- A client tool has no `execute` on the route, so `streamText` stops at the call.
+- Browser tools from [WebMCP](../tools/webmcp) reach the route the same way.
+
+Pass the tools to `useChat`, and to `fromChatTransport` for approvals:
 
 - **Client tool:** give it a `.client()` implementation. When the route stops at the call, the client runs the tool and sends the output back.
 - **Tool approval:** a tool that needs approval on the route shows up in `interrupts`. Resolve it with `interrupt.resolveInterrupt(true)` or `false`. The answer goes back to the route, and `streamText` runs or skips the tool.
@@ -148,13 +187,75 @@ export function Chat() {
 }
 ```
 
-## 5. Show data parts as activity
+## 5. Get typed structured output
+
+When the route streams an object with `Output.object`, pass the same schema to `useChat` and to `fromChatTransport`. The answer is a `structured-output` part, with partial objects while it streams and the checked `data` at the end:
+
+```tsx
+import { useChat } from '@tanstack/ai-react'
+import { fromChatTransport } from '@tanstack/ai-ui-message-stream'
+import { DefaultChatTransport } from 'ai'
+import { z } from 'zod'
+
+const recipe = z.object({ name: z.string(), steps: z.array(z.string()) })
+
+const connection = fromChatTransport(
+  new DefaultChatTransport({ api: '/api/recipe' }),
+  { outputSchema: recipe },
+)
+
+export function Recipe() {
+  const { messages } = useChat({
+    connection,
+    threadId: 'recipe',
+    outputSchema: recipe,
+  })
+
+  return messages.flatMap((message) =>
+    message.parts.map((part, i) =>
+      part.type === 'structured-output' ? (
+        <h2 key={i}>{part.partial?.name}</h2>
+      ) : null,
+    ),
+  )
+}
+```
+
+If the text is not valid JSON for the schema, the run ends with an error. All text of the reply counts as the answer, so a route that also writes text in a step that calls tools fails the check.
+
+## 6. Load a saved chat
+
+If your app saves AI SDK messages, convert them with `fromAISDKMessages` and pass them as `initialMessages`:
+
+```tsx
+import { useChat } from '@tanstack/ai-react'
+import {
+  fromAISDKMessages,
+  fromChatTransport,
+} from '@tanstack/ai-ui-message-stream'
+import { DefaultChatTransport } from 'ai'
+import type { UIMessage } from 'ai'
+
+const connection = fromChatTransport(
+  new DefaultChatTransport({ api: '/api/chat' }),
+)
+
+export function SavedChat({ saved }: { saved: Array<UIMessage> }) {
+  return useChat({
+    connection,
+    threadId: 'support-chat',
+    initialMessages: fromAISDKMessages(saved),
+  })
+}
+```
+
+## 7. Show data parts as activity
 
 A data part with an `id` (`writer.write({ type: 'data-weather', id, data })` on the route) arrives as an activity message of type `weather`. A new value for the same id replaces it. A transient data part, or one with no `id`, arrives as a custom event named `data-<name>`.
 
-## 6. Resume a stream
+## 8. Resume a stream
 
-A dropped connection needs nothing. The connection calls `reconnectToStream` on the transport, and the reply continues.
+A dropped connection needs nothing. The connection calls `reconnectToStream` on the transport, and the reply continues. It waits longer between tries that get no new data, as the TanStack HTTP connections do. To change the waits and the limit, pass `reconnect` to `fromChatTransport`, for example `{ maxAttempts: 10 }`.
 
 To continue a reply after a page reload, add client persistence, and pass the `threadId` to `fromChatTransport`. The route must support the AI SDK resume `GET` (`/api/chat/<id>/stream`):
 

packages/ai-client/src/index.ts

@@ -175,6 +175,8 @@ export {
   StreamTruncatedError,
   DurableStreamIncompleteError,
   StreamReconnectLimitError,
+  createReconnectTracker,
+  type ReconnectTracker,
   type ConnectConnectionAdapter,
   type ChatHydrateOptions,
   type ChatHydrationResult,

packages/ai-ui-message-stream/README.md

@@ -78,15 +78,16 @@ export function Chat() {
 
 ## API
 
-| Export                                                   | Direction                                                              |
-| -------------------------------------------------------- | ---------------------------------------------------------------------- |
-| `chatParamsFromAISDKRequest(request)`                    | AI SDK `useChat` request to `chat()` params, approval answers included |
-| `toUIMessageStream(stream)`                              | `chat()` events to an AI SDK UI message stream                         |
-| `toResumableUIMessageStreamResponse(stream, durability)` | A `chat()` run that AI SDK `useChat({ resume: true })` can rejoin      |
-| `resumeUIMessageStreamResponse(durability)`              | The AI SDK resume `GET`                                                |
-| `fromChatTransport(transport, { threadId, tools })`      | AI SDK `ChatTransport` to a TanStack `useChat` connection              |
-| `fromAISDKMessages` / `toAISDKMessages`                  | Message history in each direction                                      |
-| `fromUIMessageStream(stream, ids)`                       | AI SDK UI message stream to TanStack events                            |
+| Export                                                                       | Direction                                                                  |
+| ---------------------------------------------------------------------------- | -------------------------------------------------------------------------- |
+| `chatParamsFromAISDKRequest(request)`                                        | AI SDK `useChat` request to `chat()` params, approval answers included     |
+| `toUIMessageStream(stream)`                                                  | `chat()` events to an AI SDK UI message stream                             |
+| `toResumableUIMessageStreamResponse(stream, durability)`                     | A `chat()` run that AI SDK `useChat({ resume: true })` can rejoin          |
+| `resumeUIMessageStreamResponse(durability)`                                  | The AI SDK resume `GET`                                                    |
+| `fromChatTransport(transport, { threadId, tools, outputSchema, reconnect })` | AI SDK `ChatTransport` to a TanStack `useChat` connection                  |
+| `clientToolsFromAISDKBody(body, serverTools)`                                | On an AI SDK route: the client tools of the request, plus the server tools |
+| `fromAISDKMessages` / `toAISDKMessages`                                      | Message history in each direction                                          |
+| `fromUIMessageStream(stream, ids)`                                           | AI SDK UI message stream to TanStack events                                |
 
 Text, reasoning, tools, tool approvals, activity, subagents and stream resume cross in both directions where the other side has them. Shared state goes from a TanStack server to an AI SDK client only, because the AI SDK client does not send state.
 

packages/ai-ui-message-stream/package.json

@@ -49,11 +49,6 @@
     "@tanstack/ai-client": "workspace:^",
     "ai": "^7.0.0"
   },
-  "peerDependenciesMeta": {
-    "@tanstack/ai-client": {
-      "optional": true
-    }
-  },
   "devDependencies": {
     "@ag-ui/core": "1.0.0",
     "@tanstack/ai": "workspace:*",

packages/ai-ui-message-stream/src/client-tools.ts

@@ -0,0 +1,60 @@
+import { jsonSchema, tool } from 'ai'
+import type { ToolSet } from 'ai'
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+/**
+ * Add the tools that a TanStack `useChat` client declares (`clientTools` in
+ * the request body, sent by `fromChatTransport`) to the tools of an AI SDK
+ * route. A client tool has no `execute`, so `streamText` stops at its call
+ * and the client runs it. A server tool wins on a name clash.
+ *
+ * ```ts
+ * const body = await request.json()
+ * const result = streamText({
+ *   model,
+ *   messages: await convertToModelMessages(body.messages),
+ *   tools: clientToolsFromAISDKBody(body, { weather }),
+ * })
+ * ```
+ *
+ * @throws An error when `clientTools` is present but is not a list of
+ *   `{ name, description, parameters }` with a JSON Schema object.
+ */
+export function clientToolsFromAISDKBody<TServerTools extends ToolSet>(
+  body: unknown,
+  serverTools: TServerTools,
+): TServerTools & ToolSet {
+  const declared = isRecord(body) ? body.clientTools : undefined
+  if (declared === undefined) return serverTools
+  if (!Array.isArray(declared)) {
+    throw new Error('`clientTools` must be an array.')
+  }
+  // `fromEntries` defines own properties, so a `__proto__` name is inert.
+  const clientTools: ToolSet = Object.fromEntries(
+    declared.map((entry: unknown) => {
+      if (
+        !isRecord(entry) ||
+        typeof entry.name !== 'string' ||
+        entry.name === '' ||
+        typeof entry.description !== 'string' ||
+        !isRecord(entry.parameters)
+      ) {
+        throw new Error(
+          'Each client tool must have a `name`, a `description` and JSON Schema `parameters`.',
+        )
+      }
+      // No `execute`: `streamText` stops at the call and the client runs it.
+      return [
+        entry.name,
+        tool({
+          description: entry.description,
+          inputSchema: jsonSchema(entry.parameters),
+        }),
+      ]
+    }),
+  )
+  return { ...clientTools, ...serverTools }
+}