@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 the @hono/node-server adapter as the router works with standard Web Request/Response:

import { XRPCRouter } from '@atcute/xrpc-server';
import { serve } from '@hono/node-server';

const router = new XRPCRouter();

// ... add handlers ...

serve({ fetch: router.fetch, port: 3000 }, (info) => {
    console.log(`listening on port ${info.port}`);
});

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', description: `post not found` });
        }

        return json(post);
    },
});

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

subscriptions

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

runtime	adapter package
Bun	`@atcute/xrpc-server-bun`
Node.js	`@atcute/xrpc-server-node`
Deno	`@atcute/xrpc-server-deno`
Cloudflare Workers	`@atcute/xrpc-server-cloudflare`

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',
                description: `cursor is too old`,
            });
        }

        // ...
    },
});

service authentication

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

verifying incoming JWTs:

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

const jwtVerifier = new ServiceJwtVerifier({
    serviceDid: 'did:web:my-service.example.com',
    resolver: new CompositeDidDocumentResolver({
        methods: {
            plc: new PlcDidDocumentResolver(),
            web: new WebDidDocumentResolver(),
        },
    }),
});

const verifyServiceAuth = async (request: Request, lxm: string): Promise<VerifiedJwt> => {
    const authHeader = request.headers.get('authorization');
    if (!authHeader?.startsWith('Bearer ')) {
        throw new AuthRequiredError({ description: `missing or invalid authorization header` });
    }

    const result = await jwtVerifier.verify(authHeader.slice(7), { lxm });
    if (!result.ok) {
        throw new AuthRequiredError({ description: result.error.description });
    }

    return result.value;
};

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

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!