Tetsu Adapter

The @cogitator-ai/tetsu package provides cogitatorController, a Tetsu controller with every Cogitator route. Requests and responses are Zod schemas, so Tetsu validates them, types the handlers and describes them in the OpenAPI document. Tetsu runs on Bun 1.4 or later.

InstallationCopy Anchor Link

bun add @cogitator-ai/tetsu @cogitator-ai/core @tetsujs/core @tetsujs/sse

Add @cogitator-ai/workflows and @cogitator-ai/swarms to serve the workflow and swarm endpoints, and @tetsujs/openapi for the document.

Basic SetupCopy Anchor Link

import { Agent, Cogitator } from '@cogitator-ai/core';
import { cogitatorController } from '@cogitator-ai/tetsu';
import { createApp, group } from '@tetsujs/core';
import { docs } from '@tetsujs/openapi';

const cogitator = new Cogitator({
  llm: { defaultModel: 'google/gemini-3.5-flash-lite' },
  memory: { adapter: 'memory' },
});

const chat = new Agent({
  name: 'chat',
  model: 'google/gemini-3.5-flash-lite',
  instructions: 'You are a helpful assistant.',
});

const app = createApp({
  routes: [
    group('/cogitator', { children: [cogitatorController({ cogitator, agents: { chat } })] }),
    docs({ info: { title: 'Agents', version: '1.0.0' } }),
  ],
});

Bun.serve({ ...app, port: 3000 });

Long runs on BunCopy Anchor Link

Bun.serve closes a connection that stays silent for idleTimeout seconds, 10 by default, so without help a run that waits on a slow tool or model for longer ends with the connection closed. The controller keeps both kinds of answer open without any setting: the JSON routes (run, resume, workflow and swarm run) lift the idle timeout for their own request once its body is validated, and SSE streams write a heartbeat comment every sseHeartbeatMs (5 s by default). If you set idleTimeout, keep it above sseHeartbeatMs and at 5 seconds or more, since Bun cuts connections at once below that whatever is written. See Long-Running Runs for proxies and load balancers.

The routes are the common endpoints under the group's prefix. The document is at /openapi.json and the page at /docs.

OptionsCopy Anchor Link

interface CogitatorDeps {
  cogitator: Cogitator;
  agents?: Record<string, Agent>;
  workflows?: Record<string, Workflow>;
  swarms?: Record<string, SwarmConfig>;
  auth?: Authenticate | CallerHook;
  authorizeThread?: (auth: AuthContext | undefined, threadId: string) => boolean | Promise<boolean>;
  websocket?: boolean | { path?: `/${string}` };
  until?: AbortSignal | (() => AbortSignal | undefined);
  sseHeartbeatMs?: number; // SSE heartbeat interval in ms, default 5000, 0 turns it off
}

ErrorsCopy Anchor Link

Every error is answered in Tetsu's envelope { status, message, error }. A request that fails its schema gets 422 VALIDATION_FAILED with every issue. A CogitatorError keeps its status and code, for example 429 LLM_RATE_LIMITED with Retry-After. Any other error answers 500 INTERNAL_SERVER_ERROR and goes to the application's reportError, never to the client.

cogitatorErrors() returns the onError hook the controller uses. Mount one on the application to map CogitatorErrors from your own routes too.

AuthenticationCopy Anchor Link

auth runs as a beforeParse hook, before the body is read. Return the caller, or undefined for 401; throw an HttpError for another status. /health and /ready stay open.

cogitatorController({
  cogitator,
  agents: { chat },
  auth: (ctx) => {
    const token = ctx.req.headers.get('authorization')?.replace(/^Bearer /, '');
    const user = token ? sessions.find(token) : undefined;
    return user && { userId: user.id };
  },
});

The caller's userId reaches every run, so tools see it as context.userId, and scopes runs and thread endpoints to the caller.

To document the scheme, build the hook with callerHook(), wrap it in secured(), and reuse it on your own routes, where the caller is ctx.cogitatorAuth:

import { callerHook } from '@cogitator-ai/tetsu';
import { secured } from '@tetsujs/openapi';

const signedIn = secured(callerHook(authenticate), {
  name: 'bearerAuth',
  scheme: { type: 'http', scheme: 'bearer' },
  error: 'UNAUTHORIZED',
});

cogitatorController({ cogitator, agents, auth: signedIn });

Users and ThreadsCopy Anchor Link

Multiple usersCopy Anchor Link

A thread belongs to the user who created it, through a run or POST /threads/:id/messages. Runs and the thread endpoints let only that user in: another user's thread answers 403 THREAD_ACCESS_DENIED and stays untouched, and a stream on it ends with an error event carrying that code. Callers without a userId share the threads that have no owner. When several people share a server, have auth return a userId.

authorizeThreadCopy Anchor Link

On top of ownership, authorizeThread guards the thread endpoints and every run that names a threadId, before the model is called:

cogitatorController({
  cogitator,
  agents,
  auth,
  authorizeThread: (auth, threadId) => threadId.startsWith(`${auth?.userId}:`),
});

Use it for rules ownership does not express, such as thread ids your server hands out.

ApprovalsCopy Anchor Link

A tool with requiresApproval pauses the run before it executes. POST /agents/:name/run answers with status: 'paused' and pendingApprovals (toolCallId, toolName, arguments, description, sideEffects); the checkpoint stays on the server, stored by the runtime per thread. A paused stream sends approval-required with the same calls right before finish.

curl -X POST localhost:3000/cogitator/agents/support/resume \
  -H 'content-type: application/json' \
  -d '{ "threadId": "thread_…", "decisions": { "call_1": { "approved": true } } }'

POST /agents/:name/resume takes { threadId, decisions?, defaultDecision? }, where a decision is { approved, reason? } by tool call id. Approved calls run, declined ones answer the model with the reason, and calls left undecided pause the run again. /agents/:name/resume/stream streams the rest of the run, and the WebSocket accepts { type: 'resume', id, payload: { name, threadId, decisions? } }. Only the user the run belongs to may resume it (403 THREAD_ACCESS_DENIED; authorizeThread is checked too), and a thread with nothing paused answers 409 RUN_NOT_PAUSED.

StreamingCopy Anchor Link

/stream routes use @tetsujs/sse: keep-alive comments every sseHeartbeatMs, backpressure, and the run is aborted when the client leaves. Events follow the shared protocol; an agent with reasoning: { summary: true } also streams reasoning-start, reasoning-delta and reasoning-end, closed before text, tool calls and finish, and POST /agents/:name/run returns the summary as reasoning. Validation, 401, 403 and 404 are answered as JSON before the stream opens.

WebSocketCopy Anchor Link

cogitatorController({ cogitator, agents, auth, websocket: true });

The handshake runs auth. Send { type: 'run', id, payload: { type: 'agent', name, input, threadId? } } and receive event frames (token, reasoning, tool-call, tool-result, then complete or cancelled). { type: 'resume', id, payload: { name, threadId, decisions? } } continues a run paused for approvals. { type: 'stop' } cancels the run, { type: 'ping' } is answered with pong, and errors arrive as { type: 'error', error, code } without closing the socket.

ShutdownCopy Anchor Link

Pass draining from @tetsujs/lifecycle as until, so a stopping server ends open streams and sockets and clients reconnect elsewhere:

let shutdown: ReturnType<typeof onShutdownSignals> | undefined;

const app = createApp({
  routes: cogitatorController({ cogitator, agents, until: () => shutdown?.draining }),
});
shutdown = onShutdownSignals(Bun.serve({ ...app }));

ExampleCopy Anchor Link

examples/integrations/08-tetsu-server.ts is an app with its own users and its own MCP server: the agent reaches the app's tasks through MCP tools and acts for the signed-in user.

bun examples/integrations/08-tetsu-server.ts