⚡ langgraph-zod

The runtime type system & structural API boundary layer for LangGraphJS agents.

---

🌟 Why langgraph-zod?

While TypeScript provides compile-time safety, LLMs operate in the untyped world of unstructured text and JSON strings. langgraph-zod bridges that gap by deriving production-ready runtime boundary schemas directly from your LangGraph state channel definitions.

                    ┌─────────────────────────────────┐
                    │      Incoming HTTP / REST Payload│
                    └────────────────┬────────────────┘
                                     │
                        ┌────────────▼────────────┐
                        │   inputSchema.parse()   │ (Validates & Coerces)
                        └────────────┬────────────┘
                                     │
                        ┌────────────▼────────────┐
                        │ LangGraph Execution Loop│
                        └────────────┬────────────┘
                                     │
                        ┌────────────▼────────────┐
                        │  outputSchema.parse()   │ (Strips Internal Keys)
                        └────────────┬────────────┘
                                     │
                    ┌────────────────▼────────────────┐
                    │     Clean Public REST API Output │
                    └─────────────────────────────────┘

---

🚀 Key Features

• 🛡️ Boundary Validation (createGraphSchemas): Derives inputSchema (partial with optional required fields) and outputSchema (omitting internal scratchpads/keys) from state definitions.
• 📡 Streaming-First SSE Validation (v0.2.0): Incremental SSE parser + TransformStream that validates graph.streamEvents() payloads on the fly, drops malformed chunks, and handles backpressure — with <1ms per-chunk validation latency.
• 🌐 Framework-Agnostic AgentBoundary (v0.2.0): Wrap any async function or async generator — LangGraph, Vercel AI SDK, AutoGen TS, or plain custom agents — with the same input/output validation boundary.
• 🧰 Vercel AI SDK Adapter (v0.2.0): Validate tool() arguments/results and streamText() params/fullStream parts without taking a hard dependency on the ai package.
• 🔮 Zod v3 & v4 Resilient JSON Schema (v0.2.0): All internal AST inspection flows through a safe introspection layer (zod-ast) — with support for discriminated unions, intersections, recursive ZodLazy, tuples, records, and more.
• 📈 Production Telemetry (v0.2.0): Structural OpenTelemetry hooks (validation.error_count, channel.failed_key, latency histograms) plus Pino/Winston formatters for SchemaValidationError.
• 📦 Zod-Native Channel Reducers (zodChannel): Type-safe channel reducers for LangGraph Annotation.Root state (arrays, sums, latest value, custom accumulators).
• 🔄 LLM Output Parsing & Self-Healing (safeParseLLMOutput): Extracts JSON from markdown blocks, validates against Zod schemas, and retries with repair functions.
• ⚡ HTTP Framework Adapters: Turn any LangGraph agent into a validated API endpoint for Next.js App Router, Hono, or Express/Fastify in 3 lines — now with SSE streaming variants (createNextStreamHandler, createHonoStreamHandler).
• 📋 OpenAPI 3.1 & JSON Schema Generator: Automatically output interactive OpenAPI specifications with /invoke and /batch endpoints.
• 📜 Schema Registry & Breaking Change Diffing (GraphSchemaRegistry): Version your agent schemas and detect breaking API changes in CI before deployment.
• ⚙️ Composable Middleware Pipeline: Hook node-level logging, retries, and strict schema validation before and after node execution.

---

📦 Installation

# pnpm
pnpm add langgraph-zod zod @langchain/langgraph

# npm
npm install langgraph-zod zod @langchain/langgraph

# yarn
yarn add langgraph-zod zod @langchain/langgraph

---

⚡ Quickstart

1. Derive Boundary Schemas

import { z } from "zod";
import { createGraphSchemas } from "langgraph-zod";

// Define your agent channels with Zod
const AgentChannels = {
  userQuery: z.string().describe("The primary prompt from the user"),
  chatHistory: z.array(z.string()).default([]),
  internalScratchpad: z.string().optional().describe("Internal agent thoughts"),
  retryCount: z.number().default(0),
};

// Derive runtime input, output, and full state boundary schemas
export const { inputSchema, outputSchema, stateSchema, validateInput } =
  createGraphSchemas(AgentChannels, {
    omitOutputKeys: ["internalScratchpad", "retryCount"],
    requiredInputKeys: ["userQuery"],
  });

2. Wrap your LangGraph Execution Loop

import { createValidatedGraph } from "langgraph-zod";
import { myLangGraphAgent } from "./agent";

export const agentApi = createValidatedGraph(
  async (input) => {
    return await myLangGraphAgent.invoke(input);
  },
  {
    inputSchema,
    outputSchema,
    strict: true,
  }
);

const response = await agentApi.invoke({ userQuery: "Analyze revenue" });
// response.data contains validated query and chatHistory, with internalScratchpad stripped!

3. Expose as Next.js Route Handler

// app/api/agent/route.ts
import { createNextRouteHandler } from "langgraph-zod/adapters";
import { agentApi } from "@/lib/agent";

export const { POST, GET } = createNextRouteHandler({ graph: agentApi });

4. Stream validated agent events (SSE) — v0.2.0

// app/api/agent/stream/route.ts
import { createNextStreamHandler } from "langgraph-zod/adapters";
import { z } from "zod";

export const { POST } = createNextStreamHandler({
  graph: compiledGraph, // any graph exposing streamEvents()
  schemas,              // optional: validates the inbound POST payload
  eventSchemas: {       // per-event data-payload schemas
    on_chat_model_stream: z.object({ chunk: z.any() }).passthrough(),
    on_tool_end: z.object({ output: z.string() }).passthrough(),
  },
  onInvalidEvent: (info) => console.warn("dropped event", info),
});

Invalid or malformed events are dropped before they reach the client; the pull-based ReadableStream applies natural backpressure to slow consumers. For lower-level control, use createSSEStreamTransformer() (SSE wire format) or validateEventStream() (object-mode event streams) from langgraph-zod/streaming.

5. Wrap any agent framework — v0.2.0

import { AgentBoundary } from "langgraph-zod";

const boundary = new AgentBoundary({
  inputSchema: z.object({ prompt: z.string() }),
  outputSchema: z.object({ text: z.string() }),
});

// Any async function — Vercel AI SDK, AutoGen TS, custom agents:
export const run = boundary.wrapAsyncFn(async (input) => myAgent(input));

// Any async generator — every yielded chunk is validated:
export const stream = boundary.wrapGenerator(async function* (input) {
  yield* myAgent.stream(input);
});

6. Observe validation in production — v0.2.0

import { createValidationTelemetry, pinoValidationErrorSerializer } from "langgraph-zod";
import { trace, metrics } from "@opentelemetry/api";

const telemetry = createValidationTelemetry({
  tracer: trace.getTracer("my-agent"),
  meter: metrics.getMeter("my-agent"),
});

// Emits span + counters + duration histogram with
// validation.error_count / channel.failed_key attributes on failure:
const result = telemetry.traceValidation("validateInput", () =>
  schemas.validateInput(payload)
);

// Pino: pino({ serializers: { err: pinoValidationErrorSerializer } })
// Winston: winston.format(winstonValidationErrorFormat)

---

🛠️ CLI Usage

# Export OpenAPI 3.1 Specification
npx langgraph-zod openapi ./src/schemas.ts --out openapi.json

# Check for breaking schema changes in CI
npx langgraph-zod diff ./schemas.v1.ts ./schemas.v2.ts

---

🔗 Repository

https://github.com/shahrryyar/langgraph-zod

---

📜 License

ISC © 2026