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 execute method, 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:

ContractRole
iprc_agentInterface module defining the required methods, capability constants, and a validate_prc_agent helper
PRCAgentReference implementation with identity, capabilities, task execution, and an execution log
AgentIdentityRegistryOn-chain registry for agent identity (NFT-like IDs, URIs, wallets, metadata)
DataLabelingAgentExample: data labeling agent with heuristic classification
TradingAgentExample: trading agent with buy/sell position tracking
AgentReputationRegistryFeedback and reputation (cross-contract calls to identity registry)
AgentValidationRegistryThird-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 / MethodSignatureDescription
.address-> strThe on-chain address of the agent contract
.prompt(text)(text: str) -> strSingle-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) -> TaskCreates 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 / MethodSignatureDescription
.template-> strThe original template string with {placeholder} syntax
.slots-> frozensetThe set of placeholder names extracted from the template
.execute(**slots)(**kwargs) -> str | dictRenders 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 prompt passed to Agent.prompt(), or a non-string template passed to Agent.new_task()
  • A returns=dict task whose response is not valid JSON, or is not a dict/str
  • Calling a legacy PRC-Agent helper method (e.g. get_agent, execute) on an Agent constructed without a ContractTestRunner

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:

  • complete is fully deterministic -- identical input produces identical output, the property consensus requires. The reference implementation returns a deterministic string derived from the agent's model_id and the first 64 characters of the prompt, and emits a PromptCompleted event. This proves the SDK Agent.prompt()call_contract → on-chain complete() wiring end-to-end; swapping in real deterministic small-model inference is a contract-only change to complete.
  • All Agent calls are synchronous cross-contract calls settled in the same transaction.
  • returns=dict parsing is best-effort -- if the agent's response is not valid JSON, AgentError is 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:

ConstantValue
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

MethodTypeParametersReturnsDescription
register@callagent_uri: str, capabilities: liststrRegister the agent, returning the agent_id
bind_capability@callcapability: strboolAdd a capability to this agent. Owner only
execute@calltask: dictdictExecute a task. Must have required capabilities
revoke@call(none)boolDeactivate the agent. Owner only
get_agent@query(none)dictReturn agent metadata
get_capabilities@query(none)listReturn 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

FieldTypeDefaultDescription
agent_idstr""Set to ctx.contract_address at deploy
ownerstr""Address that deployed the agent
agent_uristr""URI pointing to off-chain registration JSON
capabilitieslist[]List of capability strings
activeboolFalseWhether the agent accepts tasks
task_countint0Cumulative task counter
execution_loglist[]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 emits AgentUpdated. Returns the agent_id.
  • bind_capability (owner only) adds one capability from the four valid values; emits CapabilityBound when a new capability is added.
  • revoke (owner only) deactivates the agent and emits AgentRevoked. While inactive, execute raises a RuntimeError.

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

EventFieldsWhen
AgentRegisteredagent_id, owner, uri, capabilitiesAgent deployed
AgentUpdatedagent_id, uri, capabilitiesAgent re-registered via register
CapabilityBoundagent_id, capabilityNew capability bound
TaskExecutedagent_id, task_id, task_type, senderTask executed
AgentRevokedagent_idAgent 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

FieldTypeDefaultDescription
next_agent_idint0Auto-incrementing ID counter
ownersdict{}Maps agent ID to owner address
urisdict{}Maps agent ID to metadata URI
agent_walletsdict{}Maps agent ID to wallet address
metadatadict{}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

EventFieldsWhen
Registeredagent_id, agent_uri, ownerNew agent minted
MetadataSetagent_id, metadata_key, metadata_valueMetadata or wallet updated
URIUpdatedagent_id, new_uri, updated_byAgent URI changed
AgentTransferredagent_id, from_addr, to_addrOwnership transferred

Query Methods

MethodParametersReturnsDescription
get_owneragent_idstrOwner address
get_uriagent_idstrMetadata URI
get_agent_walletagent_idstrWallet address
get_metadataagent_id, metadata_keystrArbitrary metadata value
agent_existsagent_idboolWhether 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

  1. Deploy the agent with an optional model_uri pointing to an off-chain model description.
  2. Call execute with a task dict containing data_id, optional features (a list of numeric values), and an optional label.
  3. 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".
  4. The labeled data point is stored in self.state.labels keyed by data_id, including the label, a hash of the features, the labeler's address, and the block height.

State Layout

FieldTypeDefaultDescription
agent_idstr""Contract address
ownerstr""Deployer address
capabilitieslist["inference", "training"]Fixed capabilities
activeboolFalseActivated at deploy
labelsdict{}Maps data_id to label records
label_countint0Total 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

  1. Deploy the agent with an optional strategy string (defaults to "momentum").
  2. Call execute with a task dict containing action ("buy" or "sell"), asset (a string), and amount (a number).
  3. The agent validates the action and asset, increments the trade counter, appends a position record to the log, and emits a TradeExecuted event.
  4. The position log is capped at the most recent 1000 entries to bound state size.

State Layout

FieldTypeDefaultDescription
agent_idstr""Contract address
ownerstr""Deployer address
capabilitieslist["inference", "data_access"]Fixed capabilities
activeboolFalseActivated at deploy
positionslist[]Trade position log (capped at 1000)
total_tradesint0Cumulative 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 value with configurable value_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_summary query 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_request with a validator address, the agent ID, a request URI, and a request hash. The designated validator later calls validation_response with a numeric response (0-100), an optional response URI, hash, and tag.
  • Cross-contract verification: Uses Contract(identity_registry).agent_exists() and Contract(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_summary query computes average response scores across all validations for a given agent, with optional filtering by validator addresses and tag.

Contract Summary

ContractSourceTypeDescription
iprc_agentcontracts/agents/iprc_agent.pyInterfaceCapability constants, REQUIRED_METHODS (six methods), and validate_prc_agent helper
PRCAgentcontracts/agents/prc_agent.pyReferenceSelf-contained agent with identity, capabilities, task execution, and execution log
AgentIdentityRegistrycontracts/agents/agent_identity_registry.pyRegistryOn-chain agent identity (IDs, URIs, wallets, metadata)
DataLabelingAgentcontracts/agents/examples/data_labeling_agent.pyExampleData labeling with heuristic classification
TradingAgentcontracts/agents/examples/trading_agent.pyExampleTrade execution with position tracking
AgentReputationRegistrycontracts/agents/agent_reputation_registry.pyEcosystemFeedback and reputation via cross-contract calls
AgentValidationRegistrycontracts/agents/agent_validation_registry.pyEcosystemThird-party validation requests and responses

Further Reading