Fastify Adapter

The @cogitator-ai/fastify package provides cogitatorPlugin — a fastify-plugin compatible with Fastify 5.x. It leverages Fastify's native ecosystem: @fastify/swagger for OpenAPI, @fastify/rate-limit for throttling, and @fastify/websocket for real-time communication.

InstallationCopy Anchor Link

pnpm add @cogitator-ai/fastify fastify

# optional — install for Swagger UI, rate limiting, or WebSocket
pnpm add @fastify/swagger @fastify/swagger-ui
pnpm add @fastify/rate-limit
pnpm add @fastify/websocket

Basic SetupCopy Anchor Link

import Fastify from 'fastify';
import { Cogitator, Agent, tool } from '@cogitator-ai/core';
import { cogitatorPlugin } from '@cogitator-ai/fastify';
import { z } from 'zod';

const fastify = Fastify({ logger: true });

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

const multiply = tool({
  name: 'multiply',
  description: 'Multiply two numbers',
  parameters: z.object({ a: z.number(), b: z.number() }),
  execute: async ({ a, b }) => ({ result: a * b }),
});

const mathAgent = new Agent({
  name: 'math',
  description: 'Math assistant',
  model: 'openai/gpt-6.1-sol',
  instructions: 'You are a math assistant. Use the multiply tool for multiplication.',
  tools: [multiply],
});

await fastify.register(cogitatorPlugin, {
  cogitator,
  agents: { math: mathAgent },
  prefix: '/cogitator',
  enableSwagger: true,
  enableWebSocket: true,
});

await fastify.listen({ port: 3000 });

After startup, all Cogitator endpoints are available under /cogitator and Swagger UI is at /cogitator/docs. The plugin is wrapped with fastify-plugin, so fastify.cogitator is visible to the rest of your app, while the hooks it adds (auth, rate limit, error handler) apply only to the Cogitator routes.

Plugin OptionsCopy Anchor Link

interface CogitatorPluginOptions {
  cogitator: Cogitator;
  agents?: Record<string, Agent>;
  workflows?: Record<string, Workflow>;
  swarms?: Record<string, SwarmConfig>;
  prefix?: string; // default: '/cogitator'
  auth?: AuthFunction;
  rateLimit?: RateLimitConfig;
  enableSwagger?: boolean;
  enableWebSocket?: boolean;
  swagger?: SwaggerConfig;
  websocket?: WebSocketConfig; // { path? }, default '/ws'
  sseHeartbeatMs?: number; // SSE heartbeat interval in ms, default 5000, 0 turns it off
}

enableSwagger and enableWebSocket default to off.

AuthenticationCopy Anchor Link

The auth function receives a Fastify request and returns an AuthContext. It runs as an onRequest hook:

await fastify.register(cogitatorPlugin, {
  cogitator,
  agents: { math: mathAgent },
  auth: async (request) => {
    const token = request.headers.authorization?.replace('Bearer ', '');
    if (!token) return undefined;

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

Returning undefined lets the request through unauthenticated; throwing rejects it with 401 UNAUTHORIZED. The auth context is available on request.cogitatorAuth (with request.cogitatorRequestId and request.cogitatorStartTime), its userId is passed to agent and swarm runs and scopes the thread routes, and — because it is an onRequest hook — it also guards the WebSocket upgrade.

Rate LimitingCopy Anchor Link

When rateLimit is provided, the plugin registers @fastify/rate-limit automatically:

await fastify.register(cogitatorPlugin, {
  cogitator,
  agents: { math: mathAgent },
  rateLimit: {
    max: 100,
    timeWindow: '1 minute',
  },
});

keyGenerator and errorResponseBuilder are passed through to @fastify/rate-limit. If it is not installed, the plugin logs a warning and continues without throttling. Limited requests receive 429 with code RATE_LIMIT_EXCEEDED.

Swagger & OpenAPICopy Anchor Link

When enableSwagger is true, the plugin registers @fastify/swagger and @fastify/swagger-ui (a missing package logs a warning and skips them). The spec is built by @fastify/swagger from the Fastify route schemas, so it describes the generic routes (/agents/{name}/run, …) with their validated bodies rather than one path per agent. Swagger UI is at <prefix>/docs and the JSON spec at <prefix>/docs/json; there is no /openapi.json route.

await fastify.register(cogitatorPlugin, {
  cogitator,
  agents: { math: mathAgent },
  enableSwagger: true,
  swagger: {
    title: 'My AI API',
    description: 'Production agent runtime',
    version: '2.0.0',
    contact: { name: 'Team', email: 'team@example.com' },
    servers: [{ url: 'https://api.example.com', description: 'Production' }],
  },
});

The swagger options fill the document's info and servers; servers defaults to [{ url: '/' }]. The document declares six tags: agents, threads, tools, workflows, swarms, and health.

WebSocketCopy Anchor Link

When enableWebSocket is true, the plugin registers @fastify/websocket and exposes a WebSocket endpoint:

await fastify.register(cogitatorPlugin, {
  cogitator,
  agents: { math: mathAgent },
  enableWebSocket: true,
  websocket: {
    path: '/ws', // relative to the prefix: ws://host/cogitator/ws
  },
});

Run an agent with { "type": "run", "id": "r1", "payload": { "type": "agent", "name": "math", "input": "6 * 7?" } } (add threadId to continue a thread). Events arrive as { "type": "event", "id": "r1", "payload": { "type": "token" | "reasoning" | "tool-call" | "tool-result" | "complete" | "cancelled", ... } }, failures as { "type": "error", "id": "r1", "error": "..." }. Resume a run paused for approval with { "type": "resume", "id": "r2", "payload": { "name": "math", "threadId": "...", "decisions": { "<toolCallId>": { "approved": true } } } }, send { "type": "stop" } to abort the run and { "type": "ping" } to get a pong.

One run per connection at a time — a second run or resume answers A run is already in progress — and only agent runs are supported. { "type": "subscribe", "channel": "agent:math" } delivers the events of runs of that agent started from the same user's other connections (with channel set on each frame) until unsubscribe; up to 64 channels per connection. A subscriber never receives another user's runs; without auth all connections share one anonymous user and see every run. Closing the socket aborts its run.

StreamingCopy Anchor Link

The streaming endpoints use SSE, identical to other adapters: text-* blocks closed around tool calls, tool-call-* events whose id matches tool-result.toolCallId, then finish and [DONE] (or an error event). FastifyStreamWriter hijacks the reply and keeps headers set by your hooks (CORS, rate limit). Disconnecting aborts the run:

const response = await fetch('http://localhost:3000/cogitator/agents/math/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ input: 'What is 42 * 17?' }),
});

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

Accessing the ContextCopy Anchor Link

Inside custom Fastify routes, the Cogitator context is available via the cogitator decorator:

fastify.get('/custom', async (request, reply) => {
  const { runtime, agents, workflows, swarms } = fastify.cogitator;
  const result = await runtime.run(agents.math, { input: '2 + 2' });
  return { answer: result.output };
});

ErrorsCopy Anchor Link

Schema validation failures and malformed JSON return 400 INVALID_INPUT (with the validation details), unknown agents, workflows and swarms 404 NOT_FOUND, and a failed workflow 500 WORKFLOW_FAILED. The thread routes return 503 UNAVAILABLE when the runtime has no memory and 403 THREAD_ACCESS_DENIED for another user's thread; workflow and swarm routes return 501 UNIMPLEMENTED without their packages. CogitatorErrors keep their message, code and mapped status (for example LLM_RATE_LIMITED → 429). Any other error is logged through request.log and answered as 500 with the message Internal server error (code INTERNAL_ERROR).

Exported SchemasCopy Anchor Link

The package exports JSON Schema objects for request/response validation:

import {
  AgentRunRequestSchema,
  AgentRunResponseSchema,
  AgentResumeRequestSchema,
  AddMessageRequestSchema,
  WorkflowRunRequestSchema,
  SwarmRunRequestSchema,
} from '@cogitator-ai/fastify';

AgentRunRequestSchema and SwarmRunRequestSchema require an input with more than whitespace (the RUN_INPUT_SCHEMA of @cogitator-ai/server-shared), AgentResumeRequestSchema a threadId with optional decisions / defaultDecision (the swarm timeout must be greater than 0 and at most MAX_RUN_TIMEOUT_MS), AddMessageRequestSchema requires a non-empty content and a user/assistant/system role, and WorkflowRunRequestSchema.options only allows maxConcurrency/maxIterations (integers ≥ 1) and checkpoint (additionalProperties: false). These can be used in custom Fastify route schemas for native validation.