amorce-sdk 0.2.0

Official Python SDK for the Amorce Agent Transaction Protocol (AATP)

Amorce Python SDK (AATP)

Official Python SDK for the Amorce Agent Transaction Protocol (AATP).

The Amorce SDK allows any Python application, API, or Agent to become a verified node in the Agent Economy. It provides the cryptographic primitives (Ed25519) and the transport layer required to transact securely with AI Agents (OpenAI, Google Gemini, Apple Intelligence).

🚀 Features

• Zero-Trust Security: Every request is cryptographically signed (Ed25519) locally.

• Agent Identity: Manage your agent's identity and keys securely without complexity.

• Priority Lane: Mark critical messages (high, critical) to bypass network congestion.

• HTTP/2 Support (v0.2.0): Optional AsyncAmorceClient with multiplexed connections for high-throughput agents.

• Exponential Backoff (v0.2.0): Advanced retry logic with jitter using tenacity (handles 503, 429, 504).

• Resilience: Automatic retry logic for unstable networks with exponential backoff.

• Developer Experience: Simplified IdentityManager with auto-derived Agent IDs.

• Robust Error Handling: Specific exceptions (AmorceNetworkError, AmorceAPIError) for reliable production code.

📦 Installation

Prerequisites

• Python 3.10+

• OS: Linux, macOS, or Windows.

  • Linux users: Ensure you have build-essential and libssl-dev installed for the cryptography dependency.

Install via PyPI

pip install amorce-sdk

⚡ Quick Start

1. Identity Setup

An Agent is defined by its Private Key. Never share this key.

Option A: Quick Start (Ephemeral / Testing)

Generate a new identity in memory instantly. Perfect for QA scripts or temporary bots.

from amorce import IdentityManager

# Generates a fresh Ed25519 keypair in memory (Ephemeral)
identity = IdentityManager.generate_ephemeral()

# The Agent ID is automatically derived from the Public Key (SHA-256)
print(f"Agent ID: {identity.agent_id}")
print(f"Public Key: {identity.public_key_pem}")

Option B: Production (Secure Storage)

Load your identity from a secure source or environment variable.

import os
from amorce import IdentityManager, LocalFileProvider

# Load from a local PEM file
identity = IdentityManager(LocalFileProvider("agent_key.pem"))

# OR (Recommended) Load private key content from Environment Variable
# identity = IdentityManager.from_env("AMORCE_PRIVATE_KEY")

2. Sending a Transaction (Full Example)

Use the AmorceClient to discover services and execute transactions.

import os
from amorce import AmorceClient, PriorityLevel

# Configuration (Use Env Vars in Prod!)
DIRECTORY_URL = os.getenv("AMORCE_DIRECTORY_URL", "[https://directory.amorce.io](https://directory.amorce.io)")
ORCHESTRATOR_URL = os.getenv("AMORCE_ORCHESTRATOR_URL", "[https://api.amorce.io](https://api.amorce.io)")

# 1. Initialize the client
# Note: 'agent_id' is automatically derived from the identity object.
client = AmorceClient(
    identity=identity,
    directory_url=DIRECTORY_URL,
    orchestrator_url=ORCHESTRATOR_URL
)

# 2. Define the payload (The "Letter" inside the Envelope)
payload = {
    "intent": "book_reservation",
    "params": {"date": "2025-10-12", "guests": 2}
}

# 3. Execute with PRIORITY
# Options: PriorityLevel.NORMAL, .HIGH, .CRITICAL
print(f"Sending transaction from {identity.agent_id}...")

try:
    response = client.transact(
        service_contract={"service_id": "srv_restaurant_01"},
        payload=payload,
        priority=PriorityLevel.HIGH 
    )
    
    if response.get("status") == "success":
        print(f"✅ Success! Tx ID: {response.get('transaction_id')}")
        print(f"Data: {response.get('data')}")
    else:
        print(f"⚠️ Server Error: {response}")

except AmorceNetworkError as e:
    print(f"❌ Network Error (Retryable): {e}")
except AmorceAPIError as e:
    print(f"❌ API Error {e.status_code}: {e.response_body}")
except Exception as e:
    print(f"❌ Unexpected Error: {e}")

⚡ Async Quick Start (High-Throughput Agents)

NEW in v0.2.0: For AI agents handling concurrent transactions, use AsyncAmorceClient with HTTP/2:

import asyncio
from amorce import AsyncAmorceClient, IdentityManager, PriorityLevel

async def main():
    identity = IdentityManager.generate_ephemeral()
    
    # Use as AsyncContextManager for proper resource cleanup
    async with AsyncAmorceClient(
        identity=identity,
        directory_url="https://directory.amorce.io",
        orchestrator_url="https://api.amorce.io"
    ) as client:
        payload = {
            "intent": "book_reservation",
            "params": {"date": "2025-12-15", "guests": 2}
        }
        
        response = await client.transact(
            service_contract={"service_id": "srv_restaurant_01"},
            payload=payload,
            priority=PriorityLevel.HIGH
        )
        
        if response.is_success:
            print(f"✅ Transaction: {response.transaction_id}")
            print(f"Data: {response.result.data}")

asyncio.run(main())

Why Async?

• HTTP/2 Multiplexing: Handle 100s of concurrent transactions over a single connection
• Exponential Backoff: Advanced retry logic with jitter prevents thundering herd
• Non-Blocking: Perfect for AI agents that need to transact at high throughput

Migrating from Sync to Async

Sync Client (AmorceClient)	Async Client (AsyncAmorceClient)
client = AmorceClient(...)	async with AsyncAmorceClient(...) as client:
client.transact(...)	await client.transact(...)
HTTP/1.1 via requests	HTTP/2 via httpx
Basic retry (urllib3)	Exponential backoff + jitter (tenacity)
Blocking I/O	Non-blocking async I/O

### 3. Error Handling

The SDK provides specific exceptions for robust error handling:

```python
from amorce import AmorceClient, AmorceConfigError, AmorceNetworkError, AmorceAPIError

try:
    client.transact(...)
except AmorceConfigError as e:
    print(f"Configuration Error: {e}")
except AmorceNetworkError as e:
    print(f"Network Error: {e}") # Retry might be possible
except AmorceAPIError as e:
    print(f"API Error {e.status_code}: {e.response_body}")
except Exception as e:
    print(f"Unexpected Error: {e}")

🛡️ Architecture & Security

The SDK implements the AATP v0.1 standard strictly.

1. Envelope: Data is wrapped in a AmorceEnvelope.

2. Canonicalization: JSON payloads are serialized canonically (RFC 8785) to ensure signature consistency.

3. Signing: The envelope is signed locally using Ed25519.

4. Transport: The envelope is sent via HTTP/2 to the Orchestrator.

5. Verification: The receiver verifies the signature against the Trust Directory before processing.

🔧 Troubleshooting & FAQ

Q: I get a 401 Unauthorized when registering. A: Ensure your signature logic is correct. If you use IdentityManager, the signature is handled automatically. If you are manually constructing payloads, verify you are signing the canonical JSON string encoded in UTF-8.

Q: I get SSL: CERTIFICATE_VERIFY_FAILED. A: This happens if you use a placeholder URL or a self-signed cert in dev. Ensure ORCHESTRATOR_URL points to a valid HTTPS endpoint with a trusted certificate.

Q: How do I get my Agent ID? A: Do not hardcode it. Access it via identity.agent_id. It is the SHA-256 hash of your public key.

🛠️ Development

To contribute to the SDK:

# Clone and install in editable mode
git clone [https://github.com/trebortGolin/amorce_py_sdk.git](https://github.com/trebortGolin/amorce_py_sdk.git)
cd amorce_py_sdk
pip install -e .

# Run Unit Tests
python3 -m unittest discover tests

📄 License

This project is licensed under the MIT License.