@backendkit-labs/代理网络

状态 : stable ——保持并做好生产准备. TriggerBus (cron/webhook/BullMQ)为: experimental 并在试验阶段被冻结.

HTTP/ WebSocket 服务器层 @ backendkit-labs/ 代理核心。 。 。 .

环绕 AgentEngine 在一个有流式WebSocket会话,JWT认证,Prometheus度量衡,速率限制,实时显示器总线,以及灵活的触发系统(cron jobs,webhoks,BullMQ队列)的快取5服务器中.

目录

• 安装
• 快开始
• 代理服务器选项
• HTTP 和 WebSocket 路线
• WebSocket 协议
• 引擎阵型
• 会话管理器
• JWT 认证
• 普罗米修斯度量衡
• 限制费率
• 会话元数据存储
• 监视器Bus
• 触发Bus — cron, webhooks, 队列
• 无头运输
• 全面生产实例

安装

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

可选融合(仅安装所需):

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

快开始

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

代理服务器选项

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 和 WebSocket 路线

WebSocket 协议

/chat - 聊天会

客户端 – 服务器 :

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

服务器 – 客户端 (编) AgentEvent 边框 :

{ "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 ——实时观察员

连接以观察所有会话而不发送任何消息 :

{ "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 是函数 (sessionId: string, transport: Transport) => AgentEngine。每个进入的WebSocket连接都获得自己的孤立引擎实例.

最小工厂

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

每会场记忆的工厂

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

使用 JWT 的工厂

当 JWT 认证被启用时 sessionId 带有认证用户的上下文。 也可以输入用户特定数据:

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 创建、跟踪和清理会话。 它是由内部创建的 AgentServer,但也可以单独使用.

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 认证

JwtAuth 验证 Authorization 页眉或 ?token= 查询参数(用于无法设置自定义头的浏览器 WebSocket连接).

严格模式

对于其他人接触到的服务器,打开两个开关:

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 忽略 ?token= (一个在URL中的标志在日志,代理和历史中登陆). 浏览器发送 作为 WebSocket 子协议的符号 : new WebSocket(url, ['bearer', jwt])· ; useAgentSession({ url, token }) 内 agent-web-ui 为了你.
• strict: true 除非 cors 列出它,添加 X-Content-Type-Options, (中文(简体) ), X-Frame-Options: DENY 财务报告和审定财务报表 Referrer-Policy: no-referrer他们不肯听从 host 无 auth。 。 。 .

设置

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

发布令牌( 您的登录端点)

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>

用户会议隔离

何时 sessionIdClaim 实值 'sub',每个用户自动获得自己的会话。 两个相同的请求总是打在同一会话:

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

自定义认证钩钩

如果您使用不同的认证系统( API 密钥, OAuth 令牌) , 则执行 AuthHook 直接 :

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

普罗米修斯度量衡

要求数 prom-client 将安装。 显示 a /metrics 端点有标准普罗米修斯计数器和直方图.

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

暴露的衡量标准

直接使用计量工具

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

限制费率

要求数 @fastify/rate-limit全球适用于所有路线:

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

当超过限制时,服务器会回复:

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

会话元数据存储

默认情况下,会话元数据生活在内存中,并在重启时丢失. 使用 RedisSessionMetaStore 用于跨越重启和水平缩放的持久元数据.

内存( 默认)

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

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

Redis 商店( 需要 ioredis 或 node- redis)

RedisSessionMetaStore 是鸭式的 —— 它接受任何执行的 Redis 客户端 .get(), (中文(简体) ), .set(),以及 .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
  }),
});

监视器Bus

MonitorBus 将所有会话活动广播到连接 /monitor WebSocket观察者. 由Kanban董事会使用于 @backendkit-labs/agent-web-ui。 。 。 .

独立使用

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

触发Bus — cron, webhooks, 队列

TriggerBus 让自动代理运行没有 WebSocket 客户端。 Kanban 显示器显示运行情况 .

克龙触发器( 要求) 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 触发

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

测试一个网点 :

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" } }'

队列触发器( BallMQ —— 需要) 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
    },
  });

发布一份工作(从系统中的任何地方) :

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

队列连接选项

// 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

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 是一个 Transport 缓冲所有 AgentEvent 对象在内存中而不是在套接字上流出。 内部使用 TriggerBus 用于自动运行.

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

全面生产实例

完整的生产服务器, 包括 JWT auth, Redis 会话, Prometheus 度量衡, cron 任务, webhook 触发器, 以及 BullMQ 队列处理 :

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