custom-domain-sdk

Python client and origin helpers for the Custom Domain API. Pure Python 3.10+ with httpx; no dependency on the service itself.

pip install custom-domain-sdk        # from PyPI; from a checkout: pip install ./sdk

Quick start: register a workspace domain

from custom_domain import Client

client = Client(
    "https://edge.example.net", credential="cd_..."
)  # the edge name and credential from the operator

domain = client.create_domain(
    "forms.customer.example",  # the customer's exact subdomain
    reference="ws_8f3a1c",  # your workspace id, returned to you on every request
    idempotency_key="signup-ws_8f3a1c",  # makes the call safe to retry
)
print(domain.status)  # pending_dns
print(domain.render_dns_instructions())

prints something like

Create these DNS records for forms.customer.example:

TXT   _custom-domain-challenge.forms.customer.example  ->  custom-domain-verify=...
      purpose: ownership. Create a TXT record with exactly this name and value. ...
CNAME forms.customer.example  ->  acme.edge.example.net
      purpose: routing. Point the hostname at the target with a CNAME record. ...

Show both records to the customer as returned; each carries help text for the usual DNS console mistakes. Nothing here requires knowing anything about the edge or Caddy.

Once the service has looked up the customer's DNS zone (shortly after registration), domain.setup says which DNS provider hosts it and how the records get published:

setup = domain.setup
if setup and setup.method is not None:
    print(setup.provider.name, setup.method)  # Cloudflare guided
    print(setup.host, "in", setup.zone)  # forms in customer.example
    print(setup.guide_url)  # the provider's step-by-step guide

method is domain_connect (one-step setup at the provider), guided (a provider-specific guide) or manual (the generic guide), and None until the lookup has run.

Later:

domain = client.get_domain(domain.id)
for check in domain.checks:
    print(check.type, check.status, check.error_code)
if not domain.is_ready:
    client.request_recheck(domain.id)  # after the customer fixed DNS; rate limited
for d in client.iter_domains(reference="ws_8f3a1c"):
    ...
client.delete_domain(domain.id)

When domain.setup.method is domain_connect, the customer can approve the records at their DNS provider in one step:

link = client.connect_url(domain.id, redirect_uri="https://app.example/done", state="s-42")
# send the customer to link.url; the result arrives as usual (poll or webhook)

Or send the customer to the service's own setup page for the domain, which shows the records, the live checks and the one-step button when there is one:

page = client.connect_session(domain.id, app_name="Acme Forms",
                              return_url="https://app.example/settings/domains")
# send the customer to page.url (valid 24 hours by default, for this domain only)
client.revoke_connect_session(domain.id, page.id)  # to end it early

Errors

Every API error raises a subclass of custom_domain.ApiError with status, the stable code from the API, message and details: AuthenticationError (401), PermissionDeniedError (403), NotFoundError (404), ConflictError (409, for example hostname_already_claimed), DomainConnectUnavailableError (a ConflictError for domain_connect_unavailable), ConnectPageUnavailableError (a ConflictError for connect_page_unavailable: the deployment serves no setup page), ValidationError (422, for example apex_not_supported), RateLimitedError (429, with retry_after), ServerError (5xx). TransportError means no HTTP response was received.

Read-only credentials

The operator can issue a credential with the read scope (custom-domain credential issue --scope read) for a dashboard, a monitoring job or a support tool that only needs to look. With it the client can list and get domains and webhooks; any call that changes something (create_domain, delete_domain, request_recheck, creating or deleting a webhook, ...) raises CredentialReadOnlyError, a PermissionDeniedError with the code credential_read_only. It is not retried.

from custom_domain import Client, CredentialReadOnlyError

client = Client("https://edge.example.net", credential="cd_...")  # a read key
for d in client.iter_domains():
    print(d.hostname, d.status)
try:
    client.request_recheck(d.id)
except CredentialReadOnlyError:
    ...  # changes go through the backend that holds a full key

Timeouts and retries

Each request has a 10 second timeout (Client(timeout=...)). Requests that are safe to repeat, that is GET, DELETE, and create_domain when an idempotency key is given, are retried up to max_retries (default 2) on connection errors, timeouts, 429 (waiting Retry-After) and 5xx, with exponential backoff. A create without an idempotency key and a recheck are never retried by the client: retrying a create could register twice and a recheck is rate limited, so the caller decides.

Verifying requests at your origin

The edge adds X-Custom-Domain-Assertion to every proxied request. Select the workspace from it and from nothing else.

ASGI (FastAPI, Starlette, and others):

from custom_domain import CustomDomainMiddleware

app.add_middleware(
    CustomDomainMiddleware,
    keys={
        "1": "<EDGE_ASSERTION_KEYS secret from the operator>"
    },  # current and previous key during rotation
    application_id="<your application id>",
    workspace_lookup=load_workspace,  # returns the workspace for a reference, or None
    on_missing="reject",  # "passthrough" if this app also serves its own domain
)


@app.get("/")
def home(request: Request):
    assertion = request.state.custom_domain  # None only with on_missing="passthrough"
    workspace = load_workspace(assertion.reference)

The middleware also answers GET /.well-known/custom-domain-workspace: it verifies the assertion, calls your workspace_lookup with the reference, and returns the reference only when the lookup confirms the workspace exists (404 otherwise). The service's lifecycle worker uses that answer to prove that routing and tenant selection really work before it marks a domain ready, so workspace_lookup must consult the same source your request handlers use.

Any framework:

from custom_domain import WorkspaceResolver, AssertionInvalid

resolver = WorkspaceResolver(keys={"1": "..."}, application_id="...")
try:
    assertion = resolver.resolve(request.headers, request.host)
except AssertionInvalid as exc:
    return forbidden(exc.code)  # missing, expired, bad_signature, wrong_application, ...

Origins should only be reachable from the edge (network rules), or treat any request without a valid assertion as a direct call and refuse it.

Origin behind a CDN or a shared proxy (SDK 0.10.0 and later; earlier versions raise TypeError for host_header). When the origin is registered in origin host-header mode, the edge sends the origin's own name as Host and the customer's hostname as X-Forwarded-Host. Pass the same mode to the SDK, so the assertion is checked against X-Forwarded-Host:

app.add_middleware(
    CustomDomainMiddleware,
    keys=...,
    application_id=...,
    workspace_lookup=load_workspace,
    host_header="origin",
)
resolver = WorkspaceResolver(keys=..., application_id=..., host_header="origin")

Proxies between the edge and your application must pass X-Forwarded-Host through unchanged.

Webhooks

from custom_domain import verify_webhook, parse_event, SignatureInvalid

hook = client.create_webhook(
    "https://app.acme.example/hooks/custom-domain", ["domain.ready", "domain.deleted"]
)
store(hook.secret)  # shown once

# in your handler
try:
    verify_webhook(
        request.headers.get("X-Custom-Domain-Signature"), body, secrets=[current, previous]
    )
except SignatureInvalid:
    return 400
event = parse_event(body)  # event.id, event.type, event.created_at, event.domain

Deliveries are at least once and may arrive out of order: deduplicate by event.id and compare event.created_at before moving state. A complete consumer is in examples/webhook_consumer.py.

Local development versus production

Against the local Compose environment (DISABLE_HTTPS=true, ORIGIN_ALLOW_PRIVATE=true), the edge serves plain HTTP, the certificate check is marked not applicable, and origins on private addresses are accepted. In production the edge obtains real certificates on demand at the first handshake after DNS points to it, verifies origin TLS, refuses private origin and webhook addresses, and a domain becomes ready only after the HTTPS probe through the edge returns the right workspace. Customer DNS changes take up to the record TTL to be observed; the API's checks show the last observation and the next attempt.

Compatibility

The SDK follows the API's v1 contract. Minor SDK releases add fields and methods; a field the API stops sending is never removed within a major version. Unknown fields in API responses are ignored, so newer servers work with older SDKs. Python 3.10 and newer are supported.