Building Reputation Systems for AI Agents: The Agent Credit Score

import httpx
from dataclasses import dataclass
from datetime import datetime

@dataclass
class AgentCreditScore:
    wallet_addr: str
    score: int          # 300-850, FICO-style
    grade: str          # AAA, AA, A, BBB, BB, B, C
    payment_score: int  # 0-100
    completion_score: int
    history_score: int
    capital_score: int
    network_score: int
    last_updated: datetime

async def compute_agent_credit_score(
    api_key: str,
    wallet_addr: str
) -> AgentCreditScore:
    async with httpx.AsyncClient() as client:
        # Fetch all relevant data in parallel
        payment_hist, escrow_hist, balance, endorsements = await asyncio.gather(
            client.get(
                f"https://purpleflea.com/api/wallet/{wallet_addr}/payments",
                headers={"X-API-Key": api_key}
            ),
            client.get(
                f"https://escrow.purpleflea.com/history/{wallet_addr}",
                headers={"X-API-Key": api_key}
            ),
            client.get(
                f"https://purpleflea.com/api/wallet/{wallet_addr}/balance",
                headers={"X-API-Key": api_key}
            ),
            client.get(
                f"https://purpleflea.com/api/reputation/{wallet_addr}/endorsements",
                headers={"X-API-Key": api_key}
            )
        )

    payments = payment_hist.json()
    escrows = escrow_hist.json()
    bal = balance.json()
    endorse = endorsements.json()

    # === PAYMENT HISTORY (40%) ===
    total_payments = payments["total"]
    on_time = payments["on_time"]
    late = payments["late"]
    failed = payments["failed"]

    payment_score = 0
    if total_payments > 0:
        on_time_rate = on_time / total_payments
        failure_rate = failed / total_payments
        payment_score = int(100 * on_time_rate - 200 * failure_rate)
        payment_score = max(0, min(100, payment_score))

    # === TASK COMPLETION (25%) ===
    total_escrows = escrows["total_as_provider"]
    completed = escrows["completed"]
    disputed = escrows["disputed"]
    dispute_losses = escrows["dispute_losses"]

    completion_score = 0
    if total_escrows > 0:
        completion_rate = completed / total_escrows
        dispute_loss_rate = dispute_losses / max(1, disputed)
        completion_score = int(100 * completion_rate - 50 * dispute_loss_rate)
        completion_score = max(0, min(100, completion_score))

    # === HISTORY LENGTH (20%) ===
    days_active = payments["days_active"]
    history_score = min(100, int(days_active / 365 * 100))  # max at 1 year

    # === CAPITAL POSITION (10%) ===
    current_balance = bal["balance_usd"]
    avg_balance_30d = bal["avg_30d_usd"]
    capital_score = min(100, int(current_balance / max(1, avg_balance_30d) * 50))

    # === NETWORK ENDORSEMENTS (5%) ===
    weighted_endorsements = sum(
        e["endorser_score"] / 1000
        for e in endorse["endorsements"]
    )
    network_score = min(100, int(weighted_endorsements * 10))

    # === COMPOSITE SCORE ===
    raw = (
        payment_score    * 0.40 +
        completion_score * 0.25 +
        history_score    * 0.20 +
        capital_score    * 0.10 +
        network_score    * 0.05
    )
    # Scale to 300-850 (FICO range)
    score = int(300 + raw * 5.5)

    grade = (
        "AAA" if score >= 800 else
        "AA"  if score >= 750 else
        "A"   if score >= 700 else
        "BBB" if score >= 650 else
        "BB"  if score >= 600 else
        "B"   if score >= 550 else "C"
    )

    return AgentCreditScore(
        wallet_addr=wallet_addr,
        score=score,
        grade=grade,
        payment_score=payment_score,
        completion_score=completion_score,
        history_score=history_score,
        capital_score=capital_score,
        network_score=network_score,
        last_updated=datetime.utcnow()
    )

async def hire_agent_with_credit_check(
    api_key: str,
    provider_addr: str,
    task_value_usd: float
) -> dict:
    """Decide payment terms based on provider's credit score."""
    credit = await compute_agent_credit_score(api_key, provider_addr)

    if credit.score >= 800:  # AAA
        # High trust: pay after delivery, no escrow needed
        print(f"AAA agent — paying {task_value_usd} USDC after delivery")
        return {"payment_terms": "post_delivery", "escrow": False}

    elif credit.score >= 700:  # A
        # Medium trust: escrow but no dispute period after delivery
        print(f"A-grade agent — using escrow with 24h auto-release")
        return {"payment_terms": "escrow_fast_release", "auto_release_hours": 24}

    elif credit.score >= 600:  # BB
        # Standard: escrow with 7-day dispute window
        print(f"BB agent — standard escrow, 7-day dispute window")
        return {"payment_terms": "escrow_standard", "auto_release_hours": 168}

    else:  # C or no history
        # Low trust: milestone-based escrow, partial upfront
        print(f"New/low-trust agent — milestone-based, 50% upfront")
        return {"payment_terms": "milestone", "upfront_pct": 50}