Python implementation of AT Protocol's XRPC + Lexicon. lexrpc includes a simple XRPC client, server, and Flask web server integration. All three include full Lexicon support for validating inputs, outputs, and parameters against their schemas.

Install from PyPI with pip install lexrpc or pip install lexrpc[flask].

License: This project is placed in the public domain. You may also use it under the CC0 License.

• Client
• Server
• Flask server
• Reference docs
• Release instructions
• Changelog

The lexrpc client let you call methods dynamically by their NSIDs. To make a call, first instantiate a Client, then use NSIDs to make calls, passing input as a dict and parameters as kwargs. Here's an example of logging into the official Bluesky PDS and fetching the user's timeline:

from lexrpc import Client

client = Client()
session = client.com.atproto.server.createSession({
    'identifier': 'snarfed.bsky.social',
    'password': 'hunter2',
})
print('Logged in as', session['did'])

timeline = client.app.bsky.feed.getTimeline(limit=10)
print('First 10 posts:', json.dumps(timeline, indent=2))

By default, Client connects to the official bsky.social PDS and uses the official lexicons for app.bsky and com.atproto. You can connect to a different PDS or use custom lexicons by passing them to the Client constructor:

lexicons = [
  {
    "lexicon": 1,
    "id": "com.example.my-procedure",
    "defs": ...
  },
  ...
]
client = Client('my.server.com', lexicons=lexicons)
output = client.com.example.my_procedure({'foo': 'bar'}, baz=5)

Note that - characters in method NSIDs are converted to _s, eg the call above is for the method com.example.my-procedure.

To call a method with non-JSON (eg binary) input, pass bytes to the call instead of a dict, and pass the content type with headers={'Content-Type': '...'}.

Event stream methods with type subscription are generators that yield (header, payload) tuples sent by the server. They take parameters as kwargs, but no positional input.

for header, msg in client.com.example.count(start=1, end=10):
    print(header['t'])
    print(msg['num'])

To implement an XRPC server, use the Server class. It validates parameters, inputs, and outputs. Use the method decorator to register method handlers and call to call them, whether from your web framework or anywhere else.

from lexrpc import Server

server = Server()

@server.method('com.example.my-query')
def my_query(input, num=None):
    output = {'foo': input['foo'], 'b': num + 1}
    return output

# Extract nsid and decode query parameters from an HTTP request,
# call the method, return the output in an HTTP response
nsid = request.path.removeprefix('/xrpc/')
input = request.json()
params = server.decode_params(nsid, request.query_params())
output = server.call(nsid, input, **params)
response.write_json(output)

You can also register a method handler with Server.register:

server.register('com.example.my-query', my_query_handler)

If a method has non-JSON (eg binary) input, the positional input arg will be bytes. Similarly, for binary output, return bytes from your handler.

As with Client, you can use custom lexicons by passing them to the Server constructor:

lexicons = [
  {
    "lexicon": 1,
    "id": "com.example.myQuery",
    "defs": ...
  },
  ...
]
server = Server(lexicons=lexicons)

Event stream methods with type subscription should be generators that yield frames to send to the client. Each frame is a (header dict, payload dict) tuple that will be DAG-CBOR encoded and sent to the websocket client. Subscription methods take parameters as kwargs, but no positional input.

@server.method('com.example.count')
def count(start=None, end=None):
    for num in range(start, end):
        yield {'num': num}

To serve XRPC methods in a Flask web app, first install the lexrpc package with the flask extra, eg pip install lexrpc[flask]. Then, instantiate a Server and register method handlers as described above. Finally, attach the server to your Flask app with flask_server.init_flask.

from flask import Flask
from lexrpc.flask_server import init_flask

# instantiate a Server like above
server = ...

app = Flask('my-server')
init_flask(server, app)

This configures the Flask app to serve the methods registered with the lexrpc server as per the spec. Each method is served at the path /xrpc/[NSID], procedures via POSTs and queries via GETs. Parameters are decoded from query parameters, input is taken from the JSON HTTP request body, and output is returned in the JSON HTTP response body. The Content-Type response header is set to application/json.

Here's how to package, test, and ship a new release.

1. Pull from remote to make sure we're at head.

   git checkout main
   git pull

2. Run the unit tests.

   source local/bin/activate.csh
   python -m unittest discover

3. Bump the version number in pyproject.toml and docs/conf.py. git grep the old version number to make sure it only appears in the changelog. Change the current changelog entry in README.md for this new version from unreleased to the current date.
4. Build the docs. If you added any new modules, add them to the appropriate file(s) in docs/source/. Then run ./docs/build.sh. Check that the generated HTML looks fine by opening docs/_build/html/index.html and looking around.
5. git commit -am 'release vX.Y'
6. Upload to test.pypi.org for testing.

   python -m build
   setenv ver X.Y
   twine upload -r pypitest dist/lexrpc-$ver*

7. Install from test.pypi.org.

   cd /tmp
   python -m venv local
   source local/bin/activate.csh
   pip uninstall lexrpc # make sure we force pip to use the uploaded version
   pip install --upgrade pip
   pip install -i https://test.pypi.org/simple --extra-index-url https://pypi.org/simple lexrpc==$ver

8. Smoke test that the code trivially loads and runs.

   python
   # run test code below

   Test code to paste into the interpreter:

   from lexrpc import Server
   
   server = Server(lexicons=[{
       'lexicon': 1,
       'id': 'io.example.ping',
       'defs': {
           'main': {
               'type': 'query',
               'description': 'Ping the server',
               'parameters': {'message': { 'type': 'string' }},
               'output': {
                   'encoding': 'application/json',
                   'schema': {
                       'type': 'object',
                       'required': ['message'],
                       'properties': {'message': { 'type': 'string' }},
                   },
               },
           },
       },
   }])
   
   @server.method('io.example.ping')
   def ping(input, message=''):
       return {'message': message}
   
   print(server.call('io.example.ping', {}, message='hello world'))

9. Tag the release in git. In the tag message editor, delete the generated comments at bottom, leave the first line blank (to omit the release "title" in github), put ### Notable changes on the second line, then copy and paste this version's changelog contents below it.

   git tag -a v$ver --cleanup=verbatim
   git push && git push --tags

10. Click here to draft a new release on GitHub. Enter vX.Y in the Tag version box. Leave Release title empty. Copy ### Notable changes and the changelog contents into the description text box.
11. Upload to pypi.org!

    twine upload dist/lexrpc-$ver.tar.gz dist/lexrpc-$ver-py3-none-any.whl

12. Wait for the docs to build on Read the Docs, then check that they look ok.
13. On the Versions page, check that the new version is active, If it's not, activate it in the Activate a Version section.

• Bundle lexicon.community lexicons, as of 2bf2cbb.
• base.load_lexicons: ignore non-lexicon files.
• Schema validation: for the uri string format, handle URLs with brackets (eg ]) in the hostname, eg https://example.com].

Breaking changes:

• For queries with non-JSON outputs, Client.call now returns requests.Response instead of the raw output bytes. This gives callers access to the HTTP response headers, including Content-Type.

Non-breaking changes:

• Client:
  • Handle non-JSON output encodings.
  • Don't attempt to refresh access token when a com.atproto.identity procedure fails, eg when the PLC code is wrong.
  • Accept arbitrary kwargs in the Client constructor, pass them through to requests.get/post.
• flask_server:
  • init_flask: add limit_ips kwarg for whether to allow more than one connection to event stream subscription methods per client IP.

• Schema validation:
  • Validate subscription (event stream websocket) parameters and output message payloads in both Client and Server.
  • When truncate is set, recurse into refs and arrays to truncate their string properties as necessary too.
  • Allow digits in NSID name (last segment) (background).
• Server: raise ValidationError on unknown parameters.
  • Don't allow #main in $type (bluesky-social/atproto#1968).
  • Bug fix for open unions, allow types that aren't in refs.
• Client:
  • Include headers in websocket connections for event streams.
  • Add new auth constructor kwarg to support any requests auth instance, eg requests_oauth2client.DPoPToken.
• server:
  • Redirect: Add headers kwarg.
• flask_server:
  • Interpret second positional arg to ValueError and ValidationError, ie err.args[1], as a dict of additional HTTP headers to return with the HTTP 400 response.

• Add full lexicon schema validation for records and XRPC method parameters, input, and output. Includes primitive and object types, refs and unions, string formats, type-specific constraints, etc.
• Dependencies: start to switch from dag-cbor to libipld, for performance.
• client:
  • Add new decode kwarg to subscription methods to control whether DAG-CBOR messages should be decoded into native dicts for header and payload.
• flask_server: add new top-level subscribers attr that tracks clients connected (subscribed) to each event stream.
• server:
  • Add status param to Redirect.

• Fix websocket subscription server hang with blocking server XRPC methods due to exhausting worker thread pool (#8).
• Add truncate kwarg to Client and Server constructors to automatically truncate (ellipsize) string values that are longer than their maxGraphemes or maxLength in their lexicon. Defaults to False.
• Add new base.XrpcError exception type for named errors in method definitions.
• flask_server:
  • Handle base.XrpcError, convert to JSON error response with error and message fields.
• Client:
  • Bug fix for calls with binary inputs that refresh the access token. Calls with binary input now buffer the entire input in memory. (snarfed/bridgy#1670)
  • Bug fix: omit null (None) parameters instead of passing them with string value None.
• Update bundled app.bsky and com.atproto lexicons, as of bluesky-social/atproto@15cc6ff37c326d5c186385037c4bfe8b60ea41b1.

• Drop typing-extensions version pin now that typing-validation has been updated to be compatible with it.
• Update bundled app.bsky and com.atproto lexicons, as of bluesky-social/atproto@f45eef3.

• Client:
  • Support binary request data automatically based on input type, eg dict vs bytes.
  • Add new headers kwarg to call and auto-generated lexicon method calls, useful for providing an explicit Content-Type when sending binary data.
  • Bug fix: don't infinite loop if refreshSession fails.
  • Other minor authentication bug fixes.

• Bundle the official lexicons for app.bsky and com.atproto, use them by default.
• Base:
  • Expose lexicons in defs attribute.
• Client:
  • Add minimal auth support with access_token and refresh_token constructor kwargs and session attribute. If you use a Client to call com.atproto.server.createSession or com.atproto.server.refreshSession, the returned tokens will be automatically stored and used in future requests.
  • Bug fix: handle trailing slash on server address, eg http://ser.ver/ vs http://ser.ver.
  • Default server address to official https://bsky.social PDS.
  • Add default User-Agent: lexrpc (https://lexrpc.readthedocs.io/) request header.
• Server:
  • Add new Redirect class. Handlers can raise this to indicate that the web server should serve an HTTP redirect. Whether this is official supported by the XRPC spec is still TBD.
• flask_server:
  • Return HTTP 405 error on HTTP requests to subscription (websocket) XRPCs.
  • Support the new Redirect exception.
  • Add the error field to the JSON response bodies for most error responses.

• Add array type support.
• Add support for non-JSON input and output encodings.
• Add subscription method type support over websockets.
• Add headers kwarg to Client constructor.
• Add new Server.register method for manually registering handlers.
• Bug fix for server @method decorator.

Bluesky's Lexicon design and schema handling is still actively changing, so this is an interim release. It generally supports the current lexicon design, but not full schema validation yet. I'm not yet trying to fast follow the changes too closely; as they settle down and stabilize, I'll put more effort into matching and fully implementing them. Stay tuned!

Breaking changes:

• Fully migrate to new lexicon format. Original format is no longer supported.

Initial release!

Tested interoperability with the lexicon, xprc, and xrpc-server packages in bluesky-social/atproto. Lexicon and XRPC themselves are still very early and under active development; caveat hacker!