@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!