Skip to main content

Vercel AI SDK Integration

Enforce Control Zero policies in applications built with the Vercel AI SDK.

Overview​

The Vercel AI SDK provides a unified interface for building AI-powered applications in TypeScript/JavaScript. Control Zero integrates through middleware that intercepts model calls and tool invocations before they execute.

Installation​

npm install @controlzero/sdk ai

Setup​

import { Client } from '@controlzero/sdk';
import { generateText, streamText, tool, gateway } from 'ai';

const cz = new Client({
  policy: {
    rules: [
      { allow: 'llm:generate', reason: 'LLM calls are permitted' },
      { allow: 'search_web:*', reason: 'Web search is permitted' },
      { deny: 'read_database:*', reason: 'Database access is blocked' },
    ],
  },
});

Basic Usage​

Enforce Before Generation​

Pass raiseOnDeny: true so a deny decision throws and blocks the AI call. Without it, guard() returns the decision but does not stop execution.

async function generate(prompt: string, model: string = 'openai/gpt-5.4') {
  // Enforce policy before the AI call
  await cz.guard('llm', {
    method: 'generate',
    args: { model },
    raiseOnDeny: true,
  });

  const result = await generateText({
    model: gateway(model),
    prompt,
  });

  return result.text;
}

Enforce Tool Calls​

AI SDK v6 tool definitions use inputSchema, not parameters.

import { z } from 'zod';

async function runAgent(prompt: string) {
  // Enforce the LLM call
  await cz.guard('llm', {
    method: 'generate',
    args: { model: 'openai/gpt-5.4' },
    raiseOnDeny: true,
  });

  const result = await generateText({
    model: gateway('openai/gpt-5.4'),
    prompt,
    tools: {
      searchWeb: tool({
        description: 'Search the web for information',
        inputSchema: z.object({ query: z.string() }),
        execute: async ({ query }) => {
          await cz.guard('search_web', { method: 'execute', raiseOnDeny: true });
          return `Results for: ${query}`;
        },
      }),
      readDatabase: tool({
        description: 'Query the database',
        inputSchema: z.object({ sql: z.string() }),
        execute: async ({ sql }) => {
          await cz.guard('read_database', { method: 'execute', raiseOnDeny: true });
          return `DB results for: ${sql}`;
        },
      }),
    },
  });

  return result.text;
}

Streaming with Enforcement​

async function streamResponse(prompt: string) {
  await cz.guard('llm', {
    method: 'generate',
    args: { model: 'openai/gpt-5.4' },
    raiseOnDeny: true,
  });

  const result = streamText({
    model: gateway('openai/gpt-5.4'),
    prompt,
  });

  return result.toUIMessageStreamResponse();
}

Next.js Route Handler Example​

// app/api/chat/route.ts
import { Client } from '@controlzero/sdk';
import { streamText, gateway } from 'ai';

const cz = new Client({ policyFile: './controlzero.yaml' });

export async function POST(req: Request) {
  const { messages } = await req.json();

  // Enforce before streaming. raiseOnDeny throws on deny and returns 500
  // unless you catch and convert.
  await cz.guard('llm', {
    method: 'generate',
    args: { model: 'openai/gpt-5.4' },
    raiseOnDeny: true,
  });

  const result = streamText({
    model: gateway('openai/gpt-5.4'),
    messages,
  });

  return result.toUIMessageStreamResponse();
}

First-party middleware: vercelAiMiddleware​

For teams who want one-line governance across every LLM call (instead of a guard() per route), the Node SDK ships a middleware factory that plugs directly into the AI SDK's wrapLanguageModel API.

npm install @controlzero/sdk ai
import { Client } from '@controlzero/sdk';
import { vercelAiMiddleware } from '@controlzero/sdk/integrations/vercelAi';
import { generateText, wrapLanguageModel, gateway } from 'ai';

const cz = new Client({ policyFile: './controlzero.yaml' });

// Wrap any Vercel AI SDK language model. Every generate/stream call is
// now policy-checked before going to the provider, and audited after.
const governedModel = wrapLanguageModel({
  model: gateway('openai/gpt-5.4'),
  middleware: vercelAiMiddleware(cz, { agentId: 'chat-backend' }),
});

const result = await generateText({
  model: governedModel,
  prompt: "Summarize today's board deck.",
});

What the middleware does on every call:

  1. Extracts the model + provider from the AI SDK request.
  2. Calls cz.guard('llm', { method: 'generate', args: { model }, context: { tags: { provider } }, raiseOnDeny: true }).
  3. On allow, passes the request through; on deny, throws before the provider is hit.
  4. After the call, records tokens, latency, and decision metadata to the audit trail.

Use this when you want governance to be invisible to feature code and enforced consistently across every AI SDK call path.

Example Policy​

{
  "name": "vercel-ai-policy",
  "rules": [
    {
      "effect": "allow",
      "action": "llm:generate",
      "resource": "model/gpt-5.4"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/search_web"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/read_database"
    }
  ]
}

Python: ControlZeroVercelAIMiddleware​

If you are building a Python backend that calls Vercel AI SDK endpoints (or any OpenAI-compatible endpoint from Python), use the first-party Python middleware.

from controlzero import Client
from controlzero.integrations.vercel_ai import ControlZeroVercelAIMiddleware

cz = Client(api_key="cz_live_your_key_here")
middleware = ControlZeroVercelAIMiddleware(cz, agent_id="my-python-backend")

# Before calling the AI endpoint
result = middleware.before_call(model="gpt-5.4", action="chat")
if not result["allowed"]:
    raise PermissionError(f"Blocked: {result['decision'].reason}")

# Call the model...
response = call_your_ai_endpoint(...)

# After the call -- logs tokens and latency to the audit trail
middleware.after_call(
    model="gpt-5.4",
    action="chat",
    input_tokens=response.usage.input_tokens,
    output_tokens=response.usage.output_tokens,
    duration_ms=elapsed_ms,
)

Or wrap an async generate function end-to-end:

import asyncio

async def my_generate(prompt: str) -> str:
    # ... your model call here
    return response_text

governed_generate = middleware.wrap_generate(my_generate)

# Every call is now governed
result = await governed_generate("Summarize this document")

Install: pip install controlzero (no extra packages needed).

Next Steps​