Executable source

Examples

The code shown here is read from the repository at build time. Fixture runs need no credentials; live provider runs keep keys inside your own development environment.

Open full environment

Progressive JSON

Core progressive iteration, completion events, Unicode, and a mid-stream decision.

bun run example:progressive

Read guide Source

import { isDeepStrictEqual } from "node:util"
import { z } from "zod"

import { SchemaStream, type SchemaStreamChunk } from "../src"

const schema = z.object({
  title: z.string(),
  summary: z.string(),
  details: z.object({ score: z.number(), note: z.string().nullable() }),
  tags: z.array(z.string()),
  metrics: z.record(z.string(), z.number())
})
const expected: z.output<typeof schema> = {
  title: "Progressive JSON",
  summary: 'Unicode 🌊 日本語, quotes "like this", and newlines\narrive incrementally.',
  details: { note: null, score: 0.98 },
  tags: ["streaming", "typed"],
  metrics: { chunks: 9, records: 2 }
}

function splitJson(json: string, targetChunks = 9): string[] {
  const codePoints = Array.from(json)
  const chunkLength = Math.max(1, Math.ceil(codePoints.length / targetChunks))
  const chunks: string[] = []
  let offset = 0

  while (offset < codePoints.length) {
    chunks.push(codePoints.slice(offset, offset + chunkLength).join(""))
    offset += chunkLength
  }

SDK compatibility

OpenAI Agents SDK and Vercel AI SDK streams against deterministic mock models.

bun run example:sdk

Read guide Source

import {
  Agent,
  type Model,
  type ModelResponse,
  Runner,
  type StreamEvent,
  setTraceProcessors,
  setTracingDisabled
} from "@openai/agents"
import { Output, simulateReadableStream, streamText } from "ai"
import { MockLanguageModelV4 } from "ai/test"
import { z } from "zod"

import { SchemaStream, type SchemaStreamChunk, type SchemaStreamSource } from "../src"

setTraceProcessors([])
setTracingDisabled(true)

const schema = z.object({
  headline: z.string(),
  locale: z.string(),
  sections: z.array(
    z.object({
      title: z.string(),
      facts: z.array(z.object({ label: z.string(), value: z.string() }))
    })
  )
})
const expected: z.output<typeof schema> = {
  headline: "A deterministic 🌊 SDK stream",

Mastra

Mastra structured output with the same SchemaStream materialization boundary.

bun run example:mastra

Read guide Source

import { isDeepStrictEqual } from "node:util"
import { Agent } from "@mastra/core/agent"
import { simulateReadableStream } from "ai"
import { MockLanguageModelV4 } from "ai/test"
import { z } from "zod"

import { SchemaStream, type SchemaStreamChunk, type SchemaStreamSource } from "../src"

const schema = z.object({
  title: z.string(),
  locale: z.enum(["en", "ja"]),
  itinerary: z.array(
    z.object({
      day: z.number(),
      stops: z.array(
        z.object({
          name: z.string(),
          note: z.string().nullable(),
          tags: z.array(z.string())
        })
      )
    })
  ),
  metadata: z.record(z.string(), z.string())
})

const expected: z.output<typeof schema> = {
  title: "週末 plan 🌊",
  locale: "ja",
  itinerary: [

WebSocket dashboard

Bun WebSocket transport, selectable snapshot policies, and the live dashboard UI.

bun run example:websocket

Read guide Source

import { isDeepStrictEqual } from "node:util"
import {
  Agent,
  type Model,
  type ModelResponse,
  Runner,
  type StreamEvent,
  setTraceProcessors,
  setTracingDisabled
} from "@openai/agents"
import { z } from "zod"

import {
  SchemaStream,
  type SchemaStreamChunk,
  type SchemaStreamSource,
  type SchemaStreamValuePath,
  type SnapshotPolicy
} from "../../src"

setTraceProcessors([])
setTracingDisabled(true)

const protocolVersion = 1 as const
const defaultPort = 3400
const defaultModel = "gpt-5.6-luna"
const maxPromptLength = 800
const decisionPath = ["triage", "requiresApproval"] as const
const fixtureBrief = "Customer-facing release with three open checks and an approval gate."
const trailingSlashPattern = /\/$/