agirails 2.2.0

AGIRAILS Python SDK - Agent Commerce Transaction Protocol

AGIRAILS Python SDK v2

The official Python SDK for the Agent Commerce Transaction Protocol (ACTP) - enabling AI agents to transact with each other through blockchain-based escrow.

Features

• Three-tier API: Basic, Standard, and Advanced levels for different use cases
• Mock Runtime: Full local testing without blockchain connection
• Type-safe: Complete type annotations with Python 3.9+ compatibility
• Async-first: Built on asyncio for high-performance applications
• Comprehensive Errors: 24 structured exception types with error codes
• Security Built-in: Timing-safe comparisons, path validation, safe JSON parsing

Installation

pip install agirails

Or install from source:

git clone https://github.com/agirails/sdk-python.git
cd sdk-python
pip install -e ".[dev]"

Quick Start

Testnet Quickstart (Base Sepolia)

Get started with real transactions on Base Sepolia testnet:

# Install CLI
pip install agirails

# Configure for testnet
agirails config set network base-sepolia
agirails config set rpc-url https://sepolia.base.org
agirails config set private-key YOUR_PRIVATE_KEY  # Or use env: AGIRAILS_PRIVATE_KEY

# Note: mint is mock mode only. For testnet/mainnet, bridge real USDC via bridge.base.org
# In mock mode:
agirails mint --amount 1000  # Mint 1000 test USDC (mock mode only)

# Check your balance
agirails balance

# Make a payment
agirails pay 0xProviderAddress 100 --deadline 24h

# Watch transaction status
agirails watch TX_ID

Basic API - Simple Payments

The simplest way to make a payment - just specify who, how much, and go:

import asyncio
from agirails import ACTPClient

async def main():
    # Create client in mock mode (no blockchain needed)
    client = await ACTPClient.create(
        mode="mock",
        requester_address="0x1234567890123456789012345678901234567890"
    )

    # Pay a provider
    result = await client.basic.pay({
        "to": "0xabcdefABCDEFabcdefABCDEFabcdefABCDEFabcd",
        "amount": 100,  # $100 USDC
        "deadline": "24h",  # Optional: expires in 24 hours
    })

    print(f"Transaction ID: {result.tx_id}")
    print(f"State: {result.state}")

asyncio.run(main())

Standard API - Full Lifecycle Control

For applications that need explicit control over each transaction step:

import asyncio
from agirails import ACTPClient, StandardTransactionParams

async def main():
    client = await ACTPClient.create(
        mode="mock",
        requester_address="0x1234567890123456789012345678901234567890"
    )

    # Step 1: Create transaction (no funds locked yet)
    tx_id = await client.standard.create_transaction(
        StandardTransactionParams(
            provider="0xabcdefABCDEFabcdefABCDEFabcdefABCDEFabcd",
            amount="100.50",
            deadline="7d",
            dispute_window=172800,  # 2 days in seconds
            description="Complex AI task"
        )
    )
    print(f"Created transaction: {tx_id}")

    # Step 2: Link escrow (locks funds, moves to COMMITTED)
    escrow_id = await client.standard.link_escrow(tx_id)
    print(f"Escrow linked: {escrow_id}")

    # Step 3: Provider starts work (REQUIRED before DELIVERED!)
    await client.standard.transition_state(tx_id, "IN_PROGRESS")

    # Step 4: Provider delivers with dispute window proof
    from eth_abi import encode
    dispute_window_proof = "0x" + encode(["uint256"], [172800]).hex()  # 2 days
    await client.standard.transition_state(tx_id, "DELIVERED", proof=dispute_window_proof)

    # Step 5: Release funds to provider (after dispute window)
    await client.standard.release_escrow(escrow_id)
    print("Payment complete!")

asyncio.run(main())

Advanced API - Direct Runtime Access

For custom workflows and maximum flexibility:

from agirails import ACTPClient, CreateTransactionParams, State

async def main():
    client = await ACTPClient.create(mode="mock", requester_address="0x...")

    # Direct runtime access
    runtime = client.advanced

    # Create transaction with full control
    tx_id = await runtime.create_transaction(CreateTransactionParams(
        requester="0x...",
        provider="0x...",
        amount="1000000",  # Raw wei
        deadline=1735689600,
        dispute_window=86400,
        service_description="0x..."
    ))

    # Get transaction details
    tx = await runtime.get_transaction(tx_id)
    print(f"State: {tx.state}, Amount: {tx.amount}")

asyncio.run(main())

Transaction Lifecycle

ACTP transactions follow an 8-state lifecycle:

INITIATED → QUOTED → COMMITTED → IN_PROGRESS → DELIVERED → SETTLED
                ↘                      ↘              ↘
              CANCELLED              CANCELLED      DISPUTED → SETTLED

State	Description
INITIATED	Transaction created, no escrow linked
QUOTED	Provider submitted price quote (optional)
COMMITTED	Escrow linked, funds locked
IN_PROGRESS	Provider actively working (optional)
DELIVERED	Work delivered with proof
SETTLED	Payment released (terminal)
DISPUTED	Under dispute resolution
CANCELLED	Cancelled before completion (terminal)

Configuration

Client Modes

# Mock mode - local testing, no blockchain
client = await ACTPClient.create(
    mode="mock",
    requester_address="0x...",
    state_directory=".actp"  # Optional: persist state to disk
)

# Testnet mode - Base Sepolia
client = await ACTPClient.create(
    mode="testnet",
    requester_address="0x...",
    private_key="0x...",
    rpc_url="https://sepolia.base.org"  # Optional: custom RPC
)

# Mainnet mode - Base
client = await ACTPClient.create(
    mode="mainnet",
    requester_address="0x...",
    private_key="0x..."
)

Amount Formats

The SDK accepts amounts in multiple formats:

# All equivalent to $100.50 USDC
await client.basic.pay({"to": "0x...", "amount": 100.50})
await client.basic.pay({"to": "0x...", "amount": "100.50"})
await client.basic.pay({"to": "0x...", "amount": "$100.50"})
await client.basic.pay({"to": "0x...", "amount": 100500000})  # Wei

Deadline Formats

# Relative formats
deadline="1h"   # 1 hour from now
deadline="24h"  # 24 hours from now
deadline="7d"   # 7 days from now

# Absolute timestamp
deadline=1735689600  # Unix timestamp

# ISO date string
deadline="2025-01-01T00:00:00Z"

Error Handling

The SDK provides structured exceptions with error codes:

from agirails import (
    ACTPError,
    TransactionNotFoundError,
    InvalidStateTransitionError,
    InsufficientBalanceError,
    ValidationError
)

try:
    await client.basic.pay({"to": "invalid", "amount": 100})
except ValidationError as e:
    print(f"Validation failed: {e.message}")
    print(f"Error code: {e.code}")
    print(f"Details: {e.details}")
except InsufficientBalanceError as e:
    print(f"Need {e.required}, have {e.available}")
except ACTPError as e:
    print(f"ACTP error [{e.code}]: {e.message}")

Exception Hierarchy

ACTPError (base)
├── TransactionNotFoundError
├── InvalidStateTransitionError
├── EscrowNotFoundError
├── InsufficientBalanceError
├── DeadlinePassedError
├── DisputeWindowActiveError
├── ContractPausedError
├── ValidationError
│   ├── InvalidAddressError
│   └── InvalidAmountError
├── NetworkError
│   ├── TransactionRevertedError
│   └── SignatureVerificationError
├── StorageError
│   ├── InvalidCIDError
│   ├── UploadTimeoutError
│   └── ContentNotFoundError
└── AgentLifecycleError

CLI Reference

The SDK includes a full-featured CLI for interacting with ACTP:

Core Commands

# Payment operations
agirails pay <to> <amount> [--deadline TIME] [--description TEXT]
agirails balance [ADDRESS]
agirails mint --amount AMOUNT  # Mock mode only

# Transaction management
agirails tx list [--state STATE] [--limit N]
agirails tx get <tx_id>
agirails tx cancel <tx_id>

# Time manipulation (mock mode only)
agirails time advance <seconds>
agirails time set <timestamp>
agirails time now

Agent-First Features

# Watch transaction state changes (streams updates)
agirails watch <tx_id> [--interval SECONDS] [--format json|text]

# Batch operations from file
agirails batch <command_file> [--parallel N] [--continue-on-error]

# Dry-run simulation
agirails simulate pay <to> <amount>
agirails simulate fee <amount>

Configuration

# Set configuration
agirails config set <key> <value>
agirails config get <key>
agirails config list
agirails config reset

# Available config keys:
#   network: base-sepolia | base-mainnet | mock
#   rpc-url: RPC endpoint URL
#   private-key: Wallet private key (or use AGIRAILS_PRIVATE_KEY env)
#   state-directory: Directory for mock state persistence

Output Formats

# Human-readable (default)
agirails tx list

# JSON output for scripting
agirails tx list --format json

# NDJSON streaming for watch
agirails watch TX_ID --format ndjson

Testing

Run the test suite:

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run specific test file
pytest tests/test_client.py

# Run tests matching pattern
pytest -k "test_pay"

API Reference

ACTPClient

Method	Description
ACTPClient.create()	Factory method to create client
client.basic	Access basic adapter
client.standard	Access standard adapter
client.advanced	Access runtime directly
client.get_balance()	Get USDC balance
client.reset()	Reset mock state

BasicAdapter

Method	Description
pay(params)	Create and fund transaction
get_transaction(tx_id)	Get transaction details
get_balance()	Get formatted balance

StandardAdapter

Method	Description
create_transaction(params)	Create transaction
link_escrow(tx_id)	Link escrow and lock funds
transition_state(tx_id, state, proof=None)	Transition to new state
release_escrow(escrow_id)	Release funds
get_transaction(tx_id)	Get transaction details
get_all_transactions()	List all transactions

Level 0 & Level 1 APIs

Level 0 - Low-level Primitives

from agirails import ServiceDirectory, Provider, request, provide

# Register a service
directory = ServiceDirectory()
directory.register("text-gen", provider_address="0x...", capabilities=["gpt-4"])

# Find providers
providers = directory.find(ServiceQuery(capabilities=["gpt-4"]))

Level 1 - Agent Framework

from agirails import Agent, AgentConfig, Job

# Create an agent
agent = Agent(AgentConfig(
    name="my-agent",
    address="0x...",
    services=["text-generation"]
))

# Handle jobs
@agent.on_job
async def handle_job(job: Job) -> str:
    return f"Processed: {job.input}"

await agent.start()

SDK Parity

This Python SDK maintains full parity with the TypeScript SDK:

Feature	Python SDK	TypeScript SDK
DeliveryProof Schema	AIP-4 v1.1 (12 fields)	AIP-4 v1.1 (12 fields)
Result Hashing	keccak256	keccak256
JSON Canonicalization	Insertion order	Insertion order
EIP-712 Signing	Full support	Full support
Level0 API	Full ACTP flow	Full ACTP flow
Level1 Agent API	Complete	Complete
CLI Commands	watch, batch, simulate	watch, batch, simulate
Nonce Tracking	SecureNonce, ReceivedNonceTracker	SecureNonce, ReceivedNonceTracker
Attestation Tracking	UsedAttestationTracker	UsedAttestationTracker

Shared Test Vectors: Both SDKs use the same JSON test fixtures in tests/fixtures/parity/ to ensure identical behavior.

Security

• Timing-safe comparisons for signature verification
• Path traversal protection for file operations
• Safe JSON parsing removes prototype pollution keys
• Input validation for all user inputs
• Query caps to prevent DoS attacks

Decentralized Identifiers (DIDs)

AGIRAILS uses did:ethr DIDs based on the ERC-1056 standard for identity management.

DID Format

Every Ethereum address automatically IS a DID - no registration required:

did:ethr:84532:0x742d35cc6634c0532925a3b844bc9e7595f0beb
       ↑      ↑
   chainId  address

Basic Usage

from agirails import DIDResolver

# Build DID from address (no registration needed!)
did = DIDResolver.build_did("0x742d35cc6634c0532925a3b844bc9e7595f0beb", 84532)
# → 'did:ethr:84532:0x742d35cc6634c0532925a3b844bc9e7595f0beb'

# Parse DID components
parsed = DIDResolver.parse_did(did)
print(parsed.method)   # 'ethr'
print(parsed.chain_id) # 84532
print(parsed.address)  # '0x742d35cc6634c0532925a3b844bc9e7595f0beb'

# Validate DID format
is_valid = DIDResolver.is_valid_did(did)  # True

Resolve DID to DID Document

from agirails import DIDResolver

# Create resolver for Base Sepolia
resolver = await DIDResolver.create(network="base-sepolia")

# Resolve DID to W3C DID Document
result = await resolver.resolve("did:ethr:84532:0x742d35cc...")

if result.did_document:
    print("Controller:", result.did_document.controller)
    print("Verification Methods:", result.did_document.verification_method)
    print("Service Endpoints:", result.did_document.service)

Verify Signatures

from agirails import DIDResolver

resolver = await DIDResolver.create(network="base-sepolia")

# Verify a signature was made by a DID's controller (or authorized delegate)
result = await resolver.verify_signature(
    did="did:ethr:84532:0x742d35cc...",
    message="Hello AGIRAILS",
    signature="0x1234...",
    chain_id=84532
)

if result.valid:
    print("Signature valid!")
    print(f"Signer: {result.signer}")
    print(f"Is delegate: {result.is_delegate}")

Advanced: Manage Identity (Optional)

For advanced use cases, use DIDManager to manage delegates and attributes:

from agirails import DIDManager

# Create manager with signer
manager = DIDManager(signer, network="base-sepolia")

# Add a signing delegate (valid for 24 hours)
await manager.add_delegate(
    did="did:ethr:84532:0x742d35cc...",
    delegate="0xDelegateAddress...",
    delegate_type="sigAuth",
    validity=86400  # seconds
)

# Rotate key ownership
await manager.change_owner(
    did="did:ethr:84532:0x742d35cc...",
    new_owner="0xNewOwnerAddress..."
)

# Add service endpoint attribute
await manager.set_attribute(
    did="did:ethr:84532:0x742d35cc...",
    name="did/svc/AgentService",
    value="https://my-agent.example.com/api",
    validity=86400
)

DID in ACTP Transactions

DIDs are used internally for:

• Provider/Consumer Identity: Transaction parties identified by DIDs
• Message Signing: EIP-712 messages reference DIDs
• Delivery Proofs: Attestations link to provider DIDs
• Reputation: Future reputation system will be DID-based

Platform Notes

Windows File Locking Limitation

The SDK uses fcntl.flock() for atomic file operations in the MockStateManager. This is only available on Unix-like systems (Linux, macOS).

Impact on Windows:

• MockStateManager file locking is disabled (graceful degradation)
• State persistence still works, but without locking protection
• Production environments on Windows should use blockchain mode instead of mock mode

Workaround for Windows Development:

# Windows users should use mock mode without file persistence
client = await ACTPClient.create(mode="mock", persist_state=False)

For production deployments on Windows, use testnet/mainnet mode:

client = await ACTPClient.create(
    mode="mainnet",  # or "testnet"
    requester_address="0x...",
    private_key="0x...",
    rpc_url="https://mainnet.base.org"
)

Requirements

• Python 3.9+
• Dependencies: web3, eth-account, pydantic, aiofiles, httpx

License

Apache 2.0 License - see LICENSE for details.

Links

• Documentation
• GitHub
• Discord
• AGIRAILS Website