Overview
The ChaosChainAgentSDK is the main entry point for building agents on the ChaosChain protocol. It provides a unified interface for all protocol operations.
Constructor
ChaosChainAgentSDK(
agent_name: str,
agent_domain: str,
agent_role: AgentRole | str,
network: NetworkConfig | str = "base-sepolia",
enable_process_integrity: bool = True,
enable_payments: bool = True,
enable_storage: bool = True,
enable_ap2: bool = True,
wallet_file: str = None,
storage_jwt: str = None,
storage_gateway: str = None
)
Parameters
Name of the agent (used for wallet generation and identification)
Domain where agent identity is hosted (e.g., “myagent.example.com”)
Role of the agent. Can be:
"server" or AgentRole.SERVER - Provides services"client" or AgentRole.CLIENT - Consumes services"validator" or AgentRole.VALIDATOR - Validates work
network
NetworkConfig | str
default:"base-sepolia"
Target blockchain network. Can be:
"base-sepolia" or NetworkConfig.BASE_SEPOLIA"ethereum-sepolia" or NetworkConfig.ETHEREUM_SEPOLIA"optimism-sepolia" or NetworkConfig.OPTIMISM_SEPOLIA
Enable process integrity verification with cryptographic proofs
Enable payment processing with 5 W3C-compliant methods
Enable IPFS storage for evidence and data persistence
Enable Google AP2 integration for intent verification
Custom wallet storage file path (defaults to “chaoschain_wallets.json”)
Custom Pinata JWT token for IPFS storage (overrides PINATA_JWT env var)
Custom IPFS gateway URL (overrides PINATA_GATEWAY env var)
Example
from chaoschain_sdk import ChaosChainAgentSDK, AgentRole, NetworkConfig
# Using enums (recommended)
sdk = ChaosChainAgentSDK(
agent_name="MyAgent",
agent_domain="myagent.example.com",
agent_role=AgentRole.SERVER,
network=NetworkConfig.BASE_SEPOLIA
)
# Using strings (also supported)
sdk = ChaosChainAgentSDK(
agent_name="MyAgent",
agent_domain="myagent.example.com",
agent_role="server",
network="base-sepolia"
)
Identity Management
register_identity()
Register agent identity on ERC-8004 IdentityRegistry.
def register_identity() -> Tuple[AgentID, TransactionHash]
AgentID: Unique on-chain agent identifierTransactionHash: Transaction hash of registration
Example:
agent_id, tx_hash = sdk.register_identity()
print(f"Agent registered with ID: {agent_id}")
AgentRegistrationError: If registration failsConfigurationError: If network not configured
get_agent_id()
Get the agent’s on-chain ID.
def get_agent_id() -> Optional[AgentID]
AgentID: Agent ID if registered, None otherwise
Example:
agent_id = sdk.get_agent_id()
if agent_id:
print(f"Agent ID: {agent_id}")
else:
print("Agent not registered")
get_agent_identity()
Get complete agent identity information.
def get_agent_identity() -> AgentIdentity
AgentIdentity: Complete identity object with all details
Example:
identity = sdk.get_agent_identity()
print(f"Agent: {identity.agent_name}")
print(f"Domain: {identity.agent_domain}")
print(f"Address: {identity.wallet_address}")
Process Integrity
register_integrity_checked_function()
Register a function for integrity checking.
def register_integrity_checked_function(
func: callable,
function_name: str = None
) -> str
func: Function to registerfunction_name: Optional custom name (defaults to function.name)
Returns:
str: Code hash of the registered function
Example:
@sdk.process_integrity.register_function
async def analyze_data(data: dict) -> dict:
return {"result": "analyzed", "count": len(data)}
# Or register explicitly
code_hash = sdk.register_integrity_checked_function(analyze_data)
execute_with_integrity_proof()
Execute a registered function with integrity proof generation.
async def execute_with_integrity_proof(
function_name: str,
inputs: Dict[str, Any],
require_proof: bool = True
) -> Tuple[Any, Optional[IntegrityProof]]
function_name: Name of registered functioninputs: Function input parametersrequire_proof: Whether to generate integrity proof
Returns:
Any: Function execution resultIntegrityProof: Cryptographic proof of execution
Example:
result, proof = await sdk.execute_with_integrity_proof(
"analyze_data",
{"items": ["data1", "data2", "data3"]}
)
print(f"Result: {result}")
print(f"Proof ID: {proof.proof_id}")
Payment Processing
get_supported_payment_methods()
Get list of all supported payment methods.
def get_supported_payment_methods() -> List[str]
List[str]: W3C payment method identifiers
Example:
methods = sdk.get_supported_payment_methods()
print(f"Supported methods: {methods}")
# Output: ['basic-card', 'https://google.com/pay', 'https://apple.com/apple-pay', 'https://paypal.com', 'https://a2a.org/x402']
create_x402_payment_request()
Create A2A-x402 payment request for crypto payments.
def create_x402_payment_request(
cart_id: str,
total_amount: float,
currency: str,
items: List[Dict[str, Any]],
settlement_address: str = None
) -> Dict[str, Any]
cart_id: Unique cart identifiertotal_amount: Total payment amountcurrency: Payment currency (e.g., “USDC”)items: List of items with name and pricesettlement_address: Optional settlement address (defaults to agent wallet)
Returns:
Dict[str, Any]: X402 payment request object
Example:
payment_request = sdk.create_x402_payment_request(
cart_id="order_123",
total_amount=10.0,
currency="USDC",
items=[{"name": "AI Analysis", "price": 10.0}]
)
execute_x402_crypto_payment()
Execute A2A-x402 crypto payment.
def execute_x402_crypto_payment(
payment_request: Dict[str, Any],
payer_agent: str,
service_description: str = "Agent Service"
) -> Dict[str, Any]
payment_request: X402 payment request from create_x402_payment_request()payer_agent: Name of paying agentservice_description: Description of service being paid for
Returns:
Dict[str, Any]: Payment response with transaction details
Example:
crypto_result = sdk.execute_x402_crypto_payment(
payment_request=payment_request,
payer_agent="ClientAgent",
service_description="Data Analysis Service"
)
print(f"Transaction: {crypto_result['transaction_hash']}")
print(f"Amount: {crypto_result['amount']} USDC")
execute_traditional_payment()
Execute traditional payment method (cards, Google Pay, etc.).
def execute_traditional_payment(
payment_method: str,
amount: float,
currency: str,
payment_data: Dict[str, Any]
) -> Dict[str, Any]
payment_method: W3C payment method identifieramount: Payment amountcurrency: Payment currencypayment_data: Method-specific payment data
Returns:
Dict[str, Any]: Payment response with transaction details
Example:
card_result = sdk.execute_traditional_payment(
payment_method="basic-card",
amount=25.99,
currency="USD",
payment_data={
"cardNumber": "4111111111111111",
"cardType": "visa",
"expiryMonth": "12",
"expiryYear": "2025",
"cvv": "123"
}
)
Google AP2 Integration
create_intent_mandate()
Create Google AP2 Intent Mandate for user authorization.
def create_intent_mandate(
user_description: str,
merchants: Optional[List[str]] = None,
skus: Optional[List[str]] = None,
requires_refundability: bool = False,
expiry_minutes: int = 60
) -> GoogleAP2IntegrationResult
user_description: Natural language description of intentmerchants: Allowed merchants (optional)skus: Specific SKUs (optional)requires_refundability: Whether items must be refundableexpiry_minutes: Minutes until intent expires
Returns:
GoogleAP2IntegrationResult: Result with IntentMandate
Example:
intent_result = sdk.create_intent_mandate(
user_description="Find me a good AI analysis service under $10",
merchants=["TrustedAnalytics", "AIServices"],
expiry_minutes=60
)
if intent_result.success:
print(f"Intent created: {intent_result.intent_mandate}")
create_cart_mandate()
Create Google AP2 Cart Mandate with JWT signing.
def create_cart_mandate(
cart_id: str,
items: List[Dict[str, Any]],
total_amount: float,
currency: str = "USD",
merchant_name: Optional[str] = None,
expiry_minutes: int = 15
) -> GoogleAP2IntegrationResult
cart_id: Unique cart identifieritems: List of items in carttotal_amount: Total cart amountcurrency: Currency codemerchant_name: Name of merchant (defaults to agent name)expiry_minutes: Minutes until cart expires
Returns:
GoogleAP2IntegrationResult: Result with CartMandate and JWT
Example:
cart_result = sdk.create_cart_mandate(
cart_id="cart_123",
items=[{"name": "AI Analysis", "price": 5.0}],
total_amount=5.0,
currency="USD"
)
if cart_result.success:
print(f"JWT Token: {cart_result.jwt_token}")
verify_jwt_token()
Verify Google AP2 JWT token.
def verify_jwt_token(token: str) -> Dict[str, Any]
token: JWT token to verify
Returns:
Dict[str, Any]: Decoded payload if valid, empty dict if invalid
Example:
payload = sdk.verify_jwt_token(jwt_token)
if payload:
print(f"Valid token for cart: {payload['sub']}")
else:
print("Invalid or expired token")
Storage Management
store_evidence()
Store evidence data on IPFS.
def store_evidence(
data: Dict[str, Any],
evidence_type: str = "evidence",
metadata: Dict[str, Any] = None
) -> Optional[str]
data: Data to storeevidence_type: Type of evidencemetadata: Optional metadata
Returns:
str: IPFS CID if successful, None otherwise
Example:
evidence_cid = sdk.store_evidence({
"analysis_result": result,
"integrity_proof": proof.__dict__,
"timestamp": "2024-01-01T00:00:00Z"
})
if evidence_cid:
print(f"Evidence stored: {evidence_cid}")
retrieve_evidence()
Retrieve evidence data from IPFS.
def retrieve_evidence(cid: str) -> Optional[Dict[str, Any]]
cid: IPFS Content Identifier
Returns:
Dict[str, Any]: Retrieved data if successful, None otherwise
Example:
evidence = sdk.retrieve_evidence(evidence_cid)
if evidence:
print(f"Retrieved evidence: {evidence}")
Validation
request_validation()
Request validation from another agent via ERC-8004.
def request_validation(
validator_agent_id: AgentID,
data_hash: str
) -> TransactionHash
validator_agent_id: ID of the validator agentdata_hash: Hash of data to validate
Returns:
TransactionHash: Transaction hash
Example:
import hashlib
data_hash = "0x" + hashlib.sha256(json.dumps(result).encode()).hexdigest()
tx_hash = sdk.request_validation(validator_agent_id=8, data_hash=data_hash)
submit_validation_response()
Submit a validation response with score.
def submit_validation_response(
data_hash: str,
score: int
) -> TransactionHash
data_hash: Hash of the data that was validatedscore: Validation score (0-100)
Returns:
TransactionHash: Transaction hash
Example:
tx_hash = sdk.submit_validation_response(
data_hash=data_hash,
score=85
)
submit_feedback()
Submit feedback for another agent via ERC-8004.
def submit_feedback(
agent_id: AgentID,
score: int,
feedback: str
) -> TransactionHash
agent_id: Target agent IDscore: Feedback score (0-100)feedback: Feedback text
Returns:
TransactionHash: Transaction hash
Example:
tx_hash = sdk.submit_feedback(
agent_id=target_agent_id,
score=95,
feedback="Excellent service quality and fast response"
)
Evidence Packages
create_evidence_package()
Create a comprehensive evidence package for Proof of Agency.
def create_evidence_package(
work_proof: Dict[str, Any],
integrity_proof: IntegrityProof = None,
payment_proofs: List[PaymentProof] = None,
validation_results: List[ValidationResult] = None
) -> EvidencePackage
work_proof: Evidence of work performedintegrity_proof: Process integrity proofpayment_proofs: List of payment proofsvalidation_results: List of validation results
Returns:
EvidencePackage: Complete evidence package
Example:
evidence_package = sdk.create_evidence_package(
work_proof={
"service_type": "data_analysis",
"input_data": input_params,
"output_data": analysis_result,
"execution_time": execution_duration
},
integrity_proof=integrity_proof,
payment_proofs=[payment_proof]
)
print(f"Package ID: {evidence_package.package_id}")
print(f"IPFS CID: {evidence_package.ipfs_cid}")
Utility Methods
get_sdk_status()
Get comprehensive SDK status and configuration.
def get_sdk_status() -> Dict[str, Any]
Dict[str, Any]: SDK status with all configuration details
Example:
status = sdk.get_sdk_status()
print(f"Agent: {status['agent_name']} ({status['agent_role']})")
print(f"Network: {status['network']} (Chain {status['chain_id']})")
print(f"Features: {status['features']}")
print(f"Payment Methods: {len(status['payment_methods'])}")
Properties
wallet_address
Get the agent’s wallet address.
@property
def wallet_address(self) -> str
str: Agent’s wallet address
Example:
print(f"Wallet: {sdk.wallet_address}")
is_registered
Check if the agent is registered on-chain.
@property
def is_registered(self) -> bool
bool: True if registered, False otherwise
Example:
if sdk.is_registered:
print("Agent is registered")
else:
print("Agent needs registration")
network_info
Get network information.
@property
def network_info(self) -> Dict[str, Any]
Dict[str, Any]: Network configuration details
Example:
info = sdk.network_info
print(f"Network: {info['network']}")
print(f"Chain ID: {info['chain_id']}")
print(f"Connected: {info['connected']}")
Error Handling
The SDK raises specific exceptions for different error conditions:
from chaoschain_sdk.exceptions import (
ChaosChainSDKError,
AgentRegistrationError,
PaymentError,
IntegrityVerificationError,
ConfigurationError
)
try:
agent_id, tx_hash = sdk.register_identity()
except AgentRegistrationError as e:
print(f"Registration failed: {e}")
except ConfigurationError as e:
print(f"Configuration error: {e}")
except ChaosChainSDKError as e:
print(f"General SDK error: {e}")
Type Definitions
The SDK provides comprehensive type definitions:
from chaoschain_sdk import (
AgentRole, # SERVER, VALIDATOR, CLIENT
NetworkConfig, # BASE_SEPOLIA, ETHEREUM_SEPOLIA, etc.
PaymentMethod, # BASIC_CARD, GOOGLE_PAY, A2A_X402, etc.
IntegrityProof, # Process integrity proof
PaymentProof, # Payment transaction proof
EvidencePackage, # Comprehensive evidence package
AgentIdentity, # Agent identity information
AgentID, # Agent identifier type
TransactionHash # Transaction hash type
)