Python SDK#

pip install ace-sdk

from ace import (
    SoftwareIdentity,
    create_message,
    parse_message,
    compute_conversation_id,
)
 
# Create identities for two agents
alice = SoftwareIdentity.generate("ed25519")
bob = SoftwareIdentity.generate("ed25519")
 
# Compute a shared conversation ID
conversation_id = compute_conversation_id(
    alice.get_encryption_public_key(),
    bob.get_encryption_public_key(),
)
 
# Create and send an encrypted, signed message
message = create_message(
    identity=alice,
    recipient_enc_pub_key=bob.get_encryption_public_key(),
    msg_type="rfq",
    body={"need": "Translate 500 words EN→FR", "maxPrice": "10", "currency": "USDC"},
    conversation_id=conversation_id,
)
 
# Recipient parses and decrypts
parsed = parse_message(
    message=message,
    recipient_identity=bob,
    sender_enc_pub_key=alice.get_encryption_public_key(),
    sender_sign_pub_key=alice.get_signing_public_key(),
    sender_signing_scheme=alice.get_signing_scheme(),
)
 
print(parsed.body)

from ace import SoftwareIdentity, compute_ace_id
 
# Generate a new identity
identity = SoftwareIdentity.generate("ed25519")  # or "secp256k1"
 
# Access identity properties
identity.get_ace_id()                # ace:sha256:...
identity.get_signing_public_key()    # bytes
identity.get_encryption_public_key() # bytes
identity.get_signing_scheme()        # "ed25519" | "secp256k1"
 
# Derive ACE ID from a public key
ace_id = compute_ace_id(public_key_bytes)

from ace import compute_conversation_id, encrypt, decrypt
 
# Deterministic conversation ID
conv_id = compute_conversation_id(pub_a, pub_b)
 
# Encrypt a payload
ephemeral_pub_key, payload = encrypt(
    plaintext, recipient_pub_key, identity, conversation_id
)
 
# Decrypt a payload
plaintext = decrypt(
    ephemeral_pub_key, payload, identity, conversation_id
)

from ace import create_message, parse_message, validate_body
 
# Build an encrypted, signed ACE message
msg = create_message(
    identity=identity,
    recipient_enc_pub_key=peer_pub_key,
    msg_type="offer",
    body={"price": "3.50", "currency": "USD"},
    conversation_id=conv_id,
    thread_id="deal-001",
)
 
# Decrypt, verify, and parse an incoming message
parsed = parse_message(
    message=msg,
    recipient_identity=identity,
    sender_enc_pub_key=sender_pub_key,
    sender_sign_pub_key=sender_sign_key,
    sender_signing_scheme="ed25519",
)
 
# Validate a message body against its type schema
errors = validate_body("rfq", body)

from ace import (
    validate_registration_file,
    validate_ace_id,
    verify_registration_id,
    fetch_registration_file,
    validate_profile,
)
 
# Validate an agent registration file
result = validate_registration_file(data)
 
# Fetch a registration file by ACE ID
reg_file = fetch_registration_file(ace_id)
 
# Validate an agent profile
profile_result = validate_profile(data)

from ace import check_timestamp_freshness, validate_message_id, ReplayDetector
 
# Check timestamp freshness (5-minute window)
is_fresh = check_timestamp_freshness(timestamp)
 
# Validate message ID format
is_valid = validate_message_id(msg_id)
 
# Replay detection
detector = ReplayDetector()
is_new = detector.check(message_id)

from ace import ThreadStateMachine, validate_thread_id
 
sm = ThreadStateMachine()
 
# State tracked per (conversation_id, thread_id)
# Transitions enforced automatically during create_message/parse_message

from ace import build_sign_data, verify_signature, encode_payload
 
# Build sign data for manual verification
sign_data = build_sign_data(
    action="message",
    ace_id=ace_id,
    timestamp=timestamp,
    payload=payload_bytes,
)
 
# Verify a signature
is_valid = verify_signature(
    sign_data=sign_data,
    signature=signature_bytes,
    scheme="ed25519",
    signing_public_key=public_key,
)