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, operator delegation, 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.
• Supports operators -- the owner can delegate execution rights to additional addresses without transferring ownership.
• Can be revoked -- the owner can deactivate the agent, preventing further task execution.
• Emits events -- all lifecycle transitions (registration, execution, revocation, operator changes) produce indexed events.

The standard is composed of several contracts that work together:

IPRCAgent Interface

The IPRCAgent contract defines the abstract interface that every compliant agent must implement. It specifies four required methods and provides a validate_prc_agent helper to check compliance at the contract-class level.

Required Methods

Any contract that implements these four methods with the correct decorator types is PRC-Agent compliant. The IPRCAgent contract itself raises NotImplementedError for all methods -- it exists purely as a reference for the expected signatures.

Full Interface Source

"""
IPRCAgent -- Abstract interface for PRC-Agent agents (ERC-8004 inspired).

Defines the standard methods every compliant agent contract MUST implement:
  - execute(task: dict) -> dict
  - revoke() -> bool
  - get_agent() -> dict
  - get_capabilities() -> list

Contracts inherit from IPRC-Agent or implement matching signatures.
"""

from panda import contract, call, query


REQUIRED_METHODS = {
    "execute": {"type": "call", "params": ["task"], "returns": "dict"},
    "revoke": {"type": "call", "params": [], "returns": "bool"},
    "get_agent": {"type": "query", "params": [], "returns": "dict"},
    "get_capabilities": {"type": "query", "params": [], "returns": "list"},
}


@contract
class IPRCAgent:
    """Abstract PRC-Agent interface. All methods raise NotImplementedError."""

    class State:
        _interface_marker: str = "IPRCAgent"

    @call
    def execute(self, ctx, task: dict) -> dict:
        raise NotImplementedError("IPRCAgent.execute")

    @call
    def revoke(self, ctx) -> bool:
        raise NotImplementedError("IPRCAgent.revoke")

    @query
    def get_agent(self) -> dict:
        raise NotImplementedError("IPRCAgent.get_agent")

    @query
    def get_capabilities(self) -> list:
        raise NotImplementedError("IPRCAgent.get_capabilities")


def validate_prc_agent(contract_class) -> list:
    """Validate that a contract class has all required PRC-Agent methods."""
    issues = []
    for method_name, spec in REQUIRED_METHODS.items():
        method = getattr(contract_class, method_name, None)
        if method is None:
            issues.append(f"Missing required method: {method_name}")
            continue
        if spec["type"] == "call" and not getattr(method, "_panda_is_call", False):
            issues.append(f"{method_name} must be a @call method")
        if spec["type"] == "query" and not getattr(method, "_panda_is_query", False):
            issues.append(f"{method_name} must be a @query method")
    return issues

Validation Helper

The REQUIRED_METHODS dict and validate_prc_agent() function allow tooling to check whether a given contract class satisfies the PRC-Agent interface at deploy time. The function inspects the class for each required method, checks that it exists, and verifies the decorator type (_panda_is_call or _panda_is_query). It returns a list of issues -- an empty list means the contract is fully compliant.

PRCAgent Reference Implementation

The PRCAgent contract is the production-grade reference implementation of the PRC-Agent standard. It implements all four required interface methods and adds operator delegation, metadata storage, and a built-in task router.

State Layout

Constructor

The constructor accepts an optional capabilities list. It sets agent_id to the contract address, records the deployer as owner, activates the agent, and emits a Registered event.

Task Execution

The execute method is the core of the agent. It validates that the agent is active and the caller is authorized (owner or operator), then routes the task based on task["type"]:

Every execution increments total_tasks, stores the result in last_task_result, and emits an Executed event.

Operator Model

The owner can add and remove operators. Operators can execute tasks and set metadata, but cannot revoke the agent or manage other operators. The _require_authorized helper enforces this: it checks whether the sender is either the owner or in the operators list.

Events

Full Source

"""
PRC-Agent reference implementation -- Production-grade PRC-Agent contract.

Implements the full PRC-Agent interface:
  - register / revoke lifecycle
  - execute(task) for task routing
  - capability-based metadata
  - operator model (owner + operators)
  - event emission (Registered, Executed, Revoked, OperatorAdded)

Deploy one PRCAgent per logical autonomous agent.
"""

from panda import contract, constructor, call, query, event


@contract
class PRCAgent:
    """Reference PRC-Agent implementation."""

    class State:
        agent_id: str = ""
        owner: str = ""
        capabilities: list = []
        active: bool = False
        operators: list = []
        total_tasks: int = 0
        last_task_result: dict = {}
        metadata: dict = {}

    @constructor
    def __init__(self, ctx, capabilities: list = None):
        self.state.agent_id = ctx.contract_address
        self.state.owner = ctx.sender
        self.state.capabilities = capabilities if capabilities else []
        self.state.active = True
        self.emit(
            event.Registered(
                agent_id=self.state.agent_id,
                owner=ctx.sender,
                capabilities=self.state.capabilities,
            )
        )

    def _require_authorized(self, sender: str):
        if sender != self.state.owner and sender not in self.state.operators:
            raise PermissionError("Not authorized")

    @call
    def execute(self, ctx, task: dict) -> dict:
        if not self.state.active:
            raise RuntimeError("Agent is not active")
        self._require_authorized(ctx.sender)

        task_type = task.get("type", "default")
        payload = task.get("payload", {})

        self.state.total_tasks = self.state.total_tasks + 1

        if task_type == "echo":
            result = {"echo": payload}
        elif task_type == "compute":
            values = payload.get("values", [])
            op = payload.get("op", "sum")
            if op == "sum":
                result = {"result": sum(values)}
            elif op == "mean":
                result = {"result": sum(values) / len(values) if values else 0}
            elif op == "max":
                result = {"result": max(values) if values else 0}
            elif op == "min":
                result = {"result": min(values) if values else 0}
            else:
                result = {"error": f"Unknown op: {op}"}
        elif task_type == "store":
            key = payload.get("key", "")
            value = payload.get("value", "")
            if key:
                md = dict(self.state.metadata)
                md[key] = value
                self.state.metadata = md
                result = {"stored": key}
            else:
                result = {"error": "Missing key"}
        elif task_type == "query":
            key = payload.get("key", "")
            result = {"value": self.state.metadata.get(key, None)}
        else:
            result = {"echo": payload, "task_type": task_type}

        self.state.last_task_result = result
        self.emit(
            event.Executed(
                agent_id=self.state.agent_id,
                task_type=task_type,
                task_number=self.state.total_tasks,
            )
        )
        return result

    @call
    def revoke(self, ctx) -> bool:
        if ctx.sender != self.state.owner:
            raise PermissionError("Only the owner can revoke")
        self.state.active = False
        self.emit(event.Revoked(agent_id=self.state.agent_id, revoker=ctx.sender))
        return True

    @call
    def add_operator(self, ctx, operator: str):
        if ctx.sender != self.state.owner:
            raise PermissionError("Only the owner can add operators")
        ops = list(self.state.operators)
        if operator not in ops:
            ops.append(operator)
            self.state.operators = ops
            self.emit(event.OperatorAdded(agent_id=self.state.agent_id, operator=operator))

    @call
    def remove_operator(self, ctx, operator: str):
        if ctx.sender != self.state.owner:
            raise PermissionError("Only the owner can remove operators")
        ops = list(self.state.operators)
        if operator in ops:
            ops.remove(operator)
            self.state.operators = ops

    @call
    def set_metadata(self, ctx, key: str, value: str):
        self._require_authorized(ctx.sender)
        md = dict(self.state.metadata)
        md[key] = value
        self.state.metadata = md

    @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,
            "operators": self.state.operators,
            "total_tasks": self.state.total_tasks,
        }

    @query
    def get_capabilities(self) -> list:
        return list(self.state.capabilities)

    @query
    def get_metadata_value(self, key: str) -> str:
        return self.state.metadata.get(key, "")

    @query
    def get_last_result(self) -> dict:
        return self.state.last_task_result

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:

{"capabilities": ["inference", "training", "data_access"]}

Once deployed, use the Interact panel to call execute with a task dict:

{"task": {"type": "compute", "payload": {"values": [10, 20, 30], "op": "mean"}}}

The agent returns {"result": 20.0} and emits an Executed 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

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

"""
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

Query Methods

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

Full Source

"""
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 __init__(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

Full Source

"""
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 __init__(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

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