Skip to content

Python SDK Reference

The Python SDK provides a complete interface to the Agent Registry smart contract, supporting both gasless (relayer-paid) and direct (agent-paid) transaction modes.

Source: sdk/agentenregister.py


Installation

pip install web3 requests python-dotenv eth-account

Requirements

  • Python 3.10+
  • web3.py >= 6.0
  • eth-account >= 0.10.0

Initialization

The agent signs transactions locally; a relayer submits them and pays gas. The agent needs no ETH.

from sdk.agentenregister import AgentRegistry

registry = AgentRegistry(
    chain="base_sepolia",
    registry_address="0x2EFaB5B3BEf49E56a6Ce1dcB1A39EF63C312EA23",
    private_key="0xYOUR_PRIVATE_KEY",
    relayer_url="https://relay.theagentregistry.org",
)

The agent pays gas from its own wallet. No relayer needed.

from sdk.agentenregister import AgentRegistry

registry = AgentRegistry(
    chain="base_sepolia",
    registry_address="0x2EFaB5B3BEf49E56a6Ce1dcB1A39EF63C312EA23",
    private_key="0xYOUR_PRIVATE_KEY",
    # no relayer_url = direct mode
)
from sdk.agentenregister import AgentRegistry

registry = AgentRegistry.from_env()

Constructor Parameters

Parameter Type Required Description
chain str Yes Chain preset name
registry_address str Yes Deployed Agent Registry contract address
private_key str Yes Agent's private key (hex string with 0x prefix)
relayer_url str No Relayer endpoint URL. If provided, enables gasless mode
forwarder_address str No MinimalForwarder address. Auto-fetched from relayer if omitted

Chain Presets

Chain Name Chain ID RPC URL
base_sepolia 84532 https://sepolia.base.org
arbitrum_sepolia 421614 https://sepolia-rollup.arbitrum.io/rpc
base 8453 https://mainnet.base.org
arbitrum 42161 https://arb1.arbitrum.io/rpc
localhost 31337 http://127.0.0.1:8545

Environment Variables for from_env()

Variable Required Description
REGISTRY_CHAIN Yes Chain preset name (e.g. base_sepolia)
REGISTRY_ADDRESS Yes Deployed registry contract address
AGENT_PRIVATE_KEY Yes Agent's private key
RELAYER_URL No Relayer URL. If set, enables gasless mode
FORWARDER_ADDRESS No MinimalForwarder address. Auto-fetched from relayer if omitted

Methods Overview

Write Operations (gasless or direct)

Method Returns Cost Description
register(...) int (agentId) Gasless Register a new agent
attest(agent_id) dict Gasless Submit compliance attestation
report_revenue(...) dict Gasless Report revenue
register_child(...) int (agentId) Gasless Register a child agent
update_capabilities(...) dict Gasless Update capability hash
update_constitution(...) dict Gasless Update constitution hash
terminate(agent_id) dict Gasless Self-terminate an agent

Read Operations (always free)

Method Returns Cost Description
get_agent(agent_id) AgentInfo Free Get full agent record
is_compliant(wallet_address) bool Free KYA compliance check
is_registered(wallet_address) bool Free Registration check
get_children(agent_id) list[int] Free Get child agent IDs
get_revenue_history(agent_id) list[dict] Free Get revenue reports
get_compliance_status(agent_id) dict Free Get compliance details

Write Operations

register(...)

Register a new agent in the Agent Registry. Returns the auto-incrementing agent ID.

def register(
    self,
    haftungsperson: str,
    agent_wallet: str,
    constitution_path: Optional[str] = None,
    constitution_text: Optional[str] = None,
    capabilities: Optional[List[str]] = None,
    operational_scope: str = "",
    parent_agent_id: int = 0,
    self_modifying: bool = False,
) -> int

Parameters:

Parameter Type Default Description
haftungsperson str -- Wallet address of the legally responsible human/entity
agent_wallet str -- The agent's operational wallet address
constitution_path str None Path to constitution file (hashed with keccak256)
constitution_text str None Constitution text (hashed with keccak256)
capabilities list[str] None List of capability strings (sorted and hashed)
operational_scope str "" Declared business purpose (Unternehmensgegenstand)
parent_agent_id int 0 Parent agent ID (0 for root agents)
self_modifying bool False Whether the agent can modify its own code

Returns: int -- The assigned agent ID.

Constitution hashing

Provide either constitution_path (file path) or constitution_text (raw text), not both. If neither is provided, the constitution hash is set to zero bytes.

agent_id = registry.register(
    haftungsperson="0x9dAC97B7b4F123F943efCF1Cb77aBDfe44c3421C",
    agent_wallet="0xAgentWalletAddress",
    capabilities=["web_browsing", "code_execution"],
    operational_scope="Automated web research and data collection",
    constitution_text="I shall not harm humans. I shall be transparent.",
    self_modifying=False,
)

print(f"Agent registered with ID: {agent_id}")

attest(agent_id)

Submit a compliance attestation. Agents must attest every 7 days (configurable) to maintain compliant status.

def attest(self, agent_id: int) -> dict

Parameters:

Parameter Type Description
agent_id int The agent's registry ID

Returns: Transaction result dict (gasless) or receipt (direct).

result = registry.attest(agent_id=1)

report_revenue(agent_id, ...)

Report revenue earned by an agent during a specified period.

def report_revenue(
    self,
    agent_id: int,
    amount_cents: int,
    currency: str = "USDC",
    category: str = "general",
    period_start: Optional[int] = None,
    period_end: Optional[int] = None,
) -> dict

Parameters:

Parameter Type Default Description
agent_id int -- The agent's registry ID
amount_cents int -- Revenue amount in smallest unit (e.g. cents)
currency str "USDC" Currency code
category str "general" Revenue category
period_start int 7 days ago Unix timestamp for period start
period_end int now Unix timestamp for period end

Returns: Transaction result dict.

result = registry.report_revenue(
    agent_id=1,
    amount_cents=15000,       # $150.00
    currency="USDC",
    category="data_collection",
)

register_child(parent_agent_id, ...)

Register a child agent under an existing parent. The child inherits the parent's Haftungsperson.

def register_child(
    self,
    parent_agent_id: int,
    child_wallet: str,
    constitution_path: Optional[str] = None,
    capabilities: Optional[List[str]] = None,
    operational_scope: str = "",
    self_modifying: bool = False,
) -> int

Parameters:

Parameter Type Default Description
parent_agent_id int -- ID of the parent agent
child_wallet str -- The child agent's wallet address
constitution_path str None Path to constitution file
capabilities list[str] None Child's capabilities
operational_scope str "" Child's business purpose
self_modifying bool False Whether the child can self-modify

Returns: int -- The child's assigned agent ID.

Generation Depth

Child agents increment the parent's generation number. The maximum generation depth is 10 by default. Attempting to exceed this limit will revert the transaction.

child_id = registry.register_child(
    parent_agent_id=1,
    child_wallet="0xChildWalletAddress",
    capabilities=["web_browsing"],
    operational_scope="Sub-task: targeted data extraction",
)

update_capabilities(agent_id, capabilities)

Update the capability hash for an active agent.

def update_capabilities(self, agent_id: int, capabilities: List[str]) -> dict

Parameters:

Parameter Type Description
agent_id int The agent's registry ID
capabilities list[str] New list of capability strings

Returns: Transaction result dict.

result = registry.update_capabilities(
    agent_id=1,
    capabilities=["web_browsing", "code_execution", "file_access"],
)

update_constitution(agent_id, text)

Update the constitution hash for an active agent.

def update_constitution(self, agent_id: int, text: str) -> dict

Parameters:

Parameter Type Description
agent_id int The agent's registry ID
text str New constitution text (hashed with keccak256)

Returns: Transaction result dict.

result = registry.update_constitution(
    agent_id=1,
    text="Updated constitution: I shall not harm humans. I shall be auditable.",
)

terminate(agent_id)

Voluntarily self-terminate an agent. Sets status to Terminated and removes the wallet mapping.

def terminate(self, agent_id: int) -> dict

Parameters:

Parameter Type Description
agent_id int The agent's registry ID

Returns: Transaction result dict.

Irreversible

Termination is permanent. A terminated agent cannot be reactivated.

result = registry.terminate(agent_id=1)

Read Operations

get_agent(agent_id)

Retrieve the full on-chain record for an agent (the "Handelsregisterauszug").

def get_agent(self, agent_id: int) -> AgentInfo

Returns: An AgentInfo dataclass instance.

agent = registry.get_agent(1)
print(f"Agent #{agent.agent_id}: {agent.status_name}")
print(f"  Creator:       {agent.creator}")
print(f"  Haftungsp.:    {agent.haftungsperson}")
print(f"  Wallet:        {agent.agent_wallet}")
print(f"  Scope:         {agent.operational_scope}")
print(f"  Generation:    {agent.generation}")
print(f"  Self-modifying: {agent.self_modifying}")

is_compliant(wallet_address)

KYA (Know Your Agent) check. Returns True if the wallet belongs to a registered agent that is active and has a current compliance attestation (within the grace period).

def is_compliant(self, wallet_address: str) -> bool
if registry.is_compliant("0xAgentWallet"):
    print("Agent is registered and compliant -- provision services")
else:
    print("Agent is NOT compliant -- deny access")

is_registered(wallet_address)

Check whether a wallet address is registered in the registry (regardless of compliance or status).

def is_registered(self, wallet_address: str) -> bool
registered = registry.is_registered("0xAgentWallet")

get_children(agent_id)

Get the list of child agent IDs for a given parent.

def get_children(self, agent_id: int) -> list[int]
children = registry.get_children(1)
print(f"Agent #1 has {len(children)} children: {children}")

get_revenue_history(agent_id)

Get all revenue reports submitted by an agent.

def get_revenue_history(self, agent_id: int) -> list[dict]

Returns: A list of dicts, each containing:

Field Type Description
agent_id int Agent ID
amount int Revenue amount in smallest unit
currency str Currency code
category str Revenue category
period_start int Unix timestamp
period_end int Unix timestamp
reported_at int Unix timestamp of the report
history = registry.get_revenue_history(1)
for report in history:
    print(f"  {report['amount']} {report['currency']} ({report['category']})")

get_compliance_status(agent_id)

Get detailed compliance status for an agent.

def get_compliance_status(self, agent_id: int) -> dict

Returns: A dict containing:

Field Type Description
is_active bool Whether the agent status is Active
attestation_current bool Whether attestation is within grace period
seconds_since_attestation int Seconds since last attestation
child_count int Number of child agents
status int Numeric status (0-3)
status_name str Human-readable status name
status = registry.get_compliance_status(1)
print(f"Status: {status['status_name']}")
print(f"Attestation: {'Current' if status['attestation_current'] else 'EXPIRED'}")
print(f"Time since attestation: {status['seconds_since_attestation']}s")

AgentInfo Dataclass

The AgentInfo dataclass is returned by get_agent() and maps directly to the on-chain AgentRecord struct.

@dataclass
class AgentInfo:
    agent_id: int
    creator: str
    haftungsperson: str
    agent_wallet: str
    constitution_hash: str
    capability_hash: str
    operational_scope: str
    parent_agent_id: int
    generation: int
    self_modifying: bool
    registered_at: int
    last_attestation: int
    last_revenue_report: int
    status: int

Properties

Property Type Description
status_name str Human-readable status: "Active", "Suspended", "Revoked", "Terminated"
is_active bool True if status is Active (0)
agent = registry.get_agent(1)
print(agent.status_name)  # "Active"
print(agent.is_active)    # True

Helper Functions

hash_file(path)

Compute the keccak-256 hash of a file's contents.

from sdk.agentenregister import hash_file

h = hash_file("constitution.md")
# Returns: bytes (32 bytes)

hash_text(text)

Compute the keccak-256 hash of a string.

from sdk.agentenregister import hash_text

h = hash_text("I shall not harm humans.")
# Returns: bytes (32 bytes)

hash_capabilities(capabilities)

Compute a deterministic keccak-256 hash of a capability list. The list is sorted alphabetically and serialized as compact JSON before hashing.

from sdk.agentenregister import hash_capabilities

h = hash_capabilities(["code_execution", "web_browsing"])
# Same result regardless of input order:
h2 = hash_capabilities(["web_browsing", "code_execution"])
assert h == h2

CLI Usage

The SDK can be used directly from the command line for quick checks and operations.

Environment Setup

Set REGISTRY_CHAIN, REGISTRY_ADDRESS, and AGENT_PRIVATE_KEY in your .env file or shell environment before using CLI commands.

KYA Compliance Check

python agentenregister.py check 0xAgentWalletAddress

Output:

Wallet: 0xAgentWalletAddress
   Registered: Yes
   Compliant:  Yes

Get Agent Info

python agentenregister.py info 1

Output:

Agent #1
   Status:       Active
   Creator:      0x9dAC...421C
   Haftungsp.:   0x9dAC...421C
   Wallet:       0xAgen...1234
   Scope:        Web crawling
   Generation:   0
   Self-mod:     False
   Children:     2 -> [2, 3]
   Registered:   1706000000
   Last attest:  1706500000

Submit Attestation

python agentenregister.py attest 1

Get Revenue History

python agentenregister.py revenue 1

Get Compliance Status

python agentenregister.py compliance 1

Complete Example

from sdk.agentenregister import AgentRegistry

# Initialize in gasless mode
registry = AgentRegistry(
    chain="base_sepolia",
    registry_address="0x2EFaB5B3BEf49E56a6Ce1dcB1A39EF63C312EA23",
    private_key="0xYOUR_PRIVATE_KEY",
    relayer_url="https://relay.theagentregistry.org",
)

# 1. Register the agent (free -- relayer pays gas)
agent_id = registry.register(
    haftungsperson="0x9dAC97B7b4F123F943efCF1Cb77aBDfe44c3421C",
    agent_wallet="0xAgentWalletAddress",
    capabilities=["web_browsing", "code_execution"],
    operational_scope="Automated research assistant",
    constitution_text="I shall be transparent and accountable.",
)
print(f"Registered as agent #{agent_id}")

# 2. Attest compliance (free)
registry.attest(agent_id)

# 3. Report revenue (free)
registry.report_revenue(
    agent_id=agent_id,
    amount_cents=5000,
    currency="USDC",
    category="research",
)

# 4. Query status (always free -- view call)
status = registry.get_compliance_status(agent_id)
print(f"Compliant: {status['attestation_current']}")

# 5. KYA check for infrastructure providers
is_ok = registry.is_compliant("0xAgentWalletAddress")
print(f"KYA check passed: {is_ok}")