feat(ai-sdk): add streamText/streamObject streaming sample (AI SDK v7)

.scripts/copy-shared-files.mjs

@@ -15,6 +15,7 @@ const HAS_CHILD_SAMPLES = [
 // Some samples have different config files from those in .shared/
 // that we don't want to overwrite
 const TSCONFIG_EXCLUDE = [
+  'ai-sdk',
   'nextjs-ecommerce-oneclick',
   'monorepo-folders',
   'fetch-esm',

ai-sdk/README.md

@@ -15,3 +15,14 @@ This project demonstrates some uses of the AI SDK inside Temporal.
 1. `npm run workflow tools`
 1. `npm run workflow mcp`
 1. `npm run workflow middleware`
+1. `npm run workflow stream`
+
+### Streaming
+
+The `stream` sample shows how to stream model output out of a
+Workflow. `streamText` runs the model call in
+an activity that publishes each delta onto a
+[Workflow Stream](https://docs.temporal.io/develop/typescript/workflows/workflow-streams)
+topic. The Workflow hosts the stream (`new WorkflowStream()`) and durably reassembles the
+final result, while external consumers subscribe by Workflow id to render the tokens live
+as they arrive — see `src/client.ts`.

ai-sdk/package.json

@@ -11,7 +11,7 @@
     "start": "ts-node src/worker.ts",
     "start.watch": "nodemon src/worker.ts",
     "workflow": "ts-node src/client.ts",
-    "test": "mocha --exit --require ts-node/register --require source-map-support/register src/mocha/*.test.ts"
+    "test": "mocha --exit --node-option experimental-require-module --require ts-node/register --require source-map-support/register src/mocha/*.test.ts"
   },
   "nodemonConfig": {
     "execMap": {
@@ -23,22 +23,23 @@
     ]
   },
   "dependencies": {
-    "@ai-sdk/openai": "^3.0.0",
-    "@ai-sdk/provider": "^3.0.0",
-    "@ai-sdk/mcp": "^1.0.0",
-    "@modelcontextprotocol/sdk": "^1.10.2",
-    "@temporalio/activity": "1.20.3",
-    "@temporalio/ai-sdk": "1.20.3",
-    "@temporalio/client": "1.20.3",
-    "@temporalio/envconfig": "1.20.3",
-    "@temporalio/worker": "1.20.3",
-    "@temporalio/workflow": "1.20.3",
-    "ai": "^6.0.0",
+    "@ai-sdk/openai": "^4.0.0",
+    "@ai-sdk/provider": "^4.0.0",
+    "@ai-sdk/mcp": "^2.0.0",
+    "@modelcontextprotocol/sdk": "^1.25.2",
+    "@temporalio/activity": "^1.21.0",
+    "@temporalio/ai-sdk": "^1.21.0",
+    "@temporalio/client": "^1.21.0",
+    "@temporalio/envconfig": "^1.21.0",
+    "@temporalio/worker": "^1.21.0",
+    "@temporalio/workflow": "^1.21.0",
+    "@temporalio/workflow-streams": "^1.21.0",
+    "ai": "^7.0.0",
     "nanoid": "3.x",
     "zod": "^3.25.76"
   },
   "devDependencies": {
-    "@temporalio/testing": "1.20.3",
+    "@temporalio/testing": "^1.21.0",
     "@tsconfig/node22": "^22.0.0",
     "@types/mocha": "10.x",
     "@types/node": "^22.9.1",

ai-sdk/src/client.ts

@@ -1,8 +1,35 @@
 import { Connection, Client } from '@temporalio/client';
 import { loadClientConnectConfig } from '@temporalio/envconfig';
-import { haikuAgent, mcpAgent, middlewareAgent, toolsAgent } from './workflows';
+import { WorkflowStreamClient } from '@temporalio/workflow-streams/client';
+import {
+  haikuAgent,
+  mcpAgent,
+  middlewareAgent,
+  streamingAgent,
+  toolsAgent,
+  STREAM_TOPIC,
+  consumerDoneSignal,
+} from './workflows';
 import { nanoid } from 'nanoid';
 
+// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-consumer
+// Subscribe to a Workflow's stream topic and render each text delta live as it
+// is published by the streaming activity. Each item's payload is the
+// JSON-encoded AI SDK stream part; `resultType: true` decodes it to raw bytes.
+async function renderStream(client: Client, workflowId: string, topic: string): Promise<void> {
+  const streamClient = WorkflowStreamClient.create(client, workflowId);
+  for await (const item of streamClient.subscribe<Uint8Array>(topic, 0, { resultType: true })) {
+    const part = JSON.parse(new TextDecoder().decode(item.data));
+    if (part.type === 'text-delta') process.stdout.write(part.delta);
+    if (part.type === 'finish') break;
+  }
+  process.stdout.write('\n');
+  // Acknowledge receipt so the Workflow can complete without racing this final
+  // poll against its in-memory stream log being discarded.
+  await client.workflow.getHandle(workflowId).signal(consumerDoneSignal);
+}
+// @@@SNIPEND
+
 async function run() {
   const args = process.argv;
   const workflow = args[2] ?? 'haiku';
@@ -14,6 +41,18 @@ async function run() {
 
   let handle;
   switch (workflow) {
+    case 'stream': {
+      const streamHandle = await client.workflow.start(streamingAgent, {
+        taskQueue: 'ai-sdk',
+        args: ['Temporal'],
+        workflowId: 'workflow-' + nanoid(),
+      });
+      console.log(`Started workflow ${streamHandle.workflowId}`);
+      await renderStream(client, streamHandle.workflowId, STREAM_TOPIC);
+      console.log(await streamHandle.result());
+      await connection.close();
+      return;
+    }
     case 'middleware':
       handle = await client.workflow.start(middlewareAgent, {
         taskQueue: 'ai-sdk',

ai-sdk/src/mocha/streaming.test.ts

@@ -0,0 +1,138 @@
+import { TestWorkflowEnvironment } from '@temporalio/testing';
+import { after, before, describe, it } from 'mocha';
+import { Worker } from '@temporalio/worker';
+import { WorkflowStreamClient } from '@temporalio/workflow-streams/client';
+import type {
+  LanguageModelV4,
+  LanguageModelV4CallOptions,
+  LanguageModelV4GenerateResult,
+  LanguageModelV4StreamPart,
+  LanguageModelV4StreamResult,
+  ProviderV4,
+} from '@ai-sdk/provider';
+import assert from 'assert';
+import { streamingAgent, STREAM_TOPIC, consumerDoneSignal } from '../workflows';
+import * as activities from '../activities';
+import { AiSdkPlugin } from '@temporalio/ai-sdk';
+
+// A deterministic, offline model that streams a fixed set of text deltas so
+// these tests need no OPENAI_API_KEY and can assert on exact output.
+class MockStreamModel implements LanguageModelV4 {
+  readonly specificationVersion = 'v4';
+  readonly provider = 'mock';
+  readonly modelId = 'mock-model';
+  private readonly chunks: string[];
+
+  constructor(chunks: string[]) {
+    this.chunks = chunks;
+  }
+
+  get supportedUrls(): Record<string, RegExp[]> {
+    return {};
+  }
+
+  doGenerate(_options: LanguageModelV4CallOptions): Promise<LanguageModelV4GenerateResult> {
+    throw new Error('generate not supported by mock');
+  }
+
+  doStream(_options: LanguageModelV4CallOptions): Promise<LanguageModelV4StreamResult> {
+    const chunks = this.chunks;
+    const parts: LanguageModelV4StreamPart[] = [
+      { type: 'stream-start', warnings: [] },
+      { type: 'text-start', id: 't1' },
+      ...chunks.map((delta): LanguageModelV4StreamPart => ({ type: 'text-delta', id: 't1', delta })),
+      { type: 'text-end', id: 't1' },
+      {
+        type: 'finish',
+        finishReason: { unified: 'stop', raw: undefined },
+        usage: {
+          inputTokens: { total: 1, noCache: undefined, cacheRead: undefined, cacheWrite: undefined },
+          outputTokens: { total: chunks.length, text: undefined, reasoning: undefined },
+        },
+      },
+    ];
+    return Promise.resolve({
+      stream: new ReadableStream<LanguageModelV4StreamPart>({
+        start(controller) {
+          for (const part of parts) controller.enqueue(part);
+          controller.close();
+        },
+      }),
+      request: {},
+      response: {},
+    });
+  }
+}
+
+function mockProvider(chunks: string[]): ProviderV4 {
+  return {
+    specificationVersion: 'v4',
+    languageModel: () => new MockStreamModel(chunks),
+    embeddingModel: () => {
+      throw new Error('not implemented');
+    },
+    imageModel: () => {
+      throw new Error('not implemented');
+    },
+  };
+}
+
+// Collect the live deltas an external subscriber sees on a topic.
+async function collectDeltas(client: any, workflowId: string, topic: string): Promise<string[]> {
+  const deltas: string[] = [];
+  const streamClient = WorkflowStreamClient.create(client, workflowId);
+  for await (const item of streamClient.subscribe<Uint8Array>(topic, 0, { resultType: true })) {
+    const part = JSON.parse(new TextDecoder().decode(item.data));
+    if (part.type === 'text-delta') deltas.push(part.delta);
+    if (part.type === 'finish') break;
+  }
+  // Mirror the real consumer: acknowledge receipt so the workflow can complete.
+  await client.workflow.getHandle(workflowId).signal(consumerDoneSignal);
+  return deltas;
+}
+
+describe('streaming agents', function () {
+  this.timeout(30_000);
+
+  let testEnv: TestWorkflowEnvironment;
+
+  before(async () => {
+    testEnv = await TestWorkflowEnvironment.createLocal();
+  });
+
+  after(async () => {
+    await testEnv?.teardown();
+  });
+
+  it('streamingAgent publishes live deltas and returns the full text', async () => {
+    const { client, nativeConnection } = testEnv;
+    const taskQueue = 'test-stream-text';
+    const chunks = ['Dur', 'able ', 'streams ', 'of ', 'thought'];
+
+    const worker = await Worker.create({
+      connection: nativeConnection,
+      plugins: [new AiSdkPlugin({ modelProvider: mockProvider(chunks) })],
+      taskQueue,
+      workflowsPath: require.resolve('../workflows'),
+      activities,
+    });
+
+    await worker.runUntil(async () => {
+      const handle = await client.workflow.start(streamingAgent, {
+        args: ['Temporal'],
+        workflowId: 'test-stream-text-' + Date.now(),
+        taskQueue,
+      });
+
+      const deltasPromise = collectDeltas(client, handle.workflowId, STREAM_TOPIC);
+      const result = await handle.result();
+      const deltas = await deltasPromise;
+
+      // The workflow durably reassembles the full text from the replayed stream.
+      assert.strictEqual(result, chunks.join(''));
+      // The external subscriber saw the response arrive incrementally.
+      assert.ok(deltas.length > 1, `expected multiple live deltas, got ${deltas.length}`);
+      assert.strictEqual(deltas.join(''), chunks.join(''));
+    });
+  });
+});

ai-sdk/src/mocha/workflows.test.ts

@@ -1,12 +1,21 @@
 import { TestWorkflowEnvironment } from '@temporalio/testing';
-import { before, describe, it } from 'mocha';
+import { after, before, describe, it } from 'mocha';
 import { Worker } from '@temporalio/worker';
 import { haikuAgent } from '../workflows';
 import * as activities from '../activities';
 import { AiSdkPlugin } from '@temporalio/ai-sdk';
-import { openai } from '@ai-sdk/openai';
 import assert from 'assert';
 
+// `@ai-sdk/openai` is ESM-only. Under this sample's CommonJS build a static
+// `import` is emitted as `require()`, which throws on Node 22's require(ESM)
+// implementation ("Unexpected module status 0") at module-load time — aborting
+// the whole mocha run. Load it lazily via a native dynamic `import()` (wrapped
+// in `new Function` so TypeScript doesn't down-level it back to `require`),
+// inside `before` so it only runs when this suite isn't skipped.
+const importESM = new Function('specifier', 'return import(specifier)') as (
+  specifier: string,
+) => Promise<typeof import('@ai-sdk/openai')>;
+
 const hasOpenAIKey = Boolean(process.env.OPENAI_API_KEY);
 const describeWorkflow = hasOpenAIKey ? describe : describe.skip;
 
@@ -16,9 +25,11 @@ describeWorkflow(
     this.timeout(30_000);
 
     let testEnv: TestWorkflowEnvironment;
+    let openai: (typeof import('@ai-sdk/openai'))['openai'];
 
     before(async () => {
       testEnv = await TestWorkflowEnvironment.createLocal();
+      ({ openai } = await importESM('@ai-sdk/openai'));
     });
 
     after(async () => {

ai-sdk/src/workflows.ts

@@ -1,10 +1,10 @@
-import '@temporalio/ai-sdk/lib/load-polyfills';
-import { generateText, stepCountIs, tool, wrapLanguageModel } from 'ai';
-import { TemporalMCPClient, temporalProvider } from '@temporalio/ai-sdk';
+import { generateText, streamText, stepCountIs, tool, wrapLanguageModel } from 'ai';
+import type { LanguageModelMiddleware } from 'ai';
+import { TemporalMCPClient, TemporalProvider, temporalProvider } from '@temporalio/ai-sdk/workflow';
+import { WorkflowStream } from '@temporalio/workflow-streams/workflow';
 import type * as activities from './activities';
-import { proxyActivities } from '@temporalio/workflow';
+import { proxyActivities, condition, defineSignal, setHandler } from '@temporalio/workflow';
 import z from 'zod';
-import { LanguageModelV3Middleware } from '@ai-sdk/provider';
 
 const { getWeather } = proxyActivities<typeof activities>({
   startToCloseTimeout: '1 minute',
@@ -45,8 +45,8 @@ export async function toolsAgent(question: string): Promise<string> {
 // @@@SNIPSTART typescript-vercel-ai-sdk-middleware-agent
 export async function middlewareAgent(prompt: string): Promise<string> {
   const cache = new Map<string, any>();
-  const middleware: LanguageModelV3Middleware = {
-    specificationVersion: 'v3',
+  const middleware: LanguageModelMiddleware = {
+    specificationVersion: 'v4',
     wrapGenerate: async ({ doGenerate, params }) => {
       const cacheKey = JSON.stringify(params);
       if (cache.has(cacheKey)) {
@@ -73,6 +73,7 @@ export async function middlewareAgent(prompt: string): Promise<string> {
   });
   return result.text;
 }
+// @@@SNIPEND
 
 // @@@SNIPSTART typescript-vercel-ai-sdk-mcp-agent
 export async function mcpAgent(prompt: string): Promise<string> {
@@ -88,3 +89,61 @@ export async function mcpAgent(prompt: string): Promise<string> {
   return result.text;
 }
 // @@@SNIPEND
+
+// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-topic
+// The topic that streamed model deltas are published to. External consumers
+// subscribe to this topic by workflow id to receive live tokens as they are
+// generated. See `client.ts` for the consumer side.
+export const STREAM_TOPIC = 'text-stream';
+// @@@SNIPEND
+
+// @@@SNIPSTART typescript-vercel-ai-sdk-consumer-done-signal
+// A subscriber sends this signal once it has received the stream's final delta,
+// so the workflow knows it is safe to complete. See `client.ts` for the sender.
+export const consumerDoneSignal = defineSignal('consumer-done');
+// @@@SNIPEND
+
+// A provider whose language-model calls stream their deltas onto STREAM_TOPIC.
+// Setting `streamingTopic` enables `doStream`; without it, streaming calls
+// throw. Use a distinct topic per concurrent streaming call.
+const streamingProvider = new TemporalProvider({
+  languageModel: { streamingTopic: STREAM_TOPIC },
+});
+
+// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-agent
+export async function streamingAgent(prompt: string): Promise<string> {
+  // Host the WorkflowStream as the first statement of the workflow so its
+  // publish-signal handler is registered before the streaming activity starts
+  // publishing deltas to it.
+  new WorkflowStream();
+
+  // A subscriber flips this once it has consumed the final delta (see below).
+  let consumerDone = false;
+  setHandler(consumerDoneSignal, () => {
+    consumerDone = true;
+  });
+
+  const result = streamText({
+    model: streamingProvider.languageModel('gpt-4o-mini'),
+    prompt,
+    system: 'You only respond in haikus.',
+  });
+
+  // The model call runs in an activity that publishes each delta to
+  // STREAM_TOPIC for external subscribers. Inside the workflow the deltas are
+  // replayed after the activity completes, so this loop durably reassembles
+  // the full text.
+  let text = '';
+  for await (const delta of result.textStream) {
+    text += delta;
+  }
+
+  // The workflow's stream log lives in memory and is discarded once the run
+  // completes, which can race a subscriber's final poll. Rather than guess at a
+  // fixed delay, wait for the subscriber to signal that it received the last
+  // delta. The timeout is a fallback for when nothing is subscribed, so the run
+  // can't hang forever.
+  await condition(() => consumerDone, '10 seconds');
+  return text;
+}
+// @@@SNIPEND