johnandersen777.bsky.social/atlibs

Switch repository

Loading…

# Calls

A module plans calls; the host makes them. Where a call goes is in the call itself: the session's PDS,
a service through the PDS, a service directly, or a public read of some account's PDS.

## The call builder

`rust/src/abi/src/lib.rs`

```rust
    pub fn params(mut self, params: Value) -> Self {
        self.params = Some(params);
        self
    }
    pub fn input(mut self, input: Value) -> Self {
        self.input = Some(input);
        self
    }
    /// A did#service_id the PDS proxies to (atproto-proxy); signed out, a query goes to it directly.
    pub fn proxy(mut self, proxy: impl Into<String>) -> Self {
        self.proxy = Some(proxy.into());
        self
    }
    /// An https origin called directly, or a DID whose PDS (did#service_id: that service) is read.
    pub fn service(mut self, service: impl Into<String>) -> Self {
        self.service = Some(service.into());
        self
    }
    /// A did#service_id served at `endpoints` ([{ kind: iroh|webrtc|https, address }], as its record lists
    /// them): the host picks one it can reach and sends a service-auth token aud did#service_id (never seen here).
    pub fn endpoints(mut self, did_service: impl Into<String>, endpoints: Value) -> Self {
        self.service = Some(did_service.into());
        self.endpoints = Some(endpoints);
        self
    }
    /// Keep the call open: messages arrive as `Event::Message` until `Event::Closed`.
    pub fn subscribe(mut self) -> Self {
        self.subscribe = true;
        self
    }
```

| call | goes to | needs |
|---|---|---|
| `Call::new(id, nsid).params(..)` (GET) or `.input(..)` (POST; `.bytes(..)`: a binary body) | the session's PDS | writes: `repo:<collection>`; blobs: `blob:`; repo reads: nothing |
| `.proxy("did:web:svc#id")` | the PDS forwards it (`atproto-proxy`) | `rpc:<nsid>?aud=<that did#id>` |
| `.service("https://svc")` | that origin, with a service-auth token the host mints | `rpc:<nsid>?aud=did:web:<host>` |
| `.service("did:plc:...")` with `.params(..)` | a public read of that account's PDS | nothing |
| `.service("did:web:svc#id")` with `.params(..)` | a public read of that service: the endpoint its DID document names | nothing |
| `.endpoints("did:plc:...#id", endpoints)` | one of `endpoints` the host reaches (a transport it was built with, else https), with a service-auth token aud `did#id` | `rpc:<nsid>?aud=<that did#id>` |

Signed out (no session), a `.proxy(..)` query goes to the service's endpoint directly instead, as the
`did#id` service read does: a module that proxies public reads (an AppView's) works signed out with no
change. Both are queries only (a GET; `.input(..)` is refused, signed out "needs a session") and never
carry credentials. In a page, CORS applies: the service must answer the page's origin, which the PDS
proxy does not need.

`hello` reads anyone's records signed out, so it names the account as the service:

`examples/rust/hello/src/lib.rs`

```rust
        Event::Start => Handle::new().call(
            Call::new("list", "com.atproto.repo.listRecords")
                .params(json!({ "repo": run.args.did, "collection": collection, "limit": "100" }))
                .service(run.args.did.as_str()),
        ),
```

## Bytes

Most XRPC bodies are JSON; blobs are not. A call's input `{ "$bytes": "<base64>" }` (the atproto JSON
form of bytes) goes as those raw bytes, with `encoding` as its content-type (default
`application/octet-stream`). An answer whose content-type is neither JSON nor text (`getBlob`, a CAR)
comes back as `{ "$bytes": "<base64>" }`, base64 with no padding.

`rust/src/abi/src/lib.rs`

```rust
    pub fn bytes(mut self, base64: impl Into<String>, encoding: impl Into<String>) -> Self {
```

```js run
import { runModule } from "./javascript/src/run-module/index.js";
import { hostCall, sessionXrpc } from "./javascript/src/run-module/host-call.js";

// upload two bytes, then read a blob back; the session stands in for a PDS
const handler = ({ event, state }) => {
	if (event.type === "start") return { calls: [{ id: "up", nsid: "com.atproto.repo.uploadBlob", input: { $bytes: "AAE" }, encoding: "image/png" }, { id: "get", nsid: "com.atproto.sync.getBlob", params: { did: "did:plc:x", cid: "b" } }] };
	const got = { ...state, [event.id]: event.ok ? event.output : event.error };
	return { state: got, result: got };
};
const session = { fetchHandler: async (path, init) => path.includes("uploadBlob")
	? Response.json({ sent: [init.headers["content-type"], [...init.body]] })
	: new Response(Uint8Array.of(7, 8, 9), { headers: { "content-type": "image/png" } }) };
const ctx = { wasm: { run: async ({ input }) => handler(input) } };
const { result } = await runModule(ctx, { module: new Uint8Array() }, "go", {}, hostCall({ ctx, xrpc: sessionXrpc(session) }));
console.log(JSON.stringify({ up: result.up, get: result.get }));
```

```output
{"up":{"sent":["image/png",[0,1]]},"get":{"$bytes":"BwgJ"}}
```

An event carries its call's output whole, and a run's events and calls are capped at the budget's
`ioBytes` (1 MiB by default): a module that moves blob parts of a few MB needs its host to raise it,
up to 16 MiB ([limits](../reference/limits.md)).

## Subscriptions, timers and input

`.subscribe()` keeps a call open: the host holds the WebSocket and each message arrives as
`Event::Message`, then `Event::Closed`. `send` writes up it, `cancel` closes it. `echo` subscribes to a
relay, stops at a count or a timer, and echoes host input up the stream:

`examples/rust/echo/src/lib.rs`

```rust
fn watch_handler(run: Run<Args>) -> Result<Handle, String> {
    let args = run.args;
    let limit = number(args.limit.as_ref(), 10, 1000).map_err(|e| format!("limit {e}"))?;
    let mut state: State = abi::state(run.state)?;
    Ok(match run.event {
        Event::Start => {
            if !args.service.starts_with("https://") {
                return Err("service must be an https origin".into());
            }
            let seconds = number(args.seconds.as_ref(), 10, 50).map_err(|e| format!("seconds {e}"))?;
            Handle::new()
                .call(Call::new("who", "com.atproto.repo.describeRepo").params(json!({ "repo": args.did })))
                .call(Call::new(FEED, "com.atproto.sync.subscribeRepos").service(args.service).subscribe())
                .timer(STOP, seconds * 1000)
                .state(json!(state))
        }
        Event::Result { id, result, .. } if id == "who" => {
            state.handle = result.output().ok().and_then(|o| o["handle"].as_str()).map(String::from);
            Handle::new().state(json!(state))
        }
        Event::Message { data, .. } => {
            state.count += 1;
            state.last = Some(json!({ "$type": data["$type"], "seq": data["seq"] }));
            if state.count >= limit { finish(&state, "limit").cancel(FEED) } else { Handle::new().state(json!(state)) }
        }
        Event::Input { data } => Handle::new().send(FEED, data),
        Event::Timer { .. } => finish(&state, "time"),
        Event::Closed { error, .. } => finish(&state, error.as_deref().unwrap_or("closed")),
        _ => Handle::new(),
    })
}
```

```bash run
node --test examples/rust/echo/test.mjs 2>&1 | grep -E '^ℹ fail'
```

```output
ℹ fail 0
```

Live, against the Bluesky relay:

```bash run=network
atlibs run examples/rust/echo watch '{"did":"did:plc:ocnuqjlzgohypj3zoznermms","service":"https://bsky.network","limit":"2"}' | jq -c .result
```

```output
{"count":2,"handle":…,"last":…,"stopped":"limit"}
```

Subscriptions with a session get a service-auth token; a browser cannot set WebSocket headers, so there
it goes in the query string.

## Host calls

NSIDs under `com.fedfork.exec.` are answered by the host, never a PDS. List them in `calls` like any other.

| NSID | what |
|---|---|
| `com.fedfork.exec.attest` | the host signs a record with its attestation key |
| `com.fedfork.exec.didkey.*` | the module's own did:key (module.json `didkey`): service-auth tokens, signatures |
| `com.fedfork.exec.iroh.connect` | an iroh connection, when the host has the iroh transport |
| `com.fedfork.exec.webrtc.connect` | a WebRTC datachannel, when the host has the WebRTC transport |
| `com.fedfork.exec.http.fetch` | a plain HTTP request (not XRPC) to an origin the module declares and the host allows ([below](#plain-http)) |
| `com.fedfork.exec.held.{put,append,get,delete}` | a value the host holds for the run, by name, so it need not ride in state ([below](#held-values)) |

A host without a transport refuses its calls; a module that requires one declares it as a hook, so the
run fails at load instead ([Transports](../transports.md)).

## Plain HTTP

Inference APIs and remote MCP servers are not XRPC. A module reaches them with
`com.fedfork.exec.http.fetch`: it lists that NSID in `calls` and the origins it needs in module.json
`http` (`"http": ["https://api.anthropic.com"]`; `"*"` is any origin the host allows). The host passes
its own allowlist: origins, each with headers it adds, such as an API key. A request goes out only to
an origin in both; the host's headers replace the module's of the same name, so the wasm never sees a
credential. Anything else fails as a call error (`com.fedfork.exec.http.fetch: this host does not allow
https://...`). `http` is a grant field: a pin covers it.

`rust/src/abi/src/lib.rs`

```rust
    pub fn http(id: impl Into<String>, method: &str, url: &str, headers: Value, body: Option<String>) -> Self {
```

The input is `{ method?, url, headers?, body? }` (method GET, HEAD, POST, PUT, PATCH or DELETE; body
text, or any other JSON, which goes as its JSON with [held values](#held-values) spliced in). The output is `{ status, headers, body }` for any status: header names lowercase, the body
text. Redirects are not followed (a 3xx comes back as is; in a browser, status 0). Responses are not
streamed: the body arrives whole, an event stream (SSE) too once the server ends it, so ask an
inference API for a non-streaming response. The hosts: `runModule`'s `http` option (and
`runCompose`'s, for every instance), `ATLIBS_HTTP` for both CLIs ([environment](../reference/env.md)).

```js run
import { runModule } from "./javascript/src/run-module/index.js";

// a module that asks for one URL, and a fetch that shows what the host sent
const handler = ({ event }) => event.type === "start"
	? { calls: [{ id: "q", nsid: "com.fedfork.exec.http.fetch", input: { url: "https://api.example/v1/models", headers: { "x-api-key": "guess" } } }] }
	: { result: event.ok ? event.output : event.error, done: true };
const ctx = {
	wasm: { run: async ({ input }) => handler(input) },
	http: { fetch: async (url, init) => new Response(JSON.stringify(init.headers), { headers: { "content-type": "application/json" } }) },
};
const mod = { module: new Uint8Array(), calls: ["com.fedfork.exec.http.fetch"], http: ["https://api.example"] };
const http = { "https://api.example": { headers: { "x-api-key": "the host's key" } } };
console.log(JSON.stringify((await runModule(ctx, mod, "go", {}, async () => {}, { http })).result));
console.log(JSON.stringify((await runModule(ctx, mod, "go", {}, async () => {})).result));
```

```output
{"status":200,"headers":{"content-type":"application/json"},"body":"{\"x-api-key\":\"the host's key\"}"}
"com.fedfork.exec.http.fetch: this host does not allow https://api.example"
```

In a page, the browser's CORS rules apply too: the origin must answer the page's origin.

### Other URL schemes

A URL that is not http or https goes to the host's fetcher for its scheme: a [transport](../transports.md)'s,
such as `webrtc:///<ref>/shell/mcp` or `iroh:///<ref>/shell/mcp` for the MCP server in a socialweb.computer
VM. Both lists name the scheme as `<scheme>://*` (`"http": ["webrtc://*"]`; a module's `"*"` covers http and
https only), and the host's headers and service auth never go with these requests. A host without a fetcher
for it fails the call: `com.fedfork.exec.http.fetch: no fetcher for iroh:`. The hosts: `runModule`'s and
`runCompose`'s `fetchers` option, `{ "webrtc:": async ({ method, url, headers, body, signal }) => ({ status,
headers, body }) }` (a transport plugin's `fetchers()`, spread in); in Rust, the transports' `schemes()` (see
[Transports](../transports.md#fetch-a-url)); `ATLIBS_HTTP` allows them (`{"iroh://*": {}}`).

```js run
import { runModule } from "./javascript/src/run-module/index.js";

const handler = ({ event }) => event.type === "start"
	? { calls: [{ id: "q", nsid: "com.fedfork.exec.http.fetch", input: { method: "POST", url: "webrtc:///e30/shell/mcp", body: "{}" } }] }
	: { result: event.ok ? event.output : event.error, done: true };
const ctx = { wasm: { run: async ({ input }) => handler(input) } };
const mod = { module: new Uint8Array(), calls: ["com.fedfork.exec.http.fetch"], http: ["webrtc://*"] };
// a stub transport: answers every request itself
const fetchers = { "webrtc:": async ({ method, url }) => ({ status: 200, headers: { "content-type": "text/plain" }, body: `${method} ${url}` }) };
console.log(JSON.stringify((await runModule(ctx, mod, "go", {}, async () => {}, { http: { "webrtc://*": {} }, fetchers })).result));
console.log(JSON.stringify((await runModule(ctx, mod, "go", {}, async () => {}, { http: { "webrtc://*": {} } })).result));
```

```output
{"status":200,"headers":{"content-type":"text/plain"},"body":"POST webrtc:///e30/shell/mcp"}
"com.fedfork.exec.http.fetch: no fetcher for webrtc:"
```

### Service auth instead of a key

An HTTP API that accepts an atproto service-auth token (co/core's
`https://cocore.dev/api/v1`: aud `did:web:cocore.dev`, lxm `dev.cocore.inference.dispatch`; its docs say
aud `did:web:console.cocore.dev`, which it answers `401 BadJwtAudience` today)
needs no key at all: the host's grant for that origin names the token, `{ "https://cocore.dev": {
"serviceAuth": { "aud": "did:web:cocore.dev", "lxm": "dev.cocore.inference.dispatch" } } }`.
Before each request the host calls `com.atproto.server.getServiceAuth` with that aud and lxm on the run's
own session (the one its XRPC calls use) and sends `Authorization: Bearer <token>`, replacing any the
module set; the request runs as the signed-in user, and the wasm never sees the token. A token is minted per
request, not reused (one may carry a `jti` nonce the service checks).

The mint is a call like any other, held to the module's session: under OAuth the PDS issues it only when the
session's scope has `rpc:<lxm>?aud=<aud>` (or `aud=*`), and so do atlibs' hosts (`scopeAllows`), so the module's
session record lists it. A permission's aud is a DID with a `#service` fragment or `*`, and a PDS drops a bare
DID from the grant: co/core's token aud is the bare `did:web:cocore.dev`, so its session asks
`rpc:dev.cocore.inference.dispatch?aud=*`. In a composition the host mints with the instance's own session
(`callFor`), as a lone run does with its call: the module lists no `getServiceAuth` and no edge routes it. Signed out, or
with a scope that lacks it, the fetch fails as a call error saying which token it needed and the scope
that grants it (`com.fedfork.exec.http.fetch: https://cocore.dev needs a service-auth token (aud ..., lxm
...) from the run's session: ...`). An app password session may mint any lxm.

```js run
import { runModule } from "./javascript/src/run-module/index.js";

const handler = ({ event }) => event.type === "start"
	? { calls: [{ id: "q", nsid: "com.fedfork.exec.http.fetch", input: { method: "POST", url: "https://cocore.dev/api/v1/chat/completions", body: { model: "stub", messages: [] } } }] }
	: { result: event.ok ? event.output.body : event.error, done: true };
const ctx = { wasm: { run: async ({ input }) => handler(input) }, http: { fetch: async (url, init) => new Response(init.headers.authorization) } };
const mod = { module: new Uint8Array(), calls: ["com.fedfork.exec.http.fetch"], http: ["https://cocore.dev"] };
const http = { "https://cocore.dev": { serviceAuth: { aud: "did:web:cocore.dev", lxm: "dev.cocore.inference.dispatch" } } };
// the run's session: here a stub that answers getServiceAuth
const session = async (nsid, { params }) => ({ token: `${nsid} for ${params.lxm}` });
console.log((await runModule(ctx, mod, "go", {}, session, { http })).result);
```

```output
Bearer com.atproto.server.getServiceAuth for dev.cocore.inference.dispatch
```

## Held values

A module's memory between events is its state, and the state rides in every event's input. A value
that only grows (an agent's transcript, sent whole with every inference request) would pass through
the wasm every event until it hit `ioBytes`. The host holds it instead, by name:

| NSID | input | output |
|---|---|---|
| `com.fedfork.exec.held.put` | `{ name, value }` | `{}` |
| `com.fedfork.exec.held.append` | `{ name, items }`: items added to the list `name` (a new one if absent) | `{ length, bytes }`: the list's length and JSON size (an empty append asks for them) |
| `com.fedfork.exec.held.get` | `{ name }` | `{ value }` (null if absent) |
| `com.fedfork.exec.held.delete` | `{ name }` | `{}` |

`{ "$held": name }` anywhere in a `com.fedfork.exec.http.fetch` JSON body, or in the run's result, is
replaced by that value: the host assembles the request (or the result) and the bytes never enter the
wasm. These calls are answered as they are planned, in order, so an append and the fetch after it in the
same output see each other; their results arrive as events like any call's. List them in `calls`.

`{ "$held": [names] }` is those held lists, concatenated (a system message held apart from the transcript,
then the transcript). The host keeps copies: what a module put or appended, and what a get answered, never
changes after.

Held values belong to one run: a composition's instances each have their own, nothing is shared across
runs or modules, and they are dropped when the run ends (after the result is filled in). Together they
are bounded by the budget's `heldBytes` (16 MiB by default; a host may raise it): past it, a put or append
fails as a call error.

A host may start a run with held values already set, so a large input (earlier turns of a conversation)
never rides in an event either: `runModule`'s `held` option, `{ name: value }` (`runCompose`'s: the root's
run), `ATLIBS_HELD` for both CLIs ([environment](../reference/env.md)). They count against `heldBytes`;
past it, or with a bad name, the run fails before it starts (`held: ...`). The module finds them by name
(an append to that list extends it).

```js run
import { runModule } from "./javascript/src/run-module/index.js";

const H = "com.fedfork.exec.held.";
// one event appends a message and posts the whole list; the host fills in both
const handler = ({ event }) => event.type === "start" ? {
	calls: [
		{ id: "a", nsid: `${H}append`, input: { name: "messages", items: [{ role: "user", content: "hi" }] } },
		{ id: "q", nsid: "com.fedfork.exec.http.fetch", input: { method: "POST", url: "https://api.example/v1", body: { messages: { $held: "messages" } } } },
	],
	result: { messages: { $held: "messages" } },
} : {};
const ctx = { wasm: { run: async ({ input }) => handler(input) }, http: { fetch: async (url, init) => (console.log(init.body), new Response("{}")) } };
const mod = { module: new Uint8Array(), calls: [`${H}append`, "com.fedfork.exec.http.fetch"], http: ["https://api.example"] };
const ran = await runModule(ctx, mod, "go", {}, async () => {}, { http: { "https://api.example": {} } });
console.log(JSON.stringify(ran.result));
```

```output
{"messages":[{"role":"user","content":"hi"}]}
{"messages":[{"role":"user","content":"hi"}]}
```

## Writes

A module may update or delete only records it created through this host: the host keeps a ledger of
them (`managed_records` in its storage). To manage every record in a collection, list it in
`anyRecords`:

`examples/rust/statusphere/module.json`

```json
	"anyRecords": [
		"xyz.statusphere.status"
	]
```

And it may call only what `calls` lists:

`examples/rust/statusphere/module.json`

```json
	"calls": [
		"com.atproto.repo.listRecords",
		"com.atproto.repo.createRecord",
		"com.atproto.repo.deleteRecord"
	],
```

## Budgets

A run has a budget of calls in flight, total calls, subscriptions, events and wall time. When it runs
out, the run ends with an error naming it. Numbers: [limits](../reference/limits.md).

Previous: [Write a module](write.md) | Next: [Test a module](test.md)