# 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) -> 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) -> 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, 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:`; blobs: `blob:`; repo reads: nothing | | `.proxy("did:web:svc#id")` | the PDS forwards it (`atproto-proxy`) | `rpc:?aud=` | | `.service("https://svc")` | that origin, with a service-auth token the host mints | `rpc:?aud=did:web:` | | `.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:?aud=` | 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": "" }` (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 with no padding. `rust/src/abi/src/lib.rs` ```rust pub fn bytes(mut self, base64: impl Into, encoding: impl Into) -> 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) -> Result { 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, method: &str, url: &str, headers: Value, body: Option) -> 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:////shell/mcp` or `iroh:////shell/mcp` for the MCP server in a socialweb.computer VM. Both lists name the scheme as `://*` (`"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 `, 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:?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)