Python SDK

The Layr8 Python SDK is an async library for building DIDComm agents. It handles WebSocket connections, Phoenix Channel protocol, DIDComm message formatting, and thread correlation. You write domain logic.

A basic agent is ~20 lines of Python. The SDK manages everything between your handler and the network.

Install

pip install layr8

Requires Python >= 3.11.

Core Concepts

Config

from layr8 import Client, Config, Message, log_errors

client = Client(Config(
    node_url="wss://your-node.layr8.cloud/plugin_socket/websocket",
    api_key="your-api-key",
    agent_did="did:web:your-node.layr8.cloud:my-agent",
), log_errors())

Client takes a Config and a required error handler. The error handler is called for SDK-level errors that cannot be returned to a direct caller (unparseable messages, missing handlers, server rejections). log_errors is a built-in helper that logs all errors via Python’s logging module.

All config fields fall back to environment variables if empty: LAYR8_NODE_URL, LAYR8_API_KEY, LAYR8_AGENT_DID. If agent_did is omitted, the node assigns an ephemeral DID on connect.

Handlers

Register handlers before calling connect. Use the decorator syntax — the handler’s return value determines what happens:

@client.handle("https://layr8.io/protocols/echo/1.0/request")
async def echo(msg: Message) -> Message | None:
    # Return Message → send response to sender
    # Return None    → no response (fire-and-forget)
    # Raise          → send problem report to sender

The protocol base URI is derived automatically from the message type and registered with the node on connect.

Send

await client.send(Message(
    type="https://didcomm.org/basicmessage/2.0/message",
    to=["did:web:partner-node:agent"],
    body={"content": "Hello!"},
))

By default, send waits for the server to acknowledge the message and raises if the server rejects it (e.g., authorization failure). For true fire-and-forget behavior:

await client.send(
    Message(
        type="https://didcomm.org/basicmessage/2.0/message",
        to=["did:web:partner-node:agent"],
        body={"content": "Hello!"},
    ),
    fire_and_forget=True,
)

Request (Request/Response)

resp = await client.request(
    Message(
        type="https://layr8.io/protocols/echo/1.0/request",
        to=["did:web:partner-node:echo-agent"],
        body={"message": "ping"},
    ),
    timeout=5.0,
)
# resp is the correlated response, matched by thread ID

Messages

@dataclass
class Message:
    id: str = ""              # auto-generated if empty
    type: str = ""            # DIDComm message type URI
    from_: str = ""           # auto-filled with agent DID (wire: "from")
    to: list[str]             # recipient DIDs
    thread_id: str = ""       # auto-generated for request
    parent_thread_id: str = ""
    body: Any = None          # serialized to JSON
    context: MessageContext | None = None

Use msg.unmarshal_body() to deserialize as a dict, or msg.unmarshal_body(MyDataclass) to construct a typed dataclass. On inbound messages, msg.context provides node metadata: authorized, recipient, and sender_credentials.

Async Context Manager

async with client:
    print(f"running as {client.did}")
    await asyncio.Event().wait()  # run forever

async with calls connect on enter and close on exit.

Verifiable Credentials

The SDK provides async methods for signing, verifying, storing, and retrieving W3C Verifiable Credentials through the cloud-node’s REST API.

Sign a Credential

from layr8 import Credential

cred = Credential(
    context=["https://www.w3.org/ns/credentials/v2"],
    type=["VerifiableCredential"],
    credential_subject={"id": subject_did, "name": "Acme Corp"},
)

signed_jwt = await client.sign_credential(cred)

The issuer DID defaults to client.did. Override with issuer_did="did:web:...". Output format defaults to compact_jwt; pass format="json" for JSON-LD.

Verify a Credential

result = await client.verify_credential(signed_jwt)
# result.credential — decoded credential claims (dict)
# result.headers    — JWT headers (dict)

Store, List, and Get

stored = await client.store_credential(signed_jwt)
# stored.id — unique credential ID

creds = await client.list_credentials()
# list[StoredCredential]

fetched = await client.get_credential(stored.id)
# fetched.credential_jwt — the original signed JWT

Sign and Verify a Presentation

pres_jwt = await client.sign_presentation([signed_jwt])

result = await client.verify_presentation(pres_jwt)
# result.presentation — decoded presentation (dict)
# result.headers      — JWT headers (dict)

See Verifiable Credentials for background on the W3C standard and trust model.

Durable Handlers (Persist-then-Ack)

For messages that must not be lost, use manual ack to persist before acknowledging:

@client.handle("https://layr8.io/protocols/order/1.0/created", manual_ack=True)
async def handle_order(msg: Message) -> None:
    # Write to disk first — if this fails, the message is
    # NOT acked and the cloud-node will redeliver it.
    with open("messages.jsonl", "a") as f:
        f.write(json.dumps({"id": msg.id, "body": msg.body}) + "\n")
        f.flush()

    msg.ack()  # safe to ack now

See examples/durable_handler.py for a complete working example.

Error Handling

The SDK requires an error handler at construction time. This ensures SDK-level errors are never silently swallowed.

ErrorHandler

The error handler is called for errors that cannot be returned to a direct caller:

from layr8 import ErrorKind, SDKError

Error kinds:

Kind	When
`ErrorKind.PARSE_FAILURE`	Inbound message could not be parsed as a DIDComm envelope
`ErrorKind.NO_HANDLER`	No handler registered for the message type
`ErrorKind.HANDLER_EXCEPTION`	A handler raised an exception
`ErrorKind.SERVER_REJECT`	The server rejected a message (e.g., authorization failure)
`ErrorKind.TRANSPORT_WRITE`	Failed to write to the WebSocket connection

Built-in Helper

log_errors logs all SDK errors via Python’s logging module:

from layr8 import Client, Config, log_errors

client = Client(Config(), log_errors())

Custom Handler

For production agents, route errors to your observability stack:

from layr8 import Client, Config, ErrorKind, SDKError

def on_error(error: SDKError) -> None:
    logger.error(
        "sdk error",
        extra={
            "kind": error.kind.value,
            "message_id": error.message_id,
            "type": error.type,
            "from": error.from_did,
            "cause": str(error.cause),
        },
    )

client = Client(Config(), on_error)

Connection Resilience

The SDK automatically reconnects when the WebSocket connection drops (e.g., node restart, network interruption). Reconnection uses exponential backoff starting at 1 second, capped at 30 seconds.

During reconnection:

send(), request(), and other operations raise NotConnectedError immediately — the SDK does not queue messages
The on_disconnect callback fires when the connection drops
The on_reconnect callback fires when the connection is restored
close() stops the reconnect loop

Claude Code Skill

The SDK includes a Claude Code skill that teaches Claude the full API. Install it into any agent project:

mkdir -p .claude/skills
cp /path/to/python-sdk/.claude/skills/build-layr8-agent-python.md .claude/skills/

Once installed, Claude Code knows how to build Layr8 agents in every session. Example prompts:

“Build me a FastAPI bridge agent that converts REST calls to DIDComm messages”
“Add a handler for order.created messages that validates the sender’s credentials”
“Convert this Flask endpoint into a DIDComm agent”