Express Adapter

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.

InstallationCopy Anchor Link

pnpm add @cogitator-ai/express express

Basic SetupCopy Anchor Link

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');
});

ConfigurationCopy Anchor Link

The CogitatorServerConfig interface accepts:

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 HandlingCopy Anchor Link

• 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.
• CogitatorErrors 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.

AuthenticationCopy Anchor Link

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

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 LimitingCopy Anchor Link

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.

CORSCopy Anchor Link

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.

StreamingCopy Anchor Link

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):

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));
}

WebSocketCopy Anchor Link

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

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):

{
  "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 UtilitiesCopy Anchor Link

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

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