@atcute/xrpc-server

web framework for XRPC servers.

npm install @atcute/xrpc-server

prerequisites

this framework relies on schemas generated by @atcute/lex-cli, you'd need to follow its quick start guide on how to set it up.

for these examples, we'll use a simple query operation that greets a name:

// file: lexicons/com/example/greet.json
{
    "lexicon": 1,
    "id": "com.example.greet",
    "defs": {
        "main": {
            "type": "query",
            "parameters": {
                "type": "params",
                "required": ["name"],
                "properties": {
                    "name": { "type": "string" }
                }
            },
            "output": {
                "encoding": "application/json",
                "schema": {
                    "type": "object",
                    "required": ["message"],
                    "properties": {
                        "message": { "type": "string" }
                    }
                }
            }
        }
    }
}

usage

handling requests

use addQuery() for queries (GET) and addProcedure() for procedures (POST). handlers receive typed params and input, and return responses using the json() helper:

import { XRPCRouter, json } from '@atcute/xrpc-server';
import { cors } from '@atcute/xrpc-server/middlewares/cors';

import { ComExampleGreet, ComExampleCreatePost } from './lexicons/index.js';

const router = new XRPCRouter({ middlewares: [cors()] });

router.addQuery(ComExampleGreet, {
    async handler({ params }) {
        return json({ message: `hello ${params.name}!` });
    },
});

router.addProcedure(ComExampleCreatePost, {
    async handler({ input }) {
        const post = await db.createPost(input);
        return json(post);
    },
});

export default router;

serving the router

on Deno, Bun or Cloudflare Workers, you can export the router directly and expect it to work out of the box.

for Node.js, you'll need a fetch-to-Node adapter like @remix-run/node-fetch-server since the router works with standard Web Request/Response:

import * as http from 'node:http';
import { createRequestListener } from '@remix-run/node-fetch-server';
import { XRPCRouter } from '@atcute/xrpc-server';

const router = new XRPCRouter();

// ... add handlers ...

const server = http.createServer(createRequestListener(router.fetch));
server.listen(3000, () => {
    console.log('listening on port 3000');
});

standalone handlers

if you only need a single XRPC operation, you can skip creating a router and export a handler directly:

import { createXrpcHandler, json } from '@atcute/xrpc-server';
import { AppBskyFeedGetFeedSkeleton } from '@atcute/bluesky';

export default createXrpcHandler({
    lxm: AppBskyFeedGetFeedSkeleton,
    async handler({ params }) {
        return json({ feed: [] });
    },
});

requests should be routed to /xrpc/<nsid>.

error handling

throw XRPCError in handlers to return error responses:

import { XRPCError } from '@atcute/xrpc-server';

router.addQuery(ComExampleGetPost, {
    async handler({ params, request }) {
        const session = await getSession(request);
        if (!session) {
            throw new XRPCError({ status: 401, error: 'AuthenticationRequired' });
        }

        const post = await db.getPost(params.uri);
        if (!post) {
            throw new XRPCError({ status: 400, error: 'InvalidRequest', message: `post not found` });
        }

        return json(post);
    },
});

convenience subclasses are also available: InvalidRequestError, AuthRequiredError, ForbiddenError, RateLimitExceededError, InternalServerError, UpstreamFailureError, NotEnoughResourcesError, UpstreamTimeoutError.

AuthRequiredError accepts a wwwAuthenticate option that auto-formats an RFC 7235 WWW-Authenticate header on the response (and appends access-control-expose-headers so browsers can read it from CORS responses):

import { AuthRequiredError } from '@atcute/xrpc-server';

throw new AuthRequiredError({
    message: 'invalid token',
    wwwAuthenticate: { scheme: 'Bearer', params: { error: 'BadJwtSignature' } },
});

observing errors

for logs or metrics, use the onError / onSocketError router options. these are fire-and-forget hooks invoked alongside response generation, and they are NOT called for client-induced errors (aborted requests, XRPCError, XRPCSubscriptionError) — only for bugs worth reporting:

const router = new XRPCRouter({
    onError({ error, request }) {
        reportToSentry(error, { url: request.url });
    },
    onSocketError({ error, request }) {
        reportToSentry(error, { url: request.url, subscription: true });
    },
});

health check

the router can optionally answer /xrpc/_health if you pass handleHealthCheck. this endpoint is non-standard — consumers decide the response shape and status.

const router = new XRPCRouter({
    async handleHealthCheck() {
        const healthy = await pingDatabase();
        return Response.json({ status: healthy ? 'ok' : 'degraded' }, { status: healthy ? 200 : 503 });
    },
});

subscriptions

subscriptions provide real-time streaming over WebSocket. they require a runtime-specific adapter:

here's an example using Bun:

import { XRPCRouter } from '@atcute/xrpc-server';
import { createBunWebSocket } from '@atcute/xrpc-server-bun';

import { ComExampleSubscribe } from './lexicons/index.js';

const ws = createBunWebSocket();

const router = new XRPCRouter({ websocket: ws.adapter });

router.addSubscription(ComExampleSubscribe, {
    async *handler({ params, signal }) {
        // yield messages until the client disconnects
        while (!signal.aborted) {
            const events = await getNewEvents(params.cursor);

            for (const event of events) {
                yield event;
            }

            await sleep(1000);
        }
    },
});

export default ws.wrap(router);

the handler is an async generator that yields messages. each yielded value is encoded as a CBOR frame and sent to the client. the signal is aborted when the client disconnects.

for subscription errors, use XRPCSubscriptionError:

import { XRPCSubscriptionError } from '@atcute/xrpc-server';

router.addSubscription(ComExampleSubscribe, {
    async *handler({ params }) {
        if (params.cursor && isCursorTooOld(params.cursor)) {
            throw new XRPCSubscriptionError({
                error: 'FutureCursor',
                message: `cursor is too old`,
            });
        }

        // ...
    },
});

backpressure is handled per adapter: each adapter's create*WebSocket() factory accepts highWaterMark / lowWaterMark options (default 250 KB / 50 KB) that throttle the send loop when the outgoing buffer grows. the Cloudflare Workers adapter does not apply backpressure — the runtime does not expose the outgoing WebSocket buffer.

service authentication

the @atcute/xrpc-server/auth subpackage provides utilities for service-to-service authentication using JWTs.

verifying incoming JWTs:

import { ServiceJwtVerifier } from '@atcute/xrpc-server/auth';
import {
    CompositeDidDocumentResolver,
    PlcDidDocumentResolver,
    WebDidDocumentResolver,
} from '@atcute/identity-resolver';

const jwtVerifier = new ServiceJwtVerifier({
    // list of audience values this service accepts. during the transition to proposal 0014
    // (atproto service auth audience), configure both the bare DID and the DID-with-service-ref
    // form so that tokens from older issuers keep working alongside new ones.
    acceptAudiences: ['did:web:my-service.example.com', 'did:web:my-service.example.com#atproto_pds'],
    resolver: new CompositeDidDocumentResolver({
        methods: {
            plc: new PlcDidDocumentResolver(),
            web: new WebDidDocumentResolver(),
        },
    }),
});

router.addQuery(ComExampleProtectedEndpoint, {
    async handler({ request }) {
        const auth = await jwtVerifier.verifyRequest(request, { lxm: 'com.example.protectedEndpoint' });
        return json({ caller: auth.issuer });
    },
});

verifyRequest parses the Authorization: Bearer header, verifies the token, and throws an AuthRequiredError with a populated WWW-Authenticate: Bearer error="…" challenge on every failure path. it forwards request.signal into DID resolution so aborted requests don't keep network calls alive.

additional options tune verification:

const jwtVerifier = new ServiceJwtVerifier({
    acceptAudiences: [...],
    resolver: ...,
    maxAge: 300, // max token lifetime window in seconds (default 300)
    clockLeeway: 5, // leeway applied to nbf/exp comparisons (default 5)
    didResolveTimeout: 5000, // issuer resolution timeout in ms (default 5000; null to disable)
    replayStore: { // optional replay protection; requires jti in tokens
        async check({ iss, jti }, ttlSeconds) {
            const key = `${iss}:${jti}`;
            const isNew = await redis.setnx(key, '1');
            if (isNew === 1) await redis.expire(key, ttlSeconds);
            return isNew === 1;
        },
    },
});

issuer DID resolution precedes signature verification, so forged tokens can trigger requests to the did:web host in iss. didResolveTimeout limits each resolution's duration; it does not restrict destinations. configure the resolver's fetch to reject non-public addresses (see server-side resolution and SSRF). caching can reduce repeated lookups, though signature failures trigger an uncached retry.

creating outgoing JWTs:

import { createServiceJwt } from '@atcute/xrpc-server/auth';

const jwt = await createServiceJwt({
    keypair: myServiceKeypair,
    issuer: 'did:web:my-service.example.com',
    audience: 'did:plc:targetservice',
    lxm: 'com.example.someEndpoint',
});

// use jwt in Authorization header when calling other services

internal calls

you can make typed calls to your own endpoints using @atcute/client:

import { Client, ok } from '@atcute/client';

const client = new Client({
    handler(pathname, init) {
        return router.fetch(new Request(new URL(pathname, 'http://localhost'), init));
    },
});

const data = await ok(
    client.get('com.example.greet', {
        params: { name: 'world' },
    }),
);

console.log(data.message); // fully typed!