@cogitator-ai/openai-compat

OpenAI-compatible server for Cogitator agents. Register your agents and every client that speaks the OpenAI API (the official SDKs, Open WebUI, LibreChat, LangChain, Vercel AI SDK and the like) talks to them through POST /v1/chat/completions or POST /v1/responses, with the agent's name as the model. The Assistants API is served too, for existing integrations.

Full guide: cogitator.app/docs/integrations/openai-compat

The Assistants API is deprecated. OpenAI sunset it on 2026-08-26 and current clients use Chat Completions or Responses. The /v1/assistants, /v1/threads and run endpoints keep working, but new integrations should use Chat Completions or Responses.

Installation

pnpm add @cogitator-ai/openai-compat @cogitator-ai/core

# optional: the OpenAI SDK for client code
pnpm add openai

ioredis and pg are optional peer dependencies, needed only for RedisThreadStorage / PostgresThreadStorage.

Features

• Chat Completions - POST /v1/chat/completions, streaming and not, with tool calls, usage and images
• Responses - POST /v1/responses with the streaming events of the API, previous_response_id, function calls, and GET / DELETE of stored responses
• Agents as models - model names a registered agent, which answers with its own instructions, model and tools. GET /v1/models lists them
• Client functions - Functions a client declares come back as tool calls, and the outputs it sends resume the same run
• Assistants API (deprecated) - Assistants, threads, messages, runs, run steps and files over pluggable storage (in-memory, Redis, PostgreSQL)
• OpenAI errors - Every error in the OpenAI error format, with the parameter at fault
• Authentication - Optional API keys (constant-time check, /health stays public), with threads and runs scoped to the key that created them
• Abort on disconnect - A client that goes away stops the agent run
• CORS - Configurable cross-origin requests

There are no vector stores, and code_interpreter and file_search assistant tools are stored but not executed.

Quick Start

import { createOpenAIServer } from '@cogitator-ai/openai-compat';
import { Agent, Cogitator, tool } from '@cogitator-ai/core';
import { z } from 'zod';

const lookupOrder = tool({
  name: 'lookup_order',
  description: 'Find an order by id',
  parameters: z.object({ id: z.string() }),
  execute: async ({ id }) => ({ id, status: 'shipped' }),
});

const support = new Agent({
  name: 'support',
  model: 'openai/gpt-6-luna',
  instructions: 'You are the support agent of an online shop.',
  tools: [lookupOrder],
});

const server = createOpenAIServer(new Cogitator(), {
  port: 8080,
  agents: { support },
  apiKeys: ['sk-my-secret-key'],
});

await server.start();
console.log(server.getBaseUrl()); // http://localhost:8080/v1

import OpenAI from 'openai';

const openai = new OpenAI({ baseURL: 'http://localhost:8080/v1', apiKey: 'sk-my-secret-key' });

const completion = await openai.chat.completions.create({
  model: 'support',
  messages: [{ role: 'user', content: 'Where is order A-1?' }],
});
console.log(completion.choices[0].message.content);

const response = await openai.responses.create({ model: 'support', input: 'Where is order A-1?' });
console.log(response.output_text);

Point Open WebUI, LibreChat or any other OpenAI client at http://localhost:8080/v1 with the API key, and pick the agent from the model list.

Agents as models

agents maps model ids to Cogitator agents. A request whose model is not one of them answers 404 with code model_not_found. GET /v1/models lists the agents, and the cogitator model of the Assistants API.

For each request the server runs the agent with:

• its own instructions first, then the client's system / developer messages (or the Responses instructions) after them,
• the request's temperature, top_p, token limit, stop and response format,
• its own tools, plus the functions the client declares in tools (a client function with the name of an agent tool is left out).

The conversation the client sends is replayed to the agent: the last user message is the input and the turns before it come as a transcript, so the APIs stay stateless and nothing is written to the agent's memory. Only the images of the last user message reach the model.

Client functions

When the model calls a function the client declared, the turn ends with the call (finish_reason: 'tool_calls', or a function_call output item) and the run waits. When the client sends the outputs back for every call (a tool message, or a function_call_output item), the same run goes on from where it stopped, with the outputs as the function results. The call ids the client gets are generated by the server, so only the client that got them can answer. A run waits 10 minutes, or until 1000 newer runs wait, and outputs for a run that no longer waits are replayed to the agent as part of the conversation instead.

Agent tools that need an approval (requiresApproval) are not run when nothing approves them (guardrails.onToolApproval): the model is told the tool was not run.

A tool that asks a question (context.interrupt()) has no way to reach the client over these APIs, which only carry function calls. The request fails instead (500, server_error) with a message that names each question, such as paused to ask confirm_plan ("Go ahead with the migration?"), rather than answering with an empty tool_calls turn or resuming without an answer.

tool_choice: 'none' runs the agent without any tool. required and { type: 'function', ... } become the run's toolChoice: the model must call a tool (or the named one) before it answers, and the run goes back to auto once it has.

Chat Completions

const stream = await openai.chat.completions.create({
  model: 'support',
  messages: [
    { role: 'system', content: 'Answer in one sentence.' },
    { role: 'user', content: 'Where is order A-1?' },
  ],
  stream: true,
  stream_options: { include_usage: true },
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta.content ?? '');
}

| Parameter | Support | | --------------------------------------------- | -------------------------------------------------------------------------------------------------------- | | model | A registered agent | | messages | system, developer, user (text and image_url parts), assistant (text and tool_calls), tool | | stream, stream_options.include_usage | Chunks with delta.content and delta.tool_calls, the finish reason, then the usage chunk and [DONE] | | tools, tool_choice, parallel_tool_calls | Function tools, see Client functions | | temperature, top_p | Passed to the agent | | max_completion_tokens (max_tokens) | The agent's output token limit | | stop | Stop sequences | | response_format | text, json_object, json_schema | | n | Only 1 |

finish_reason is stop, tool_calls, length when the answer hit the token limit, or content_filter when the provider withheld it. A refusal comes back in message.refusal. usage carries prompt_tokens_details.cached_tokens and completion_tokens_details.reasoning_tokens. Other parameters are ignored.

Responses

const first = await openai.responses.create({
  model: 'support',
  instructions: 'Answer in one sentence.',
  input: 'Where is order A-1?',
});

const next = await openai.responses.create({
  model: 'support',
  previous_response_id: first.id,
  input: 'And order B-2?',
});

const stream = openai.responses.stream({ model: 'support', input: 'Summarize our talk' });
stream.on('response.output_text.delta', (event) => process.stdout.write(event.delta));
const final = await stream.finalResponse();

| Parameter | Support | | --------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------ | | model | A registered agent | | input | A string, or message items (input_text, input_image with image_url, output_text), function_call and function_call_output items | | instructions | After the agent's own instructions, for this response only | | previous_response_id | Continues the conversation of a stored response | | tools, tool_choice, parallel_tool_calls | Function tools only | | temperature, top_p, max_output_tokens | Passed to the agent | | text.format | text, json_object, json_schema | | store | Keep the response (default true) | | stream | The Responses streaming events | | metadata | Returned with the response |

A streamed response sends response.created, response.in_progress, the output items (response.output_item.added, response.content_part.added, response.output_text.delta, response.output_text.done, response.content_part.done, response.output_item.done, and response.function_call_arguments.delta / .done for function calls), then response.completed, response.incomplete or response.failed, every event with its sequence_number. A response cut by max_output_tokens is incomplete with incomplete_details.reason: 'max_output_tokens' (content_filter when the provider withheld it).

Stored responses live in memory (maxStoredResponses, default 1000, the oldest dropped first): GET /v1/responses/{id}, GET /v1/responses/{id}/input_items and DELETE /v1/responses/{id}. previous_response_id carries the conversation, not the earlier instructions. Built-in tools (web search, file search, computer use), background and file inputs are not supported.

Assistants API (deprecated)

import OpenAI from 'openai';

const openai = new OpenAI({
  baseURL: 'http://localhost:8080/v1',
  apiKey: 'sk-my-secret-key',
});

const assistant = await openai.beta.assistants.create({
  name: 'Math Tutor',
  instructions: 'You help with math problems',
  model: 'ollama/llama3.2:latest',
});

const thread = await openai.beta.threads.create();

await openai.beta.threads.messages.create(thread.id, {
  role: 'user',
  content: 'What is 2 + 2?',
});

const run = await openai.beta.threads.runs.createAndPoll(thread.id, {
  assistant_id: assistant.id,
});

const messages = await openai.beta.threads.messages.list(thread.id);
console.log(run.status, messages.data[0].content); // newest message first

// Streaming
const stream = openai.beta.threads.runs
  .stream(thread.id, { assistant_id: assistant.id })
  .on('textDelta', (delta) => process.stdout.write(delta.value ?? ''));
await stream.finalRun();

The Assistants API runs assistants the client creates, not the registered agents. Assistant model values are Cogitator model strings (openai/gpt-6.1-sol, ollama/llama3.2:latest, ...). The cogitator model listed by GET /v1/models maps to the server's defaultModel, and runs that use it fail when no defaultModel is configured. Server-side tools are available to every run.

OpenAI Server

The OpenAIServer exposes Cogitator agents as an OpenAI-compatible REST API.

Configuration

import { OpenAIServer } from '@cogitator-ai/openai-compat';

const server = new OpenAIServer(cogitator, {
  port: 8080,
  host: '0.0.0.0',

  apiKeys: ['sk-key1', 'sk-key2'],

  agents: { support, researcher },

  tools: [calculator, datetime, webSearch],

  logging: true,

  cors: {
    origin: ['http://localhost:3000', 'https://myapp.com'],
    methods: ['GET', 'POST', 'DELETE', 'OPTIONS'],
  },
});

Configuration Options

| Option | Type | Default | Description | | ---------------------------------- | ------------------------------- | -------------------------------------- | --------------------------------------------------------------------------- | | port | number | 8080 | Port to listen on (0 picks a free port) | | host | string | '127.0.0.1' | Host to bind to. A public host needs apiKeys | | apiKeys | string[] | [] | API keys for authentication, scoping threads and runs. Empty disables auth | | allowUnauthenticatedPublicAccess | boolean | false | Serve a public host without apiKeys (behind a gateway that authenticates) | | sseHeartbeatMs | number | 5000 | Heartbeat comment interval on streams, 0 turns it off | | agents | Record<string, Agent> | {} | Agents served by Chat Completions and Responses, by model id | | maxRequestBodyBytes | number | 20 MB | Body limit of /v1/chat/completions and /v1/responses | | maxStoredResponses | number | 1000 | Responses kept in memory for previous_response_id and GET | | tools | Tool[] | [] | Server-side tools available to every Assistants API run | | defaultModel | string | - | Model the Assistants API uses for the cogitator model id | | storage | ThreadStorage | in-memory | Persistence backend (connect before passing) | | maxFileSize | number | 512 MB | Upload limit for POST /v1/files, in bytes | | logging | boolean | false | Enable Fastify request logging (JSON logs) | | cors.origin | string \| string[] \| boolean | no CORS | Origins browsers may call from. Without cors no cross-origin page can | | cors.methods | string[] | ['GET', 'POST', 'DELETE', 'OPTIONS'] | Allowed HTTP methods |

Exposure

The server binds 127.0.0.1 by default, so only this machine reaches it. Without apiKeys anyone who reaches it can create assistants with any instructions and run the server's tools, so start() refuses a public host (0.0.0.0, :: or an external address) unless apiKeys are set or allowUnauthenticatedPublicAccess: true says a gateway in front authenticates callers. CORS is off unless cors.origin names the origins browsers may call from, so a web page on another origin cannot drive the server from a visitor's browser. Run streams write a heartbeat comment every sseHeartbeatMs while a run is silent, so proxies do not cut a run that waits on a slow tool.

Server Lifecycle

await server.start(); // throws if already started

console.log(server.getUrl()); // uses the bound port, so `port: 0` works
console.log(server.getBaseUrl()); // getUrl() + '/v1'

console.log(server.isRunning());

const adapter = server.getAdapter(); // the OpenAIAdapter behind the routes
const fastify = server.getFastify(); // underlying Fastify instance

await server.stop();

For tests without a listening socket, await server.waitUntilReady() and then call server.getFastify().inject(...).

Authentication

With apiKeys set, every request except GET /health needs Authorization: Bearer <key>. A missing header answers 401 with code missing_api_key; a malformed header or unknown key answers 401 with code invalid_api_key.

Threads and runs per key

A request that authenticates with a key carries request.apiKeyOwner, the sha256 hex digest of the key (the key itself is never stored). Threads (POST /v1/threads, POST /v1/threads/runs) keep that owner in storage, and runs keep it in memory. Another key gets 404 not_found for the thread, its messages, its runs and their steps, the same answer as for an id that does not exist, so it cannot tell that the thread exists. Listing the runs of a thread shows only the runs of the caller's key.

• Threads stored before this scoping have no owner and stay reachable by every key, as before. Runs created on them still belong to the key that created them.
• Without apiKeys nothing is scoped.
• Assistants, files and stored responses are shared by every key.

apiKeyOwner(key) (exported) gives the owner of a key, for instance to list its threads from storage (see Listing threads).

Health Check

curl http://localhost:8080/health
# {"status":"ok"}

OpenAI Adapter (Assistants API, deprecated)

The OpenAIAdapter provides in-process access without running a server. The server uses one internally (server.getAdapter()).

import { createOpenAIAdapter } from '@cogitator-ai/openai-compat';

const adapter = createOpenAIAdapter(cogitator, {
  tools: [calculator], // server-side Cogitator tools
  defaultModel: 'openai/gpt-6-luna', // resolves the `cogitator` model id
  storage, // ThreadStorage, default: in-memory
  maxStoredRuns: 10_000, // finished runs kept in memory (default 10 000)
});

Assistant Management

const assistant = await adapter.createAssistant({
  model: 'openai/gpt-6.1-sol',
  name: 'Code Helper',
  description: 'Writes TypeScript',
  instructions: 'You help write code',
  temperature: 0.7,
  response_format: { type: 'json_object' },
  metadata: { category: 'development' },
});

const fetched = await adapter.getAssistant(assistant.id); // undefined if missing

const updated = await adapter.updateAssistant(assistant.id, {
  name: 'Code Expert',
  temperature: 0.5,
});

const all = await adapter.listAssistants();

const deleted = await adapter.deleteAssistant(assistant.id); // boolean

Thread Operations

const thread = await adapter.createThread({ project: 'demo' }); // metadata

const fetched = await adapter.getThread(thread.id);

await adapter.updateThread(thread.id, { metadata: { stage: 'review' } }); // merged into existing metadata

const message = await adapter.addMessage(thread.id, {
  role: 'user',
  content: 'Hello, how are you?',
  metadata: { source: 'web' },
});

const messages = await adapter.listMessages(thread.id, {
  limit: 20,
  order: 'asc',
  after: 'msg_abc123',
  before: 'msg_xyz789',
  run_id: 'run_123',
});

const msg = await adapter.getMessage(thread.id, 'msg_abc123');

await adapter.deleteThread(thread.id);

Message content is a string or an array of parts: { type: 'text', text }, { type: 'image_url', image_url: { url } } or { type: 'image_file', image_file: { file_id } }.

Run Execution

const run = await adapter.createRun(thread.id, {
  assistant_id: assistant.id,
  model: 'openai/gpt-6.1-sol',
  instructions: 'Be concise',
  temperature: 0.5,
  additional_messages: [{ role: 'user', content: 'Extra context' }],
  metadata: { source: 'api' },
});

const status = adapter.getRun(thread.id, run.id); // synchronous

const runs = adapter.listRuns(thread.id); // newest first

const cancelled = adapter.cancelRun(thread.id, run.id); // throws if the run already finished

for await (const { event, data } of adapter.streamRunEvents(run.id)) {
  console.log(event, data);
}

createRun returns the queued run immediately and executes it in the background. It rejects with an InvalidRequestError when the assistant or thread is missing, the thread already has an active run (one active run per thread) or max_prompt_tokens is not a positive integer.

Runs use the whole thread as context (earlier messages are replayed to the agent), receive an abort signal on cancel, and honour additional_instructions, response_format (json_object / json_schema), max_completion_tokens, top_p, tool_choice (none or a specific function), parallel_tool_calls and truncation_strategy (last_messages). max_prompt_tokens is a best-effort budget (estimated at about 4 characters per token, instructions included): the oldest replayed messages are dropped until the prompt fits, and when the last user message alone does not fit the run ends incomplete with incomplete_details: { reason: 'max_prompt_tokens' }. Image parts (image_url, and image_file uploads with a png/jpg/jpeg/gif/webp extension) are passed to the model.

streamRunEvents(runId, fromIndex = 0) replays the run's event log from fromIndex and ends after the next done event. getRunEventCursor(runId) returns the current log position (use it before submitToolOutputs to stream only the continuation), and getStreamEmitter(runId) exposes the raw EventEmitter. Token deltas are emitted only for runs created with stream: true.

Each thread, message and run method takes the caller's owner as its last argument (adapter.getThread(id, owner), adapter.createRun(threadId, request, owner), ...): a thread or run of another owner is treated as missing. Leave it out for unscoped access, which is what the server does without apiKeys.

Runs live in the adapter's memory, unlike assistants, threads, messages and files, which go to storage. Poll, cancel or submit tool outputs to a run on the same process that created it: with several processes, run a single instance or route each thread to the same one (sticky routing).

Run Steps

Every run records its steps, as the Assistants API does:

• a tool_calls step for each model turn that called tools, with each call's id, name, arguments and output (null until the output is known). A step whose calls wait on the client stays in_progress while the run is in requires_action, and completes with the submitted outputs after submit_tool_outputs.
• a message_creation step for each assistant message, with its message_id. A run writes one message per model turn that answered with text: a turn that writes "Let me check." and then calls a tool stores that text as its own completed message (and completes its step) before the tool_calls step opens, so steps never overlap and the stored messages hold exactly the streamed text. A run without stream: true stores the same messages and steps, taking each turn's text as the run's middleware leaves it, also when the turn calls a client function and the run waits in requires_action.

Steps carry created_at / completed_at / failed_at / cancelled_at / expired_at (unix seconds) and usage: the tokens of the model call that produced the step, when the agent run reports them (a turn that wrote text and then called tools counts toward its message, the first step it produced), or the run's usage when the run has a single step. When the run is cancelled, expires or fails, the steps still in progress end cancelled, expired or failed (with the run's last_error), and a message the run had started streaming is kept with status: 'incomplete' and incomplete_details.reason run_cancelled, run_expired or run_failed. thread.message.completed goes out once the message is stored, so a client that reads the message on that event finds it, and a run reaches requires_action only once the messages before the pause are stored. A store that fails, or a thread deleted while the run writes to it, fails the run: that message ends incomplete with run_failed and its message_creation step failed.

const steps = await openai.beta.threads.runs.steps.list(run.id, { thread_id: thread.id });

for (const step of steps.data) {
  if (step.step_details.type === 'tool_calls') {
    for (const call of step.step_details.tool_calls) {
      if (call.type === 'function') console.log(call.function.name, call.function.output);
    }
  }
}

const inProcess = adapter.listRunSteps(thread.id, run.id); // oldest first, undefined for an unknown run
const one = adapter.getRunStep(thread.id, run.id, steps.data[0].id);

Steps are kept with their run, so they live in memory and leave with it (maxStoredRuns). The include[] parameter of the step endpoints is accepted and ignored, since there are no file_search results.

Tool Outputs

Server-side tools run inside Cogitator. Assistant tools of type function are executed by the API client: when the model calls one, the run moves to requires_action with the pending calls, and continues after all outputs are submitted (runs waiting longer than 10 minutes expire). Outputs must cover every pending call; missing or unknown tool_call_ids are rejected. With the OpenAI SDK use submitToolOutputsAndPoll / submitToolOutputsStream.

Server-side tools that need approval (requiresApproval, or the guardrails) come to the client the same way, as calls of functions it never declared: the run waits in requires_action instead of completing without them. The output decides the call: {"approved": true} (or approve) runs it, {"approved": false, "reason": "..."} declines it with the reason, and any other text declines it with that text, so nothing ambiguous runs a server tool. A configured guardrails.onToolApproval decides server tools itself and they never reach the client. While the run waits it is paused rather than running, so the run timeout (limits.defaultTimeout, 120 s by default) does not count the client's time: only expires_at (10 minutes per wait) does.

A server-side tool that asks a question (context.interrupt()) fails the run (status: 'failed'), with last_error.message naming each question, since submit_tool_outputs can carry function outputs only. The run never waits in requires_action with no calls to submit.

const run = adapter.getRun(thread.id, runId);

if (run?.status === 'requires_action') {
  const toolCalls = run.required_action?.submit_tool_outputs.tool_calls ?? [];

  const outputs = await Promise.all(
    toolCalls.map(async (call) => ({
      tool_call_id: call.id,
      output: await executeMyTool(call.function.name, call.function.arguments),
    }))
  );

  await adapter.submitToolOutputs(thread.id, runId, {
    tool_outputs: outputs,
  });
}

Thread Manager

The ThreadManager handles storage for threads, messages, assistants, and files. Get the adapter's instance with adapter.getThreadManager(), or create one directly.

import { ThreadManager } from '@cogitator-ai/openai-compat';

const manager = new ThreadManager(); // or new ThreadManager(storage)

Assistant Storage

interface StoredAssistant {
  id: string;
  name: string | null;
  description?: string | null;
  model: string;
  instructions: string | null;
  tools: AssistantTool[];
  metadata: Record<string, string>;
  temperature?: number;
  top_p?: number;
  response_format?: ResponseFormat;
  created_at: number;
}

const assistant = await manager.createAssistant({
  model: 'openai/gpt-6.1-sol',
  name: 'Helper',
  instructions: 'Be helpful',
});

const fetched = await manager.getAssistant(assistant.id);
const updated = await manager.updateAssistant(assistant.id, { name: 'Expert' });
const all = await manager.listAssistants();
await manager.deleteAssistant(assistant.id);

Thread Storage

const thread = await manager.createThread({ key: 'value' });
const fetched = await manager.getThread(thread.id);
await manager.updateThread(thread.id, { metadata: { key: 'other' } });
await manager.deleteThread(thread.id);

Message Operations

const message = await manager.addMessage(thread.id, {
  role: 'user',
  content: 'Hello!',
});

const assistantMsg = await manager.addAssistantMessage(
  thread.id,
  'Hi there!',
  assistant.id,
  run.id
  // optional 5th argument: message id to reuse
);

const messages = await manager.listMessages(thread.id, {
  limit: 50,
  order: 'desc',
});

const one = await manager.getMessage(thread.id, message!.id);

const llmMessages = await manager.getMessagesForLLM(thread.id); // { role, content, images? }[]

addMessage / addAssistantMessage resolve to undefined when the thread does not exist.

File Management

const file = await manager.addFile(Buffer.from('file content'), 'document.txt', 'assistants');

const fetched = await manager.getFile(file.id); // StoredFile with `content: Buffer`

const all = await manager.listFiles();

await manager.deleteFile(file.id);

Persistent Storage

By default, ThreadManager uses in-memory storage. For production, use Redis or PostgreSQL backends.

Storage Backends

import {
  ThreadManager,
  InMemoryThreadStorage,
  RedisThreadStorage,
  PostgresThreadStorage,
  createThreadStorage,
} from '@cogitator-ai/openai-compat';

In-Memory (Default)

const manager = new ThreadManager();
// equivalent to:
const explicit = new ThreadManager(new InMemoryThreadStorage());

Redis Storage

Requires the ioredis peer dependency:

pnpm add ioredis

const storage = new RedisThreadStorage({
  host: 'localhost', // default: 'localhost'
  port: 6379, // default: 6379
  keyPrefix: 'cogitator:openai:', // default: 'cogitator:openai:'
  ttl: 86400, // seconds, default: 86400 (24h); 0 disables expiry
});
await storage.connect();

const manager = new ThreadManager(storage);

// When done:
await storage.disconnect();

With a connection URL (takes precedence over host / port):

const storage = new RedisThreadStorage({
  url: 'redis://user:pass@localhost:6379/0',
});

PostgreSQL Storage

Requires the pg peer dependency:

pnpm add pg

const storage = new PostgresThreadStorage({
  connectionString: 'postgresql://user:pass@localhost:5432/db',
  schema: 'public', // default: 'public'
  tableName: 'openai_compat_data', // default: 'openai_compat_data'
});
await storage.connect(); // creates the table and its indexes if missing

const manager = new ThreadManager(storage);

// When done:
await storage.disconnect();

schema and tableName must be plain SQL identifiers (/^[a-zA-Z_][a-zA-Z0-9_]*$/); the constructor throws otherwise.

Factory Function

const memory = createThreadStorage(); // or createThreadStorage({ type: 'memory' })

const redis = createThreadStorage({
  type: 'redis',
  host: 'localhost',
  port: 6379,
});
await redis.connect?.();

const postgres = createThreadStorage({
  type: 'postgres',
  connectionString: 'postgresql://localhost/db',
});
await postgres.connect?.();

The factory returns an unconnected ThreadStorage; call connect() before use.

Using with the Adapter or Server

import { OpenAIAdapter, RedisThreadStorage, createOpenAIServer } from '@cogitator-ai/openai-compat';
import { Cogitator } from '@cogitator-ai/core';

const storage = new RedisThreadStorage({ host: 'localhost' });
await storage.connect();

const cogitator = new Cogitator({ llm: { defaultModel: 'openai/gpt-6.1-sol' } });

// In-process adapter backed by Redis
const adapter = new OpenAIAdapter(cogitator, { tools: [], storage });

// Or the REST server
const server = createOpenAIServer(cogitator, { storage });

Storage is the single source of truth for assistants, threads, messages and files (no in-process cache), so several server instances can share one Redis or PostgreSQL backend. Runs are not persisted: they stay in the memory of the instance that executes them. Redis listings use SCAN + MGET. ioredis and pg are loaded on connect(), with an install hint if missing.

ThreadStorage Interface

All storage backends implement this interface:

interface ThreadStorage {
  // Threads
  saveThread(id: string, thread: StoredThread): Promise<void>;
  loadThread(id: string): Promise<StoredThread | null>;
  deleteThread(id: string): Promise<boolean>;
  listThreads(options?: ListStoredThreadsOptions): Promise<StoredThreadPage>;

  // Assistants
  saveAssistant(id: string, assistant: StoredAssistant): Promise<void>;
  loadAssistant(id: string): Promise<StoredAssistant | null>;
  deleteAssistant(id: string): Promise<boolean>;
  listAssistants(): Promise<StoredAssistant[]>;

  // Files
  saveFile(id: string, file: StoredFile): Promise<void>;
  loadFile(id: string): Promise<StoredFile | null>;
  deleteFile(id: string): Promise<boolean>;
  listFiles(): Promise<StoredFile[]>;

  // Lifecycle (optional)
  connect?(): Promise<void>;
  disconnect?(): Promise<void>;
}

StoredThread is { thread: Thread; messages: Message[]; owner?: string }, owner being the digest of the API key that created the thread. StoredFile is { id, content: Buffer, filename, created_at, purpose? }.

Listing threads

The Assistants API has no route that lists threads, so listThreads is for operators, on the storage or on ThreadManager:

import { apiKeyOwner } from '@cogitator-ai/openai-compat';

const first = await storage.listThreads({ owner: apiKeyOwner('sk-key1'), limit: 50 });
const next = await storage.listThreads({
  owner: apiKeyOwner('sk-key1'),
  limit: 50,
  cursor: first.nextCursor,
});
const unowned = await storage.listThreads({ owner: null }); // threads stored before key scoping

| Option | Default | Meaning | | -------- | -------- | ------------------------------------------------------------------------------ | | owner | all | Threads of this owner only, null for the threads without one | | limit | 50 | Threads per page, from 1 to 500 (RangeError otherwise) | | cursor | - | nextCursor of the page before, with the same filters and order | | order | 'desc' | By creation time, then id: desc starts with the newest threads, asc oldest |

A page is { threads, nextCursor? }, nextCursor being absent on the last page. A cursor the storage did not hand out throws TypeError. What a page costs depends on the backend:

• In-memory: filters and sorts the threads of the process.
• PostgreSQL: WHERE, ORDER BY and LIMIT in SQL, served by the partial index idx_<table>_thread_owner on (owner, creation time, id) that connect() creates. Ids compare with COLLATE "C", the code unit order the other backends use, whatever the database collation. A listing without owner reads every thread row.
• Redis: threads are plain keys with a TTL and no index, so every page reads all thread keys (SCAN + MGET, messages included) and sorts them in memory. That is fine for an occasional listing, use PostgreSQL when there are many threads.

Custom Storage Implementation

import type { ThreadStorage, StoredThread } from '@cogitator-ai/openai-compat';

class MyCustomStorage implements ThreadStorage {
  async saveThread(id: string, thread: StoredThread): Promise<void> {
    // Your implementation
  }
  // ... implement all methods
}

const manager = new ThreadManager(new MyCustomStorage());

Supported Endpoints

All endpoints except /health live under /v1. List endpoints for assistants, messages and runs accept limit (1-100, default 20), order (asc / desc, default desc), after and before, and return { object: 'list', data, first_id, last_id, has_more }. after gives the items right after the cursor and before the items right before it, both in list order, and has_more tells whether more items lie beyond the page, away from the cursor. GET /v1/files accepts limit (1-10 000, default 10 000), order (default desc) and after, plus a purpose filter.

Models

| Method | Endpoint | Description | | ------ | ---------------- | ----------------------------------------------------- | | GET | /v1/models | Lists the registered agents and the cogitator model | | GET | /v1/models/:id | Gets one model | | GET | /health | Health check (public, no auth) |

Chat Completions and Responses

| Method | Endpoint | Description | | ------ | ------------------------------- | ------------------------------------------- | | POST | /v1/chat/completions | Chat completion (SSE with stream: true) | | POST | /v1/responses | Create a response (SSE with stream: true) | | GET | /v1/responses/:id | Get a stored response | | GET | /v1/responses/:id/input_items | List the input items of a response | | DELETE | /v1/responses/:id | Delete a stored response |

The Assistants API endpoints below are deprecated.

Assistants

| Method | Endpoint | Description | | ------ | -------------------- | ---------------- | | POST | /v1/assistants | Create assistant | | GET | /v1/assistants | List assistants | | GET | /v1/assistants/:id | Get assistant | | POST | /v1/assistants/:id | Update assistant | | DELETE | /v1/assistants/:id | Delete assistant |

Threads

| Method | Endpoint | Description | | ------ | ----------------- | ------------------------------------------- | | POST | /v1/threads | Create thread (optional initial messages) | | GET | /v1/threads/:id | Get thread | | POST | /v1/threads/:id | Update thread metadata (merged) | | DELETE | /v1/threads/:id | Delete thread |

Messages

| Method | Endpoint | Description | | ------ | ---------------------------------- | ------------------------------------- | | POST | /v1/threads/:id/messages | Add message (user or assistant) | | GET | /v1/threads/:id/messages | List messages (also filters run_id) | | GET | /v1/threads/:id/messages/:msg_id | Get message |

Runs

| Method | Endpoint | Description | | ------ | -------------------------------------------------- | ------------------- | | POST | /v1/threads/runs | Create thread + run | | POST | /v1/threads/:id/runs | Create run | | GET | /v1/threads/:id/runs | List runs | | GET | /v1/threads/:id/runs/:run_id | Get run status | | POST | /v1/threads/:id/runs/:run_id/cancel | Cancel run | | POST | /v1/threads/:id/runs/:run_id/submit_tool_outputs | Submit tool outputs | | GET | /v1/threads/:id/runs/:run_id/steps | List run steps | | GET | /v1/threads/:id/runs/:run_id/steps/:step_id | Get run step |

The create and submit_tool_outputs endpoints stream SSE when the body has stream: true. Run steps are listed like the other lists (limit, order, after, before). There is no GET /v1/threads, as in the OpenAI API: operators list threads from storage.

Files

| Method | Endpoint | Description | | ------ | ----------------------- | ------------------------------------------------- | | POST | /v1/files | Upload file (multipart/form-data) | | GET | /v1/files | List files (purpose, limit, order, after) | | GET | /v1/files/:id | Get file metadata | | GET | /v1/files/:id/content | Download file content | | DELETE | /v1/files/:id | Delete file |

Uploads take a file part and an optional purpose field (assistants, assistants_output, batch, batch_output, fine-tune, fine-tune-results, vision; default assistants).

Error Handling

The server returns OpenAI-compatible error responses:

interface OpenAIError {
  error: {
    message: string;
    type: string;
    param?: string;
    code?: string;
  };
}

formatOpenAIError(code, message, type?, param?) builds this shape if you add your own routes via server.getFastify().

Error Types

| HTTP Status | Type | Code | When | | ----------- | ----------------------- | ------------------------------------ | ------------------------------------------------------ | | 400 | invalid_request_error | invalid_request, missing_file | Invalid parameters, unknown assistant, thread busy | | 401 | invalid_request_error | missing_api_key, invalid_api_key | Missing, malformed or unknown API key | | 404 | invalid_request_error | not_found | Unknown thread, message, run, assistant, file or route | | 429 | rate_limit_error | rate_limit_exceeded | Errors thrown with status 429 | | 500 | server_error | internal_error | Unhandled errors |

Every 5xx answer says Internal server error; the details are logged on the server, never sent to the client. The adapter refuses invalid requests (unknown assistant or thread, a busy thread, a run in the wrong state, missing or unknown tool outputs, an invalid max_prompt_tokens) by throwing InvalidRequestError (exported, with an optional param), which the server answers as 400 invalid_request; any other error from the adapter is a server failure.

A run that fails during execution does not produce an HTTP error: it ends with status: 'failed' and last_error: { code: 'server_error', message }. message is Internal server error unless the cause is a CogitatorError or an InvalidRequestError, whose message is kept; the original error is logged.

Client-Side Error Handling

try {
  const run = await openai.beta.threads.runs.create(threadId, {
    assistant_id: 'invalid-id',
  });
} catch (error) {
  if (error instanceof OpenAI.APIError) {
    console.log(error.status); // 400
    console.log(error.message);
    console.log(error.code); // 'invalid_request'
  }
}

Run Status

Runs go through the following states:

type RunStatus =
  | 'queued'
  | 'in_progress'
  | 'requires_action'
  | 'cancelling'
  | 'cancelled'
  | 'failed'
  | 'completed'
  | 'incomplete'
  | 'expired';

Status Flow

queued → in_progress → completed
                     → failed
                     → requires_action → in_progress → ...
                                       → expired (no outputs within 10 minutes)
                     → incomplete (last message exceeds max_prompt_tokens, or the answer hit max_completion_tokens)

queued / in_progress / requires_action → cancelling → cancelled

A run is incomplete with incomplete_details.reason: 'max_prompt_tokens' when the last message alone does not fit in max_prompt_tokens (see Run Execution), and with 'max_completion_tokens' when the answer stopped at the output token limit (max_completion_tokens, or the model's own). The answer is then kept as a message with status: 'incomplete' and incomplete_details.reason: 'max_tokens', and the stream sends thread.message.incomplete and thread.run.incomplete instead of the completed events. An answer the provider's content filter withheld is an incomplete message with reason content_filter in a completed run.

Polling for Completion

import OpenAI from 'openai';

async function waitForRun(
  openai: OpenAI,
  threadId: string,
  runId: string
): Promise<OpenAI.Beta.Threads.Run> {
  const terminalStates = ['completed', 'failed', 'cancelled', 'expired', 'requires_action'];

  while (true) {
    const run = await openai.beta.threads.runs.retrieve(runId, { thread_id: threadId });

    if (terminalStates.includes(run.status)) {
      return run;
    }

    await new Promise((r) => setTimeout(r, 1000));
  }
}

The SDK's createAndPoll / submitToolOutputsAndPoll do the same.

SSE Streaming

Runs created with stream: true answer with Server-Sent Events and stream tokens as they are generated.

Streaming with OpenAI SDK

const run = await openai.beta.threads.runs.create(threadId, {
  assistant_id: assistant.id,
  stream: true,
});

for await (const event of run) {
  if (event.event === 'thread.message.delta') {
    for (const content of event.data.delta.content ?? []) {
      if (content.type === 'text' && content.text?.value) {
        process.stdout.write(content.text.value);
      }
    }
  }
}

Create Thread and Run with Streaming

const run = await openai.beta.threads.createAndRun({
  assistant_id: assistant.id,
  thread: {
    messages: [{ role: 'user', content: 'Hello!' }],
  },
  stream: true,
});

for await (const event of run) {
  console.log(event.event, event.data);
}

Stream Events

The server emits these events:

type EmittedEvent =
  | { event: 'thread.run.created'; data: Run }
  | { event: 'thread.run.queued'; data: Run }
  | { event: 'thread.run.in_progress'; data: Run }
  | { event: 'thread.run.requires_action'; data: Run }
  | { event: 'thread.run.completed'; data: Run }
  | { event: 'thread.run.failed'; data: Run }
  | { event: 'thread.run.cancelling'; data: Run }
  | { event: 'thread.run.cancelled'; data: Run }
  | { event: 'thread.run.expired'; data: Run }
  | { event: 'thread.run.incomplete'; data: Run }
  | { event: 'thread.run.step.created'; data: RunStep }
  | { event: 'thread.run.step.in_progress'; data: RunStep }
  | { event: 'thread.run.step.delta'; data: RunStepDelta }
  | { event: 'thread.run.step.completed'; data: RunStep }
  | { event: 'thread.run.step.failed'; data: RunStep }
  | { event: 'thread.run.step.cancelled'; data: RunStep }
  | { event: 'thread.run.step.expired'; data: RunStep }
  | { event: 'thread.message.created'; data: Message }
  | { event: 'thread.message.in_progress'; data: Message }
  | { event: 'thread.message.delta'; data: MessageDelta }
  | { event: 'thread.message.completed'; data: Message }
  | { event: 'thread.message.incomplete'; data: Message }
  | { event: 'done'; data: '[DONE]' };

A stream ends with done after the run finishes or pauses on requires_action; continue a paused run with submit_tool_outputs and stream: true, which streams only the events after the submission.

Step events come in the order of the OpenAI API: thread.run.step.created and thread.run.step.in_progress open a step, before the thread.message.* events of its message. A tool_calls step opens with an empty tool_calls list, gets one thread.run.step.delta per call (so the SDK's toolCallCreated fires), and thread.run.step.completed carries the outputs. A run that waits on the client streams thread.run.step.delta and then thread.run.requires_action, and the stream after submit_tool_outputs starts with thread.run.in_progress and the completed step.

Message Delta Format

All text deltas target content index 0, so SDK helpers accumulate them into a single text part whose message id matches the stored message.

interface MessageDelta {
  id: string;
  object: 'thread.message.delta';
  delta: {
    content?: {
      index: number;
      type: 'text';
      text?: { value?: string };
    }[];
  };
}

Raw SSE Endpoint

For non-SDK usage, the SSE stream is available at:

POST /v1/threads/{thread_id}/runs
Content-Type: application/json

{
  "assistant_id": "asst_xxx",
  "stream": true
}

Response format (Server-Sent Events):

event: thread.run.created
data: {"id":"run_xxx","status":"queued",...}

event: thread.run.queued
data: {"id":"run_xxx","status":"queued",...}

event: thread.run.in_progress
data: {"id":"run_xxx","status":"in_progress",...}

event: thread.message.created
data: {"id":"msg_xxx","status":"in_progress",...}

event: thread.message.in_progress
data: {"id":"msg_xxx","status":"in_progress",...}

event: thread.message.delta
data: {"id":"msg_xxx","object":"thread.message.delta","delta":{"content":[{"index":0,"type":"text","text":{"value":"Hello"}}]}}

event: thread.message.delta
data: {"id":"msg_xxx","object":"thread.message.delta","delta":{"content":[{"index":0,"type":"text","text":{"value":" world"}}]}}

event: thread.message.completed
data: {"id":"msg_xxx","status":"completed",...}

event: thread.run.completed
data: {"id":"run_xxx","status":"completed",...}

event: done
data: [DONE]

Type Reference

Server & Adapter

import type {
  OpenAIServerConfig,
  AuthConfig,
  OpenAIAdapterOptions,
  StreamEventType,
  StreamEventData,
  StreamEmitterEvents,
  RunStreamEvent,
  LLMThreadMessage,
  CreateAssistantParams,
  UpdateAssistantParams,
} from '@cogitator-ai/openai-compat';

import {
  COGITATOR_MODEL_ID,
  formatOpenAIError,
  InvalidRequestError,
} from '@cogitator-ai/openai-compat';

Core Types

import type {
  OpenAIError,
  ListResponse,
  Assistant,
  AssistantTool,
  FunctionDefinition,
  ResponseFormat,
  JsonSchema,
  CreateAssistantRequest,
  UpdateAssistantRequest,
} from '@cogitator-ai/openai-compat';

Thread Types

import type { Thread, ToolResources, CreateThreadRequest } from '@cogitator-ai/openai-compat';

Message Types

import type {
  Message,
  MessageContent,
  TextContent,
  TextAnnotation,
  ImageFileContent,
  ImageUrlContent,
  Attachment,
  CreateMessageRequest,
  MessageContentPart,
} from '@cogitator-ai/openai-compat';

Run Types

import type {
  Run,
  RunStatus,
  RequiredAction,
  ToolCall,
  RunError,
  Usage,
  ToolChoice,
  IncompleteDetails,
  TruncationStrategy,
  CreateRunRequest,
  SubmitToolOutputsRequest,
  ToolOutput,
} from '@cogitator-ai/openai-compat';

Run Step Types

import type { RunStep, StepDetails, StepToolCall, RunStepDelta } from '@cogitator-ai/openai-compat';

File Types

import type { FileObject, FilePurpose, UploadFileRequest } from '@cogitator-ai/openai-compat';

Stream Types

import type {
  StreamEvent,
  MessageDelta,
  MessageContentDelta,
  RunStepDelta,
} from '@cogitator-ai/openai-compat';

Storage Types

import type {
  ThreadStorage,
  StoredThread,
  StoredAssistant,
  StoredFile,
  RedisThreadStorageConfig,
  PostgresThreadStorageConfig,
} from '@cogitator-ai/openai-compat';

import {
  InMemoryThreadStorage,
  RedisThreadStorage,
  PostgresThreadStorage,
  createThreadStorage,
} from '@cogitator-ai/openai-compat';

Examples

Chat Bot with Memory

import { createOpenAIServer } from '@cogitator-ai/openai-compat';
import { Cogitator } from '@cogitator-ai/core';
import OpenAI from 'openai';

const cogitator = new Cogitator({
  llm: { defaultModel: 'ollama/llama3.2:latest' },
});

const server = createOpenAIServer(cogitator, { port: 8080 });
await server.start();

const openai = new OpenAI({
  baseURL: server.getBaseUrl(),
  apiKey: 'not-needed',
});

const assistant = await openai.beta.assistants.create({
  name: 'Chat Bot',
  instructions: 'You are a friendly chat bot. Remember previous messages.',
  model: 'ollama/llama3.2:latest',
});

const thread = await openai.beta.threads.create();

async function chat(message: string): Promise<string> {
  await openai.beta.threads.messages.create(thread.id, {
    role: 'user',
    content: message,
  });

  const run = await openai.beta.threads.runs.createAndPoll(thread.id, {
    assistant_id: assistant.id,
  });
  if (run.status !== 'completed') throw new Error(run.last_error?.message ?? run.status);

  const messages = await openai.beta.threads.messages.list(thread.id, {
    limit: 1,
    order: 'desc',
  });

  const content = messages.data[0].content[0];
  return content.type === 'text' ? content.text.value : '';
}

console.log(await chat('Hi, my name is Alex'));
console.log(await chat('What is my name?'));

Server-Side Tools

import { createOpenAIServer } from '@cogitator-ai/openai-compat';
import { Cogitator, tool } from '@cogitator-ai/core';
import { z } from 'zod';
import OpenAI from 'openai';

const lookupOrder = tool({
  name: 'lookup_order',
  description: 'Look up an order by id',
  parameters: z.object({
    orderId: z.string().describe('Order id'),
  }),
  execute: async ({ orderId }) => ({ orderId, status: 'shipped' }),
});

const cogitator = new Cogitator({
  llm: { defaultModel: 'openai/gpt-6.1-sol' },
});

const server = createOpenAIServer(cogitator, {
  port: 8080,
  tools: [lookupOrder],
});

await server.start();

const openai = new OpenAI({
  baseURL: server.getBaseUrl(),
  apiKey: 'not-needed',
});

const assistant = await openai.beta.assistants.create({
  name: 'Support Bot',
  instructions: 'Use lookup_order to answer questions about orders.',
  model: 'openai/gpt-6.1-sol',
});
// Server-side tools (`tools` on the server) are available to every run without
// being declared on the assistant. Declare `function` tools only for functions
// your client executes (they surface as `requires_action`).

Client-Executed Function Tools

const assistant = await openai.beta.assistants.create({
  model: 'openai/gpt-6.1-sol',
  tools: [
    {
      type: 'function',
      function: {
        name: 'get_weather',
        description: 'Current weather for a city',
        parameters: {
          type: 'object',
          properties: { city: { type: 'string' } },
          required: ['city'],
        },
      },
    },
  ],
});

let run = await openai.beta.threads.runs.createAndPoll(thread.id, {
  assistant_id: assistant.id,
});

while (run.status === 'requires_action' && run.required_action) {
  const tool_outputs = run.required_action.submit_tool_outputs.tool_calls.map((call) => ({
    tool_call_id: call.id,
    output: JSON.stringify({ city: JSON.parse(call.function.arguments).city, tempC: 21 }),
  }));
  run = await openai.beta.threads.runs.submitToolOutputsAndPoll(run.id, {
    thread_id: thread.id,
    tool_outputs,
  });
}

File Upload

import fs from 'node:fs';
import OpenAI from 'openai';

const openai = new OpenAI({
  baseURL: 'http://localhost:8080/v1',
  apiKey: 'key',
});

const file = await openai.files.create({
  file: fs.createReadStream('data.csv'),
  purpose: 'assistants',
});

console.log('Uploaded:', file.id);

const content = await openai.files.content(file.id);
console.log('Content:', await content.text());

await openai.files.delete(file.id);

Multi-Model Setup

const cogitator = new Cogitator({
  llm: { defaultModel: 'ollama/llama3.2:latest' },
});

const server = createOpenAIServer(cogitator, { port: 8080 });
await server.start();

const openai = new OpenAI({
  baseURL: server.getBaseUrl(),
  apiKey: 'not-needed',
});

const localAssistant = await openai.beta.assistants.create({
  name: 'Local Assistant',
  model: 'ollama/llama3.2:latest',
  instructions: 'Fast local responses',
});

const cloudAssistant = await openai.beta.assistants.create({
  name: 'Cloud Assistant',
  model: 'openai/gpt-6.1-sol',
  instructions: 'Complex reasoning tasks',
});

License

MIT