# Tetsu Adapter

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

> Serve Cogitator agents, threads, workflows and swarms from a Tetsu application on Bun — one controller with validated requests, SSE, WebSocket and an OpenAPI document.

The `@cogitator-ai/tetsu` package provides `cogitatorController`, a [Tetsu](https://tetsujs.com) 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**.

## Installation

```bash
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 Setup

```typescript
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 Bun

`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](/docs/server-adapters#long-running-runs) for proxies and load balancers.

The routes are the [common endpoints](/docs/server-adapters#common-endpoints) under the group's prefix. The document is at `/openapi.json` and the page at `/docs`.

## Options

```typescript
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
}
```

## Errors

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.

| Status | `error`                 | When                                                        |
| ------ | ----------------------- | ----------------------------------------------------------- |
| 401    | `UNAUTHORIZED`          | `auth` returned `undefined`                                 |
| 403    | `THREAD_ACCESS_DENIED`  | The thread belongs to another user                          |
| 403    | `THREAD_FORBIDDEN`      | `authorizeThread` refused the thread                        |
| 404    | `AGENT_NOT_FOUND`, …    | No agent, workflow or swarm with that name                  |
| 409    | `BLACKBOARD_DISABLED`   | The swarm has no blackboard                                 |
| 409    | `RUN_NOT_PAUSED`        | A resume named a thread without a paused run                |
| 501    | `PACKAGE_NOT_INSTALLED` | `@cogitator-ai/workflows` or `@cogitator-ai/swarms` missing |
| 503    | `MEMORY_NOT_CONFIGURED` | A thread endpoint on a runtime without memory               |

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

## Authentication

`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.

```typescript
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`:

```typescript
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 Threads

### Multiple users

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`.

### `authorizeThread`

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

```typescript
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.

## Approvals

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`.

```bash
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`.

## Streaming

`/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](/docs/server-adapters#streaming-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.

## WebSocket

```typescript
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](#approvals). `{ type: 'stop' }` cancels the run, `{ type: 'ping' }` is answered with `pong`, and errors arrive as `{ type: 'error', error, code }` without closing the socket.

## Shutdown

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

```typescript
let shutdown: ReturnType<typeof onShutdownSignals> | undefined;

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

## Example

`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.

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