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¶
Requirements
- Python 3.10+
web3.py>= 6.0eth-account>= 0.10.0
Initialization¶
The agent signs transactions locally; a relayer submits them and pays gas. The agent needs no ETH.
The agent pays gas from its own wallet. No relayer needed.
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.
Parameters:
| Parameter | Type | Description |
|---|---|---|
agent_id |
int |
The agent's registry ID |
Returns: Transaction result dict (gasless) or receipt (direct).
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.
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.
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.
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.
Read Operations¶
get_agent(agent_id)¶
Retrieve the full on-chain record for an agent (the "Handelsregisterauszug").
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).
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).
get_children(agent_id)¶
Get the list of child agent IDs for a given parent.
get_revenue_history(agent_id)¶
Get all revenue reports submitted by an agent.
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.
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) |
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¶
Output:
Get Agent Info¶
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¶
Get Revenue History¶
Get Compliance Status¶
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}")