Runtime Middleware

CopilotKit Runtime provides middleware hooks that allow you to intercept, modify, and monitor HTTP requests and responses. This is essential for authentication, logging, request transformation, and response analysis.

Overview

Middleware in CopilotKit operates at two lifecycle stages:

Before Request - Runs before the request reaches the handler (authentication, validation, request modification)
After Request - Runs after the handler returns a response (logging, analytics, response processing)

All middleware runs in the same process as your runtime for optimal performance and simplicity.

Middleware Types

Before Request Middleware

Intercepts requests before they reach the agent runtime. Can modify the request or return early with a custom response.

import { CopilotRuntime } from "@copilotkit/runtime";
import type { BeforeRequestMiddlewareFn } from "@copilotkit/runtime";

const beforeMiddleware: BeforeRequestMiddlewareFn = async ({ 
  runtime, 
  request, 
  path 
}) => {
  // Access request properties
  console.log("Request path:", path);
  console.log("Request method:", request.method);
  
  // Optionally modify and return a new request
  const modifiedRequest = new Request(request.url, {
    ...request,
    headers: {
      ...Object.fromEntries(request.headers),
      "X-Custom-Header": "value"
    }
  });
  
  return modifiedRequest;
};

const runtime = new CopilotRuntime({
  agents: { default: myAgent },
  beforeRequestMiddleware: beforeMiddleware
});

After Request Middleware

Runs after the handler completes. Useful for logging, analytics, and response processing.

import type { AfterRequestMiddlewareFn } from "@copilotkit/runtime";

const afterMiddleware: AfterRequestMiddlewareFn = async ({ 
  runtime, 
  response, 
  path,
  messages,
  threadId,
  runId
}) => {
  // Access response data
  console.log("Response status:", response.status);
  console.log("Thread ID:", threadId);
  console.log("Run ID:", runId);
  
  // Process SSE messages if available
  if (messages) {
    console.log(`Processed ${messages.length} messages`);
    messages.forEach(msg => {
      console.log(`Event: ${msg.event}, Data:`, msg.data);
    });
  }
};

const runtime = new CopilotRuntime({
  agents: { default: myAgent },
  afterRequestMiddleware: afterMiddleware
});

Common Use Cases

Authentication

Verify user identity before processing requests:

const authMiddleware: BeforeRequestMiddlewareFn = async ({ request }) => {
  const authHeader = request.headers.get("Authorization");
  
  if (!authHeader) {
    throw new Response("Unauthorized", { 
      status: 401,
      headers: { "WWW-Authenticate": "Bearer" }
    });
  }
  
  const token = authHeader.replace("Bearer ", "");
  
  try {
    const user = await verifyToken(token);
    
    // Add user info to request for downstream handlers
    const modifiedRequest = new Request(request.url, request);
    modifiedRequest.headers.set("X-User-Id", user.id);
    modifiedRequest.headers.set("X-User-Role", user.role);
    
    return modifiedRequest;
  } catch (error) {
    throw new Response("Invalid token", { status: 401 });
  }
};

Request Logging

Log all incoming requests for debugging and monitoring:

const loggingMiddleware: BeforeRequestMiddlewareFn = async ({ 
  request, 
  path 
}) => {
  const startTime = Date.now();
  
  console.log({
    timestamp: new Date().toISOString(),
    method: request.method,
    path,
    headers: Object.fromEntries(request.headers),
    userAgent: request.headers.get("User-Agent")
  });
  
  // Store start time for duration calculation
  request.headers.set("X-Start-Time", startTime.toString());
  
  return request;
};

Rate Limiting

Implement rate limiting based on user or IP:

const rateLimiter = new Map<string, { count: number; resetAt: number }>();

const rateLimitMiddleware: BeforeRequestMiddlewareFn = async ({ request }) => {
  const userId = request.headers.get("X-User-Id") || "anonymous";
  const now = Date.now();
  
  const limit = rateLimiter.get(userId);
  
  if (limit) {
    if (now < limit.resetAt) {
      if (limit.count >= 100) {
        throw new Response("Rate limit exceeded", { 
          status: 429,
          headers: {
            "Retry-After": Math.ceil((limit.resetAt - now) / 1000).toString()
          }
        });
      }
      limit.count++;
    } else {
      // Reset limit
      rateLimiter.set(userId, { count: 1, resetAt: now + 60000 });
    }
  } else {
    rateLimiter.set(userId, { count: 1, resetAt: now + 60000 });
  }
  
  return request;
};

Request Validation

Validate request body structure before processing:

const validationMiddleware: BeforeRequestMiddlewareFn = async ({ 
  request, 
  path 
}) => {
  if (path === "/api/copilotkit" && request.method === "POST") {
    try {
      const body = await request.clone().json();
      
      // Validate required fields
      if (!body.messages || !Array.isArray(body.messages)) {
        throw new Response(
          JSON.stringify({ error: "Invalid request: messages array required" }),
          { status: 400, headers: { "Content-Type": "application/json" } }
        );
      }
      
      // Validate message structure
      for (const msg of body.messages) {
        if (!msg.role || !msg.content) {
          throw new Response(
            JSON.stringify({ error: "Invalid message structure" }),
            { status: 400, headers: { "Content-Type": "application/json" } }
          );
        }
      }
    } catch (error) {
      if (error instanceof Response) throw error;
      throw new Response("Invalid JSON", { status: 400 });
    }
  }
  
  return request;
};

Analytics and Monitoring

Track agent usage, performance, and errors:

const analyticsMiddleware: AfterRequestMiddlewareFn = async ({ 
  response,
  path,
  messages,
  threadId,
  runId
}) => {
  const startTime = parseInt(response.headers.get("X-Start-Time") || "0");
  const duration = startTime ? Date.now() - startTime : 0;
  
  // Track metrics
  await analytics.track({
    event: "agent_request_completed",
    properties: {
      path,
      status: response.status,
      duration,
      threadId,
      runId,
      messageCount: messages?.length || 0,
      success: response.ok
    }
  });
  
  // Track specific event types from SSE stream
  if (messages) {
    const eventTypes = messages.map(m => m.event);
    const toolCalls = eventTypes.filter(t => t === "TOOL_CALL_STREAMING_START").length;
    
    await analytics.track({
      event: "agent_tool_usage",
      properties: {
        threadId,
        runId,
        toolCallCount: toolCalls
      }
    });
  }
};

Error Handling

Catch and log errors consistently:

const errorHandlingMiddleware: AfterRequestMiddlewareFn = async ({ 
  response,
  path,
  threadId,
  runId
}) => {
  if (!response.ok) {
    // Log error details
    console.error({
      timestamp: new Date().toISOString(),
      path,
      status: response.status,
      threadId,
      runId,
      statusText: response.statusText
    });
    
    // Send error to monitoring service
    await errorMonitoring.captureError({
      level: "error",
      message: `Agent request failed: ${response.status}`,
      context: { path, threadId, runId }
    });
  }
};

Response Caching

Cache responses for improved performance:

const cache = new Map<string, { response: Response; expiresAt: number }>();

const cachingMiddleware: BeforeRequestMiddlewareFn = async ({ 
  request, 
  path 
}) => {
  if (request.method !== "GET") return request;
  
  const cacheKey = `${path}:${request.url}`;
  const cached = cache.get(cacheKey);
  
  if (cached && Date.now() < cached.expiresAt) {
    // Return cached response
    return cached.response.clone();
  }
  
  return request;
};

const cacheStorageMiddleware: AfterRequestMiddlewareFn = async ({ 
  response,
  path
}) => {
  if (response.ok && response.headers.get("Content-Type")?.includes("application/json")) {
    const cacheKey = `${path}:${response.url}`;
    cache.set(cacheKey, {
      response: response.clone(),
      expiresAt: Date.now() + 60000 // 1 minute TTL
    });
  }
};

SSE Stream Processing

The after middleware provides parsed SSE messages for stream analysis:

const streamAnalysisMiddleware: AfterRequestMiddlewareFn = async ({ 
  messages,
  threadId,
  runId
}) => {
  if (!messages) return;
  
  // Analyze the event stream
  const eventCounts = new Map<string, number>();
  
  for (const message of messages) {
    const count = eventCounts.get(message.event) || 0;
    eventCounts.set(message.event, count + 1);
    
    // Extract specific data from events
    if (message.event === "STATE_SNAPSHOT") {
      console.log("State snapshot:", message.data);
    }
    
    if (message.event === "TOOL_CALL_RESULT") {
      console.log("Tool result:", message.data);
    }
  }
  
  console.log(`Stream analysis for run ${runId}:`, {
    totalEvents: messages.length,
    eventBreakdown: Object.fromEntries(eventCounts)
  });
};

Middleware Composition

Combine multiple middleware functions for complex workflows:

// Compose multiple before middleware
const composeBefore = (...middlewares: BeforeRequestMiddlewareFn[]): 
  BeforeRequestMiddlewareFn => {
  return async (params) => {
    let request = params.request;
    
    for (const middleware of middlewares) {
      const result = await middleware({ ...params, request });
      if (result instanceof Request) {
        request = result;
      } else if (result instanceof Response) {
        throw result; // Early exit with response
      }
    }
    
    return request;
  };
};

// Compose multiple after middleware
const composeAfter = (...middlewares: AfterRequestMiddlewareFn[]): 
  AfterRequestMiddlewareFn => {
  return async (params) => {
    for (const middleware of middlewares) {
      await middleware(params);
    }
  };
};

// Use composed middleware
const runtime = new CopilotRuntime({
  agents: { default: myAgent },
  beforeRequestMiddleware: composeBefore(
    authMiddleware,
    rateLimitMiddleware,
    loggingMiddleware
  ),
  afterRequestMiddleware: composeAfter(
    analyticsMiddleware,
    errorHandlingMiddleware,
    cacheStorageMiddleware
  )
});

Middleware Parameters

BeforeRequestMiddlewareParameters

interface BeforeRequestMiddlewareParameters {
  runtime: CopilotRuntime;  // Access to runtime instance
  request: Request;          // Incoming HTTP request
  path: string;             // Request path (e.g., "/api/copilotkit")
}

AfterRequestMiddlewareParameters

interface AfterRequestMiddlewareParameters {
  runtime: CopilotRuntime;  // Access to runtime instance
  response: Response;        // HTTP response from handler
  path: string;             // Request path
  messages?: Message[];      // Parsed SSE messages (if SSE response)
  threadId?: string;        // Thread ID from RUN_STARTED event
  runId?: string;           // Run ID from RUN_STARTED event
}

Best Practices

Keep It Fast

Middleware runs on every request - keep operations lightweight and avoid blocking I/O

Handle Errors

Always catch errors in middleware to prevent request failures from uncaught exceptions

Clone Requests

Use request.clone() if you need to read the request body, as bodies can only be read once

Async-First

All middleware is async - use await for database queries, API calls, and other I/O operations

Testing Middleware

Test your middleware in isolation:

import { describe, it, expect, vi } from "vitest";

describe("authMiddleware", () => {
  it("should reject requests without authorization header", async () => {
    const request = new Request("https://example.com/api/copilotkit");
    
    await expect(
      authMiddleware({ 
        runtime: mockRuntime, 
        request, 
        path: "/api/copilotkit" 
      })
    ).rejects.toThrow("Unauthorized");
  });
  
  it("should add user info to request headers", async () => {
    const request = new Request("https://example.com/api/copilotkit", {
      headers: { "Authorization": "Bearer valid-token" }
    });
    
    const result = await authMiddleware({ 
      runtime: mockRuntime, 
      request, 
      path: "/api/copilotkit" 
    });
    
    expect(result.headers.get("X-User-Id")).toBe("123");
    expect(result.headers.get("X-User-Role")).toBe("admin");
  });
});

Shared State

Access and modify agent state in middleware

Multi-Agent

Route requests to specific agents in middleware

​Runtime Middleware

​Overview

​Middleware Types

​Before Request Middleware

​After Request Middleware

​Common Use Cases

​Authentication

​Request Logging

​Rate Limiting

​Request Validation

​Analytics and Monitoring

​Error Handling

​Response Caching

​SSE Stream Processing

​Middleware Composition

​Middleware Parameters

​BeforeRequestMiddlewareParameters

​AfterRequestMiddlewareParameters

​Best Practices