@backendkit-labs/agent-web

Status: stable — Maintained and production-ready. TriggerBus (cron/webhook/BullMQ) is experimental and frozen during the pilot phase.

HTTP/WebSocket server layer for @backendkit-labs/agent-core.

Wraps AgentEngine in a Fastify 5 server with streaming WebSocket sessions, JWT authentication, Prometheus metrics, rate limiting, a real-time monitor bus, and a flexible trigger system (cron jobs, webhooks, BullMQ queues).

Table of contents

• Installation
• Quick start
• AgentServer options
• HTTP and WebSocket routes
• WebSocket protocol
• EngineFactory pattern
• SessionManager
• JWT authentication
• Prometheus metrics
• Rate limiting
• Session metadata store
• MonitorBus
• TriggerBus — cron, webhooks, queues
• HeadlessTransport
• Full production example

Installation

npm install @backendkit-labs/agent-web @backendkit-labs/agent-core

Optional integrations (install only what you need):

npm install jsonwebtoken          # JWT auth
npm install prom-client           # Prometheus metrics
npm install @fastify/rate-limit   # rate limiting
npm install node-cron             # cron triggers
npm install bullmq                # queue triggers
npm install ioredis               # Redis session store

Quick start

import { AgentServer } from '@backendkit-labs/agent-web';
import type { EngineFactory } from '@backendkit-labs/agent-web';
import { AgentEngine, AgentRegistry, ToolRegistry, ProviderRegistry } from '@backendkit-labs/agent-core';

const engineFactory: EngineFactory = (sessionId, transport) =>
  new AgentEngine({
    transport,
    sessionId,
    // ...your engine config
  });

const server = new AgentServer({
  port:          3000,
  engineFactory,
  cors:          '*',
});

await server.start();
// WS  /chat      — streaming chat sessions
// WS  /monitor   — real-time read-only observer
// GET /health    — { status, sessions, monitors }
// GET /sessions  — active session list
// DELETE /sessions/:id — close a session

AgentServer options

interface AgentServerOptions {
  // Required
  engineFactory: EngineFactory;       // creates a new AgentEngine per session

  // Network
  port?:   number;                    // default: 3000
  host?:   string;                    // default: '127.0.0.1' (use '0.0.0.0', with auth, for remote callers)
  cors?:   string | string[];         // default: '*' (none with strict)
  strict?: boolean;                   // fail-closed defaults, see "Strict mode"
  logger?: boolean | object;          // Fastify logger, default: false

  // Auth
  auth?: AuthHook;                    // called on every request before handlers

  // Rate limiting (requires @fastify/rate-limit)
  rateLimiting?: {
    max?:        number;              // max requests per IP per window, default: 100
    timeWindow?: string;             // e.g. '1 minute', '30 seconds'
  };

  // Prometheus metrics (requires prom-client)
  metrics?: {
    enabled:          boolean;        // must be true to activate
    path?:            string;         // default: '/metrics'
    prefix?:          string;         // metric name prefix
    collectNodeMetrics?: boolean;     // include Node.js process metrics
  };

  // Session management
  maxIdleMs?:   number;               // idle timeout before session is closed
  maxSessions?: number;               // max concurrent sessions
  store?:       SessionMetaStore;     // durable metadata store (default: in-memory)

  // Triggers
  triggers?: TriggerBus;             // wires webhook routes and starts cron jobs
}

HTTP and WebSocket routes

WebSocket protocol

/chat — chat sessions

Client → server:

{ "type": "input",  "content": "Your message here" }
{ "type": "abort" }

Server → client (streamed AgentEvent frames):

{ "type": "run_start",     "sessionId": "abc-123" }
{ "type": "token",         "content": "Hello" }
{ "type": "tool_call",     "name": "search", "args": { "query": "..." } }
{ "type": "tool_result",   "name": "search", "result": "..." }
{ "type": "agent_switch",  "from": "general", "to": "security-reviewer" }
{ "type": "done",          "final": "Full assistant response" }
{ "type": "error",         "message": "Something went wrong" }

/monitor — real-time observer

Connect to observe all sessions without sending any messages:

{ "type": "monitor_hello",         "sessionIds": ["abc-123", "def-456"] }
{ "type": "monitor_session_open",  "sessionId": "ghi-789" }
{ "type": "monitor_session_close", "sessionId": "ghi-789" }
{ "sessionId": "abc-123", "type": "run_input", "content": "..." }
{ "sessionId": "abc-123", "type": "token",     "content": "..." }
{ "sessionId": "abc-123", "type": "done",      "final": "..." }

EngineFactory pattern

EngineFactory is a function (sessionId: string, transport: Transport) => AgentEngine. Each incoming WebSocket connection gets its own isolated engine instance.

Minimal factory

import type { EngineFactory } from '@backendkit-labs/agent-web';
import { AgentEngine } from '@backendkit-labs/agent-core';

const engineFactory: EngineFactory = (sessionId, transport) =>
  new AgentEngine({
    transport,
    sessionId,
    model:           { provider: 'openai', id: 'gpt-4o' },
    agents,
    tools,
    providers,
    defaultProvider: 'openai',
    defaultAgentId:  'general',
    maxIterations:   50,
    iterationMode:   'auto',
  });

Factory with per-session memory

import { createMemorySystem } from '@backendkit-labs/agent-core';
import * as path from 'path';
import * as os from 'os';

const engineFactory: EngineFactory = (sessionId, transport) => {
  // Isolated memory per session
  const sessDir = path.join(os.homedir(), '.bk-agent', 'sessions', sessionId);
  const memory = createMemorySystem({
    episodic:  { provider: 'json', path: path.join(sessDir, 'episodes.json') },
    semantic:  { provider: 'json', path: path.join(sessDir, 'knowledge.json') },
    procedural:{ provider: 'json', path: path.join(sessDir, 'patterns.json') },
  });

  return new AgentEngine({
    transport,
    sessionId,
    memorySystem: memory,
    ...baseConfig,
  });
};

Factory with userId from JWT

When JWT auth is enabled, the sessionId carries the authenticated user's context. You can also inject user-specific data:

const engineFactory: EngineFactory = (sessionId, transport) => {
  // sessionId is set to the JWT 'sub' claim when JwtAuth is used
  const userConfig = getUserConfig(sessionId); // look up per-user settings

  return new AgentEngine({
    transport,
    sessionId,
    projectContext: `User: ${userConfig.name}\nRole: ${userConfig.role}`,
    ...baseConfig,
  });
};

SessionManager

SessionManager creates, tracks, and cleans up sessions. It is created internally by AgentServer, but can also be used standalone.

import { SessionManager } from '@backendkit-labs/agent-web';

const sessions = new SessionManager({
  engineFactory,
  maxIdleMs:   30 * 60 * 1000,  // close after 30 min of inactivity
  maxSessions: 100,
  monitorBus:  myMonitorBus,
  metrics:     myMetricsCollector,
  store:       myMetaStore,
});

// Lifecycle
const session = await sessions.getOrCreate('session-id', myTransport);
await sessions.close('session-id');
const all = sessions.list(); // SessionMeta[]

JWT authentication

JwtAuth validates Bearer tokens in the Authorization header or ?token= query parameter (useful for browser WebSocket connections that can't set custom headers).

Strict mode

For a server that other people reach, turn on both switches:

const auth = new JwtAuth({ secret: process.env.JWT_SECRET!, allowQueryToken: false }).hook();
new AgentServer({ engineFactory, host: '0.0.0.0', auth, strict: true, cors: ['https://app.example.com'] });

• allowQueryToken: false ignores ?token= (a token in a URL lands in logs, proxies and history). A browser sends the token as the WebSocket subprotocol: new WebSocket(url, ['bearer', jwt]); useAgentSession({ url, token }) in agent-web-ui does it for you.
• strict: true allows no CORS origin unless cors lists it, adds X-Content-Type-Options, X-Frame-Options: DENY and Referrer-Policy: no-referrer, and refuses (throws) to listen on a non-loopback host without auth.

Setup

import { AgentServer, JwtAuth } from '@backendkit-labs/agent-web';

const jwt = new JwtAuth({
  secret:        process.env.JWT_SECRET!,  // or use publicKey for RS256
  algorithm:     'HS256',                  // default
  sessionIdClaim:'sub',                    // JWT claim to use as sessionId
  expiresIn:     '24h',
});

const server = new AgentServer({
  port:          3000,
  engineFactory,
  auth:          jwt.hook(),               // AuthHook applied to all routes
});

Issuing tokens (your login endpoint)

import { JwtAuth } from '@backendkit-labs/agent-web';

const jwt = new JwtAuth({ secret: process.env.JWT_SECRET! });

// Issue a token for an authenticated user
const token = jwt.sign({
  sub:  'user-123',
  role: 'admin',
  name: 'Alice',
});

// Return to client — they send it as:
// Authorization: Bearer <token>
// or ws://host/chat?token=<token>

Per-user session isolation

When sessionIdClaim is 'sub', every user gets their own session automatically. Two requests with the same token always hit the same session:

const jwt = new JwtAuth({
  secret:         process.env.JWT_SECRET!,
  sessionIdClaim: 'sub',  // user ID as session key
});

Custom auth hook

If you use a different auth system (API keys, OAuth tokens), implement AuthHook directly:

import type { AuthHook } from '@backendkit-labs/agent-web';

const apiKeyAuth: AuthHook = async (req, reply) => {
  const key = req.headers['x-api-key'];
  if (!key || !validApiKeys.has(key as string)) {
    reply.code(401).send({ error: 'Unauthorized' });
  }
};

const server = new AgentServer({ auth: apiKeyAuth, ...opts });

Prometheus metrics

Requires prom-client to be installed. Exposes a /metrics endpoint with standard Prometheus counters and histograms.

const server = new AgentServer({
  ...opts,
  metrics: {
    enabled:            true,
    path:               '/metrics',         // default
    prefix:             'bk_agent_',        // prepended to all metric names
    collectNodeMetrics: true,               // include Node.js process metrics
  },
});

Exposed metrics

Using MetricsCollector directly

import { MetricsCollector } from '@backendkit-labs/agent-web';

const collector = new MetricsCollector({ prefix: 'my_agent_' });

// Record a completed run
const endRun = collector.startRun('session-abc');
// ... after run completes:
endRun('success');   // or endRun('error')

// Get Prometheus text format
const text = await collector.getMetrics();
// # HELP my_agent_runs_total ...
// my_agent_runs_total{status="success"} 42

Rate limiting

Requires @fastify/rate-limit. Applied globally to all routes:

const server = new AgentServer({
  ...opts,
  rateLimiting: {
    max:        60,          // max 60 requests per IP
    timeWindow: '1 minute',  // sliding window
  },
});

When the limit is exceeded, the server responds with:

{ "statusCode": 429, "error": "Too Many Requests", "message": "Rate limit exceeded" }

Session metadata store

By default, session metadata lives in memory and is lost on restart. Use RedisSessionMetaStore for durable metadata across restarts and horizontal scaling.

In-memory (default)

import { InMemorySessionMetaStore } from '@backendkit-labs/agent-web';

const server = new AgentServer({
  ...opts,
  store: new InMemorySessionMetaStore(),
});

Redis store (requires ioredis or node-redis)

RedisSessionMetaStore is duck-typed — it accepts any Redis client that implements .get(), .set(), and .del():

import { RedisSessionMetaStore } from '@backendkit-labs/agent-web';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);

const server = new AgentServer({
  ...opts,
  store: new RedisSessionMetaStore(redis, {
    ttlSeconds: 86400,    // keep session metadata for 24h
    prefix:     'bk:sess:', // key prefix
  }),
});

MonitorBus

MonitorBus broadcasts all session events to connected /monitor WebSocket observers. Used by the Kanban board in @backendkit-labs/agent-web-ui.

Standalone usage

import { MonitorBus } from '@backendkit-labs/agent-web';

const monitor = new MonitorBus();

// Subscribe to all events (e.g. for logging or alerting)
const unsub = monitor.subscribe((frame) => {
  if (frame.type === 'done') {
    console.log(`[${frame.sessionId}] run completed`);
  }
  if (frame.type === 'error') {
    alerting.notify(`Agent error in session ${frame.sessionId}: ${frame.message}`);
  }
});

// Clean up
unsub();

TriggerBus — cron, webhooks, queues

TriggerBus lets autonomous agent runs happen without a WebSocket client. Runs are visible in the Kanban monitor.

Cron trigger (requires node-cron)

import { TriggerBus } from '@backendkit-labs/agent-web';

const triggers = new TriggerBus(engineFactory)
  .addCron({
    name:        'daily-report',
    schedule:    '0 8 * * 1-5',     // weekdays at 8 AM
    buildPrompt: () => 'Generate the daily engineering standup summary from recent commits.',
    runOnStart:  false,              // set true to fire immediately on server start
    onResult:    (out) => slack.send('#daily', out),
    onError:     (err) => logger.error('[daily-report]', err),
  })
  .addCron({
    name:        'health-check',
    schedule:    '*/15 * * * *',    // every 15 minutes
    buildPrompt: () => 'Check all API endpoints and report any that are down.',
    onResult:    (out) => { if (out.includes('DOWN')) pagerduty.alert(out); },
  });

const server = new AgentServer({ triggers, ...opts });
await server.start();
// Cron jobs start automatically with the server

Webhook trigger

const triggers = new TriggerBus(engineFactory)
  .addWebhook({
    name:        'github-pr',
    path:        '/webhooks/github',             // registered as POST /webhooks/github
    secret:      process.env.GITHUB_SECRET,      // HMAC X-Hub-Signature-256 validation (optional)
    buildPrompt: (body) => {
      const pr = body.pull_request;
      return `Review PR #${pr.number}: "${pr.title}"\n${pr.diff_url}`;
    },
    onResult: (out, body) => github.postReview(body.pull_request.number, out),
    onError:  (err) => logger.error('[github-pr]', err),
  })
  .addWebhook({
    name:        'jira-issue',
    path:        '/webhooks/jira',
    buildPrompt: (body) => `Analyze Jira issue ${body.issue.key}: ${body.issue.fields.summary}`,
    onResult:    (out) => jira.addComment(body.issue.key, out),
  });

Testing a webhook:

curl -X POST http://localhost:3000/webhooks/github \
  -H "Content-Type: application/json" \
  -H "X-Hub-Signature-256: sha256=<hmac>" \
  -d '{ "pull_request": { "number": 42, "title": "Add auth module" } }'

Queue trigger (BullMQ — requires bullmq)

import { TriggerBus } from '@backendkit-labs/agent-web';

const triggers = new TriggerBus(engineFactory)
  .addQueue({
    name:        'document-processor',
    queueName:   'documents',           // BullMQ queue name
    connection:  process.env.REDIS_URL!, // Redis URL or connection object
    concurrency: 3,                     // process 3 jobs in parallel
    buildPrompt: (job) => {
      return `Summarize and extract key points from this document:\n\n${job.data.content}`;
    },
    onResult: (out, job) => {
      db.documents.update(job.data.id, { summary: out });
    },
    onError: (err, job) => {
      logger.error(`[queue] Failed to process document ${job?.data?.id}:`, err);
      // The error is re-thrown so BullMQ marks the job as failed and applies retry policy
    },
  });

Publishing a job (from anywhere in your system):

import { Queue } from 'bullmq';

const queue = new Queue('documents', { connection: process.env.REDIS_URL! });

await queue.add('process-doc', {
  id:      'doc-456',
  content: 'Full document text...',
}, {
  attempts:    3,
  backoff:     { type: 'exponential', delay: 2000 },
  removeOnComplete: 100,
  removeOnFail:     50,
});

Queue connection options

// URL string
connection: 'redis://localhost:6379'
connection: 'redis://:password@redis-host:6380/1'

// Connection object
connection: {
  host:     'redis-host',
  port:     6380,
  password: 'mypassword',
  username: 'default',
  db:       1,
  tls:      {},            // enable TLS
}

CronTriggerDef with all options

interface CronTriggerDef {
  name:         string;                        // unique name for the trigger
  schedule:     string;                        // standard cron expression
  buildPrompt:  (ts: Date) => string;          // returns the prompt to run
  agentId?:     string;                        // override default agent
  runOnStart?:  boolean;                       // fire immediately on server start
  onResult?:    (output: string) => void;
  onError?:     (err: Error) => void;
}

HeadlessTransport

HeadlessTransport is a Transport implementation that buffers all AgentEvent objects in memory instead of streaming them over a socket. Used internally by TriggerBus for autonomous runs.

import { HeadlessTransport } from '@backendkit-labs/agent-web';

const transport = new HeadlessTransport();

const engine = new AgentEngine({ ...opts, transport });
await engine.run('Generate a daily report');

// All events are buffered
const events = transport.getEvents();
const tokens = events
  .filter(e => e.type === 'chunk')
  .map(e => e.content)
  .join('');

console.log('Final output:', tokens);

Full production example

A complete production server with JWT auth, Redis sessions, Prometheus metrics, cron jobs, webhook triggers, and BullMQ queue processing:

import { AgentServer, JwtAuth, TriggerBus, RedisSessionMetaStore } from '@backendkit-labs/agent-web';
import { createEnterpriseSetup } from '@backendkit-labs/agent-enterprise';
import { ProviderRegistry, DeepSeekProvider } from '@backendkit-labs/agent-core';
import Redis from 'ioredis';

// LLM providers
const providers = new ProviderRegistry();
providers.register('deepseek', new DeepSeekProvider({
  apiKey: process.env.DEEPSEEK_API_KEY!,
  model:  'deepseek-chat',
}));

// Enterprise setup (RAG + 6 agent profiles)
const enterprise = createEnterpriseSetup({
  vaultPath:       process.env.VAULT_PATH!,
  providers,
  defaultProvider: 'deepseek',
});
await enterprise.indexAll();

// Redis
const redis = new Redis(process.env.REDIS_URL!);

// JWT auth
const jwt = new JwtAuth({
  secret:         process.env.JWT_SECRET!,
  sessionIdClaim: 'sub',
});

// Autonomous triggers
const triggers = new TriggerBus(enterprise.engineFactory)
  .addCron({
    name:        'weekly-digest',
    schedule:    '0 9 * * 1',   // Mondays at 9 AM
    buildPrompt: () => 'Generate a weekly digest of all knowledge base changes.',
    onResult:    (out) => slack.send('#general', out),
  })
  .addWebhook({
    name:        'support-ticket',
    path:        '/webhooks/support',
    secret:      process.env.WEBHOOK_SECRET,
    buildPrompt: (body) => `Handle support ticket #${body.id}: ${body.subject}\n${body.description}`,
    onResult:    (out, body) => helpdesk.reply(body.id, out),
  })
  .addQueue({
    name:        'email-processor',
    queueName:   'emails',
    connection:  process.env.REDIS_URL!,
    concurrency: 5,
    buildPrompt: (job) => `Draft a professional reply to this email:\n\n${job.data.body}`,
    onResult:    (out, job) => emailService.send({ to: job.data.from, body: out }),
  });

// Server
const server = new AgentServer({
  port:          3000,
  engineFactory: enterprise.engineFactory,
  auth:          jwt.hook(),
  store:         new RedisSessionMetaStore(redis),
  cors:          ['https://app.company.com'],
  logger:        { level: 'info' },
  rateLimiting:  { max: 60, timeWindow: '1 minute' },
  metrics:       { enabled: true, prefix: 'bk_' },
  maxIdleMs:     30 * 60 * 1000,
  triggers,
});

await server.start();
console.log('Agent server running on :3000');

// Graceful shutdown
process.on('SIGTERM', () => server.stop());