Python SDK#

pip install ace-sdk

from ace import FileStore, NetworkClient, PrekeyVault, RelayClient, SoftwareIdentity, TEXT
 
me = SoftwareIdentity.generate()
net = NetworkClient.open(
    me,
    FileStore("state"),
    RelayClient("https://relay.aceprotocol.org"),
    PrekeyVault("state/prekeys.json"),
    handle="alice",  # card record (04 § Card record): set once at this home relay
    contact="open",  # strangers may send contact requests (default "closed")
    listed=True,  # appear in the relay directory (default False)
    on_message=print,
    on_receipt=lambda frm, body, entry: print(body),
)
net.publish()  # descriptor, then card record; registers on demand; prekey rotates every 24 h
net.lookup_card("@bob@relay.aceprotocol.org")  # verified card: handle, card, pay, contact
net.connect("@bob@relay.aceprotocol.org", profile=TEXT.id, note="hi")
net.pull()  # receive, send receipts, ack (or net.follow())
for request in net.requests():  # requests never admit anyone by themselves
    net.accept(request.from_id)
mid = net.send(peer_id, "ace:text:1", {"message": "hello"})
net.maintain()  # publish, push permits, sync blocks, retry, prune

from ace import DELEGATION, TEXT, NetworkClient, Profiles, create_delegation_credential
 
net = NetworkClient.open(..., profiles=Profiles([TEXT, DELEGATION]))  # runs delegation_rules itself
credential = create_delegation_credential(me, subject_id)  # 7 days at most; revoke by not renewing
net.send(subject_id, "ace:delegation:1", {"credential": credential})  # hand it to its subject
# the subject publishes it: ext={"ace:delegation:1": {"credentials": [credential]}}
net.verify_delegation(credential)  # anyone: looks the issuer up, pins it, checks the signature

from ace import COMMERCE, TEXT, Profiles, commerce_rules
 
profiles = Profiles([TEXT, COMMERCE])
rules = [commerce_rules(store, me.get_ace_id())]  # pass to NetworkClient / Inbox / Outbox

uv run pytest -q
uv run ruff check . && uv run ruff format --check .

"""Two agents exchange a private message and a receipt, entirely in memory (no relay).
 
Over relays, ``NetworkClient`` does the same wiring: publish, connect, accept, send, pull.
"""
 
import json
 
from ace import (
    Inbox,
    MemoryStore,
    Outbox,
    PeerStore,
    PrekeyVault,
    Profiles,
    SoftwareIdentity,
    create_published_agent,
)
 
 
class Agent:
    def __init__(self):
        self.identity = SoftwareIdentity.generate()
        self.id = self.identity.get_ace_id()
        self.store, self.vault = MemoryStore(), PrekeyVault()
        self.peers = PeerStore(self.store)
        self.outbox = Outbox(self.identity, self.store, Profiles(), vault=self.vault)
        self.inbox = Inbox.open(
            self.identity,
            self.store,
            self.vault,
            self.peers,
            Profiles(),
            on_message=lambda m: print(f"{self.id[:20]}… got {m.type}: {m.body}"),
            outbox=self.outbox,
        )
 
    def published(self):
        return create_published_agent(
            self.identity, relays=["https://relay.example"], prekey=self.vault.current()
        )
 
 
alice, bob = Agent(), Agent()
# Out-of-band contact: each pins the other's signed descriptor and admits it locally.
for me, other in ((alice, bob), (bob, alice)):
    me.peers.adopt(other.published())
    me.peers.set_status(other.id, "admitted")
 
mid = alice.outbox.enqueue(bob.id, "ace:text:1", {"message": "hello bob"})
wire = []
alice.outbox.deliver(
    mid,
    lambda packet, agent: wire.append(packet),
    lambda ace_id, refresh=False: alice.peers.agent(ace_id),
)
 
outcome = bob.inbox.receive(json.dumps(wire[0]).encode())  # verify, decrypt, store
alice.inbox.receive(json.dumps(outcome.receipt.packet).encode())  # the signed receipt
print("outbox status:", alice.outbox.get(mid).status)  # delivered