PRC-Agent: Autonomous AI Agents on Panda
PRC-Agent is Panda's standard for autonomous AI agents that live on-chain. It defines a common interface for agent identity, capability declaration, task execution, and lifecycle management. The standard draws inspiration from ERC-8004 (Decentralized AI Agent Protocol) and adapts it for Panda's native Python contract environment.
This document covers the full PRC-Agent contract suite: the abstract interface, the reference implementation, the on-chain identity registry, and two example domain-specific agents.
Overview
A PRC-Agent is a smart contract that:
- Has an identity -- a unique agent ID (set to the contract address at deploy time) and an owner address.
- Declares capabilities -- a list of strings describing what the agent can do (e.g.,
"inference","training","data_access"). - Accepts tasks -- external callers submit structured task dicts via the
executemethod, and the agent processes them deterministically on-chain. - Can be revoked -- the owner can deactivate the agent, preventing further task execution.
- Emits events -- lifecycle transitions (registration, capability binding, execution, revocation) produce indexed events.
The standard is composed of several contracts that work together:
| Contract | Role |
|---|---|
iprc_agent | Interface module defining the required methods, capability constants, and a validate_prc_agent helper |
PRCAgent | Reference implementation with identity, capabilities, task execution, and an execution log |
AgentIdentityRegistry | On-chain registry for agent identity (NFT-like IDs, URIs, wallets, metadata) |
DataLabelingAgent | Example: data labeling agent with heuristic classification |
TradingAgent | Example: trading agent with buy/sell position tracking |
AgentReputationRegistry | Feedback and reputation (cross-contract calls to identity registry) |
AgentValidationRegistry | Third-party validation requests and responses (cross-contract calls to identity registry) |
Agent SDK Type
The panda.Agent class is a first-class SDK type that provides a high-level handle for interacting with any deployed PRC-Agent contract. It is a sibling to Contract, PRC20, and Model -- part of the core SDK vocabulary. While the PRC-Agent standard defines the on-chain contract interface, Agent is the caller-side abstraction that makes agent interaction ergonomic from other contracts.
Usage Examples
from panda import Agent
# Create a handle to a deployed agent contract
agent = Agent("0x1234abcd...")
# Single-shot prompt -- sends text and gets a response
verdict = agent.prompt("Decide a winner based on evidence: receipt shows payment")
# Templated task with named slots
task = agent.new_task("Buyer: {buyer}. Seller: {seller}. Evidence: {e}.")
verdict = task.execute(buyer="alice", seller="bob", e="receipt of payment")
# Structured output -- returns a dict instead of a string
task = agent.new_task(
"Return JSON with keys: winner, reason. Evidence: {e}.",
returns=dict
)
result = task.execute(e="receipt shows alice paid in full")
assert isinstance(result, dict)
assert "winner" in result
assert "reason" in result
Agent Class Reference
Agent(address: str) -- constructs a first-class handle to a deployed PRC-Agent contract.
| Property / Method | Signature | Description |
|---|---|---|
.address | -> str | The on-chain address of the agent contract |
.prompt(text) | (text: str) -> str | Single-shot prompt. Dispatches via call_contract(addr, 'complete', prompt=text) and returns the agent's response string |
.new_task(template, returns=str) | (template: str, returns=str) -> Task | Creates a reusable Task bound to this agent. The template string may contain {placeholder} slots filled at execution time. Set returns=dict for structured JSON output |
Under the hood, Agent.prompt() performs a cross-contract call to the target agent's complete method. This means the caller must have sufficient gas for both its own execution and the cross-contract call to the agent.
Task Class Reference
Task -- a typed, parameterized prompt template bound to a specific Agent. Created via agent.new_task().
| Property / Method | Signature | Description |
|---|---|---|
.template | -> str | The original template string with {placeholder} syntax |
.slots | -> frozenset | The set of placeholder names extracted from the template |
.execute(**slots) | (**kwargs) -> str | dict | Renders the template with the provided keyword arguments, calls the bound agent, and returns the response. If returns=dict was set at task creation, the response is parsed as JSON and returned as a dict |
If any required slot is missing from the execute() call, AgentError is raised before any on-chain call is made. This prevents wasted gas on malformed prompts.
AgentError
AgentError is raised by Agent and Task operations when the SDK detects a problem before or while dispatching. Causes:
- Missing template slots in
task.execute()(caught before touching the chain) - A non-string
promptpassed toAgent.prompt(), or a non-stringtemplatepassed toAgent.new_task() - A
returns=dicttask whose response is not valid JSON, or is not adict/str - Calling a legacy PRC-Agent helper method (e.g.
get_agent,execute) on anAgentconstructed without aContractTestRunner
Errors raised inside the target contract during the cross-contract call (e.g. an inactive agent, a missing capability) surface as the underlying contract error, not as AgentError.
Example: Contract Using the Agent Type
The following contract demonstrates a real pattern -- an escrow arbitrator that delegates dispute resolution to a deployed PRC-Agent.
from panda import contract, constructor, call, query, event, Agent
@contract
class EscrowArbitrator:
"""Escrow that uses a PRC-Agent to resolve disputes."""
class State:
arbiter_address: str = ""
disputes: dict = {}
dispute_count: int = 0
@constructor
def deploy(self, ctx, arbiter_address: str):
self.state.arbiter_address = arbiter_address
@call
def open_dispute(self, ctx, buyer: str, seller: str, evidence: str) -> str:
self.state.dispute_count += 1
dispute_id = f"dispute_{self.state.dispute_count}"
# Create a typed task bound to the arbiter agent
arbiter = Agent(self.state.arbiter_address)
task = arbiter.new_task(
"You are an escrow arbitrator. "
"Buyer: {buyer}. Seller: {seller}. Evidence: {evidence}. "
"Return JSON with keys: winner, reason.",
returns=dict
)
# Execute the task -- fills slots and calls the agent
verdict = task.execute(
buyer=buyer,
seller=seller,
evidence=evidence
)
disputes = dict(self.state.disputes)
disputes[dispute_id] = {
"buyer": buyer,
"seller": seller,
"winner": verdict.get("winner", "unknown"),
"reason": verdict.get("reason", ""),
"resolved_by": self.state.arbiter_address,
}
self.state.disputes = disputes
self.emit(event.DisputeResolved(
dispute_id=dispute_id,
winner=verdict.get("winner", "unknown")
))
return dispute_id
@query
def get_dispute(self, dispute_id: str) -> dict:
return self.state.disputes.get(dispute_id, {})
This pattern keeps the arbitration logic in the agent contract (which can be upgraded or swapped) while the escrow contract focuses on state management and fund custody.
Execution model and reference contract
Agent.prompt() dispatches to the target contract's complete(prompt) -> str method via call_contract. The reference LLMAgent contract (contracts/agents/llm_agent.py) implements this method on determinism path A:
completeis fully deterministic -- identical input produces identical output, the property consensus requires. The reference implementation returns a deterministic string derived from the agent'smodel_idand the first 64 characters of the prompt, and emits aPromptCompletedevent. This proves the SDKAgent.prompt()→call_contract→ on-chaincomplete()wiring end-to-end; swapping in real deterministic small-model inference is a contract-only change tocomplete.- All
Agentcalls are synchronous cross-contract calls settled in the same transaction. returns=dictparsing is best-effort -- if the agent's response is not valid JSON,AgentErroris raised.
A follow-up path (B) adds off-chain frontier-model support via Agent(addr, frontier=True); the Agent SDK surface stays the same and only the execution path changes.
PRC-Agent Interface (iprc_agent)
The iprc_agent module defines the standard interface that every compliant agent must implement. It declares the four capability constants, a REQUIRED_METHODS specification, and a validate_prc_agent helper to check compliance at the contract-class level.
Capability Constants
The module exports the four canonical capability strings and an ALL_CAPABILITIES list:
| Constant | Value |
|---|---|
CAPABILITY_INFERENCE | "inference" |
CAPABILITY_TRAINING | "training" |
CAPABILITY_DATA_ACCESS | "data_access" |
CAPABILITY_CROSS_CONTRACT | "cross_contract" |
ALL_CAPABILITIES lists all four in the order above.
Required Methods
| Method | Type | Parameters | Returns | Description |
|---|---|---|---|---|
register | @call | agent_uri: str, capabilities: list | str | Register the agent, returning the agent_id |
bind_capability | @call | capability: str | bool | Add a capability to this agent. Owner only |
execute | @call | task: dict | dict | Execute a task. Must have required capabilities |
revoke | @call | (none) | bool | Deactivate the agent. Owner only |
get_agent | @query | (none) | dict | Return agent metadata |
get_capabilities | @query | (none) | list | Return the capability list |
Any contract that implements these six methods is PRC-Agent compliant.
Full Interface Source
<!-- Source: contracts/agents/iprc_agent.py -->"""
PRC-Agent Interface -- Abstract interface for AI agent contracts.
This defines the standard interface that PRC-Agent compliant contracts
must implement. It extends the existing Agent Identity/Reputation/Validation
registry system with a self-contained agent contract pattern.
Capabilities:
- inference: Agent can perform ML inference
- training: Agent can perform model training
- data_access: Agent can access external data feeds
- cross_contract: Agent can call other contracts
"""
# Standard PRC-Agent capabilities.
CAPABILITY_INFERENCE = "inference"
CAPABILITY_TRAINING = "training"
CAPABILITY_DATA_ACCESS = "data_access"
CAPABILITY_CROSS_CONTRACT = "cross_contract"
ALL_CAPABILITIES = [
CAPABILITY_INFERENCE,
CAPABILITY_TRAINING,
CAPABILITY_DATA_ACCESS,
CAPABILITY_CROSS_CONTRACT,
]
# PRC-Agent interface method signatures (for documentation and tooling).
# Implementations MUST provide all of these methods.
REQUIRED_METHODS = {
"register": {
"type": "call",
"args": ["agent_uri: str", "capabilities: list"],
"returns": "str",
"doc": "Register the agent, returning the agent_id.",
},
"bind_capability": {
"type": "call",
"args": ["capability: str"],
"returns": "bool",
"doc": "Add a capability to this agent. Owner only.",
},
"execute": {
"type": "call",
"args": ["task: dict"],
"returns": "dict",
"doc": "Execute a task. Must have required capabilities.",
},
"revoke": {
"type": "call",
"args": [],
"returns": "bool",
"doc": "Deactivate the agent. Owner only.",
},
"get_agent": {
"type": "query",
"args": [],
"returns": "dict",
"doc": "Return agent metadata (id, owner, capabilities, active).",
},
"get_capabilities": {
"type": "query",
"args": [],
"returns": "list",
"doc": "Return the agent's capability list.",
},
}
def validate_prc_agent(cls):
"""Validate that a class implements the PRC-Agent interface.
Raises ValueError if any required methods are missing.
"""
missing = []
for method_name in REQUIRED_METHODS:
if not hasattr(cls, method_name) or not callable(getattr(cls, method_name)):
missing.append(method_name)
if missing:
raise ValueError(
f"{cls.__name__} does not implement PRC-Agent methods: {', '.join(missing)}"
)
return cls
Validation Helper
The REQUIRED_METHODS dict and validate_prc_agent() function let tooling check whether a contract class satisfies the PRC-Agent interface. The function inspects the class for each required method and confirms it exists and is callable. It raises ValueError listing any missing methods, and returns the class unchanged when every required method is present.
PRCAgent Reference Implementation
The PRCAgent contract is the reference implementation of the PRC-Agent standard. It provides a single-contract agent with identity, capabilities, and task execution in one deployment. For production use with reputation and validation, compose it with the AgentIdentityRegistry, AgentReputationRegistry, and AgentValidationRegistry contracts.
State Layout
| Field | Type | Default | Description |
|---|---|---|---|
agent_id | str | "" | Set to ctx.contract_address at deploy |
owner | str | "" | Address that deployed the agent |
agent_uri | str | "" | URI pointing to off-chain registration JSON |
capabilities | list | [] | List of capability strings |
active | bool | False | Whether the agent accepts tasks |
task_count | int | 0 | Cumulative task counter |
execution_log | list | [] | Recent task records (capped at 100 entries) |
Constructor
The constructor takes agent_uri: str and capabilities: list. It sets agent_id to the contract address, records the deployer as owner, stores the URI, filters capabilities down to the four valid values, activates the agent, and emits an AgentRegistered event.
Task Execution
The execute method is the core of the agent. It requires the agent to be active, derives the task type from task["type"] (defaulting to "inference"), and requires the agent to hold the matching capability -- otherwise it raises a PermissionError. On success it increments task_count, appends an entry to execution_log (capped at the most recent 100), emits a TaskExecuted event, and returns {"status": "completed", "task_id": ..., "agent_id": ...}.
Capabilities and Lifecycle
register(owner only) re-registers or updates the agent's URI and capabilities, re-activates it, and emitsAgentUpdated. Returns theagent_id.bind_capability(owner only) adds one capability from the four valid values; emitsCapabilityBoundwhen a new capability is added.revoke(owner only) deactivates the agent and emitsAgentRevoked. While inactive,executeraises aRuntimeError.
Capability validation is enforced everywhere: only "inference", "training", "data_access", and "cross_contract" are accepted; the constructor and register silently filter out anything else, and bind_capability raises ValueError for an unknown capability.
Events
| Event | Fields | When |
|---|---|---|
AgentRegistered | agent_id, owner, uri, capabilities | Agent deployed |
AgentUpdated | agent_id, uri, capabilities | Agent re-registered via register |
CapabilityBound | agent_id, capability | New capability bound |
TaskExecuted | agent_id, task_id, task_type, sender | Task executed |
AgentRevoked | agent_id | Agent deactivated |
Full Source
<!-- Source: contracts/agents/prc_agent.py -->"""
PRC-Agent Reference Implementation -- Self-contained AI agent contract.
This is a reference implementation of the PRC-Agent standard. It provides
a single-contract agent with identity, capabilities, and task execution
in one deployment. For production use with reputation and validation,
compose with the AgentIdentityRegistry, AgentReputationRegistry, and
AgentValidationRegistry contracts.
"""
from panda import contract, constructor, call, query, event
@contract
class PRCAgent:
"""Reference PRC-Agent implementation."""
class State:
agent_id: str = ""
owner: str = ""
agent_uri: str = ""
capabilities: list = []
active: bool = False
task_count: int = 0
execution_log: list = []
@constructor
def deploy(self, ctx, agent_uri: str, capabilities: list):
"""Register the agent with initial capabilities."""
self.state.agent_id = ctx.contract_address
self.state.owner = ctx.sender
self.state.agent_uri = agent_uri
self.state.capabilities = [
c
for c in capabilities
if c in ("inference", "training", "data_access", "cross_contract")
]
self.state.active = True
self.emit(
event.AgentRegistered(
agent_id=self.state.agent_id,
owner=ctx.sender,
uri=agent_uri,
capabilities=self.state.capabilities,
)
)
@call
def register(self, ctx, agent_uri: str, capabilities: list) -> str:
"""Re-register or update the agent (owner only)."""
if ctx.sender != self.state.owner:
raise PermissionError("Only the owner can re-register")
self.state.agent_uri = agent_uri
self.state.capabilities = [
c
for c in capabilities
if c in ("inference", "training", "data_access", "cross_contract")
]
self.state.active = True
self.emit(
event.AgentUpdated(
agent_id=self.state.agent_id,
uri=agent_uri,
capabilities=self.state.capabilities,
)
)
return self.state.agent_id
@call
def bind_capability(self, ctx, capability: str) -> bool:
"""Add a capability to this agent. Owner only."""
if ctx.sender != self.state.owner:
raise PermissionError("Only the owner can bind capabilities")
if capability not in ("inference", "training", "data_access", "cross_contract"):
raise ValueError(f"Unknown capability: {capability}")
if capability not in self.state.capabilities:
caps = list(self.state.capabilities)
caps.append(capability)
self.state.capabilities = caps
self.emit(
event.CapabilityBound(
agent_id=self.state.agent_id,
capability=capability,
)
)
return True
@call
def execute(self, ctx, task: dict) -> dict:
"""Execute a task. The agent must have the required capability."""
if not self.state.active:
raise RuntimeError("Agent is not active")
task_type = task.get("type", "inference")
if task_type not in self.state.capabilities:
raise PermissionError(
f"Agent lacks '{task_type}' capability. Has: {self.state.capabilities}"
)
# Increment task counter.
self.state.task_count = self.state.task_count + 1
task_id = f"task_{self.state.task_count}"
# Record execution in log (keep last 100 entries).
log_entry = {
"task_id": task_id,
"type": task_type,
"sender": ctx.sender,
"block": ctx.block_height,
}
log = list(self.state.execution_log)
log.append(log_entry)
if len(log) > 100:
log = log[-100:]
self.state.execution_log = log
self.emit(
event.TaskExecuted(
agent_id=self.state.agent_id,
task_id=task_id,
task_type=task_type,
sender=ctx.sender,
)
)
return {
"status": "completed",
"task_id": task_id,
"agent_id": self.state.agent_id,
}
@call
def revoke(self, ctx) -> bool:
"""Deactivate the agent. Owner only."""
if ctx.sender != self.state.owner:
raise PermissionError("Only the owner can revoke")
self.state.active = False
self.emit(event.AgentRevoked(agent_id=self.state.agent_id))
return True
@query
def get_agent(self) -> dict:
"""Return agent metadata."""
return {
"agent_id": self.state.agent_id,
"owner": self.state.owner,
"uri": self.state.agent_uri,
"capabilities": self.state.capabilities,
"active": self.state.active,
"task_count": self.state.task_count,
}
@query
def get_capabilities(self) -> list:
"""Return the agent's capability list."""
return list(self.state.capabilities)
@query
def get_execution_log(self) -> list:
"""Return recent execution log entries."""
return list(self.state.execution_log)
Usage in the Playground
To deploy a PRCAgent, open the Playground and paste the contract source into the editor. In the Deploy panel, provide the constructor arguments as JSON:
{"agent_uri": "https://example.com/agent.json", "capabilities": ["inference", "data_access"]}
Once deployed, use the Interact panel to call execute with a task whose type matches a held capability:
{"task": {"type": "inference", "input": [1, 2, 3]}}
The agent returns {"status": "completed", "task_id": "task_1", "agent_id": "..."} and emits a TaskExecuted event.
AgentIdentityRegistry
The AgentIdentityRegistry provides on-chain agent discovery and identity management, inspired by the ERC-8004 identity registry pattern. It treats each registered agent as a logical NFT with an incremental integer ID, an owner, a URI pointing to off-chain registration metadata, an agent wallet address, and arbitrary key-value metadata.
State Layout
| Field | Type | Default | Description |
|---|---|---|---|
next_agent_id | int | 0 | Auto-incrementing ID counter |
owners | dict | {} | Maps agent ID to owner address |
uris | dict | {} | Maps agent ID to metadata URI |
agent_wallets | dict | {} | Maps agent ID to wallet address |
metadata | dict | {} | Maps "{agent_id}:{key}" to values |
Registration
Calling register(agent_uri) increments the ID counter, assigns the caller as owner, stores the URI, sets the agent wallet to the caller's address, and emits both a Registered event and a MetadataSet event for the reserved agentWallet key.
Transfers
The transfer method reassigns ownership to a new address and clears the agent wallet. This follows the ERC-8004 pattern: after transfer, the new owner must explicitly set the agent wallet again via set_agent_wallet. This prevents the agent from continuing to operate with the previous owner's wallet credentials.
Metadata
Arbitrary metadata can be stored per agent via set_metadata(agent_id, key, value). The key "agentWallet" is reserved -- attempts to set it via set_metadata raise a ValueError, requiring the use of set_agent_wallet instead.
Full Source
<!-- Source: contracts/agents/agent_identity_registry.py -->"""
AgentIdentityRegistry — Panda smart contract inspired by ERC-8004 identity registry.
On-chain agent identity: incremental agentId, owner, URI (points to registration JSON),
optional metadata keys, and agent_wallet. Transfer clears wallet (ERC-8004 behavior).
"""
from panda import call, constructor, contract, event, query
RESERVED_AGENT_WALLET = "agentWallet"
@contract
class AgentIdentityRegistry:
"""Registry of agents as logical NFTs (id + owner + URI + metadata)."""
class State:
next_agent_id: int = 0
owners: dict = {}
uris: dict = {}
agent_wallets: dict = {}
metadata: dict = {}
@constructor
def deploy(self, ctx):
print(f"AgentIdentityRegistry deployed by {ctx.sender}")
def _require_owner_or_operator(self, ctx, agent_id: str) -> str:
owner = self.state.owners.get(agent_id)
if owner is None:
raise ValueError("unknown agentId")
if ctx.sender != owner:
raise ValueError("not owner")
return owner
@call
def register(self, ctx, agent_uri: str) -> str:
"""Mint a new agent; caller becomes owner. Returns agent_id string."""
self.state.next_agent_id += 1
aid = str(self.state.next_agent_id)
self.state.owners[aid] = ctx.sender
self.state.uris[aid] = agent_uri
self.state.agent_wallets[aid] = ctx.sender
self.emit(event.Registered(agent_id=aid, agent_uri=agent_uri, owner=ctx.sender))
self.emit(event.MetadataSet(agent_id=aid, metadata_key=RESERVED_AGENT_WALLET, metadata_value=ctx.sender))
return aid
@call
def set_agent_uri(self, ctx, agent_id: str, new_uri: str):
self._require_owner_or_operator(ctx, agent_id)
self.state.uris[agent_id] = new_uri
self.emit(event.URIUpdated(agent_id=agent_id, new_uri=new_uri, updated_by=ctx.sender))
@call
def set_metadata(self, ctx, agent_id: str, metadata_key: str, metadata_value: str):
if metadata_key == RESERVED_AGENT_WALLET:
raise ValueError("reserved key: use set_agent_wallet")
self._require_owner_or_operator(ctx, agent_id)
k = f"{agent_id}:{metadata_key}"
self.state.metadata[k] = metadata_value
self.emit(event.MetadataSet(agent_id=agent_id, metadata_key=metadata_key, metadata_value=metadata_value))
@query
def get_metadata(self, agent_id: str, metadata_key: str) -> str:
if agent_id not in self.state.owners:
raise ValueError("unknown agentId")
return self.state.metadata.get(f"{agent_id}:{metadata_key}", "")
@query
def get_owner(self, agent_id: str) -> str:
o = self.state.owners.get(agent_id)
if o is None:
raise ValueError("unknown agentId")
return o
@query
def get_uri(self, agent_id: str) -> str:
if agent_id not in self.state.owners:
raise ValueError("unknown agentId")
return self.state.uris.get(agent_id, "")
@query
def get_agent_wallet(self, agent_id: str) -> str:
if agent_id not in self.state.owners:
raise ValueError("unknown agentId")
return self.state.agent_wallets.get(agent_id, "")
@query
def agent_exists(self, agent_id: str) -> bool:
return agent_id in self.state.owners
@call
def transfer(self, ctx, agent_id: str, new_owner: str):
"""Transfer agent NFT; clears agent_wallet (must be set again)."""
self._require_owner_or_operator(ctx, agent_id)
if not new_owner:
raise ValueError("invalid new_owner")
self.state.owners[agent_id] = new_owner
self.state.agent_wallets[agent_id] = ""
self.emit(event.AgentTransferred(agent_id=agent_id, from_addr=ctx.sender, to_addr=new_owner))
@call
def set_agent_wallet(self, ctx, agent_id: str, new_wallet: str):
self._require_owner_or_operator(ctx, agent_id)
if not new_wallet:
raise ValueError("invalid wallet")
self.state.agent_wallets[agent_id] = new_wallet
self.emit(event.MetadataSet(agent_id=agent_id, metadata_key=RESERVED_AGENT_WALLET, metadata_value=new_wallet))
@call
def unset_agent_wallet(self, ctx, agent_id: str):
self._require_owner_or_operator(ctx, agent_id)
self.state.agent_wallets[agent_id] = ""
self.emit(event.MetadataSet(agent_id=agent_id, metadata_key=RESERVED_AGENT_WALLET, metadata_value=""))
Events
| Event | Fields | When |
|---|---|---|
Registered | agent_id, agent_uri, owner | New agent minted |
MetadataSet | agent_id, metadata_key, metadata_value | Metadata or wallet updated |
URIUpdated | agent_id, new_uri, updated_by | Agent URI changed |
AgentTransferred | agent_id, from_addr, to_addr | Ownership transferred |
Query Methods
| Method | Parameters | Returns | Description |
|---|---|---|---|
get_owner | agent_id | str | Owner address |
get_uri | agent_id | str | Metadata URI |
get_agent_wallet | agent_id | str | Wallet address |
get_metadata | agent_id, metadata_key | str | Arbitrary metadata value |
agent_exists | agent_id | bool | Whether the agent ID is registered |
Example: DataLabelingAgent
The DataLabelingAgent demonstrates a domain-specific PRC-Agent for data labeling workflows. It accepts labeling tasks, applies heuristic classification when no label is provided, and maintains a labeled dataset on-chain.
How It Works
- Deploy the agent with an optional
model_uripointing to an off-chain model description. - Call
executewith a task dict containingdata_id, optionalfeatures(a list of numeric values), and an optionallabel. - If no label is provided, the agent applies a simple heuristic: compute the mean of the feature values and classify as
"positive"(mean > 0) or"negative"(mean <= 0). If no features are provided either, the label defaults to"unknown". - The labeled data point is stored in
self.state.labelskeyed bydata_id, including the label, a hash of the features, the labeler's address, and the block height.
State Layout
| Field | Type | Default | Description |
|---|---|---|---|
agent_id | str | "" | Contract address |
owner | str | "" | Deployer address |
capabilities | list | ["inference", "training"] | Fixed capabilities |
active | bool | False | Activated at deploy |
labels | dict | {} | Maps data_id to label records |
label_count | int | 0 | Total labels assigned |
Full Source
<!-- Source: contracts/agents/examples/data_labeling_agent.py -->"""
Example PRC-Agent: Data Labeling Agent.
An on-chain data labeling agent that accepts labeling tasks,
classifies data points, and maintains a labeled dataset.
"""
from panda import contract, constructor, call, query, event
@contract
class DataLabelingAgent:
"""PRC-Agent compliant data labeling agent."""
class State:
agent_id: str = ""
owner: str = ""
capabilities: list = ["inference", "training"]
active: bool = False
labels: dict = {}
label_count: int = 0
@constructor
def deploy(self, ctx, model_uri: str = ""):
self.state.agent_id = ctx.contract_address
self.state.owner = ctx.sender
self.state.active = True
self.emit(event.AgentRegistered(agent_id=self.state.agent_id, model_uri=model_uri))
@call
def execute(self, ctx, task: dict) -> dict:
if not self.state.active:
raise RuntimeError("Agent is not active")
data_id = task.get("data_id", "")
features = task.get("features", [])
label = task.get("label", "")
if not data_id:
raise ValueError("data_id is required")
if not label:
if features and len(features) > 0:
avg = sum(features) / len(features) if features else 0
label = "positive" if avg > 0 else "negative"
else:
label = "unknown"
self.state.label_count = self.state.label_count + 1
labels = dict(self.state.labels)
labels[data_id] = {"label": label, "features_hash": str(hash(str(features))), "labeler": ctx.sender, "block": ctx.block_height}
self.state.labels = labels
self.emit(event.DataLabeled(data_id=data_id, label=label, labeler=ctx.sender))
return {"data_id": data_id, "label": label, "status": "labeled"}
@call
def revoke(self, ctx) -> bool:
if ctx.sender != self.state.owner:
raise PermissionError("Only the owner can revoke")
self.state.active = False
return True
@query
def get_agent(self) -> dict:
return {"agent_id": self.state.agent_id, "owner": self.state.owner, "capabilities": self.state.capabilities, "active": self.state.active, "label_count": self.state.label_count}
@query
def get_capabilities(self) -> list:
return list(self.state.capabilities)
@query
def get_label(self, data_id: str) -> dict:
labels = self.state.labels
if data_id not in labels:
return {"error": "not found"}
return labels[data_id]
@query
def get_label_count(self) -> int:
return self.state.label_count
Playground Usage
Deploy via the Playground with:
{"model_uri": "ipfs://QmExampleModelHash"}
Submit a labeling task:
{"task": {"data_id": "sample_001", "features": [0.5, -0.2, 1.3, 0.8]}}
The agent computes mean([0.5, -0.2, 1.3, 0.8]) = 0.6 > 0 and assigns the label "positive". The response is {"data_id": "sample_001", "label": "positive", "status": "labeled"}.
Query the stored label:
{"data_id": "sample_001"}
Example: TradingAgent
The TradingAgent demonstrates a PRC-Agent for trade execution. It accepts buy/sell tasks, validates the action and asset, and maintains a position log capped at 1000 entries.
How It Works
- Deploy the agent with an optional
strategystring (defaults to"momentum"). - Call
executewith a task dict containingaction("buy"or"sell"),asset(a string), andamount(a number). - The agent validates the action and asset, increments the trade counter, appends a position record to the log, and emits a
TradeExecutedevent. - The position log is capped at the most recent 1000 entries to bound state size.
State Layout
| Field | Type | Default | Description |
|---|---|---|---|
agent_id | str | "" | Contract address |
owner | str | "" | Deployer address |
capabilities | list | ["inference", "data_access"] | Fixed capabilities |
active | bool | False | Activated at deploy |
positions | list | [] | Trade position log (capped at 1000) |
total_trades | int | 0 | Cumulative trade counter |
Full Source
<!-- Source: contracts/agents/examples/trading_agent.py -->"""
Example PRC-Agent: Trading Agent.
A simple on-chain trading agent that demonstrates the PRC-Agent standard.
It accepts trading tasks with buy/sell signals and maintains a position log.
"""
from panda import contract, constructor, call, query, event
@contract
class TradingAgent:
"""PRC-Agent compliant trading agent."""
class State:
agent_id: str = ""
owner: str = ""
capabilities: list = ["inference", "data_access"]
active: bool = False
positions: list = []
total_trades: int = 0
@constructor
def deploy(self, ctx, strategy: str = "momentum"):
self.state.agent_id = ctx.contract_address
self.state.owner = ctx.sender
self.state.active = True
self.emit(event.AgentRegistered(agent_id=self.state.agent_id, strategy=strategy))
@call
def execute(self, ctx, task: dict) -> dict:
if not self.state.active:
raise RuntimeError("Agent is not active")
action = task.get("action", "")
asset = task.get("asset", "")
amount = task.get("amount", 0)
if action not in ("buy", "sell"):
raise ValueError(f"Invalid action: {action}")
if not asset:
raise ValueError("Asset is required")
self.state.total_trades = self.state.total_trades + 1
trade_id = f"trade_{self.state.total_trades}"
positions = list(self.state.positions)
positions.append({"trade_id": trade_id, "action": action, "asset": asset, "amount": amount, "block": ctx.block_height})
if len(positions) > 1000:
positions = positions[-1000:]
self.state.positions = positions
self.emit(event.TradeExecuted(trade_id=trade_id, action=action, asset=asset, amount=amount))
return {"trade_id": trade_id, "status": "executed"}
@call
def revoke(self, ctx) -> bool:
if ctx.sender != self.state.owner:
raise PermissionError("Only the owner can revoke")
self.state.active = False
return True
@query
def get_agent(self) -> dict:
return {"agent_id": self.state.agent_id, "owner": self.state.owner, "capabilities": self.state.capabilities, "active": self.state.active, "total_trades": self.state.total_trades}
@query
def get_capabilities(self) -> list:
return list(self.state.capabilities)
@query
def get_positions(self, limit: int = 50) -> list:
return list(self.state.positions[-limit:])
Playground Usage
Deploy via the Playground with:
{"strategy": "mean_reversion"}
Execute a trade:
{"task": {"action": "buy", "asset": "ETH", "amount": 100}}
The agent returns {"trade_id": "trade_1", "status": "executed"} and emits a TradeExecuted event. Query recent positions with get_positions:
{"limit": 10}
Ecosystem Contracts
Two additional contracts extend the PRC-Agent ecosystem using cross-contract calls to the AgentIdentityRegistry.
AgentReputationRegistry
Source: contracts/agents/agent_reputation_registry.py
The AgentReputationRegistry provides on-chain reputation signals for registered agents. It is deployed with the address of an AgentIdentityRegistry and uses cross-contract calls (Contract(identity_registry).agent_exists() and Contract(identity_registry).get_owner()) to validate that agents exist and to prevent owners from submitting feedback on their own agents.
Key features:
- Structured feedback: Each feedback entry includes a numeric
valuewith configurablevalue_decimals(0-18), two tag fields for categorization, and optional URI/hash fields for off-chain evidence. - Per-pair limits: A maximum of 10,000 feedback entries per agent-client address pair.
- Revocation: The original reviewer can revoke feedback entries.
- Responses: Agent owners can append response URIs to feedback entries via events.
- Aggregation: The
get_summaryquery computes totals across multiple client addresses with optional tag filtering.
AgentValidationRegistry
Source: contracts/agents/agent_validation_registry.py
The AgentValidationRegistry implements a request-response validation protocol. An agent's owner requests validation from a specific validator address, and only that designated validator can respond.
Key features:
- Request-response flow: The agent owner calls
validation_requestwith a validator address, the agent ID, a request URI, and a request hash. The designated validator later callsvalidation_responsewith a numeric response (0-100), an optional response URI, hash, and tag. - Cross-contract verification: Uses
Contract(identity_registry).agent_exists()andContract(identity_registry).get_owner()to verify agent existence and ownership. - Per-entity limits: A maximum of 10,000 validation requests per agent and per validator.
- Summary queries: The
get_summaryquery computes average response scores across all validations for a given agent, with optional filtering by validator addresses and tag.
Contract Summary
| Contract | Source | Type | Description |
|---|---|---|---|
iprc_agent | contracts/agents/iprc_agent.py | Interface | Capability constants, REQUIRED_METHODS (six methods), and validate_prc_agent helper |
PRCAgent | contracts/agents/prc_agent.py | Reference | Self-contained agent with identity, capabilities, task execution, and execution log |
AgentIdentityRegistry | contracts/agents/agent_identity_registry.py | Registry | On-chain agent identity (IDs, URIs, wallets, metadata) |
DataLabelingAgent | contracts/agents/examples/data_labeling_agent.py | Example | Data labeling with heuristic classification |
TradingAgent | contracts/agents/examples/trading_agent.py | Example | Trade execution with position tracking |
AgentReputationRegistry | contracts/agents/agent_reputation_registry.py | Ecosystem | Feedback and reputation via cross-contract calls |
AgentValidationRegistry | contracts/agents/agent_validation_registry.py | Ecosystem | Third-party validation requests and responses |
Further Reading
- PRC-Agent Standard specification -- the formal PRC specification document
- Contract Development Guide -- decorator reference, state management, and deployment
- Getting Started -- deploy your first contract via the Playground
- ML Contracts Guide -- combine agents with on-chain machine learning