johnandersen777.bsky.social/atlibs-asyncai

Switch repository

Loading…

# atlibs-asyncai

The [asyncai](https://github.com/publicdomainrelay/asyncai) agent loop as an [atlibs](../atlibs) wasm
module: inference over the Anthropic Messages API or OpenAI chat completions ([co/core](https://cocore.dev),
as the signed-in user with no key, or any compatible endpoint), tools from MCP servers, until the model
stops or the step cap. It runs wherever atlibs runs modules (a page, Node, the Rust host, a composition).

```
start -> MCP initialize / notifications/initialized / tools/list (every server at once)
      -> inference -> tool_use -> MCP tools/call (in parallel) -> tool_result -> inference ...
      -> end_turn (or max_tokens, refusal, ...), or the step cap
```

## Layout

- `modules/asyncai/`: the module (`src/lib.rs`, `module.json`), its tests (`test.mjs` offline,
  `live.test.mjs` live, `code-mode.test.mjs` with atlibs-code-mode) and `mcp-stub.mjs` (a test MCP server: one JSON-RPC message in, its response out; as Streamable HTTP; as a host scheme fetcher)
- `web/asyncai/`: a page that runs the agent with a prompt and shows the transcript and the live MCP servers (`atlibs` and
  `modules/` are symlinks into `../atlibs` and `target/`)

## Build and test

Needs `../atlibs` with the plain-HTTP host call (`com.fedfork.exec.http.fetch` and its `serviceAuth`,
atlibs [calls](../atlibs/docs/modules/calls.md#plain-http)) and host-set held values (`held`). The
`webrtc://` tests (offline and live) need its scheme fetchers too (`runModule`'s `fetchers`); on a core
without them they skip, saying so.

```bash
for r in atlibs@88b2ba8 atlibs-code-mode@9f360fc; do
  git clone -q https://knotless.dev/johnandersen777.bsky.social/${r%@*} ../${r%@*} && git -C ../${r%@*} checkout -q ${r#*@}
done
(cd ../atlibs && cargo host) && (cd ../atlibs-code-mode && npm ci --silent)   # the Rust engine; the code-mode MCP server
cargo wasm
node --test modules/*/test.mjs web/*/*.test.mjs                            # offline: stubbed inference and MCP; the page in headless Chrome
node --import ../atlibs/rust/host/swap.mjs --test modules/asyncai/test.mjs  # the same on the Rust engine
node --test modules/asyncai/live.test.mjs                                  # live: DeepWiki's MCP server, inference (below), a compaction (a 5000-token window)
node --test modules/asyncai/code-mode.test.mjs                             # with ../atlibs-code-mode as the MCP server, in-process and over HTTP
```

The live test reads `ASYNCAI_ENDPOINT`, `ASYNCAI_MODEL` and `ASYNCAI_API_KEY`, else the
`ANTHROPIC_BASE_URL`, `ANTHROPIC_MODEL` and `ANTHROPIC_API_KEY` exports of `~/.local/bin/deepseek-claude`
(DeepSeek's Anthropic-compatible endpoint). The key goes only into the host's allowlist. Its co/core test
signs in with `ASYNCAI_COCORE_HANDLE` and `ASYNCAI_COCORE_PASSWORD` (an app password), else the
`ALICE_ATPROTO_*` exports of `~/.aliceoa_bsky_social_source`, and runs the `stub` model; an account that
never connected co/core answers `onboarding_required`, which passes unless `ASYNCAI_COCORE_ONBOARDED=1`.

The page: `cd web/asyncai && python3 -m http.server`, open `http://127.0.0.1:8000/` (the loopback OAuth
client). Pick the backend:

| backend | API | credentials (in the page's host allowlist, never the module) |
|---|---|---|
| co/core (the default) | OpenAI chat completions at `https://cocore.dev/api/v1` | none: sign in; the host mints a service-auth token from your session per request |
| Anthropic | the Messages API at `https://api.anthropic.com` | an API key as `x-api-key` (plus `anthropic-dangerous-direct-browser-access`) |
| custom | any base URL, in either format | a token: `x-api-key` (Anthropic format) or `Authorization: Bearer` (OpenAI format) |

Signing in asks only for `rpc:dev.cocore.inference.dispatch?aud=*` (the module's session
record, `records/asyncai.json`). Other endpoints and remote MCP servers must answer the page's origin (CORS).

## The module

`run`, input:

| field | default | what |
|---|---|---|
| `prompt` | required | the user message |
| `system` | none | the system prompt |
| `provider` | `anthropic` | the API: `anthropic` (the Messages API) or `openai` (chat completions: co/core, and any OpenAI-compatible endpoint) |
| `endpoint` | `https://api.anthropic.com`; `openai`: `https://cocore.dev/api/v1` | the base URL: `anthropic` appends `/v1/messages`, `openai` `/chat/completions` |
| `model` | `claude-opus-5-5`; `openai`: required | the model (`stub` is co/core's test model) |
| `stream` | `false` | `openai`: ask for an event stream, collected whole by the host and the module (content, tool calls by index, usage via `stream_options.include_usage`) |
| `maxTokens` | 16000 | `max_tokens` per inference call |
| `maxSteps` | 9 | inference calls per run; on the last one the model may not use tools (`tool_choice: none`, plus asyncai's "tool budget exhausted" note) |
| `mcp` | `[{ "name": "mcp" }]` (the in-process hook) | the MCP servers ([below](#mcp-servers)); an array, or its JSON |
| `dynamicMcp` | off | the model may connect servers mid-run ([below](#dynamic-servers-mcp_connect)): `true` (`["url"]`), or a list of `"url"`, URL prefixes (`webrtc:///`, `iroh:///` too) and hook NSIDs; or its JSON |
| `extra` | none | fields merged over the request body (`thinking`, `output_config`, `fallbacks`, ..., or OpenAI's `tool_choice`, `country` for co/core, ...) |
| `compact` | `client` | [compaction](#compaction): `client` (an inference call writes a summary; any endpoint, either API), `server` (Anthropic's compaction beta; `anthropic` only), or `off` (`false`) |
| `contextWindow` | 200000 | the model's context window, tokens |
| `compactAt` | 0.8 | compaction when the next request would pass this fraction of `contextWindow` |
| `compactPrompt` | asyncai's (Anthropic's default compaction prompt) | the summary instruction (`server`: the edit's `instructions`) |

Result (when the run ends): `{ text, stop, steps, usage, compactions, servers, tools, messages, error? }`. `stop` is
the last `stop_reason` (`end_turn`, `max_tokens`, `refusal`, ...), `max_steps` at the cap, or `error`
(`error` says why: an HTTP error, a refused origin). `messages` is the transcript in the provider's form,
valid to send again: a `tool_use` the cap left unanswered gets an error `tool_result` (OpenAI: a `tool`
message). `usage` sums the provider's own fields (`input_tokens`, ... or `prompt_tokens`, ...).

Earlier turns (the memory: a previous run's `result.messages`) are not an input: the host passes them as
the held value `messages` when it starts the run (`runModule`'s `held: { messages }`, `ATLIBS_HELD` for the
CLIs; atlibs [held values](../atlibs/docs/modules/calls.md#held-values)), so they never ride in an event.
The prompt is appended to them.

### The two APIs

The module keeps the transcript in the provider's own form and maps only at the edges:

| | `anthropic` (Messages API) | `openai` (chat completions) |
|---|---|---|
| system | `system` | a system message held apart (`system`) and sent first: `"messages": { "$held": ["system", "messages"] }` |
| tools | `{ name, description, input_schema }` (the MCP `inputSchema`) | `{ type: "function", function: { name, description, parameters } }` |
| a tool call | a `tool_use` block | `tool_calls[]` (`function.arguments` is JSON text) |
| its result | a `tool_result` block (`is_error`), all of a turn's in one user message | a `tool` message each (`tool_call_id`; an error's text starts `Error: `) |
| the last step | `tool_choice: { type: "none" }`, the note in the tool results' message | `tool_choice: "none"`, the note as a user message |
| stop | `stop_reason` | `finish_reason`: `stop` -> `end_turn`, `tool_calls` -> `tool_use`, `length` -> `max_tokens`, `content_filter` -> `refusal` |
| size (compaction) | `input_tokens + cache_* + output_tokens` | `total_tokens` |
| headers | `content-type`, `anthropic-version` | `content-type`, `accept` |

`servers` is the live set (the start's and those the model connected, less the disconnected and the failed
connects): `{ name, url | hook, phase: ready | skipped | error, protocolVersion, tools, error?,
dropped?, dynamic? }` (`dynamic: true`: connected with `mcp_connect`; an entry is a valid `mcp` entry again).
`tools` is what the last request offered.
`compactions` counts them; `usage` sums every inference call's, the summary requests' too.

The transcript lives in the host, not the module's state: each new message goes up once as a
`com.fedfork.exec.held.append` to the held value `messages`, each inference body names it
(`"messages": { "$held": "messages" }`) and the host splices it in, and so does the result
(atlibs [held values](../atlibs/docs/modules/calls.md#held-values)). So a long transcript never passes
back through the wasm; it is bounded by the host's `heldBytes` (16 MiB by default), past which the run
ends with `stop: error` (`transcript: ...`). A page follows it from the module's calls in `onEvent`'s
outputs: a `com.fedfork.exec.held.append` of `messages` adds its `items`, a `com.fedfork.exec.held.put` of
`messages` (a compaction) restarts it at its `value` (`web/asyncai/agent.js`; a page that only follows
appends keeps showing the whole history, while `result.messages` is the compacted transcript). 429, 5xx
and 529 are retried three times (`retry-after`, else 2 s, 8 s, 32 s).

| module.json | |
|---|---|
| `calls` | `com.fedfork.exec.http.fetch`, `com.fedfork.exec.held.append`, `com.fedfork.exec.held.put` |
| `http` | `["*", "webrtc://*", "iroh://*"]`: any http(s) origin the host allows (inference and remote MCP servers are the caller's choice; the host's allowlist bounds them), and the host transports' URLs (`*` does not cover another scheme; the host must allow `webrtc://*` / `iroh://*` and have a fetcher for it, else that server fails to connect) |
| `hooks` | `com.fedfork.exec.mcp.message` (optional) |
| `anyHooks` | `true`: a composition may route other NSIDs from it, one per extra in-process server |
| `session` | `asyncai` (`records/asyncai.json`): `rpc:dev.cocore.inference.dispatch?aud=*`, so the host may mint co/core's token; the module makes no atproto calls of its own |

The host passes `runModule` (or `runCompose`) its `http` allowlist with the inference origin's
credentials (`ATLIBS_HTTP` for the CLIs), the presets:

| preset | input | `http` |
|---|---|---|
| co/core | `provider: "openai"` (endpoint and model: `stub` or one of `/models`) | `{ "https://cocore.dev": { "serviceAuth": { "aud": "did:web:cocore.dev", "lxm": "dev.cocore.inference.dispatch" } } }`: a token minted per request from the run's session (atlibs [service auth](../atlibs/docs/modules/calls.md#service-auth-instead-of-a-key)); the account must have connected co/core once, else `inference HTTP 401: ... onboarding_required` |
| Anthropic | none | `{ "https://api.anthropic.com": { "headers": { "x-api-key": "..." } } }` |
| custom | `endpoint`, `provider` (the format), `model` | `{ "<origin>": { "headers": { "x-api-key": "..." } } }` (Anthropic format) or `{ "authorization": "Bearer ..." }` (OpenAI) |

plus each remote MCP server's origin (`"https://mcp.deepwiki.com": {}`). The wasm sets `content-type` (and
`anthropic-version: 2026-10-21`, or `accept`); a host header of the same name replaces it. Give a long run unlimited `calls`, `events` and `wallMs`
(atlibs [limits](../atlibs/docs/reference/limits.md)); the step cap bounds it.

## Compaction

asyncai (the reference loop) neither compacts nor trims; this module does, so a long run stays inside
the model's context window.

**`client`** (the default). After each inference the module keeps its size: the response's
`input_tokens + cache_creation_input_tokens + cache_read_input_tokens + output_tokens` (OpenAI:
`total_tokens`). When the next
request would pass `contextWindow * compactAt` tokens (that size plus about 4 bytes a token for the
messages not yet sent), it first asks for a summary:

1. The summary request is the next request as it would have gone (the same `system`, `tools` and
   transcript, so the cached prefix holds and the history stays append-only) with the compaction prompt as
   a text block after the last user message's tool results. It is not a step (`maxSteps`).
2. The `<summary>` (or the whole text) becomes the transcript's first message, a user message starting
   `This session continues an earlier conversation that was compacted to save context.` and ending with
   the run's `prompt` (`The user's latest request: ...`). It replaces the held transcript (`held.put`).
3. The last tool round follows it as it was: the assistant message with its `tool_use` blocks and the user
   message with their `tool_result` blocks, never one without the other. Its `thinking` blocks are dropped:
   they were bound to the history the summary replaced, which the Messages API checks (preserved thinking
   on `claude-opus-5-5` and the like).
4. The run goes on from there; the next request is not checked again.

At the start, earlier turns (the host's held `messages`) already past the threshold are summarized before
the first step, the prompt after them: the module asks their size with an empty `held.append` (it answers
`{ length, bytes }`; about 4 bytes a token) before it infers; so a resumed run's `result.messages` is the compacted memory. A summary request that
fails (an HTTP error after the retries, or no text) ends the run with `stop: error`, `error: "compaction: ..."`
(its prompt is then in the transcript's last message).

Only the last tool round is kept verbatim, not the last N: the module holds only the messages it has not
appended yet, and a held value cannot be sliced (keeping N would need the rounds in state, which rides in
every event, or a held-value slice in atlibs). Anthropic recommends summary-only compaction for preserved
thinking anyway.

**`server`**: Anthropic's [compaction at a token threshold](https://platform.claude.com/docs/en/build-with-claude/compaction-threshold)
(beta). Each request gets `anthropic-beta: compact-2026-01-12` and
`context_management: { edits: [{ type: "compact_20260112", trigger: { type: "input_tokens", value } }] }`
(`value`: the threshold, at least 50000; `instructions`: `compactPrompt`). A response with a `compaction`
block counts a compaction, and since the API ignores whatever comes before that block, the held transcript
restarts at the assistant message from it on. Other endpoints reject the field: opt in only for Anthropic's.
The compaction iteration's tokens are in the response's `usage.iterations`, not summed into `usage`. For
the on-demand form or other edits, set `context_management` (and the header, in the host's allowlist) through
`extra` with `compact: off`.

A page sees a compaction in `onEvent`'s outputs: an `http.fetch` call with id `compact.<n>` (the summary
request) and then the `held.put`; `runAgent` passes `progress(rows, { compacting })` and shows the summary as
a `compaction` row.

## MCP servers

One MCP client (it asks for protocol `2025-11-25` and speaks `2025-11-25`, `2025-06-18` and
`2025-03-26`, whichever the server answers), two transports:

| `mcp` entry (the descriptor beside `name`, or in `server`) | transport |
|---|---|
| `{ "name": "docs", "url": "https://mcp.example/mcp" }` | Streamable HTTP through `com.fedfork.exec.http.fetch` |
| `{ "name": "vm", "server": { "url": "webrtc:///<ref>/shell/mcp" } }` (or `iroh:///<ref>/shell/mcp`) | the same, over the host's transport for that scheme: a VM's guest agent, as the requester's `contract` result gives it |
| `{ "name": "code" }` or `{ "name": "code", "hook": "<nsid>" }` | the composition hook: `com.fedfork.exec.mcp.message` unless `hook` names another NSID |

One of `url` and `hook`; a `url` is `<scheme>://...`. The `{ nsid, input }` descriptor (a host MCP
transport NSID, `com.fedfork.exec.webrtc.mcp` / `iroh.mcp`) is gone: it fails with `a server is {url} or
{hook}`. At most 32 servers a run.

Over `webrtc://` and `iroh://` (atlibs: `com.fedfork.exec.http.fetch` hands any other scheme to a host
fetcher, `runModule`'s `fetchers: { "webrtc:": ... }`, a transport plugin's `fetchers()`) the client is the
same Streamable HTTP one. The guest agent answers stateless JSON (no `mcp-session-id`, `202` to a
notification); each HTTP/1.1 request holds its own datachannel or stream, so parallel tool calls are the
transport's concern. Every MCP request carries a deadline (the call's `timeoutMs`): 120 s, plus the
tool call's own `waitMs` or `timeoutMs` argument (at most 600 s of it), so the guest's `read` long-poll
(`waitMs` up to 30 s) and a long `exec` (up to 10 min) finish, while a dead transport is a tool error
instead of a run held to its `wallMs`. Response headers are read whatever their case.

Tools are namespaced per server: `<name>__<tool>`, with anything outside `[A-Za-z0-9_-]` turned into `_`
(Messages API names). A name past 64 characters, or one that clashes, is left out and listed in that
server's `dropped`. Server names are 1 to 32 of `A-Z a-z 0-9 _ -`, unique. A server that fails to
initialize or list is reported (`phase: error`) and the run goes on with the others.

### Dynamic servers: `mcp_connect`

With `dynamicMcp`, the model gets two built-in tools (after the servers' tools):

- **`mcp_connect { name, server }`**: `server` is a descriptor as above (`{ url }` or `{ hook }`; the
  requester's `contract` result gives `{ url: "webrtc:///..." }` and `{ url: "iroh:///..." }`). The module sends `initialize`,
  `notifications/initialized` and `tools/list` (following its cursors); the tool result is
  `{ connected, protocolVersion, tools: [<name>__<tool>], dropped? }`, or an error (`not connected: ...`).
  The tools are offered from the next inference on, never in the turn that connected (a call of one there
  is an unknown tool). Several connects in one turn register their tools in `tool_use` order.
- **`mcp_disconnect { name }`**: any live server, the start's too; its tools are gone from the next request
  on (a call of one already out in the same turn still answers). No HTTP `DELETE` of the session.

What a descriptor may name is `dynamicMcp`'s, never the model's: `"url"` allows any URL (the host's `http`
allowlist still bounds it), a URL prefix allows that URL and those under it (`https://a.example/mcp`
allows `https://a.example/mcp/x`, not `https://a.example/mcpx`; any scheme: `webrtc:///` allows every
`webrtc://` URL, `webrtc:///<ref>` one VM's), and an NSID allows it as a `hook` (so the model cannot route
an MCP server to a hook not listed). `true` is `["url"]`. Without `dynamicMcp` the tools are not offered
and a call of one is an unknown tool. For the VM guest agent: `["webrtc:///"]` in a page,
`["iroh:///"]` in Node or the Rust host (and the host's `http` allows `webrtc://*` / `iroh://*`).

A `name` in the live set (a start server's too) is refused until it is disconnected; a server that fails
to connect leaves the live set, so the name can be tried again. Tool names that clash or pass 64
characters are `dropped`, as at the start. The live set is in the module's state, not the transcript, so a
compaction keeps it (and the tools offered), and the summary message adds `MCP servers connected with
mcp_connect, still connected (...): <names>.` A long poll (the guest agent's `read` with `waitMs`) is an
ordinary tool call, its deadline past its wait (above). The next run starts from its `mcp` again: pass `result.servers` back to keep them.

### The hook contract: `com.fedfork.exec.mcp.message`

For in-process MCP servers (another module behind a composition edge, or a caller's handler). Real MCP,
not a wrapper: the same messages a remote server gets, one per hook call.

- **Input**: one MCP JSON-RPC 2.0 message object, a request (`{ jsonrpc: "2.0", id, method, params }`) or
  a notification (no `id`). A module serving it receives the message's fields at the top of its entry's
  input (`jsonrpc`, `id`, `method`, `params`), plus the host's `caller` and the instance's own `input`.
- **Output**: the JSON-RPC response object for a request (`{ jsonrpc: "2.0", id, result }` or
  `{ jsonrpc: "2.0", id, error: { code, message } }`, `id` echoed); `null` for a notification.
- **What this client sends**, in order: `initialize` (`{ protocolVersion, capabilities: {}, clientInfo }`),
  `notifications/initialized`, `tools/list` (following `nextCursor`, at most 50 pages), then `tools/call`
  (`{ name, arguments }`, several at once). Request ids are strings (`init.0`, `list.0.0`, `tool.3`).
- **Results**: `structuredContent`, when present, goes to the model as its JSON; else the `content`
  blocks, text as is and other blocks as their JSON, joined by newlines. `isError: true` and JSON-RPC
  errors become `is_error` tool results.
- **Each message is its own run** of the serving entry: a server that needs state across messages keeps it
  itself (initialize can answer the same every time).
- **Unrouted**: the hook is optional; with nothing routing it the call answers `null`, so that server is
  skipped (`phase: skipped`) and the agent runs without its tools.
- **Several in-process servers**: each is one edge. The first can use `com.fedfork.exec.mcp.message`; each
  other gets an NSID of the composition's choosing on its own edge (asyncai declares `anyHooks`), named in
  that server's `hook`.
- **Out of scope in v1**: server-initiated requests and notifications (sampling, elicitation, roots,
  progress, `list_changed`, logging), resources, prompts, cancellation, and re-initializing an expired
  Streamable HTTP session.

A serving module declares `implements: [{ "nsid": "com.fedfork.exec.mcp.message", "entry": "mcp" }]`
(`kcp-mcp` implements `com.example.mcp.kcp`): an edge reaches that entry. The agent's `input` keys are its
module.json `config` (`mcp`, `system`, `model`, ...). A composition with two in-process servers and a remote
one (the composition.json a `compose.ts` writes, [atlibs compositions](../atlibs/docs/compositions.md#write-one)):

```json
{
	"$type": "com.fedfork.exec.composition",
	"name": "agent",
	"instances": [
		{ "name": "agent", "module": "asyncai", "input": { "mcp": [
			{ "name": "code", "hook": "com.fedfork.exec.mcp.message" },
			{ "name": "kcp", "hook": "com.example.mcp.kcp" },
			{ "name": "deepwiki", "url": "https://mcp.deepwiki.com/mcp" }
		] } },
		{ "name": "code", "module": "code-mode" },
		{ "name": "kcp", "module": "kcp-mcp" }
	],
	"edges": [
		{ "from": "agent", "nsid": "com.fedfork.exec.mcp.message", "to": "code", "reason": "code execution tools" },
		{ "from": "agent", "nsid": "com.example.mcp.kcp", "to": "kcp", "reason": "market tools" }
	]
}
```

### Streamable HTTP

Each message is a POST of its JSON to the server URL with `accept: application/json, text/event-stream`;
after initialize, `mcp-protocol-version` and (when the server gave one) `mcp-session-id` go on every
request. A JSON or an event-stream (SSE) answer both work: atlibs hands the module the whole body once
the server ends it, and the module takes the response with the request's id. A notification expects 202.
No GET stream: server-initiated messages are out of scope, as above.

## Open

- [ ] Publish this module (`atlibs publish`) once atlibs' `com.fedfork.exec.http.fetch` is committed there
- [ ] Server-initiated MCP messages (sampling, progress, `list_changed`), and an expired Streamable HTTP session re-initializing
- [ ] Keep the last N tool rounds through a compaction, not only the last: a held-value slice (or a put that splices `$held`) in atlibs
- [ ] `fallbacks: "default"` for `claude-opus-5-5` refusals: pass it in `extra` with its beta header in the host allowlist; not on by default, since other endpoints reject it