Python client and server for Bluesky/AT Protocol's XRPC + Lexicon
CC0-1.0 License
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.
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)
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.
Run the unit tests.
source local/bin/activate.csh
python -m unittest discover
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.
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.
git commit -am 'release vX.Y'
Upload to test.pypi.org for testing.
python -m build
setenv ver X.Y
twine upload -r pypitest dist/lexrpc-$ver*
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
deactivate
Smoke test that the code trivially loads and runs.
source local/bin/activate.csh
python
# run test code below
deactivate
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'))
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
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.
Upload to pypi.org!
twine upload dist/lexrpc-$ver.tar.gz dist/lexrpc-$ver-py3-none-any.whl
Wait for the docs to build on Read the Docs, then check that they look ok.
On the Versions page, check that the new version is active, If it's not, activate it in the Activate a Version section.
object
types, ref
s and union
s, string formats, type-specific constraints, etc.dag-cbor
to libipld
, for performance.client
:
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
:
status
param to Redirect
.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
.base.XrpcError
exception type for named errors in method definitions.flask_server
:
base.XrpcError
, convert to JSON error response with error
and message
fields.Client
:
None
) parameters instead of passing them with string value None
.app.bsky
and com.atproto
lexicons, as of bluesky-social/atproto@15cc6ff37c326d5c186385037c4bfe8b60ea41b1.typing-extensions
version pin now that typing-validation has been updated to be compatible with it.app.bsky
and com.atproto
lexicons, as of bluesky-social/atproto@f45eef3.Client
:
dict
vs bytes
.headers
kwarg to call
and auto-generated lexicon method calls, useful for providing an explicit Content-Type
when sending binary data.refreshSession
fails.app.bsky
and com.atproto
, use them by default.Base
:
defs
attribute.Client
:
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.http://ser.ver/
vs http://ser.ver
.https://bsky.social
PDS.User-Agent: lexrpc (https://lexrpc.readthedocs.io/)
request header.Server
:
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
:
Redirect
exception.error
field to the JSON response bodies for most error responses.subscription
method type support over websockets.headers
kwarg to Client
constructor.Server.register
method for manually registering handlers.@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:
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!