@composy/tool-server

@composy/tool-server 是 tools/* 体系里的 Node 端运行时基础包，用来把“模块注册、任务执行、HTTP 挂载、作业管理、日志事件、制品存储、可观测性”收敛成一套统一能力。

它适合这些场景：

• 需要把一组工具任务封装成可查询、可重试、可取消的作业系统。
• 需要把 UI bridge、HTTP API、SSE、WebSocket 推送挂到统一运行时里。
• 需要在 Express、Fastify、Koa 或原生 Node HTTP 中复用同一套任务执行逻辑。
• 需要为任务增加日志、事件、制品、持久化和 Prometheus 指标。

功能概览

• ToolServerRuntime 负责模块注册、任务入队、并发控制、重试、取消、排队、健康检查和作业查询。
• 内置 bridge 自动挂载 toolServerBridge，对外提供 health、runTask、listJobs、cancelJob、rerunJob、drainQueue 等能力。
• HTTP 宿主抽象支持内置 ToolHttpServer，也支持 Express、Fastify、Koa 适配器。
• HTTP 中间件内置 API Key 鉴权、限流、请求 ID、审计日志、远端地址白名单。
• 任务执行支持 runtime hooks 与 task middleware，便于统一接入审计、链路追踪、租户上下文和输出包装。
• 作业数据支持 job、log、event、artifact 的查询、保留策略和落盘。
• 可观测性内置 Prometheus metrics，支持 WebSocket 事件绑定与 SSE job stream。

目录结构

src/
  bridge/         内置 tool-server bridge
  http/           内置 HTTP server、宿主适配器、中间件、工具函数
  modules/        模块定义辅助函数
  runtime/        运行时核心、存储、传输、观测能力
  types/          对外公开的 TypeScript 类型

安装

pnpm add @composy/tool-server

如果在 monorepo 里使用 workspace 依赖：

{
  "dependencies": {
    "@composy/tool-server": "workspace:*"
  }
}

要求：

• Node.js >= 18
• TypeScript strict 模式
• ESM 项目体验最佳，但同样会产出 CJS 版本

快速开始

下面的示例定义了一个最小模块，并启动内置 HTTP 运行时：

import { createToolServer, defineToolServerModule } from '@composy/tool-server'

const demoModule = defineToolServerModule({
  id: 'demo',
  title: 'Demo Module',
  tasks: [
    {
      id: 'echo',
      title: 'Echo Message',
      async run(input: { message: string }) {
        return {
          echoed: input.message,
        }
      },
    },
  ],
})

async function main(): Promise<void> {
  const runtime = createToolServer({
    metrics: true,
    modules: [demoModule],
    port: 17702,
  })

  await runtime.start()
  console.info(runtime.getURL())
}

void main()

启动后默认会提供这些能力：

• GET /health 返回 runtime 健康摘要。
• GET /metrics 在启用 metrics: true 时返回 Prometheus 文本格式指标。
• GET /api/ui/tool-server/jobs/stream 提供 SSE 作业事件流。
• GET /api/ui/tool-server/jobs/:jobId/artifacts/:artifactId 下载指定制品。
• toolServerBridge 自动挂载一组 UI bridge 方法，供 UI 端或其他桥接调用。

核心概念

模块

模块是任务的逻辑分组，包含：

• id 模块唯一标识。
• title / description 对外展示信息。
• tasks 任务集合，支持数组写法和对象写法。
• bridges 该模块需要额外挂载的 UI bridge。
• healthCheck 模块级健康检查逻辑。
• maxConcurrent 模块级并发上限。

任务

单个任务支持这些关键能力：

• validateInput 运行前校验输入。
• validateOutput 运行后校验输出。
• concurrency.maxConcurrent 任务级并发限制。
• concurrency.lockKey 分布式锁语义的本地化实现，同一 lockKey 的任务不会并行执行。
• retry 任务级重试策略。
• run 真正的执行函数。

任务上下文

run(input, context) 里的 context 提供：

• context.update(progress, message) 更新进度和阶段消息。
• context.log(level, message, data) 记录结构化日志。
• context.event(type, data) 追加自定义事件。
• context.artifact(...) 产出制品并自动纳入管理。
• context.signal 取消与超时信号。
• context.throwIfAborted() 主动检测中止。

Runtime 能力

createToolServer(options)

最常用的入口。默认会创建内置 ToolHttpServer，并注册所有基础路由与 bridge。

常用配置包括：

• host / port 当未显式传入 server 时，控制内置 HTTP 服务器监听地址。
• basePath 给所有运行时路由增加统一前缀。
• modules 启动时批量注册的模块。
• httpMiddlewares 统一挂到所有运行时路由上的中间件。
• taskMiddlewares 包裹任务执行，可在 run 前后接入横切逻辑，也可以返回新的 output。
• hooks 提供 beforeTaskRun、afterTaskRun、onTaskError 三个轻量观察点。
• taskTimeoutMs 默认任务超时。
• concurrency.maxConcurrent 全局并发上限。
• defaultRetry 默认重试策略，可被任务或单次调用覆盖。
• jobRetention / jobLogRetention / jobEventRetention / jobArtifactRetention 各类保留策略。
• jobStore job 持久化存储。
• artifactStore job 制品存储。
• metrics 启用 Prometheus 指标。

常用实例方法

• runtime.start() 启动宿主服务器，并在需要时加载持久化 job。
• runtime.stop() 停止服务器并等待存储写入完成。
• runtime.listModules() 返回所有模块 manifest。
• runtime.getJob(jobId) 读取某个 job 当前快照。
• runtime.runTask(input) 提交一个任务，可选择 wait: true 同步等待终态。
• runtime.cancelJob({ jobId }) 取消 queued / running job。
• runtime.rerunJob({ jobId }) 以历史 job 为模板重新运行。
• runtime.listJobs(...) 分页查询作业。
• runtime.getJobLogs(...) 查询作业日志。
• runtime.getJobEvents(...) 查询作业事件。
• runtime.getJobArtifacts(...) 查询作业制品。
• runtime.getQueueState() 返回队列快照，包含 queued/running 总数、按模块聚合和按任务聚合的数据。
• runtime.pauseQueue() / runtime.resumeQueue() 暂停或恢复队列调度。
• runtime.drainQueue(...) 等待当前队列排空。

任务扩展点

taskMiddlewares 适合处理可组合的横切逻辑，声明顺序就是执行顺序；hooks 适合观察任务生命周期：

import { defineToolServerTaskMiddleware } from '@composy/tool-server'

const runtime = createToolServer({
  hooks: {
    beforeTaskRun(context) {
      context.runtime.listModules()
    },
    onTaskError(context) {
      console.warn(context.error.message)
    },
  },
  taskMiddlewares: [
    defineToolServerTaskMiddleware(async (context, next) => {
      const startedAt = Date.now()
      const output = await next()
      context.runtime.listJobs({ moduleId: context.module.id })
      return {
        output,
        durationMs: Date.now() - startedAt,
      }
    }),
  ],
})

runtime.getQueueState()

如果 middleware 抛错，runtime 会沿用既有失败与重试语义；如果 onTaskError 自身抛错，runtime 会追加 job:hook:error 事件，但不会覆盖任务原始错误。

HTTP 层

内置 ToolHttpServer

内置服务器适合这些场景：

• 本地 CLI 工具进程
• 小型 Node 服务
• 单元测试或集成测试中的嵌入式宿主

它支持：

• CORS
• 最大 body 限制
• 动态路由参数
• 按 HTTP method 分桶的路由匹配，路由数量增长时避免跨 method 线性扫描。
• port: 0 临时端口场景下，getURL() 会返回 Node 实际监听端口。
• 与 runtime 同生命周期的 start() / stop()

宿主适配器

如果你已经有自己的服务框架，可以直接接入：

import {
  createExpressToolHttpServer,
  createToolHttpApiKeyAuthMiddleware,
  createToolServer,
} from '@composy/tool-server'
import express from 'express'

const app = express()
app.use(express.json())

const runtime = createToolServer({
  httpMiddlewares: [
    createToolHttpApiKeyAuthMiddleware({
      keys: ['dev-token'],
    }),
  ],
  server: createExpressToolHttpServer(app, {
    getURL: 'http://127.0.0.1:3000',
    maxBodySize: 1024 * 1024,
  }),
})

console.info(runtime.listModules())

可用适配器：

• createExpressToolHttpServer
• createFastifyToolHttpServer
• createKoaToolHttpServer

HTTP 中间件

API Key 鉴权

import { createToolHttpApiKeyAuthMiddleware, createToolServer } from '@composy/tool-server'

const runtime = createToolServer({
  httpMiddlewares: [
    createToolHttpApiKeyAuthMiddleware({
      keys: ['local-dev-token'],
    }),
  ],
})

console.info(runtime.getQueueState())

其他中间件

• createToolHttpAuditLoggerMiddleware 记录 method、path、statusCode 和 duration。
• createToolHttpRateLimitMiddleware 基于内存的窗口限流。
• createToolHttpRemoteAddressAllowlistMiddleware 控制允许访问的 IP、前缀或 CIDR。
• createToolHttpRequestIdMiddleware 生成请求 ID 并写入响应头与 context.state。

存储与制品

job 持久化

createToolServerFileJobStore() 会把每个 job 落盘成一个 JSON 文件，适合单进程工具服务：

import {
  createToolServer,
  createToolServerFileArtifactStore,
  createToolServerFileJobStore,
} from '@composy/tool-server'

const runtime = createToolServer({
  artifactStore: createToolServerFileArtifactStore({
    dir: './.data/artifacts',
  }),
  jobStore: createToolServerFileJobStore({
    dir: './.data/jobs',
  }),
})

console.info(runtime.getURL())

制品

在任务里可以这样产出制品：

import { defineToolServerModule } from '@composy/tool-server'

const artifactModule = defineToolServerModule({
  id: 'artifact-demo',
  title: 'Artifact Demo',
  tasks: [
    {
      id: 'write-report',
      title: 'Write Report',
      async run(_input, context) {
        await context.artifact({
          contentType: 'text/plain; charset=utf-8',
          data: 'report ready',
          name: 'report.txt',
        })

        return { ok: true }
      },
    },
  ],
})

void artifactModule

runtime 会自动：

• 记录制品元数据
• 提供下载接口
• 在事件流里追加 job:artifact 事件
• 按保留策略自动清理旧制品

SSE、WebSocket 与指标

SSE

GET /api/ui/tool-server/jobs/stream 支持这些 query：

• snapshot=true 连接后先推送现有快照。
• jobId=... 只订阅某个 job。
• moduleId=... 只订阅某个模块的 job。
• taskId=... 只订阅某个任务的 job。
• terminalOnly=true 只看终态 job。
• types=job:update,job:log,job:event 指定订阅的事件类型；传 all 表示全部。

WebSocket

提供三个绑定函数：

• bindToolServerJobsToWebSocketServer
• bindToolServerJobLogsToWebSocketServer
• bindToolServerJobEventsToWebSocketServer

它们都支持：

• filter 过滤是否广播。
• format 自定义推送 payload 结构。

Prometheus

启用 metrics: true 后，会自动挂载 metrics handler。

默认指标包括：

• tool_server_uptime_ms
• tool_server_jobs_queued
• tool_server_jobs_running
• tool_server_jobs_completed
• tool_server_jobs_failed
• tool_server_jobs_started_total
• tool_server_jobs_completed_total
• tool_server_jobs_failed_total
• tool_server_job_duration_ms
• tool_server_job_queue_duration_ms

类型导出

包根入口会导出：

• 所有 runtime / job / module / HTTP 相关类型
• defineToolServerModule
• toolServerBridge
• 内置 HTTP server 与宿主适配器
• 文件存储、WebSocket 绑定和 Prometheus 指标工具

这意味着你通常只需要：

import type {
  ToolServerJobRecord,
  ToolServerRuntimeOptions,
  ToolServerTaskDefinition,
} from '@composy/tool-server'

const runtimeOptions: ToolServerRuntimeOptions = {}

const taskDefinition: ToolServerTaskDefinition<{ id: string }, ToolServerJobRecord> = {
  id: 'typed-task',
  async run() {
    throw new Error(`Implement ${runtimeOptions.basePath ?? '/'} first.`)
  },
}

void taskDefinition

构建与开发

当前包使用 @composy/pack 作为统一打包工具，.ldesign/pack.config.ts 会把 src/**/*.ts 全量构建成 ESM、CJS 与类型声明。

常用命令：

pnpm run type-check
pnpm run lint:check
pnpm run build

构建后会在 dist/ 中保留按目录拆分的产物，便于根入口和深层入口同时复用。

设计约定

• 所有 HTTP 宿主最终都会归一化为同一个 ToolHttpContext。
• job 只存在 queued、running、completed、failed 四种状态。
• retention 清理会同步清理持久化数据，避免内存与磁盘状态漂移。
• queue 会保留稳定的优先级顺序，并在调度时同时检查全局、模块、任务与锁约束。
• getQueueState() 的聚合字段由当前队列与运行中索引实时派生，不维护第二份可漂移状态。
• 指标采集会在 job 被清理时回收内部状态，避免长期运行的额外内存增长。

License

MIT

通过 @composy/cli 统一接入

接入类型：helper 统一命令：ldesign tool-server 命令别名：server 包内原生 bin：无独立 bin 当前包通过 helper 方式接入，统一命令会调用 @composy/cli 内置封装。

pnpm add -D @composy/cli
ldesign tool-server --help
ldesign tools run tool-server --help