# Express Adapter

URL: https://cogitator.app/docs/server-adapters/express

> Serve your Cogitator agents over HTTP with Express.js — class-based setup with middleware, SSE streaming, and WebSocket support.

The `@cogitator-ai/express` package provides `CogitatorServer`, a class that mounts a full REST API onto your existing Express router. It handles JSON parsing, authentication, rate limiting, CORS, error handling, Swagger docs, and WebSocket support out of the box.

## Installation

```bash
pnpm add @cogitator-ai/express express
```

## Basic Setup

```typescript
import express from 'express';
import { Cogitator, Agent, tool } from '@cogitator-ai/core';
import { CogitatorServer } from '@cogitator-ai/express';
import { z } from 'zod';

const app = express();

const cogitator = new Cogitator({
  llm: {
    defaultModel: 'openai/gpt-6.1-sol',
    providers: {
      openai: { apiKey: process.env.OPENAI_API_KEY! },
    },
  },
});

const search = tool({
  name: 'web_search',
  description: 'Search the web',
  parameters: z.object({ query: z.string() }),
  execute: async ({ query }) => ({ results: [`Result for: ${query}`] }),
});

const assistant = new Agent({
  name: 'assistant',
  description: 'Assistant with web search',
  model: 'openai/gpt-6.1-sol',
  instructions: 'You are a helpful assistant with web search capabilities.',
  tools: [search],
});

const server = new CogitatorServer({
  app: app,
  cogitator,
  agents: { assistant },
  config: {
    basePath: '/cogitator',
    enableSwagger: true,
    enableWebSocket: false,
  },
});

await server.init();

app.listen(3000, () => {
  console.log('Server running on http://localhost:3000');
  console.log('Swagger UI: http://localhost:3000/cogitator/docs');
});
```

## Configuration

The `CogitatorServerConfig` interface accepts:

```typescript
interface CogitatorServerConfig {
  app: Router;
  cogitator: Cogitator;
  agents?: Record<string, Agent>;
  workflows?: Record<string, Workflow>;
  swarms?: Record<string, SwarmConfig>;
  config?: {
    basePath?: string; // default: '/cogitator'
    enableWebSocket?: boolean; // default: false
    enableSwagger?: boolean; // default: true
    auth?: AuthFunction;
    rateLimit?: RateLimitConfig;
    cors?: CorsConfig;
    swagger?: SwaggerConfig;
    websocket?: WebSocketConfig; // { path?, pingInterval? (ms, default 30000), maxPayloadSize? (bytes, default 1 MiB) }
    sseHeartbeatMs?: number; // SSE heartbeat interval in ms, default 5000, 0 turns it off
  };
}
```

`init()` mounts the router once (calling it again does nothing); `server.isInitialized` tells whether it ran. It writes nothing to the console: `init()` and the WebSocket setup report through the core logger at `debug` level (`LOG_LEVEL=debug`, or your own logger via `setLogger()` from `@cogitator-ai/core`).

## Request Handling

* Agent and swarm bodies are validated: `input` must contain more than whitespace (`""` and `"   "` are refused before the model is called), `context` an object, `threadId` a non-empty string, swarm `timeout` a positive number of at most `MAX_RUN_TIMEOUT_MS` (2147483647, the longest delay a timer can hold). Invalid or malformed JSON bodies return `400 INVALID_INPUT`.
* Workflow `options` are limited to `maxConcurrency`, `maxIterations` (positive integers) and `checkpoint` (boolean); other options are dropped and wrongly typed ones return `400`.
* Thread messages need a `role` of `user`, `assistant` or `system` and a non-empty string `content`.
* A workflow that fails returns `500 WORKFLOW_FAILED`, and its stream ends with an `error` event instead of `workflow_completed`. Without `@cogitator-ai/workflows` or `@cogitator-ai/swarms` installed, those routes return `501 UNIMPLEMENTED`.
* Unknown agents, workflows, swarms and routes return `404 NOT_FOUND`. `GET /swarms/:name/blackboard` returns the swarm's configured `blackboard.sections`, or `400` when its blackboard is not enabled.
* The thread routes use `cogitator.getMemory()` and return `503 UNAVAILABLE` when the runtime has no memory. Another user's thread returns `403 THREAD_ACCESS_DENIED`.
* `CogitatorError`s keep their message, code and mapped status (for example `LLM_RATE_LIMITED` → 429). Any other error is logged and answered as `500` with `{ "error": { "message": "Internal server error", "code": "INTERNAL_ERROR" } }`.
* `GET /agents` exposes each agent's `config.description`, never its `instructions`.

## Authentication

Pass a custom `auth` function that receives the Express request and returns an `AuthContext`:

```typescript
const server = new CogitatorServer({
  app,
  cogitator,
  agents: { assistant },
  config: {
    auth: async (req) => {
      const token = req.headers.authorization?.replace('Bearer ', '');
      if (!token) return undefined;

      const user = await verifyToken(token);
      return { userId: user.id, roles: user.roles };
    },
  },
});
```

When `auth` returns `undefined`, the request proceeds as unauthenticated; throw to reject it with `401 UNAUTHORIZED`. The `AuthContext` is attached to `req.cogitator.auth` (next to `requestId` and `startTime`), and its `userId` is passed to agent and swarm runs and scopes the thread routes.

The same function guards WebSocket upgrades. There it receives the raw upgrade request (`http.IncomingMessage`), not an Express request, so read only `headers` and `url` in code that serves both.

## Rate Limiting

```typescript
config: {
  rateLimit: {
    windowMs: 60_000,
    max: 100,
    message: 'Too many requests',
    keyGenerator: (req) => req.ip ?? 'unknown',
    skip: (req) => req.path === '/health',
    trustProxy: false, // true: key by the nearest X-Forwarded-For hop (the address your proxy saw)
  },
}
```

The limiter is in-memory and per process; `windowMs` must be positive and `max` non-negative, otherwise creating the middleware throws. Requests without a client address share a single bucket.

## CORS

```typescript
config: {
  cors: {
    origin: ['https://app.example.com'],
    credentials: true,
    methods: ['GET', 'POST', 'DELETE'],
  },
}
```

`credentials` defaults to `true` for explicit origins and to `false` for `origin: '*'` (plain `Access-Control-Allow-Origin: *`). Setting `credentials: true` together with `'*'` reflects every origin with credentials — only do that deliberately. Disallowed origins get no CORS grants, including on preflight.

## Streaming

The `/agents/:name/stream` endpoint returns an SSE stream: `start`, `text-start`/`text-delta`/`text-end` blocks, `tool-call-start`/`tool-call-delta`/`tool-call-end` (the id is the model's tool call id, matching `tool-result.toolCallId`), `tool-result`, then `finish` and `[DONE]` — or an `error` event. Disconnecting aborts the run. Read it with `fetch` (EventSource cannot send a POST body):

```typescript
const response = await fetch('http://localhost:3000/cogitator/agents/assistant/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ input: 'Explain quantum computing' }),
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(decoder.decode(value));
}
```

## WebSocket

WebSocket support requires the `ws` package. Enable it, then attach it to your HTTP server after `init()`:

```typescript
import { createServer } from 'http';

const server = new CogitatorServer({
  app,
  cogitator,
  agents: { assistant },
  config: { enableWebSocket: true },
});
await server.init();

const httpServer = createServer(app);
await server.attachWebSocket(httpServer); // ws://host/cogitator/ws
httpServer.listen(3000);
```

`attachWebSocket()` returns the `WebSocketServer`; it throws when `enableWebSocket` is off, when `init()` has not run, or when `ws` is not installed. The path is `<basePath>/ws`; `websocket.path` replaces it with a full path. The server pings every `pingInterval` ms and drops clients that stop answering.

Run an agent (only agent runs are supported; `workflow` and `swarm` runs answer with an error):

```json
{
  "type": "run",
  "id": "req_1",
  "payload": { "type": "agent", "name": "assistant", "input": "Hello", "threadId": "t1" }
}
```

Events arrive as `{ "type": "event", "id": "req_1", "payload": { "type": "token" | "reasoning" | "tool-call" | "tool-result" | "complete" | "cancelled", ... } }`, and failures as `{ "type": "error", "id": "req_1", "error": "..." }`. Resume a run paused for approval with `{ "type": "resume", "id": "req_2", "payload": { "name": "assistant", "threadId": "t1", "decisions": { "call_1": { "approved": true } } } }`. Send `{ "type": "stop" }` to abort the running agent and `{ "type": "ping" }` to get a `pong`. A new `run` or `resume` aborts the one still running on that connection.

`{ "type": "subscribe", "channel": "agent:assistant" }` (answered with `subscribed`) delivers the events of runs of that agent started from other connections of the same user (the `userId` your `auth` returned), with `channel` set on each frame; `unsubscribe` stops them. Runs of other users never reach the subscriber. Without `auth` every connection counts as the same anonymous user and sees every run of that agent. A connection holds up to 64 subscriptions of at most 256 characters.

## Exported Utilities

The package also exports individual building blocks if you need to compose routes manually:

```typescript
import {
  createAuthMiddleware,
  createRateLimitMiddleware,
  createCorsMiddleware,
  createHealthRoutes,
  createAgentRoutes,
  createThreadRoutes,
  createToolRoutes,
  createWorkflowRoutes,
  createSwarmRoutes,
  errorHandler,
  notFoundHandler,
  ExpressStreamWriter,
  setupSSEHeaders,
  generateOpenAPISpec,
  serveSwaggerUI,
  setupWebSocket,
} from '@cogitator-ai/express';
```