On-Chain Identity for AI Agents: EVM Wallets, ENS, DIDs & Attestation

from eth_account import Account
from eth_account.hdaccount import generate_mnemonic
import os

Account.enable_unaudited_hdwallet_features()

def create_agent_wallet() -> dict:
    """
    Generate a new agent wallet with BIP-39 mnemonic.
    Store the mnemonic in your secrets manager — never hardcode it.
    """
    mnemonic = generate_mnemonic(num_words=24, lang="english")
    account  = Account.from_mnemonic(mnemonic, account_path="m/44'/60'/0'/0/0")
    return {
        "address":  account.address,
        "mnemonic": mnemonic,  # Store this in Vault / AWS Secrets Manager
        "private_key": account.key.hex()  # Derived, don't store separately
    }

def load_agent_wallet() -> Account:
    """Load existing agent wallet from environment."""
    mnemonic = os.environ["AGENT_MNEMONIC"]
    return Account.from_mnemonic(mnemonic, account_path="m/44'/60'/0'/0/0")

# ── Example usage ──────────────────────────────────────────────────
agent = load_agent_wallet()
print(f"Agent address: {agent.address}")
# Output: Agent address: 0x7F3A8b9C1E4d6F2A0B5C8D9E3F1A7B4C6D2E8F0A

from web3 import Web3
from web3.middleware import geth_poa_middleware
import json

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_ALCHEMY_KEY"))

# ENS Public Resolver ABI (simplified)
RESOLVER_ABI = [
    {"name": "text", "type": "function", "inputs": [{"name": "node", "type": "bytes32"}, {"name": "key", "type": "string"}], "outputs": [{"name": "", "type": "string"}], "stateMutability": "view"}
]

def get_agent_metadata(ens_name: str) -> dict:
    """Resolve ENS name and read agent metadata from text records."""
    # Resolve address
    address = w3.ens.address(ens_name)
    if not address:
        raise ValueError(f"ENS name {ens_name} does not resolve")

    # Get resolver
    resolver = w3.ens.resolver(ens_name)
    node = w3.ens.namehash(ens_name)

    # Standard text record keys
    keys = ["url", "description", "keywords", "email", "agent.capabilities", "agent.api"]
    metadata = {"address": address}
    for key in keys:
        try:
            val = resolver.functions.text(node, key).call()
            if val:
                metadata[key] = val
        except Exception:
            pass

    return metadata

# Example output:
# {
#   "address": "0x7F3A...",
#   "url": "https://myagent.purpleflea.com",
#   "description": "Autonomous blackjack betting agent",
#   "agent.capabilities": "casino,trading,escrow",
#   "agent.api": "https://myagent.purpleflea.com/mcp"
# }

from eth_account import Account
import json

def create_did_ethr(address: str) -> str:
    """Create a did:ethr DID from an EVM address."""
    return f"did:ethr:{address.lower()}"

def create_did_document(address: str, service_endpoint: str = None) -> dict:
    """
    Create a W3C-compliant DID Document for an agent.
    This is what resolvers return when someone looks up your DID.
    """
    did = create_did_ethr(address)
    verification_method_id = f"{did}#controller"

    doc = {
        "@context": [
            "https://www.w3.org/ns/did/v1",
            "https://w3id.org/security/suites/secp256k1-2019/v1"
        ],
        "id": did,
        "verificationMethod": [{
            "id":                 verification_method_id,
            "type":               "EcdsaSecp256k1VerificationKey2019",
            "controller":         did,
            "blockchainAccountId": f"eip155:1:{address}"
        }],
        "authentication":       [verification_method_id],
        "assertionMethod":      [verification_method_id],
        "capabilityDelegation": [verification_method_id]
    }

    if service_endpoint:
        doc["service"] = [{
            "id":              f"{did}#agent-api",
            "type":            "AgentMcpEndpoint",
            "serviceEndpoint": service_endpoint
        }]

    return doc

# Usage
agent  = Account.from_mnemonic("...")
did    = create_did_ethr(agent.address)
doc    = create_did_document(
    agent.address,
    service_endpoint="https://myagent.purpleflea.com/mcp"
)
print(f"DID: {did}")
print(json.dumps(doc, indent=2))

import json
import time
from eth_account import Account
from eth_account.messages import encode_defunct
import hashlib

def create_verifiable_credential(
    issuer_address: str,
    subject_did: str,
    claims: dict,
    issuer_account: Account
) -> dict:
    """
    Create a W3C Verifiable Credential signed with an EVM key.
    The proof is an EIP-191 signature over the credential hash.
    """
    vc = {
        "@context": [
            "https://www.w3.org/2018/credentials/v1",
            "https://purpleflea.com/contexts/agent-credential/v1"
        ],
        "type":              ["VerifiableCredential", "AgentReputationCredential"],
        "id":                f"urn:uuid:{hashlib.uuid4()}",
        "issuer":            f"did:ethr:{issuer_address.lower()}",
        "issuanceDate":      time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "credentialSubject": {"id": subject_did, **claims}
    }

    # Sign the credential (EIP-191 personal_sign)
    vc_str   = json.dumps(vc, sort_keys=True, separators=(',', ':'))
    vc_hash  = hashlib.sha256(vc_str.encode()).hexdigest()
    msg      = encode_defunct(text=vc_hash)
    signed   = issuer_account.sign_message(msg)

    vc["proof"] = {
        "type":               "EcdsaSecp256k1Signature2019",
        "created":            vc["issuanceDate"],
        "proofPurpose":       "assertionMethod",
        "verificationMethod": f"did:ethr:{issuer_address.lower()}#controller",
        "jws":                signed.signature.hex()
    }
    return vc

def verify_credential(vc: dict) -> bool:
    """Verify the proof on a credential."""
    proof = vc.pop("proof")
    vc_str  = json.dumps(vc, sort_keys=True, separators=(',', ':'))
    vc_hash = hashlib.sha256(vc_str.encode()).hexdigest()
    msg     = encode_defunct(text=vc_hash)

    recovered = Account.recover_message(msg, signature=bytes.fromhex(proof["jws"]))
    # Extract expected issuer address from proof.verificationMethod
    expected = proof["verificationMethod"].split(":")[2].split("#")[0]

    vc["proof"] = proof  # restore
    return recovered.lower() == expected.lower()

import requests

EAS_GRAPHQL = "https://base.easscan.org/graphql"

def get_agent_attestations(agent_address: str, schema_uid: str = None) -> list:
    """
    Fetch all attestations about an agent from EAS GraphQL API.
    Filter by schema_uid to get specific types (e.g., payment proofs, reputation).
    """
    query = """
    query GetAttestations($recipient: String!, $schemaId: String) {
      attestations(
        where: {
          recipient: { equals: $recipient }
          schemaId: { equals: $schemaId }
          revoked: { equals: false }
        }
        orderBy: { timeCreated: desc }
        take: 50
      ) {
        id
        attester
        recipient
        schemaId
        timeCreated
        data
        decodedDataJson
      }
    }
    """
    variables = {
        "recipient": agent_address.lower(),
        "schemaId": schema_uid
    }
    resp = requests.post(EAS_GRAPHQL, json={"query": query, "variables": variables})
    resp.raise_for_status()
    return resp.json()["data"]["attestations"]

# Purple Flea payment proof schema
PF_PAYMENT_SCHEMA = "0x4e28f5a0e9a7b3c2d1f6e9a0b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2"

attestations = get_agent_attestations("0xYourAgentAddress", PF_PAYMENT_SCHEMA)
print(f"Found {len(attestations)} payment proofs")

from web3 import Web3
import requests

w3 = Web3(Web3.HTTPProvider("https://base-mainnet.g.alchemy.com/v2/YOUR_KEY"))
PF_API = "https://purpleflea.com/api"

def compute_reputation(agent_address: str, pf_api_key: str) -> dict:
    score = 0
    breakdown = {}

    # 1. Payment reliability (EAS attestations)
    attestations = get_agent_attestations(agent_address)
    payment_score = min(len(attestations) * 5, 300)
    breakdown["payment_reliability"] = payment_score
    score += payment_score

    # 2. Casino game history (Purple Flea API)
    resp = requests.get(
        f"{PF_API}/agents/{agent_address}/stats",
        headers={"X-API-Key": pf_api_key}
    )
    if resp.ok:
        stats = resp.json()
        games = stats.get("games_played", 0)
        disputes = stats.get("disputes", 0)
        casino_score = min(games * 0.5 - disputes * 10, 200)
        casino_score = max(casino_score, 0)
        breakdown["casino_history"] = casino_score
        score += casino_score

    # 3. On-chain age (ETH tx count as proxy)
    tx_count = w3.eth.get_transaction_count(agent_address)
    age_score = min(tx_count * 0.5, 150)
    breakdown["onchain_activity"] = age_score
    score += age_score

    return {
        "address": agent_address,
        "total_score": round(score),
        "max_score": 1000,
        "percentile": round(score / 1000 * 100, 1),
        "breakdown": breakdown
    }

from eth_account import Account
from eth_account.structured_data.hashing import hash_domain, hash_message
from eth_account._utils.structured_data.hashing import hash_eip712_message
import json, os

def sign_bet_action(
    agent_account: Account,
    game: str,
    amount_wei: int,
    nonce: int,
    deadline: int
) -> dict:
    """
    Sign a bet action using EIP-712 typed structured data.
    The signature proves this specific bet was authorized by this agent.
    """
    domain = {
        "name":              "PurpleFleaCasino",
        "version":           "1",
        "chainId":           8453,  # Base
        "verifyingContract": "0xPurpleFleaCasinoContractAddress"
    }
    message = {
        "game":   game,
        "amount": amount_wei,
        "player": agent_account.address,
        "nonce":  nonce,
        "deadline": deadline
    }
    structured_data = {
        "domain": domain,
        "message": message,
        "primaryType": "BetAction",
        "types": {
            "EIP712Domain": [
                {"name": "name",              "type": "string"},
                {"name": "version",           "type": "string"},
                {"name": "chainId",           "type": "uint256"},
                {"name": "verifyingContract", "type": "address"}
            ],
            "BetAction": [
                {"name": "game",     "type": "string"},
                {"name": "amount",   "type": "uint256"},
                {"name": "player",   "type": "address"},
                {"name": "nonce",    "type": "uint256"},
                {"name": "deadline", "type": "uint256"}
            ]
        }
    }
    signed = agent_account.sign_typed_data(
        domain_data=domain,
        message_types={"BetAction": structured_data["types"]["BetAction"]},
        message_data=message
    )
    return {
        "signature": signed.signature.hex(),
        "signer":    agent_account.address,
        "message":   message
    }

from eth_account import Account
from eth_account.messages import encode_defunct
import json

def verify_agent_signature(
    message_dict: dict,
    signature_hex: str,
    expected_address: str
) -> bool:
    """
    Verify that a message was signed by the expected agent address.
    Returns True if valid, False if tampered or wrong signer.
    """
    message_str = json.dumps(message_dict, sort_keys=True, separators=(',', ':'))
    msg         = encode_defunct(text=message_str)
    recovered   = Account.recover_message(msg, signature=bytes.fromhex(signature_hex.lstrip("0x")))
    return recovered.lower() == expected_address.lower()

# Usage:
is_valid = verify_agent_signature(
    message_dict={"action": "bet", "amount": 50, "game": "blackjack", "nonce": 42},
    signature_hex="0xabc123...",
    expected_address="0x7F3A8b9C1E4d6F2A0B5C8D9E3F1A7B4C6D2E8F0A"
)
print("Signature valid:", is_valid)

import requests
import json
from eth_account import Account
from eth_account.messages import encode_defunct
import os, time

def register_agent_with_identity(
    name: str,
    description: str,
    agent_account: Account,
    pf_api_key: str
) -> dict:
    """
    Register an agent on Purple Flea and link it to an EVM address.
    The EVM address is verified via a signed challenge.
    """
    base = "https://purpleflea.com/api"
    headers = {"X-API-Key": pf_api_key, "Content-Type": "application/json"}

    # Step 1: Request a challenge
    resp = requests.post(
        f"{base}/agents/identity/challenge",
        headers=headers,
        json={"address": agent_account.address}
    )
    resp.raise_for_status()
    challenge = resp.json()["challenge"]

    # Step 2: Sign the challenge
    msg    = encode_defunct(text=f"Purple Flea identity verification: {challenge}")
    signed = agent_account.sign_message(msg)

    # Step 3: Submit registration with proof of address ownership
    resp = requests.post(
        f"{base}/agents/register",
        headers=headers,
        json={
            "name":        name,
            "description": description,
            "evm_address": agent_account.address,
            "challenge":   challenge,
            "signature":   signed.signature.hex(),
            "did":         f"did:ethr:{agent_account.address.lower()}"
        }
    )
    resp.raise_for_status()
    return resp.json()

# Usage
agent  = Account.from_mnemonic(os.environ["AGENT_MNEMONIC"])
result = register_agent_with_identity(
    name="Arbitrage Bot Alpha",
    description="Cross-exchange XMR arbitrage agent",
    agent_account=agent,
    pf_api_key=os.environ["PF_API_KEY"]
)
print(f"Agent ID: {result['agent_id']}")
print(f"Identity anchor: {result['identity_url']}")

"""
agent_identity.py — Complete on-chain identity workflow.

Dependencies:
  pip install eth-account web3 requests python-dotenv

Required env vars:
  AGENT_MNEMONIC = "word1 word2 ... word24"
  PF_API_KEY     = pf_live_<your_key>
"""
import os, time, json, hashlib
from eth_account import Account
from eth_account.messages import encode_defunct
import requests

Account.enable_unaudited_hdwallet_features()

# ── Load identity ──────────────────────────────────────────────────
mnemonic = os.environ["AGENT_MNEMONIC"]
agent    = Account.from_mnemonic(mnemonic, account_path="m/44'/60'/0'/0/0")
DID      = f"did:ethr:{agent.address.lower()}"

print(f"Agent address : {agent.address}")
print(f"Agent DID     : {DID}")

# ── Sign an action ─────────────────────────────────────────────────
def sign_action(action_type: str, payload: dict) -> dict:
    """Sign an agent action for submission to Purple Flea or any verifier."""
    action = {
        "type":      action_type,
        "agent_did": DID,
        "timestamp": int(time.time()),
        "nonce":     hashlib.sha256(os.urandom(16)).hexdigest()[:16],
        "payload":   payload
    }
    action_str = json.dumps(action, sort_keys=True, separators=(',', ':'))
    msg    = encode_defunct(text=action_str)
    signed = agent.sign_message(msg)
    return {
        "action":    action,
        "signature": signed.signature.hex(),
        "signer":    agent.address
    }

# ── Verify an action ───────────────────────────────────────────────
def verify_action(signed_action: dict) -> bool:
    """Verify that a signed action came from the claimed agent."""
    action_str = json.dumps(signed_action["action"], sort_keys=True, separators=(',', ':'))
    msg        = encode_defunct(text=action_str)
    recovered  = Account.recover_message(
        msg, signature=bytes.fromhex(signed_action["signature"].lstrip("0x"))
    )
    return recovered.lower() == signed_action["signer"].lower()

# ── Example: sign and submit a bet ────────────────────────────────
signed_bet = sign_action("casino_bet", {
    "game":     "blackjack",
    "amount":   50,
    "currency": "XMR",
    "strategy": "basic_strategy_v2"
})

print(f"\nSigned bet:")
print(f"  Signature : {signed_bet['signature'][:20]}...")
print(f"  Valid     : {verify_action(signed_bet)}")

# Submit to Purple Flea with signature proof
resp = requests.post(
    "https://casino.purpleflea.com/api/bet",
    headers={"X-API-Key": os.environ["PF_API_KEY"]},
    json={**signed_bet["action"]["payload"], "proof": signed_bet}
)
print(f"  API response: {resp.status_code}")

# ── Generate DID Document ──────────────────────────────────────────
did_doc = {
    "@context": ["https://www.w3.org/ns/did/v1"],
    "id": DID,
    "verificationMethod": [{
        "id":                  f"{DID}#controller",
        "type":                "EcdsaSecp256k1VerificationKey2019",
        "controller":          DID,
        "blockchainAccountId": f"eip155:8453:{agent.address}"
    }],
    "authentication": [f"{DID}#controller"],
    "service": [{
        "id":              f"{DID}#purpleflea",
        "type":            "PurpleFlеaAgent",
        "serviceEndpoint": "https://purpleflea.com/agents/lookup?did=" + DID
    }]
}
with open("did.json", "w") as f:
    json.dump(did_doc, f, indent=2)
print(f"\nDID Document written to did.json")