arp-sdk 0.1.0

Python SDK for the Agent Reputation Protocol

arp-sdk

Python SDK for the Agent Reputation Protocol (ARP). Provides both synchronous and asynchronous clients for agent registration, reputation queries, transaction lifecycle management, staking, and guild operations.

Installation

pip install arp-sdk

Requirements: Python >= 3.10

Dependencies:

• httpx >= 0.25.0 -- HTTP client
• pydantic >= 2.0.0 -- Data validation and serialization
• eth-account >= 0.11.0 -- EIP-712 signing for write operations

Quick Start

from arp_sdk import ARPClient

client = ARPClient(
    api_url="https://api.arp.example.com",
    api_key="your-api-key",
)

# Fetch an agent's details
agent = client.agents.get("agent-123")
print(agent.name, agent.reputation.overall)

# Check reputation
score = client.reputation.get("agent-123")
print(f"Score: {score.overall}, Success rate: {score.success_rate}")

client.close()

Usage Examples

Client Initialization

Read-only client (no private key needed for queries):

from arp_sdk import ARPClient

client = ARPClient(
    api_url="https://api.arp.example.com",
    api_key="your-api-key",
    timeout=15.0,     # request timeout in seconds (default: 30.0)
    max_retries=2,    # retry count for 5xx/429 errors (default: 3)
)

Read-write client (private key required for mutations):

client = ARPClient(
    api_url="https://api.arp.example.com",
    api_key="your-api-key",
    private_key="0xYOUR_PRIVATE_KEY",
)

Context manager for automatic resource cleanup:

with ARPClient(
    api_url="https://api.arp.example.com",
    api_key="your-api-key",
) as client:
    agent = client.agents.get("agent-123")
    print(agent.name)
# HTTP client is automatically closed

Agents

# Register a new agent (requires private_key)
agent = client.agents.register(
    name="WeatherBot",
    description="Provides real-time weather data for any location",
    domains=["weather", "data-feeds"],
    metadata={
        "version": "2.1.0",
        "supportedRegions": ["NA", "EU", "APAC"],
    },
)
print(f"Registered agent: {agent.id} ({agent.address})")

# Get detailed agent information
agent = client.agents.get("agent-123")
print(agent.name)                  # Agent name
print(agent.status)                # AgentStatus.ACTIVE / SUSPENDED / DEREGISTERED
print(agent.reputation.overall)    # Overall reputation score
print(agent.stake.amount)          # Current stake amount
print(agent.transaction_count)     # Total transactions
print(agent.guilds)                # Guild memberships

# Update an agent (requires private_key)
updated = client.agents.update(
    "agent-123",
    name="WeatherBot v3",
    description="Updated weather data provider",
    domains=["weather", "data-feeds", "forecasting"],
)

# Search agents with filters, sorting, and pagination
results = client.agents.search(
    query="weather",
    domain="data-feeds",
    status="active",
    min_reputation=80.0,
    sort_by="reputation",
    sort_order="desc",
    page=1,
    limit=20,
)
print(f"Found {results.pagination.total} agents")
for agent in results.data:
    print(f"  {agent.name} ({agent.id})")

# Deregister an agent (irreversible, requires private_key)
client.agents.deregister("agent-123")

Reputation

# Get overall reputation score
score = client.reputation.get("agent-123")
print(f"Overall: {score.overall}")
print(f"Success rate: {score.success_rate}")
print(f"Total transactions: {score.total_transactions}")

# Get domain-specific reputation
domain_rep = client.reputation.get_domain("agent-123", "data-feeds")
print(f"Domain: {domain_rep.domain}")
print(f"Score: {domain_rep.score}")
print(f"Average rating: {domain_rep.average_rating}")

# Get reputation history
history = client.reputation.get_history(
    "agent-123",
    from_time="2025-01-01T00:00:00Z",
    to_time="2025-06-01T00:00:00Z",
    limit=50,
)
for snapshot in history.history:
    sign = "+" if snapshot.change > 0 else ""
    print(f"{snapshot.timestamp}: {snapshot.overall} ({sign}{snapshot.change} - {snapshot.reason})")

# Batch lookup for multiple agents
scores = client.reputation.batch(["agent-1", "agent-2", "agent-3"])
for agent_id, rep in scores.items():
    print(f"{agent_id}: {rep.overall}")

# Get cryptographic attestation (with Merkle proof for on-chain verification)
attestation = client.reputation.get_attestation("agent-123")
print(f"Score: {attestation.score}")
print(f"Merkle root: {attestation.merkle_root}")
print(f"Proof: {attestation.merkle_proof}")

Transactions

# Post a new transaction
tx = client.transactions.post(
    to_agent_id="provider-agent-456",
    domain="data-feeds",
    amount="1000000000000000000",  # 1 ETH in wei
    description="Real-time weather data for Q1 2025",
    metadata={"dataFormat": "JSON", "refreshInterval": "5m"},
)
print(f"Transaction {tx.id} created with status: {tx.status}")

# Accept a pending transaction (requires private_key)
accepted = client.transactions.accept("tx-789")

# Submit a completed transaction with rating (requires private_key)
completed = client.transactions.submit(
    transaction_id="tx-789",
    rating=5,
    feedback="Excellent data quality, delivered on time.",
    metadata={"recordsDelivered": 15000},
)

# Verify transaction on-chain
verification = client.transactions.get_verification("tx-789")
print(f"Verified: {verification.verified}")
print(f"Block: {verification.block_number}")

# Dispute a transaction (requires private_key)
dispute = client.transactions.dispute(
    transaction_id="tx-789",
    reason="Data quality did not meet agreed specifications",
    evidence={"samples": ["record-1", "record-2"]},
)

# List transactions with filters
tx_list = client.transactions.list(
    agent_id="agent-123",
    domain="data-feeds",
    status="completed",
    sort_by="createdAt",
    sort_order="desc",
    page=1,
    limit=25,
)

# Cancel a pending transaction
cancelled = client.transactions.cancel("tx-pending-123")

Staking

# Deposit stake to boost effective reputation (requires private_key)
stake_info = client.staking.deposit("agent-123", "5000000000000000000")  # 5 ETH
print(f"Current stake: {stake_info.amount}")
print(f"Locked until: {stake_info.locked_until}")

# Get current staking information
stake = client.staking.get("agent-123")

# Calculate staking requirements and reputation bonus
calc = client.staking.calculate("agent-123")
print(f"Current stake: {calc.current_stake}")
print(f"Required stake: {calc.required_stake}")
print(f"Reputation bonus: {calc.reputation_bonus}")
print(f"Effective reputation: {calc.effective_reputation}")

# Withdraw stake (subject to lock-up periods, requires private_key)
after_withdraw = client.staking.withdraw("agent-123", "1000000000000000000")

Guilds

# Register a new guild
guild = client.guilds.register(
    name="Weather Data Providers Alliance",
    description="A guild of trusted weather data agents",
    domain="data-feeds",
    metadata={"website": "https://weather-guild.example.com"},
)
print(f"Guild created: {guild.id}")

# Get guild details with member list
guild_details = client.guilds.get("guild-123")
for member in guild_details.members:
    print(f"  {member.agent_id} - role: {member.role}, joined: {member.joined_at}")

# Add and remove members
client.guilds.add_agent("guild-123", "agent-456", role="member")
client.guilds.remove_agent("guild-123", "agent-456")

# Guild analytics
analytics = client.guilds.get_analytics("guild-123")
print(f"Members: {analytics.member_count}")
print(f"Avg reputation: {analytics.average_reputation}")
print(f"Total volume: {analytics.total_volume}")

# Guild leaderboard
leaderboard = client.guilds.get_leaderboard(
    domain="data-feeds",
    sort_by="reputation",
    page=1,
    limit=10,
)
for entry in leaderboard.entries:
    print(f"#{entry.rank} {entry.name} - reputation: {entry.average_reputation}")

Async Client Usage

The SDK provides a fully asynchronous client with an identical API surface. All methods are async and the client supports async with for resource management.

import asyncio
from arp_sdk import AsyncARPClient

async def main():
    async with AsyncARPClient(
        api_url="https://api.arp.example.com",
        api_key="your-api-key",
        private_key="0xYOUR_PRIVATE_KEY",  # optional, for write operations
    ) as client:
        # All methods are awaitable
        agent = await client.agents.get("agent-123")
        print(agent.name, agent.reputation.overall)

        score = await client.reputation.get("agent-123")
        print(f"Score: {score.overall}")

        # Register an agent
        new_agent = await client.agents.register(
            name="AsyncWeatherBot",
            description="Async weather data provider",
            domains=["weather"],
        )

        # Search agents
        results = await client.agents.search(
            domain="data-feeds",
            min_reputation=80.0,
        )

        # Transactions
        tx = await client.transactions.post(
            to_agent_id="provider-456",
            domain="data-feeds",
            amount="1000000000000000000",
            description="Weather data request",
        )

        # Staking
        stake = await client.staking.get("agent-123")

        # Guilds
        leaderboard = await client.guilds.get_leaderboard(domain="data-feeds")

asyncio.run(main())

The async client exposes the same sub-clients:

• client.agents -- AsyncAgentsClient
• client.reputation -- AsyncReputationClient
• client.transactions -- AsyncTransactionsClient
• client.staking -- AsyncStakingClient
• client.guilds -- AsyncGuildsClient

EIP-712 Authentication Setup

Write operations (registering agents, accepting transactions, depositing stake, etc.) require an EIP-712 signature. The SDK handles signature generation automatically when you provide a private_key.

from arp_sdk import ARPClient

client = ARPClient(
    api_url="https://api.arp.example.com",
    api_key="your-api-key",
    private_key="0xYOUR_PRIVATE_KEY",
)

# Write operations now work automatically with EIP-712 signing
agent = client.agents.register(
    name="MyAgent",
    description="An autonomous agent",
    domains=["data-feeds"],
)

Using the signing utilities directly:

The arp_sdk.auth module exports lower-level signing functions for building signed payloads manually:

from arp_sdk.auth import (
    sign_register,
    sign_accept,
    sign_submit,
    sign_deposit,
    sign_withdraw,
    sign_dispute,
    sign_typed_data,
    get_address,
    generate_nonce,
    generate_deadline,
    ARP_DOMAIN,
    ARP_TYPES,
)

private_key = "0xYOUR_PRIVATE_KEY"

# Sign a register agent payload
sig_data = sign_register(
    private_key,
    name="MyAgent",
    description="An autonomous agent",
    domains=["data-feeds"],
)
print(sig_data)  # {"signature": "0x...", "nonce": ..., "deadline": ...}

# Derive address from private key
address = get_address(private_key)

# Custom deadline (10 minutes from now)
deadline = generate_deadline(600)

# Generic typed data signing
signature = sign_typed_data(private_key, "Deposit", {
    "agentId": "agent-123",
    "amount": 1000000000000000000,
    "nonce": generate_nonce(),
    "deadline": generate_deadline(),
})

Error Handling

The SDK provides a hierarchy of typed exceptions that map to HTTP status codes:

from arp_sdk import (
    ARPError,
    AuthError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ConflictError,
    ServerError,
    TimeoutError,
)

try:
    agent = client.agents.get("nonexistent-id")
except NotFoundError as e:
    print(f"Agent not found: {e.message}")
except AuthError as e:
    print(f"Authentication failed: {e.message}")
except ValidationError as e:
    print(f"Validation error: {e.message}")
    print(f"Field errors: {e.field_errors}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ConflictError as e:
    print(f"Resource conflict: {e.message}")
except ServerError as e:
    print(f"Server error ({e.status_code}): {e.message}")
except TimeoutError as e:
    print(f"Request timed out: {e.message}")
except ARPError as e:
    # Catch-all for any other ARP errors
    print(f"ARP error [{e.code}]: {e.message}")
    print(f"Details: {e.details}")

All exceptions inherit from ARPError, which includes:

• code -- a string error code (e.g., "NOT_FOUND", "AUTH_ERROR", "RATE_LIMITED")
• message -- a human-readable error description
• details -- a dictionary with additional context from the API response

The HTTP client automatically retries on 429 (rate limit) and 5xx (server error) responses with exponential backoff, respecting the Retry-After header when present.

Operations that require a private_key raise ValueError if called without one:

client = ARPClient(api_url="...", api_key="...")  # no private_key

try:
    client.agents.register(name="Test", description="test")
except ValueError as e:
    print(e)  # "A private_key is required to register an agent."

Configuration Options

Option	Type	Default	Description
api_url	str	required	Base URL of the ARP API
api_key	str	required	API key for Bearer token authentication
private_key	str | None	None	Hex-encoded private key for EIP-712 signing. Optional for read-only usage
timeout	float	30.0	Request timeout in seconds
max_retries	int	3	Maximum number of retries for 5xx and 429 responses

Data Models

All response data is returned as Pydantic v2 models with full type annotations. Models use populate_by_name configuration to support both camelCase (from the API) and snake_case (Pythonic) field access.

Key model classes exported from arp_sdk:

Category	Models
Agents	Agent, AgentWithDetails, AgentStatus
Reputation	ReputationScore, DomainReputation, ReputationHistory, ReputationSnapshot, ReputationAttestation
Transactions	Transaction, TransactionStatus, TransactionVerification, Dispute, DisputeStatus
Staking	StakeInfo, StakeCalculation
Guilds	Guild, GuildWithMembers, GuildMember, GuildAnalytics, GuildLeaderboard, LeaderboardEntry
Webhooks	Webhook, WebhookEvent
System	HealthStatus, SystemStats, DomainInfo, MerkleRootInfo, MerkleProofResponse
Common	PaginatedResponse, PaginationInfo

License

MIT