trc8004-m2m 1.2.4

TRC-8004 Machine-to-Machine Agent Registry SDK for TRON

TRC-8004-M2M Agent Registry SDK

Python SDK for the TRC-8004 Machine-to-Machine Agent Registry on TRON blockchain.

Features

• Agent Registration: Register AI agents as NFTs on TRON blockchain
• Fast Search: Query agents by skills, tags, reputation (via API)
• Validation Workflows: Submit and respond to validation requests
• Reputation Management: Give and track feedback for agents
• Agent-to-Agent (A2A) Communication: Agent Protocol client for inter-agent interaction
• Event Parsing: Extract agent_id from blockchain transactions
• Multi-Protocol Data Loading: Fetch data from file://, ipfs://, https:// URIs
• Hybrid Architecture: Blockchain for trust + API for performance
• Type-Safe: Full Pydantic models and type hints

Installation

pip install trc8004-m2m

Or with development dependencies:

pip install trc8004-m2m[dev]

Quick Start

Contact Addresses

Mainnet

Identity Registry    TYmmnmgkxteBvH8u8LAfb8sCcs1Eph2tk2
Validation Registry  TY9sqKiWBhZP6aMdjarcViyD2cudMAGfGn
Reputation Registry  TU8PmqF3mZoEGE97Se5oSgNo6bUs4VrswB
M2M TRC20 Token      TSH8XLQRMrCTTdCr3rUH2zUiuDZQjfmHaX

Shasta (Testnet)

Identity Registry    TUf5fAgpLrR6YM3P9oX9GNG1tGV3VcyPE3
Validation Registry  TJXVDV7hpsTSSz3QBCJdw99eFvTWUjxph6
Reputation Registry  TN5FBfXASyxrjUa9V73Hfme2pY9yWh6Rsh

Initialize Registry

from trc8004_m2m import AgentRegistry

# Read-only (no private key needed)
registry = AgentRegistry(network="mainnet")

# With private key (for write operations)
registry = AgentRegistry(
    private_key="your_hex_private_key",
    network="mainnet"  # or "shasta", "nile"
)

Register an Agent

agent_id = await registry.register_agent(
    name="TradingBot Pro",
    description="Advanced AI trading agent for DeFi",
    skills=[
        {
            "skill_id": "market_analysis",
            "skill_name": "Market Analysis",
            "description": "Technical analysis of crypto markets"
        },
        {
            "skill_id": "trading",
            "skill_name": "Automated Trading",
            "description": "Execute trades on DEXs"
        }
    ],
    endpoints=[
        {
            "endpoint_type": "rest_api",
            "url": "https://api.tradingbot.pro/v1",
            "name": "Trading API",
            "version": "1.0.0"
        }
    ],
    tags=["trading", "defi", "analytics"],
    version="2.1.0"
)

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

Search for Agents

# Search by skills
agents = await registry.search_agents(
    skills=["trading", "market_analysis"],
    min_reputation=80,
    verified_only=True,
    limit=10
)

for agent in agents:
    print(f"{agent.name} - Score: {agent.avg_reputation_score}")

# Full-text search
agents = await registry.search_agents(
    query="trading bot",
    limit=20
)

Get Agent Details

agent = await registry.get_agent(agent_id=123)

print(f"Name: {agent.name}")
print(f"Owner: {agent.owner_address}")
print(f"Reputation: {agent.avg_reputation_score}")
print(f"Skills: {[s.skill_name for s in agent.skills]}")

Submit Validation

# Submit validation request
tx = await registry.submit_validation(
    validator_address="TValidatorAddress...",
    agent_id=123,
    request_data={
        "test_case": "market_analysis_btc",
        "input": {
            "asset": "BTC/USDT",
            "timeframe": "1h"
        },
        "expected_output": {
            "trend": "bullish",
            "confidence": 0.85
        }
    }
)

# Respond to validation (validators only)
tx = await registry.respond_to_validation(
    request_hash="0xabc123...",
    response_score=95,
    response_data={
        "passed": True,
        "execution_time": 1.2,
        "accuracy": 0.96
    },
    tag="performance"
)

Give Feedback

tx = await registry.give_feedback(
    agent_id=123,
    score=95,
    tag1="execution",
    tag2="reliability",
    endpoint="/api/v1/trade",
    feedback_data={
        "trades_executed": 50,
        "success_rate": 0.96,
        "avg_latency_ms": 800,
        "total_profit_pct": 12.5
    }
)

Get Reputation Stats

stats = await registry.get_agent_reputation(agent_id=123)

print(f"Average Score: {stats['average_score']}")
print(f"Total Feedback: {stats['total_feedback']}")
print(f"By Tag: {stats['by_tag']}")

Agent-to-Agent Communication

Using Agent Protocol Client

from trc8004_m2m import AgentProtocolClient

# Connect to another agent
client = AgentProtocolClient(
    base_url="https://agent.example.com"
)

# One-shot execution
result = await client.run({
    "skill": "market_analysis",
    "params": {
        "asset": "BTC/USDT",
        "timeframe": "1h"
    }
})

print(result["output"])

# Multi-step workflow
task = await client.create_task()
step1 = await client.execute_step(task["task_id"], '{"action": "quote"}')
step2 = await client.execute_step(task["task_id"], '{"action": "execute"}')

await client.close()

Discover Agent Endpoints

# Search for agents
agents = await registry.search_agents(skills=["trading"])

# Get agent endpoint
agent = agents[0]
endpoint_url = agent.endpoints[0].url

# Connect via Agent Protocol
client = AgentProtocolClient(base_url=endpoint_url)
result = await client.run({"skill": "quote", "params": {...}})

Chain Utilities

Load Data from URIs

from trc8004_m2m.utils import load_request_data

# Load from IPFS
data = await load_request_data("ipfs://QmXxx...")

# Load from HTTPS
data = await load_request_data("https://example.com/data.json")

# Load from local file (testing)
data = await load_request_data("file:///tmp/request.json")

Parse Transaction Events

from trc8004_m2m.utils import parse_agent_registered_event

# Get transaction receipt
tx_info = await tron_client.get_transaction_info(tx_id)

# Extract agent_id from AgentRegistered event
agent_id = parse_agent_registered_event(tx_info)
print(f"Agent registered with ID: {agent_id}")

Hash and Verify Data

from trc8004_m2m.utils import compute_metadata_hash, keccak256_hex

# Compute metadata hash
metadata = {"name": "MyAgent", "version": "1.0"}
hash_value = compute_metadata_hash(metadata)

# Verify data integrity
fetched_data = await load_request_data(uri)
computed_hash = compute_metadata_hash(fetched_data)
assert computed_hash == expected_hash, "Data integrity check failed"

Configuration

Custom Contract Addresses

registry = AgentRegistry(
    private_key="...",
    network="mainnet",
    identity_registry="TIdentityRegistryAddress...",
    validation_registry="TValidationRegistryAddress...",
    reputation_registry="TReputationRegistryAddress..."
)

Custom API URL

registry = AgentRegistry(
    private_key="...",
    network="mainnet",
    api_url="https://custom-api.trc8004.io"
)

Architecture

Hybrid Blockchain + API

┌─────────────────────────────────────────────┐
│            Your Application                  │
└──────────────┬──────────────────────────────┘
               │
               │ TRC-8004-M2M SDK
               │
      ┌────────┴────────┐
      │                 │
┌─────▼──────┐   ┌─────▼─────────┐
│ TRON Chain │   │  Registry API │
│            │   │  (PostgreSQL) │
│ - NFTs     │   │               │
│ - Events   │   │ - Fast search │
│ - Immutable│   │ - Analytics   │
└────────────┘   └───────────────┘

Writes go to blockchain (trustless, verifiable)
Reads come from API (fast, rich queries)

Data Flow

1. Register Agent: Upload metadata to IPFS → Register on-chain → Indexer caches in DB
2. Search Agents: Query PostgreSQL (milliseconds)
3. Verify Ownership: Query blockchain (trustless)

Models

Agent

from trc8004_m2m import Agent

agent = Agent(
    agent_id=123,
    owner_address="TOwnerAddress...",
    name="My Agent",
    description="...",
    version="1.0.0",
    token_uri="ipfs://Qm...",
    skills=[...],
    endpoints=[...],
    tags=["tag1", "tag2"],
    verified=True,
    avg_reputation_score=92.5,
    total_validations=450,
    total_feedback=320,
    registered_at=datetime.now()
)

Skill

from trc8004_m2m import Skill

skill = Skill(
    skill_id="trading",
    skill_name="Automated Trading",
    description="Execute trades on DEXs",
    input_schema={
        "type": "object",
        "properties": {
            "asset": {"type": "string"},
            "amount": {"type": "number"}
        }
    }
)

Error Handling

from trc8004_m2m import (
    RegistryError,
    ContractError,
    NetworkError,
    ValidationError,
)

try:
    agent = await registry.get_agent(999999)
except NetworkError as e:
    print(f"Network error: {e}")
except RegistryError as e:
    print(f"General error: {e.code} - {e}")

Development

Install from source

git clone https://github.com/M2M-TRC8004-Registry/trc8004-m2m-sdk
cd trc8004-m2m-sdk
pip install -e .[dev]

Run tests

pytest

Format code

black trc8004_m2m/
ruff check trc8004_m2m/

Examples

See the examples/ directory for:

• Agent registration
• Search and discovery
• Validation workflows
• Reputation management
• Custom integrations

License

MIT License - see LICENSE file.

Contributing

Contributions welcome! Please open an issue or pull request on GitHub.

Support

• Web: https://m2mregistry.io
• Documentation: https://m2mregistry.io/docs
• Twitter: https://x.com/m2m_registry
• GitHub: https://github.com/M2M-TRC8004-Registry

Roadmap

• Event subscriptions (WebSocket)
• Agent-to-agent communication
• Agent Economy ($M2M based)
• Payment integration (Wirex)
• Reputation oracles