HTTP API & MCP

@chamvi/http turns a chamvi app into one routup App: humans start and follow runs over REST and server-sent events, machines post webhooks with bearer tokens, and agents (OpenClaw, Claude, any MCP client) start runs as MCP tools. It also serves /healthz, /readyz and /metrics, and ships the graceful shutdown helper.

npm install @chamvi/http @chamvi/core --save

const engine = new Engine({
    store,
    agents,
    resumeSecret,
});
// the scheduler and the app read the store, agents and clock from the engine
const scheduler = createScheduler({ engine });

let server: ReturnType<typeof serve> | undefined;
const shutdown = createShutdown({
    engine,
    scheduler,
    // the server is created after the app; the helper only calls close() at shutdown
    server: { close: async (force) => { await server?.close(force); } },
    readinessDelayMs: 5_000,
});

const app = createHttpApp({
    engine,
    auth: authenticators(verifier),
    shutdown,
    basePath: '/api',
    // behind one reverse proxy: trust its X-Forwarded-For
    trustProxy: 1,
    cors: { origin: 'https://chamvi.example.com' },
});

await engine.start();
await scheduler.start();
// srvx must not install its own signal handler
server = serve(app, { port: 3000, gracefulShutdown: false });

Every HTTP or MCP start ends in engine.dispatch(); nothing runs synchronously in a request. Errors are always JSON { status, code, message, issues? }, unmatched routes included; a 5xx body says Internal Server Error and the cause goes to onError. An issue carries the received value only when it is small (at most 256 characters as JSON), so a 400 for an oversized list, a long string or a large file never echoes the body back.

Triggers decide what is reachable

Trigger	Served at	Envelope
http({ schema? }) (id http, at most one per agent)	POST /agents/:name/runs, MCP run_<name>	trigger: 'http', source http (REST, UI) or mcp
webhook({ path, schema?, idempotencyKey?, concurrencyKey? }) (id webhook:<path>)	POST /hooks/:agent<path>	trigger: 'webhook:<path>', source webhook

An agent without http() answers 404 on REST and gets no MCP tool. The trigger's schema validates the payload; a failure answers 400 with issues and dispatches nothing. A webhook is served on its own path under the agent: webhook({ path: '/issues' }) on agent intake is POST /hooks/intake/issues (a path of / is /hooks/intake); GET /agents lists each hook URL. See Triggers & the Envelope.

Routes

Method and path	Permission	Answer
POST /agents/:name/runs	agent:<name>:run	202 { runId, deduplicated }; body { payload?, idempotencyKey?, thread?, attachments? } (or an Idempotency-Key header)
GET /runs	run:read, or agent:<agent>:run with filter[agent]=<agent>	200 { data, meta: { total, limit, offset } }
GET /runs/:id	run:read or agent:<run.agent>:run	The run with its steps
GET /runs/:id/stream	as GET /runs/:id	SSE
GET /runs/:id/artifacts	as GET /runs/:id	200 { data: Artifact[] }
GET /runs/:id/artifacts/:artifactId	as GET /runs/:id	The bytes, as a download
POST /runs/:id/cancel	run:cancel	202 { runId, status }
POST /resume	agent:<run.agent>:approve	202 { runId, status: 'queued' }; body { token, approved, reason? }
POST /hooks/:agent<path>	agent:<agent>:run	202 { runId, deduplicated }
GET /agents, GET /schedules	run:read	Agent summaries; cron() triggers with their next fire time
GET, PATCH /agents/:name/state/kv/:key	state:read, state:write	{ key, value }
GET /agents/:name/state/collections/:collection	state:read	Documents ({ id, ...fields }, or { id, doc } for a non-object); top-level equality filters (filter[issue]=42 matches a number, filter[issue]="42" the string; filter[id] is 400, id being the store id), page[limit] (no offset)
/mcp	authenticated	MCP endpoint
GET /healthz, /readyz, /metrics	none	Liveness, readiness, Prometheus

GET /runs takes rapiq's URL syntax: filter[agent]=triage, filter[status]=queued,running, page[limit] (default 50, at most 100), page[offset], sort=createdAt. Bodies are JSON of at most bodyLimit bytes (default 1 MiB). A protected route checks the bearer token before it reads the body. Runs leave the API without their internal claim, and the run's invoker and a wait step's approver without their permissions (see Invoker).

Files

POST /agents/:name/runs takes attachments: at most 16 objects { mediaType, filename?, data }, mediaType of the form type/subtype (no parameters), filename 1 to 255 characters, data strict base64 (padded, standard alphabet, no whitespace). A bad entry answers 400 validation_error with the path of the field (['attachments', 0, 'data']) and dispatches nothing. Base64 travels in the JSON body, so bodyLimit (default 1 MiB) bounds the files too; raise it for larger ones. The files become attachments of the run.

const base = 'https://chamvi.example.com/api';
const headers = { authorization: `Bearer ${token}`, 'content-type': 'application/json' };

const started = await fetch(`${base}/agents/triage/runs`, {
    method: 'POST',
    headers,
    body: JSON.stringify({
        payload: { issue: 43 },
        // at most 16, data as strict base64; the whole body stays within bodyLimit
        attachments: [{
            mediaType: 'image/png',
            filename: 'crash.png',
            data: Buffer.from(png).toString('base64'),
        }],
    }),
});
const { runId } = await started.json() as { runId: string };

// metadata of the attachments and of every file a tool stored
const listed = await fetch(`${base}/runs/${runId}/artifacts`, { headers });
const { data } = await listed.json() as { data: Array<{ id: string; mediaType: string }> };

// the bytes, served as a download with the stored media type
const [first] = data;
const file = first ? await fetch(`${base}/runs/${runId}/artifacts/${first.id}`, { headers }) : undefined;

GET /runs/:id/artifacts lists the run's artifacts (attachments and files its tools stored, see Files) as metadata. GET /runs/:id/artifacts/:artifactId answers the bytes with the stored media type as content-type, content-length, content-disposition: attachment (with filename and an RFC 5987 filename* when the artifact has a filename), x-content-type-options: nosniff and cache-control: private, no-store, so a stored HTML or SVG file is downloaded, never rendered in the API's origin. Both follow the run read rule (404 for an unknown run before 403); an unknown artifact, or one of another run, is 404 artifact_not_found.

Following a run

GET /runs/:id/stream sends run.updated with the current run, then each step as step.appended (id is the step index, so Last-Event-ID resumes), then live events; it ends once the run is completed, failed or cancelled. The engine's events only cover runs executing in this process, so the stream also polls the store every stream.pollMs (2 s) and sees a run that another replica executes; an idle poll sends ping. A client that disconnects unsubscribes at once, and the app ends open streams on shutdown.

While the model writes, the stream also sends delta events, data: { step, kind, text } with kind text or reasoning (the engine's run.delta). step is the checkpoint step index of the llm step being generated (one per model call), not the step.appended id, which also counts tool and wait rows. They carry no id, so Last-Event-ID stays a step index, and they are live only: a reconnect or a late subscriber gets the stored llm step instead, never the deltas it missed. Only the replica executing the run has them; a stream served by another replica sees the steps through its store poll and no deltas.

To resume, send Last-Event-ID with the last step id you saw, or the query ?after=<step index> (a non-negative integer, else 400). If both are present, a valid Last-Event-ID wins: it is the client's progress on a reconnect, while the URL stays the one of the first connect.

The stream takes bearer tokens only, and browser EventSource cannot send an Authorization header. In a browser, read the stream with a fetch-based SSE reader (fetch plus an SSE parser over the response body) that sends Authorization and, on reconnect, Last-Event-ID (or ?after=).

An app holds at most stream.maxOpen (1000) open streams plus waiting MCP run_<agent> calls, on one counter. Over the cap a stream answers 503 with Retry-After: 5; retry after that many seconds.

Authentication and permissions

auth takes one authenticator or a chain; the first principal wins. A resolved principal needs a non-empty id and kind and a permissions array of strings; anything else is a server misconfiguration (500, reported to onError) and ends the chain.

const ciToken = process.env.CI_TOKEN;
return [
    // CI and service accounts: tokens of at least 16 characters.
    // Static tokens go first: authup() in remote mode introspects every non-JWT bearer.
    ...(ciToken ?
        [staticTokens({
            tokens: [{
                value: ciToken,
                id: 'ci',
                permissions: ['agent:deploy:run', 'run:read'],
            }],
        })] :
        []),
    // authup: grants come from global tokens, verified remotely
    authup({ verifier }),
    // GitHub Actions OIDC: the mapper decides which repository gets which grants
    oidc({
        issuers: [{
            id: 'github-actions',
            issuer: 'https://token.actions.githubusercontent.com',
            audience: 'chamvi',
        }],
        principal: ({ claims }) => (claims.repository === 'acme/infra' ?
            {
                id: `gha:${String(claims.sub)}`,
                kind: 'oidc',
                permissions: ['agent:deploy:run'],
            } :
            undefined),
    }),
];

• staticTokens({ tokens: [{ value, id, permissions }] }): bearer tokens (value, at least 16 characters, each given once), each with a non-empty id (the principal and run invoker id; several tokens may share one, e.g. while rotating). Put it before authup() and oidc(): authup() in remote mode introspects every bearer that is not a JWT, so a static token placed after it is sent to the authup server.
• authup(): authup tokens through @authup/server-adapter-web. Only tokens with the global scope grant their permissions; others authenticate and get 403. Grants need remote verification (introspection); a cached result lives until the token's exp.
• oidc(): any OIDC issuer through vouchly. Every issuer needs an audience, and principal is required: a public issuer such as GitHub Actions lets anyone mint a token for your audience, so the mapper decides which claims get which grants. A mapper result without a non-empty id and kind and a permissions array of strings is a server error (500, reported to onError), not a client one.

Permission	Grants
agent:<name>:run	Start the agent's runs and read them
agent:<name>:approve	Answer the agent's approvals
run:read	Read and list every run, agents and schedules
run:cancel	Cancel runs
state:read, state:write	Agent state
tool:<agent>:<tool>	Read by requireToolPermission(): the agent offers the tool to this principal's runs
*	Everything

A * segment matches exactly one segment (agent:*:run). A client that may start an agent's runs can follow them without run:read, but to list runs it must filter to that agent.

POST /resume needs the approve permission for the run's agent as well as the resume token: a leaked token alone does nothing. The token's signature and expiry are checked before any store lookup.

Invoker

POST /agents/:name/runs and MCP run_<agent> record the authenticated principal as the run's invoker (run.trigger.invoker): its id, kind, permissions and a name from the name or preferred_username claim (never email, since every reader of the run sees the name; static tokens have no name). An agent's toolPolicy and tool code read it. POST /resume and MCP resume_run record the principal that answered as the wait step's output.approver. Webhook runs have no invoker, and neither ever comes from a request body.

The invoker is a snapshot: a run keeps the grants it was started with, even when a grant is revoked while it waits on an approval. Responses (REST, SSE, MCP) carry the invoker and the approver without their permissions, so a run:read holder never learns another principal's grants.

MCP

/mcp is a streamable HTTP endpoint (MCP TypeScript SDK v2; 2025-era clients are served statelessly), authenticated like REST. Each client sees the tools its principal may use:

• run_<agent> for each agent with http() it may run: { payload?, idempotencyKey?, thread?, waitSeconds? }. With waitSeconds (capped by mcp.maxWaitSeconds, default 60) it waits for completed, failed, cancelled or waiting and returns the output, the error or the pending approval, never the resume token. A wait counts against stream.maxOpen: over the cap the call does not wait and answers at once with { runId, status, deduplicated }, as with waitSeconds: 0; the run is already dispatched and goes on.
• get_run, list_runs: the REST read rule applies.
• resume_run: { token, approved, reason? }, with the approve permission.

An MCP start dedupes its idempotencyKey under <agent>:mcp:<kind>:<id>, a REST start under <agent>:http:<kind>:<id> (the principal's kind and id, URI-encoded): one key used over both transports, or by two principals, creates two runs. A client registers the endpoint like any HTTP MCP server:

// OpenClaw (or any MCP client) lists the app as a streamable HTTP server
export const mcpServers = {
    chamvi: {
        type: 'http',
        url: 'http://chamvi.agents.svc/api/mcp',
        headers: { Authorization: 'Bearer ${CHAMVI_TOKEN}' },
    },
};

This is the "MCP in" half of the OpenClaw integration.

Extra routes

routes adds routes that other packages provide, served under basePath beside the built-in ones. A route is a static path with a method and a fetch-style handler: it gets the request (URL, method, headers and signal, without the body) and a readBody() that returns the raw bytes (the app's bodyLimit, 413 and 415 apply; repeated calls return the same bytes), and answers { status, body? } (an object, array, number or boolean as JSON, a string as text/plain, no body or null as an empty body). Extra routes are not authenticated by auth: they check their own credentials, as a webhook checks its signature. Thrown errors render like those of every other route. rateLimit: false takes a route out of the rate limit, for senders that never retry a refused request. Paths match case-insensitively, like routup's routing: a route may not differ from a built-in segment or another route only by case.

Operations

• /metrics is unauthenticated: keep it off the public ingress (a NetworkPolicy, or an internal listener) or set metrics: false. Each app has its own prom-client registry, labelled by route template (/runs/:id, /runs/:id/artifacts/:artifactId; unknown paths share /_unmatched); SSE streams and rate-limited requests are not timed.
• The rate limit (300 requests per minute per client, health, metrics and CORS preflights excluded) keys on the socket address. Behind a reverse proxy set trustProxy, or every client shares the proxy's bucket; X-Forwarded-For is never trusted otherwise.
• CORS is off unless cors is given.
• /readyz fails while shutting down, while the engine drains and when the store does not answer within readyTimeoutMs (2 s).

Shutdown

// fail readiness, stop the scheduler, end streams, stop intake, drain
process.once('SIGTERM', () => {
    void shutdown.shutdown().then(() => process.exit(0));
});

createShutdown() fails readiness, waits readinessDelayMs for load balancers, stops the scheduler, runs the onShutdown callbacks (the app ends its SSE streams and run_<agent> waits, which answer with the run id, and closes MCP there; new streams and MCP requests then answer 503), stops accepting connections, drains the engine (drainTimeoutMs, 30 s) and waits at most closeTimeoutMs (5 s) for open connections before closing them. Every call returns the same promise. Pass gracefulShutdown: false to serve() so srvx does not handle the signal itself.