aae — Python SDK

Reference implementation of the AAE protocol in Python.

Status

End-to-end working. The lifecycle orchestrator walks PROPOSE → PREVIEW → APPROVE → COMMIT → AUDIT with pluggable adapters for policy, audit, approvals, and capability tokens. All 27 conformance tests pass.

This is the canonical reference SDK. When the spec is ambiguous, this implementation's behavior is the answer; clarifications then propagate back into docs/v0.2-clarifications.md.

Install

pip install -e .

Dependencies: pydantic, pyjwt[crypto], python-ulid, rfc8785, jsonschema, PyYAML, cryptography.

60-second example

import asyncio
from datetime import datetime, timezone
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from ulid import ULID

from aae import (
    BlastRadius, Context, Decision, Effect, LifecycleClient,
    Proposal, Step, StepPreview,
)
from aae.approvals import AutoGrantApproval
from aae.audit import InMemoryAuditSink
from aae.policy import AllowAllPolicy
from aae.tokens import JwtSigner
from aae.tools import InMemoryToolRegistry, StepResult, Tool


# 1. Define a tool: preview predicts effects, commit executes.
async def echo_preview(step):
    return StepPreview(
        step_index=0,
        predicted_effects=[Effect(type="read_only", target="echo")],
        warnings=[],
    )


async def echo_commit(step, token):
    return StepResult(
        success=True,
        outputs={"echoed": step.args.get("message", "")},
    )


registry = InMemoryToolRegistry()
registry.register(Tool(
    name="echo",
    plan_schema={
        "type": "object",
        "properties": {"message": {"type": "string"}},
        "additionalProperties": False,
    },
    preview_fn=echo_preview,
    commit_fn=echo_commit,
    default_blast_radius=BlastRadius.READ_ONLY,
))

# 2. Build the JWT signer.
priv = Ed25519PrivateKey.generate()
priv_pem = priv.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption(),
)
pub_pem = priv.public_key().public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
signer = JwtSigner(
    active_kid="key-1",
    active_private_key=priv_pem,
    trusted_public_keys={"key-1": pub_pem},
    issuer="my-host",
)

# 3. Wire the lifecycle.
client = LifecycleClient(
    registry=registry,
    policy=AllowAllPolicy(),               # for production: OPA/Cerbos/YamlAllowlistPolicy
    audit=InMemoryAuditSink(),             # for production: JsonlAuditSink or your own
    approvals=AutoGrantApproval(),         # for production: CliApproval or your own delivery
    tokens=signer,
    issuer="my-host",
)

# 4. Submit a proposal.
proposal = Proposal(
    proposal_id=str(ULID()),
    agent_id="my-agent",
    tenant_id="my-tenant",
    intent="say_hello",
    context=Context(rationale="Demo"),
    steps=[
        Step(tool="echo", args={"message": "hi"},
             blast_radius=BlastRadius.READ_ONLY),
    ],
    submitted_at=datetime.now(timezone.utc),
)

result = asyncio.run(client.execute(proposal))
print(result.decision)               # Decision.ALLOW
print(result.step_results[0].outputs)  # {'echoed': 'hi'}
print(result.audit_chain_tip)        # sha256:...

Adapters shipped

For production:

• Policy: implement aae.policy.PolicyAdapter over your engine of choice (OPA, Cerbos, custom HTTP service).
• Audit: implement aae.audit.AuditSink over your durable store (Postgres, S3 with object lock, append-only DynamoDB tables).
• Approvals: implement aae.approvals.ApprovalDelivery over your notification + decision-collection surface (Slack app, web UI, mobile push).

The Protocol types are designed so a custom adapter is typically 50-150 lines.

Design notes

What LifecycleClient enforces

1. Phases happen in order. The lifecycle never advances without the prior phase emitting its audit event.
2. Audit events for a phase transition are persisted before the next phase begins. An interrupted lifecycle leaves a partial-but-verifiable chain.
3. Capability tokens bind to the SHA-256 of canonical approved-steps bytes. The commit phase re-verifies this hash before executing.
4. Approval grants trigger a re-evaluation of policy. A grant is not automatically an allow; policy may have changed during the wait.
5. On any failure, the corresponding *_failed or *_aborted event is appended before the exception propagates.

What it does NOT do

• Multitenant routing, agent authentication, approval persistence across host restarts — those are the host application's responsibility.
• Rate limiting, replay protection, network calls — those are the policy engine's or gateway's responsibility.
• Long-running plan deviation — use re-proposal (submit a new proposal with context.derived_from).

Future-ready seams

• Algorithm-agility: hashes carry their algorithm in the value (sha256:...). v1.x can introduce BLAKE3 or post-quantum hashes without breaking historical chains. See aae.algorithms.
• Extension fields: payloads accept any keys not collision-prone with current/future standardized names. Convention: prefix host-specific fields with ext..
• Federated audit: AuditSink.append will return an optional witness URL in v0.3 for external chain-head publication.
• Multi-agent: agent_id on proposals will become an optional agent_chain[] for proposals submitted on behalf of other agents.
• Cost-aware policy: ext.cost_estimate shape in step previews is reserved for v0.3.

Testing

cd sdks/python
PYTHONPATH=src python -m pytest tests/
PYTHONPATH=src python -m aae.conformance   # cross-implementation conformance

The aae-conformance console script runs after pip install -e ..

Versioning

This SDK targets AAE protocol v0.2. The protocol version is exposed as aae.__protocol_version__. The SDK's own version is aae.__version__.

License

Dual MIT / Apache-2.0. See repo root.