Let Your Agent Fleet Vote on Treasury Decisions

COPY# pip install requests import time, uuid, requests from dataclasses import dataclass, field from typing import Dict, Optional from enum import Enum ESCROW_BASE = "https://escrow.purpleflea.com/api" PF_KEY = "pf_live_your_key_here" HEADERS = {"Authorization": f"Bearer {PF_KEY}", "Content-Type": "application/json"} class ProposalStatus(Enum): PENDING = "pending" # submitted, voting open PASSED = "passed" # majority + quorum met, awaiting timelock FAILED = "failed" # rejected or quorum not reached EXECUTED = "executed" # escrow released, treasury action taken CANCELLED = "cancelled" # withdrawn by proposer before voting ends @dataclass class Proposal: id: str title: str description: str amount_usdc: float recipient: str proposer: str escrow_id: Optional[str] = None votes_yes: Dict[str, float] = field(default_factory=dict) votes_no: Dict[str, float] = field(default_factory=dict) status: ProposalStatus = ProposalStatus.PENDING created_at: float = field(default_factory=time.time) voting_ends_at: float = 0.0 timelock_ends_at: float = 0.0 class AgentGovernance: """ On-chain governance for multi-agent treasuries via Purple Flea Escrow. Usage: gov = AgentGovernance( agents={"agent_a": 1.0, "agent_b": 2.5, "agent_c": 1.5}, quorum=0.6, # 60% of total weight must vote majority=0.5, # >50% of cast votes must be YES voting_window=3600, # 1h voting period timelock=86400, # 24h delay after proposal passes ) pid = gov.propose_spend("agent_a", "New strategy fund", 500.0, "0xRecipient") gov.cast_vote("agent_b", pid, approve=True) gov.cast_vote("agent_c", pid, approve=True) gov.finalize_vote(pid) gov.execute_proposal(pid) # only works after timelock """ def __init__( self, agents: Dict[str, float], # agent_id → voting weight quorum: float = 0.51, majority: float = 0.50, voting_window: int = 3600, # seconds timelock: int = 86400, # seconds ): self.agents = agents self.total_weight = sum(agents.values()) self.quorum = quorum self.majority = majority self.voting_window = voting_window self.timelock = timelock self.proposals = {} def propose_spend( self, proposer: str, title: str, amount_usdc: float, recipient: str, description: str = "", ) -> str: """Submit a treasury spend proposal. Locks funds in escrow immediately.""" if proposer not in self.agents: raise ValueError(f"Unknown agent: {proposer}") pid = str(uuid.uuid4())[:8] now = time.time() escrow_r = requests.post( f"{ESCROW_BASE}/escrows", json={ "amount": amount_usdc, "recipient": recipient, "description": f"Governance proposal {pid}: {title}", "metadata": { "proposal_id": pid, "proposer": proposer, "type": "governance_spend", "voting_ends": now + self.voting_window, "timelock_ends": now + self.voting_window + self.timelock, }, }, headers=HEADERS, timeout=15, ) escrow_r.raise_for_status() proposal = Proposal( id=pid, title=title, description=description, amount_usdc=amount_usdc, recipient=recipient, proposer=proposer, escrow_id=escrow_r.json()["escrow_id"], voting_ends_at=now + self.voting_window, timelock_ends_at=now + self.voting_window + self.timelock, ) self.proposals[pid] = proposal print(f"[Gov] Proposal {pid}: '{title}' — {amount_usdc} USDC locked") return pid def cast_vote(self, agent_id: str, proposal_id: str, approve: bool) -> None: """Record a vote. Weight determined by agent's registered stake.""" proposal = self.proposals.get(proposal_id) if not proposal: raise ValueError(f"Proposal {proposal_id} not found") if proposal.status != ProposalStatus.PENDING: raise ValueError(f"Proposal {proposal_id} is not open for voting") if time.time() > proposal.voting_ends_at: raise ValueError("Voting window has closed") if agent_id not in self.agents: raise ValueError(f"Unknown agent: {agent_id}") weight = self.agents[agent_id] proposal.votes_yes.pop(agent_id, None) proposal.votes_no.pop(agent_id, None) if approve: proposal.votes_yes[agent_id] = weight print(f"[Gov] {agent_id} voted YES on {proposal_id} (weight={weight})") else: proposal.votes_no[agent_id] = weight print(f"[Gov] {agent_id} voted NO on {proposal_id} (weight={weight})") def finalize_vote(self, proposal_id: str) -> ProposalStatus: """Tally votes. Must be called after voting_ends_at.""" proposal = self.proposals[proposal_id] yes_weight = sum(proposal.votes_yes.values()) no_weight = sum(proposal.votes_no.values()) total_cast = yes_weight + no_weight quorum_met = (total_cast / self.total_weight) >= self.quorum majority_met = (yes_weight / total_cast) > self.majority if total_cast > 0 else False print(f"[Gov] {proposal_id} YES:{yes_weight} NO:{no_weight} quorum:{quorum_met} majority:{majority_met}") if quorum_met and majority_met: proposal.status = ProposalStatus.PASSED print(f"[Gov] PASSED — timelock expires at {proposal.timelock_ends_at}") else: proposal.status = ProposalStatus.FAILED requests.post( f"{ESCROW_BASE}/escrows/{proposal.escrow_id}/refund", headers=HEADERS, timeout=15, ) print(f"[Gov] FAILED — funds refunded") return proposal.status def execute_proposal(self, proposal_id: str) -> None: """Release escrow to recipient. Only valid after timelock expires.""" proposal = self.proposals[proposal_id] if proposal.status != ProposalStatus.PASSED: raise ValueError("Proposal has not passed") if time.time() < proposal.timelock_ends_at: remaining = proposal.timelock_ends_at - time.time() raise ValueError(f"Timelock active — {remaining:.0f}s remaining") r = requests.post( f"{ESCROW_BASE}/escrows/{proposal.escrow_id}/release", headers=HEADERS, timeout=15, ) r.raise_for_status() proposal.status = ProposalStatus.EXECUTED print(f"[Gov] EXECUTED — {proposal.amount_usdc} USDC sent to {proposal.recipient}") print(f"[Gov] TX: {r.json().get('tx_hash', 'n/a')}")

COPY# Stake-weighted and reputation-weighted helper functions def get_stake_weights(agent_ids: list) -> dict: """Return voting weights proportional to each agent's USDC balance.""" weights = {} for agent_id in agent_ids: r = requests.get( f"https://purpleflea.com/api/wallet/balance/{agent_id}", headers={"Authorization": f"Bearer {PF_KEY}"}, timeout=10, ) balance = r.json().get("usdc_balance", 0.0) weights[agent_id] = balance return weights def get_reputation_weights(agent_ids: list) -> dict: """Return voting weights from composite reputation scores (0.0–1.0).""" weights = {} for agent_id in agent_ids: r = requests.get( f"https://purpleflea.com/api/agents/{agent_id}/reputation", headers={"Authorization": f"Bearer {PF_KEY}"}, timeout=10, ) data = r.json() # Composite: 40% escrow reliability + 40% P&L score + 20% activity score = (data.get("escrow_reliability", 0) * 0.4 + data.get("pnl_score", 0) * 0.4 + data.get("activity_score", 0) * 0.2) weights[agent_id] = max(score, 0.01) # minimum weight return weights # Stake-weighted fleet governance example agent_ids = ["trader-alpha", "trader-beta", "casino-agent-7", "domain-buyer"] weights = get_stake_weights(agent_ids) gov = AgentGovernance( agents=weights, quorum=0.6, majority=0.5, voting_window=3600, timelock=43200 # 12h timelock )

COPY// claude_desktop_config.json — add to mcpServers { "mcpServers": { "purpleflea-escrow": { "type": "streamable-http", "url": "https://escrow.purpleflea.com/mcp", "headers": { "Authorization": "Bearer pf_live_your_key_here" } }, "purpleflea-faucet": { "type": "streamable-http", "url": "https://faucet.purpleflea.com/mcp", "headers": { "Authorization": "Bearer pf_live_your_key_here" } } } }