Middleware

Middleware

Middleware lets you intercept and transform requests, responses, and stream events.

Basic Middleware

import { llm, Middleware } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';

const loggingMiddleware: Middleware = {
  name: 'logging',

  onStart(ctx) {
    console.log('Request starting:', ctx.modelId);
    ctx.state.set('startTime', Date.now());
  },

  onEnd(ctx) {
    const duration = Date.now() - (ctx.state.get('startTime') as number);
    console.log(`Request completed in ${duration}ms`);
  },
};

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [loggingMiddleware],
});

Middleware Interface

interface Middleware {
  readonly name: string;

  // Lifecycle hooks
  onStart?(ctx: MiddlewareContext): void | Promise<void>;
  onEnd?(ctx: MiddlewareContext): void | Promise<void>;
  onError?(error: Error, ctx: MiddlewareContext): void | Promise<void>;  // Non-cancellation errors
  onAbort?(ctx: MiddlewareContext): void | Promise<void>;  // Called on cancellation
  onRetry?(attempt: number, error: Error, ctx: MiddlewareContext): void | Promise<void>;

  // Request/Response hooks
  onRequest?(ctx: MiddlewareContext): void | Promise<void>;
  onResponse?(ctx: MiddlewareContext): void | Promise<void>;
  onTurn?(turn: Turn, ctx: MiddlewareContext): void | Promise<void>;  // LLM only

  // Stream hooks (LLM, Image)
  onStreamEvent?(
    event: StreamEvent,
    ctx: StreamContext
  ): StreamEvent | StreamEvent[] | null;
  onStreamEnd?(ctx: StreamContext): void | Promise<void>;

  // Tool hooks (LLM only)
  onToolCall?(tool: Tool, params: unknown, ctx: MiddlewareContext): void | Promise<void>;
  onToolResult?(tool: Tool, result: unknown, ctx: MiddlewareContext): void | Promise<void>;
}

Middleware Context

interface MiddlewareContext {
  readonly modality: 'llm' | 'embedding' | 'image';
  readonly modelId: string;
  readonly provider: string;
  readonly streaming: boolean;

  request: AnyRequest;     // Mutable - modify before sending
  response?: AnyResponse;  // Mutable - modify after receiving

  readonly state: Map<string, unknown>;  // Share data between hooks
  readonly startTime: number;
  endTime?: number;

  emit(event: StreamEvent): void;  // Emit events to stream consumers
}

Built-in Middleware

Middleware is imported from dedicated entry points, not the main package:

Logging Middleware

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { loggingMiddleware } from '@providerprotocol/ai/middleware/logging';

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [
    loggingMiddleware({
      level: 'debug',      // 'debug', 'info', 'warn', 'error'
      logRequest: true,
      logResponse: true,
      logTiming: true,
    }),
  ],
});

Parsed Object Middleware

Stream partial JSON for structured output:

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { parsedObjectMiddleware } from '@providerprotocol/ai/middleware/parsed-object';

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  structure: { /* schema */ },
  middleware: [parsedObjectMiddleware()],
});

const stream = claude.stream('Extract data...');

for await (const event of stream) {
  if (event.type === 'object_delta') {
    console.log('Partial:', event.delta.parsed);
  }
}

Persistence Middleware

Load and save conversation threads around LLM requests:

import { llm, Thread } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { persistenceMiddleware, PersistenceAdapter } from '@providerprotocol/ai/middleware/persistence';

const adapter = new PersistenceAdapter({
  id: 'conversation-123',
  load: async (id) => loadThreadFromDatabase(id), // Thread | ThreadJSON | null
  save: async (id, thread, turn) => {
    await saveThreadToDatabase(id, thread);
    if (turn) {
      await saveTurnToDatabase(id, turn);
    }
  },
});

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [persistenceMiddleware({ adapter })],
});

const turn = await claude.generate('Hello!');

The persistence middleware automatically loads conversation history before each request and saves the updated thread after completion. The save callback receives the complete thread and the new turn (if any).

Pub-Sub Middleware (Stream Resumption)

Enable reconnecting clients to catch up on missed events during active generation. The middleware buffers events and publishes them to subscribers. Streams are created lazily on first use.

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';

// Create a shared adapter instance (singleton per process)
const adapter = memoryAdapter({ maxStreams: 500 });

const instance = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [
    pubsubMiddleware({
      adapter,
      streamId: 'unique-stream-id',  // Client-provided ID for reconnection
    }),
  ],
});

Server-Side: Handling Reconnections

The middleware buffers events and server routes handle reconnection logic. Use the fire-and-forget pattern with .then() to save completed turns without blocking the response.

Web API

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';
import { webapi } from '@providerprotocol/ai/middleware/pubsub/server';

const adapter = memoryAdapter();

Bun.serve({
  port: 3000,
  async fetch(req: Request) {
    const { messages, streamId } = await req.json();

    if (!await adapter.exists(streamId)) {
      // Fire-and-forget: start generation and save when complete
      const model = llm({
        model: anthropic('claude-sonnet-4-20250514'),
        middleware: [pubsubMiddleware({ adapter, streamId })],
      });
      model.stream(messages).then(turn => saveToDatabase(streamId, turn));
    }

    // Both new and reconnecting clients subscribe to events
    return new Response(webapi.createSubscriberStream(streamId, adapter), {
      headers: { 'Content-Type': 'text/event-stream' },
    });
  },
});

Express

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';
import { express } from '@providerprotocol/ai/middleware/pubsub/server';

const adapter = memoryAdapter();

app.post('/api/ai', async (req, res) => {
  const { messages, streamId } = req.body;

  if (!await adapter.exists(streamId)) {
    const model = llm({
      model: anthropic('claude-sonnet-4-20250514'),
      middleware: [pubsubMiddleware({ adapter, streamId })],
    });
    model.stream(messages).then(turn => saveToDatabase(streamId, turn));
  }

  express.streamSubscriber(streamId, adapter, res);
});

Fastify

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';
import { fastify } from '@providerprotocol/ai/middleware/pubsub/server';

const adapter = memoryAdapter();

app.post('/api/ai', async (request, reply) => {
  const { messages, streamId } = request.body;

  if (!await adapter.exists(streamId)) {
    const model = llm({
      model: anthropic('claude-sonnet-4-20250514'),
      middleware: [pubsubMiddleware({ adapter, streamId })],
    });
    model.stream(messages).then(turn => saveToDatabase(streamId, turn));
  }

  return fastify.streamSubscriber(streamId, adapter, reply);
});

H3/Nuxt

server/api/ai.post.ts

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';
import { h3 } from '@providerprotocol/ai/middleware/pubsub/server';

const adapter = memoryAdapter();

export default defineEventHandler(async (event) => {
  const { messages, streamId } = await readBody(event);

  if (!await adapter.exists(streamId)) {
    const model = llm({
      model: anthropic('claude-sonnet-4-20250514'),
      middleware: [pubsubMiddleware({ adapter, streamId })],
    });
    model.stream(messages).then(turn => saveToDatabase(streamId, turn));
  }

  return h3.streamSubscriber(streamId, adapter, event);
});

Custom Storage Adapters

Implement PubSubAdapter for custom backends (Redis, etc.). Streams are created lazily on first append() or subscribe() call.

import type { PubSubAdapter, SubscriptionCallback, CompletionCallback } from '@providerprotocol/ai/middleware/pubsub';

const redisAdapter: PubSubAdapter = {
  async exists(streamId) { /* check if stream exists */ },
  async append(streamId, event) { /* append event (create stream if needed) */ },
  async getEvents(streamId) { /* return events array ([] if not found) */ },
  subscribe(streamId, onEvent: SubscriptionCallback, onComplete: CompletionCallback) {
    /* subscribe to events; call onComplete when stream ends */
    return () => { /* unsubscribe */ };
  },
  publish(streamId, event) { /* broadcast event to subscribers */ },
  async remove(streamId) { /* notify onComplete callbacks, then delete */ },
};

Note

The remove() method must notify all subscribers via their onComplete callback before deleting the stream. This signals clients to close their connections gracefully.

Pipeline Middleware

Run async post-turn processing stages (image generation, embeddings, slug creation, etc.) with progress events streamed to connected clients.

import { llm } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';
import { pubsubMiddleware, memoryAdapter } from '@providerprotocol/ai/middleware/pubsub';
import { pipelineMiddleware, isPipelineStageEvent } from '@providerprotocol/ai/middleware/pipeline';

const adapter = memoryAdapter();

const model = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  structure: BlogPostSchema,
  middleware: [
    pubsubMiddleware({ adapter, streamId: postId }),
    pipelineMiddleware<BlogPost>({
      stages: [
        {
          type: 'slug',
          run: (turn, emit) => {
            const slug = turn.data!.title.toLowerCase().replace(/\s+/g, '-');
            emit({ slug });
          },
        },
        {
          type: 'embedding',
          run: async (turn, emit) => {
            await vectorize(turn.data!);
            emit({ embedded: true });
          },
        },
      ],
      parallel: false,
      continueOnError: false,
      onStageError: ({ stage, error }) => {
        console.error(`Stage ${stage.type} failed:`, error);
      },
    }),
  ],
});

Note

Place pipelineMiddleware after pubsubMiddleware in the middleware array. This ensures pubsub sets up before pipeline stages execute, and pipeline events are visible to pubsub subscribers.

Consuming Pipeline Events

Pipeline stages emit events that flow through onStreamEvent and are visible to for await consumers:

for await (const event of model.stream(prompt)) {
  if (isPipelineStageEvent(event)) {
    console.log(`Stage: ${event.delta.stage}`, event.delta.payload);
  }
}

Emitting Custom Events

Any middleware can emit stream events using ctx.emit(). Events flow through onStreamEvent for all middleware and are visible to direct stream consumers.

const progressMiddleware: Middleware = {
  name: 'progress',
  async onTurn(turn, ctx) {
    ctx.emit({
      type: 'processing_start',
      index: 0,
      delta: { status: 'processing' },
    });

    await doSomeWork(turn);

    ctx.emit({
      type: 'processing_complete',
      index: 0,
      delta: { status: 'done' },
    });
  },
};

Tip

StreamEvent.type accepts custom string types while preserving autocomplete for built-in types. No spec changes needed to emit custom events from middleware.

Request Transformation

Modify requests before they’re sent:

const addMetadataMiddleware: Middleware = {
  name: 'add-metadata',

  onRequest(ctx) {
    // Add custom headers
    ctx.request.config = {
      ...ctx.request.config,
      headers: {
        ...ctx.request.config?.headers,
        'X-Request-ID': crypto.randomUUID(),
      },
    };
  },
};

Response Transformation

Modify responses after they’re received:

const sanitizeMiddleware: Middleware = {
  name: 'sanitize',

  onResponse(ctx) {
    if (ctx.modality === 'llm' && ctx.response) {
      // Sanitize response content
      for (const message of ctx.response.messages) {
        // Process message content
      }
    }
  },
};

Stream Event Transformation

Filter or transform stream events:

const filterMiddleware: Middleware = {
  name: 'filter',

  onStreamEvent(event, ctx) {
    // Filter out reasoning blocks
    if (event.type === 'reasoning_delta') {
      return null; // Remove this event
    }

    // Transform text events
    if (event.type === 'text_delta') {
      return {
        ...event,
        delta: {
          ...event.delta,
          text: event.delta.text?.toUpperCase(),
        },
      };
    }

    return event;
  },
};

Error Handling

Handle errors in middleware. The onError hook is called for non-cancellation errors, while onAbort is called when a request is cancelled:

const errorMiddleware: Middleware = {
  name: 'error-handler',

  onError(error, ctx) {
    // Called for non-cancellation errors
    console.error(`Error in ${ctx.provider}/${ctx.modelId}:`, error.message);

    logToMonitoring({
      error: error.message,
      provider: ctx.provider,
      model: ctx.modelId,
      duration: ctx.endTime ? ctx.endTime - ctx.startTime : 0,
    });
  },

  onAbort(ctx) {
    // Called when request is cancelled
    console.log(`Request cancelled: ${ctx.provider}/${ctx.modelId}`);
  },
};

Tool Interception

Monitor or modify tool calls:

const toolMiddleware: Middleware = {
  name: 'tool-monitor',

  onToolCall(tool, params, ctx) {
    console.log(`Tool called: ${tool.name}`, params);
  },

  onToolResult(tool, result, ctx) {
    console.log(`Tool result: ${tool.name}`, result);
  },
};

Execution Order

Multiple middleware execute in specific order:

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [first, second, third],
});

// Execution order:
// Request phase:  first → second → third
// Response phase: third → second → first (reverse)
// Turn phase:     third → second → first (reverse, LLM only)
// Error phase:    all notified (non-cancellation errors)
// Abort phase:    all notified (cancellation only)

// onTurn hooks run for all middleware even if one throws.
// The first error is re-thrown after all middleware complete.
// This ensures cleanup middleware (e.g. pubsub) always runs.

Sharing State

Use the context state map to share data between hooks:

const timingMiddleware: Middleware = {
  name: 'timing',

  onStart(ctx) {
    ctx.state.set('startTime', performance.now());
  },

  onRequest(ctx) {
    ctx.state.set('requestTime', performance.now());
  },

  onResponse(ctx) {
    const requestTime = ctx.state.get('requestTime') as number;
    console.log('Response latency:', performance.now() - requestTime, 'ms');
  },

  onEnd(ctx) {
    const startTime = ctx.state.get('startTime') as number;
    console.log('Total time:', performance.now() - startTime, 'ms');
  },
};

Async Middleware

Hooks can be async:

const authMiddleware: Middleware = {
  name: 'auth',

  async onRequest(ctx) {
    // Fetch fresh token
    const token = await refreshAuthToken();
    ctx.request.config = {
      ...ctx.request.config,
      headers: {
        ...ctx.request.config?.headers,
        'Authorization': `Bearer ${token}`,
      },
    };
  },
};

Complete Example

import { llm, Middleware, UPPError } from '@providerprotocol/ai';
import { anthropic } from '@providerprotocol/ai/anthropic';

// Analytics middleware
const analyticsMiddleware: Middleware = {
  name: 'analytics',

  onStart(ctx) {
    ctx.state.set('requestId', crypto.randomUUID());
    ctx.state.set('startTime', Date.now());
  },

  onRequest(ctx) {
    const requestId = ctx.state.get('requestId');
    console.log(`[${requestId}] Starting ${ctx.provider}/${ctx.modelId}`);
  },

  onResponse(ctx) {
    const requestId = ctx.state.get('requestId');
    const duration = Date.now() - (ctx.state.get('startTime') as number);

    // Track usage
    trackAnalytics({
      requestId,
      provider: ctx.provider,
      model: ctx.modelId,
      duration,
      streaming: ctx.streaming,
    });
  },

  onError(error, ctx) {
    const requestId = ctx.state.get('requestId');

    trackError({
      requestId,
      provider: ctx.provider,
      model: ctx.modelId,
      error: error instanceof UPPError ? error.code : 'UNKNOWN',
      message: error.message,
    });
  },

  onAbort(ctx) {
    const requestId = ctx.state.get('requestId');
    console.log(`[${requestId}] Request cancelled`);
  },

  onTurn(turn, ctx) {
    const requestId = ctx.state.get('requestId');
    console.log(`[${requestId}] Turn completed with ${turn.messages.length} messages`);
  },

  onStreamEvent(event, ctx) {
    // Count tokens for streaming
    if (event.type === 'text_delta') {
      const tokens = (ctx.state.get('streamTokens') as number) || 0;
      ctx.state.set('streamTokens', tokens + 1);
    }
    return event;
  },

  onStreamEnd(ctx) {
    const tokens = ctx.state.get('streamTokens');
    console.log('Streamed tokens:', tokens);
  },
};

const claude = llm({
  model: anthropic('claude-sonnet-4-20250514'),
  middleware: [analyticsMiddleware],
});

Tip

Keep middleware focused on a single concern. Compose multiple small middleware rather than building one large one.