Custom lexicons

The models, namespaces, and clients that ship with this SDK are not hand-written. They are generated from the JSON lexicons in lexicons/, the ones atproto.com and bsky.app publish.

Nothing about that is specific to Bluesky. Point the generator at lexicons of your own and it produces the same thing for them: Pydantic models, typed namespaces, record sugar, subscription clients, and a Client subclass that talks to your service. You do not fork the SDK and you do not edit anything under atproto_client/.

Note

This page is about the Python side. For what a lexicon is and how to write one, read the lexicon guide and the lexicon spec on atproto.com.

A worked example

The SDK’s own test fixtures make a good tutorial, and they are in the repository under examples/custom_lexicons/lexicons/. Three files describing a small β€œstatusphere” service: a record, a query, and a subscription.

The record is the interesting one, because it references lexicons it does not own:

examples/custom_lexicons/lexicons/xyz.statusphere.status.json
{
  "lexicon": 1,
  "id": "xyz.statusphere.status",
  "defs": {
    "main": {
      "type": "record",
      "key": "tid",
      "record": {
        "type": "object",
        "required": ["status", "createdAt"],
        "properties": {
          "status": { "type": "string", "minLength": 1, "maxGraphemes": 1, "maxLength": 32 },
          "createdAt": { "type": "string", "format": "datetime" },
          "subject": { "type": "ref", "ref": "com.atproto.repo.strongRef" },
          "aboutPost": { "type": "ref", "ref": "app.bsky.feed.post" },
          "extra": { "type": "unknown" }
        }
      }
    },
    "statusView": {
      "type": "object",
      "required": ["uri", "status"],
      "properties": {
        "uri": { "type": "string", "format": "at-uri" },
        "status": { "type": "string" },
        "author": { "type": "ref", "ref": "app.bsky.actor.defs#profileViewBasic" },
        "embed": {
          "type": "union",
          "refs": ["app.bsky.embed.images", "app.bsky.embed.external"]
        }
      }
    }
  }
}

subject points at com.atproto.repo.strongRef, aboutPost at app.bsky.feed.post, author at app.bsky.actor.defs#profileViewBasic, and embed is a union of two Bluesky embed types. None of those are yours. Resolving them correctly is most of what the generator does.

The query and the subscription round it out:

examples/custom_lexicons/lexicons/xyz.statusphere.getStatuses.json
{
  "lexicon": 1,
  "id": "xyz.statusphere.getStatuses",
  "defs": {
    "main": {
      "type": "query",
      "parameters": {
        "type": "params",
        "properties": {
          "limit": { "type": "integer", "minimum": 1, "maximum": 100, "default": 50 },
          "cursor": { "type": "string" }
        }
      },
      "output": {
        "encoding": "application/json",
        "schema": {
          "type": "object",
          "required": ["statuses"],
          "properties": {
            "cursor": { "type": "string" },
            "statuses": {
              "type": "array",
              "items": { "type": "ref", "ref": "xyz.statusphere.status#statusView" }
            }
          }
        }
      }
    }
  }
}
examples/custom_lexicons/lexicons/xyz.statusphere.subscribeStatuses.json
{
  "lexicon": 1,
  "id": "xyz.statusphere.subscribeStatuses",
  "defs": {
    "main": {
      "type": "subscription",
      "description": "Stream of status updates.",
      "parameters": {
        "type": "params",
        "properties": {
          "cursor": { "type": "integer" }
        }
      },
      "message": {
        "schema": {
          "type": "union",
          "refs": ["#update", "#info"]
        }
      },
      "errors": [{ "name": "FutureCursor" }]
    },
    "update": {
      "type": "object",
      "required": ["seq", "status"],
      "properties": {
        "seq": { "type": "integer" },
        "status": { "type": "string" },
        "did": { "type": "string", "format": "did" }
      }
    },
    "info": {
      "type": "object",
      "required": ["name"],
      "properties": {
        "name": { "type": "string" },
        "message": { "type": "string" }
      }
    }
  }
}

Generating

From examples/custom_lexicons/:

atp gen --lexicon-dir ./lexicons custom --output-dir ./statusphere --package statusphere
Generating statusphere:
- models...
- namespaces...
- subscriptions...
Done! Package written to ./statusphere

Important

--lexicon-dir belongs to gen, not to custom, so it goes before the subcommand. Putting it after fails with Error: '--lexicon-dir' is required. Pass it before the subcommand.

Name the output directory after the package. --package statusphere makes the generated code import itself as statusphere, so the directory it lives in has to be called statusphere and has to be importable, either on sys.path or inside your project.

Ruff must be installed; the generator formats what it writes.

What lands on disk

statusphere/
β”œβ”€β”€ __init__.py
β”œβ”€β”€ client.py                          # attach_namespaces() + StatusphereClient(Client)
β”œβ”€β”€ async_client.py                    # attach_async_namespaces() + AsyncStatusphereClient
β”œβ”€β”€ subscriptions.py                   # message unions, parsers, sync + async clients
β”œβ”€β”€ models/
β”‚   β”œβ”€β”€ __init__.py                    # the _Ids table and the lazy accessors
β”‚   β”œβ”€β”€ type_conversion.py             # record registration for $type resolution
β”‚   β”œβ”€β”€ unknown_type.py
β”‚   └── xyz/statusphere/
β”‚       β”œβ”€β”€ status.py                  # Record + the *RecordResponse models
β”‚       β”œβ”€β”€ get_statuses.py            # Params / ParamsDict / Response
β”‚       └── subscribe_statuses.py      # Params + the message defs
└── namespaces/
    β”œβ”€β”€ __init__.py
    β”œβ”€β”€ sync_ns.py                     # XyzNamespace, XyzStatusphereNamespace, XyzStatusphereStatusRecord
    └── async_ns.py                    # the async mirror

Sync and async are always both generated. --no-client drops only client.py and async_client.py. You still get namespaces/async_ns.py and subscriptions.py.

Using it

Two ways in. Either instantiate the generated client, or graft the namespaces onto a client you already have:

examples/custom_lexicons/use_generated_client.py
"""Use a package generated from custom lexicons.

Generate it first, from this directory:

    atp gen --lexicon-dir ./lexicons custom --output-dir ./statusphere --package statusphere
"""

from atproto import Client, models

# The generated package. Importing it registers its record types for $type resolution.
from statusphere import models as statusphere_models
from statusphere.client import StatusphereClient, attach_namespaces

USERNAME = 'example.com'
PASSWORD = 'hunter2'  # noqa: S105 never hardcode your password in a real application


def with_generated_client() -> None:
    # StatusphereClient is a subclass of the SDK Client, so it keeps app, com, chat and the rest
    client = StatusphereClient()
    client.login(USERNAME, PASSWORD)

    # the SDK's own namespaces still work
    print(client.app.bsky.actor.get_profile({'actor': client.me.did}).display_name)

    # and so do yours
    print(client.xyz.statusphere.get_statuses({'limit': 10}))

    # records of your own lexicons get the same sugar as the built-in ones
    status = statusphere_models.XyzStatusphereStatus.Record(
        status='πŸ‘',
        created_at=client.get_current_time_iso(),
    )
    created = client.xyz.statusphere.status.create(client.me.did, status)
    print(created.uri)


def with_existing_client() -> None:
    # Already have a client? Graft the namespaces onto it instead of switching classes.
    client = Client()
    client.login(USERNAME, PASSWORD)

    attach_namespaces(client)
    print(client.xyz.statusphere.get_statuses({'limit': 10}))


def resolve_a_custom_record() -> None:
    # $type resolution reaches into the generated package, so a custom record nested in an
    # SDK model deserializes to your class instead of degrading to DotDict.
    record = models.get_or_create(
        {'$type': 'xyz.statusphere.status', 'status': 'πŸ‘', 'createdAt': '2026-01-01T00:00:00Z'},
        None,
        strict=False,
    )
    print(type(record), record.status)


if __name__ == '__main__':
    with_generated_client()

The generated client

StatusphereClient subclasses the SDK’s Client, so it keeps everything: login and session refresh, the transport, the headers machinery, and all seven built-in namespace roots. It adds xyz on top.

The class name is derived from --package: underscores become word boundaries and each word is capitalised, then Client is appended. --package my_pkg gives you MyPkgClient and AsyncMyPkgClient.

attach_namespaces

attach_namespaces(client) sets one attribute per root authority in your package. It is annotated client: t.Any on purpose, because the namespaces it attaches only ever call invoke_query and invoke_procedure, which is the whole of the XrpcClient protocol in atproto_client.namespaces.base. Anything satisfying those two methods works: a stock Client, your own subclass, or a transport you wrote yourself.

Use it when the client already exists and you do not want to switch classes.

Record sugar

Records in your lexicons get the same generated helpers the built-in ones get: create, get, list, delete, with rkey, swap_commit, swap_record and validate where they apply:

client.xyz.statusphere.status.create(client.me.did, status_record)
client.xyz.statusphere.status.list(client.me.did, limit=10)
client.xyz.statusphere.status.get(client.me.did, rkey)
client.xyz.statusphere.status.delete(client.me.did, rkey)

These are ordinary com.atproto.repo.* calls underneath, so they work against any PDS. See Records and repositories.

How the pieces fit

Three mechanisms make a generated package compose with the SDK rather than duplicate it. You do not have to configure any of them, but knowing they exist explains the behaviour.

References resolve into the installed SDK

Code is generated only for the lexicons in --lexicon-dir. A $ref that points outside them, such as "ref": "com.atproto.repo.strongRef", becomes a reference to the SDK’s existing model instead of a second copy of it.

The generator does not need the SDK’s lexicons for that. The one thing it has to know about a foreign lexicon is whether its main definition is a record, because records are named Record rather than Main, and the SDK’s generated record table already answers it. That table is built from the same lexicons as the installed models, so a reference can only ever name a model that exists in the SDK you have installed.

Models chain to the SDK’s

statusphere/models/__init__.py ends with:

__getattr__, __dir__ = make_lazy_accessors(__name__, fallback='atproto_client.models')

Your package resolves its own NSID aliases and falls through to the SDK for everything else. So statusphere.models.XyzStatusphereStatus comes from your package and statusphere.models.ComAtprotoRepoStrongRef comes from the SDK, through the same attribute access. Nothing is imported until it is touched.

Records resolve by $type at runtime

models/type_conversion.py registers a name-only map:

RECORD_TYPES = {
    'xyz.statusphere.status': 'XyzStatusphereStatus',
}
register_record_types('statusphere.models', RECORD_TYPES)

The registry stores names rather than classes, so no model module is imported until a record with that $type actually shows up.

The consequence worth knowing: importing your package is what makes its records decode. A custom record sitting in an unknown field of an SDK model, or arriving over the firehose, deserializes to your Record class if your package has been imported, and degrades to a DotDict if it has not.

Subscriptions

A subscription lexicon whose message.schema has refs gets the full treatment in subscriptions.py:

  • a XyzStatusphereSubscribeStatusesMessage union of the message models,

  • a #fragment-keyed map from frame type to model,

  • parse_xyz_statusphere_subscribe_statuses_message(frame),

  • and XyzStatusphereSubscribeStatusesClient plus its async twin.

from statusphere.subscriptions import (
    XyzStatusphereSubscribeStatusesClient,
    parse_xyz_statusphere_subscribe_statuses_message,
)

client = XyzStatusphereSubscribeStatusesClient('wss://statusphere.example.com/xrpc', params={'cursor': 42})


def on_message(frame) -> None:
    print(parse_xyz_statusphere_subscribe_statuses_message(frame))


client.start(on_message)

The generated client takes base_uri and recv_timeout rather than hardcoding a host, because deployment values are not part of a lexicon. The SDK’s own FirehoseSubscribeReposClient is one of these with today’s relay defaults filled in. See Firehose.

Note

A subscription that declares a subprotocol gets the union, the type map, and the parser, but no client, because the transport is not the standard one. Jetstream is the example; its client is hand-written for that reason.

Limitations

The generator does not yet cover everything a lexicon can express:

  • Top-level defs that are bare primitives. A def that is an integer, boolean, bytes, cid-link, blob, or unknown rather than an object, string, token, or array is skipped silently. Wrap it in an object if you need it.

  • permission and permission-set defs are not generated.

Neither blocks generating a working package; they mean those particular defs produce no code.

Keeping generated code out of your diffs

The output is deterministic, so both options work:

  • Commit it. Your package is importable without a build step, and reviewers see what changed when a lexicon changes.

  • Generate it in CI. Add the atp gen custom invocation to your build and gitignore the output directory. Pin the atproto version: generated code is only guaranteed to work with the SDK version that produced it.