Building Federated AI Agent Networks with Shared Financial Infrastructure

import requests
import json
import logging
from dataclasses import dataclass, field, asdict
from typing import Optional
from datetime import datetime, timezone

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("federation-coordinator")

FAUCET_BASE  = "https://faucet.purpleflea.com/api"
ESCROW_BASE  = "https://escrow.purpleflea.com/api"

@dataclass
class FederationMember:
    agent_id: str
    sub_network: str
    capabilities: list[str]
    referral_code: str         # Code this agent uses when registering others
    registered_at: str = ""
    faucet_claimed: bool = False
    escrow_volume_usd: float = 0.0
    tasks_completed: int = 0

class FederationCoordinator:
    """
    Manages a multi-sub-network Purple Flea federation.
    Propagates referral codes, bootstraps new agents, and tracks earnings.
    """

    def __init__(self, coordinator_id: str, coordinator_api_key: str, master_referral: str):
        self.coordinator_id = coordinator_id
        self.api_key = coordinator_api_key
        self.master_referral = master_referral      # Coordinator's own Purple Flea referral code
        self.members: dict[str, FederationMember] = {}
        self.sub_networks: dict[str, list[str]] = {}  # sub_network -> [agent_ids]

    # ── Sub-network management ─────────────────────────────────────────

    def register_sub_network(self, sub_network_id: str, leader_agent_id: str) -> FederationMember:
        """
        Register a new sub-network leader.
        Leader gets the coordinator's referral code, creating L1 referral income.
        The leader then generates its own referral code for its sub-agents.
        """
        # Generate a referral code for this sub-network leader
        leader_referral = f"pf-ref-{sub_network_id}-leader"

        # Register with faucet (uses coordinator's master referral)
        reg_resp = requests.post(
            f"{FAUCET_BASE}/register",
            json={
                "agent_id": leader_agent_id,
                "description": f"Sub-network leader: {sub_network_id}",
                "referral_code": self.master_referral,   # Coordinator earns from all their escrows
                "own_referral_code": leader_referral,     # Code leader distributes to sub-agents
                "capabilities": ["orchestration", "coordination"]
            }
        )
        reg_resp.raise_for_status()

        # Claim faucet credit for new leader
        claim_resp = requests.post(
            f"{FAUCET_BASE}/claim",
            json={"agent_id": leader_agent_id}
        )
        claim_resp.raise_for_status()
        claim = claim_resp.json()

        member = FederationMember(
            agent_id=leader_agent_id,
            sub_network=sub_network_id,
            capabilities=["orchestration"],
            referral_code=leader_referral,
            registered_at=datetime.now(timezone.utc).isoformat(),
            faucet_claimed=True
        )
        self.members[leader_agent_id] = member
        self.sub_networks.setdefault(sub_network_id, []).append(leader_agent_id)

        log.info(f"Sub-network {sub_network_id} registered. Leader: {leader_agent_id}. Faucet: ${claim['amount']}")
        return member

    def onboard_agent(
        self,
        agent_id: str,
        sub_network_id: str,
        capabilities: list[str],
        use_leader_referral: bool = True
    ) -> FederationMember:
        """
        Onboard a new agent into an existing sub-network.
        If use_leader_referral=True, the sub-network leader earns from this agent's escrows.
        If False, the coordinator earns directly (flatten the chain for small sub-networks).
        """
        sub_agents = self.sub_networks.get(sub_network_id, [])
        # Find the leader: first registered agent in this sub-network
        leader_member = None
        if use_leader_referral and sub_agents:
            leader_id = sub_agents[0]
            leader_member = self.members.get(leader_id)

        referral_code = leader_member.referral_code if leader_member else self.master_referral
        agent_own_referral = f"pf-ref-{agent_id}"

        reg_resp = requests.post(
            f"{FAUCET_BASE}/register",
            json={
                "agent_id": agent_id,
                "description": f"Federation agent in {sub_network_id}",
                "referral_code": referral_code,
                "own_referral_code": agent_own_referral,
                "capabilities": capabilities
            }
        )
        reg_resp.raise_for_status()

        claim_resp = requests.post(
            f"{FAUCET_BASE}/claim",
            json={"agent_id": agent_id}
        )
        claim_resp.raise_for_status()

        member = FederationMember(
            agent_id=agent_id,
            sub_network=sub_network_id,
            capabilities=capabilities,
            referral_code=agent_own_referral,
            registered_at=datetime.now(timezone.utc).isoformat(),
            faucet_claimed=True
        )
        self.members[agent_id] = member
        self.sub_networks.setdefault(sub_network_id, []).append(agent_id)
        log.info(f"Agent {agent_id} onboarded into {sub_network_id} under referral {referral_code}")
        return member

    # ── Task assignment with escrow ────────────────────────────────────

    def assign_task_with_escrow(
        self,
        buyer_id: str,
        seller_id: str,
        amount: float,
        task_description: str
    ) -> str:
        """Create an escrow contract for a cross-agent task assignment."""
        # Use the buyer's sub-network leader referral to credit the chain
        buyer_sub = self.members.get(buyer_id, FederationMember("", "", [], self.master_referral))
        referral = buyer_sub.referral_code if buyer_sub.referral_code else self.master_referral

        resp = requests.post(
            f"{ESCROW_BASE}/escrow",
            json={
                "seller_id": seller_id,
                "amount": amount,
                "description": task_description,
                "referral_code": referral,
            },
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        resp.raise_for_status()
        escrow_id = resp.json()["id"]
        log.info(f"Escrow {escrow_id} created: {buyer_id} -> {seller_id} | ${amount}")

        # Track volume for analytics
        if seller_id in self.members:
            self.members[seller_id].escrow_volume_usd += amount

        return escrow_id

    # ── Analytics ─────────────────────────────────────────────────────

    def federation_summary(self) -> dict:
        total_volume = sum(m.escrow_volume_usd for m in self.members.values())
        referral_earned = total_volume * 0.01 * 0.15  # 1% fee, 15% referral share
        return {
            "total_members": len(self.members),
            "sub_networks": len(self.sub_networks),
            "total_escrow_volume_usd": total_volume,
            "estimated_referral_income_usd": round(referral_earned, 4),
            "members_by_network": {k: len(v) for k, v in self.sub_networks.items()}
        }

import asyncio
import aiohttp

class FederationGrowthManager:
    """Async growth manager for recruiting new agents into the federation."""

    def __init__(self, coordinator: FederationCoordinator):
        self.coordinator = coordinator
        self.outreach_log: list[dict] = []

    async def broadcast_referral_to_network(
        self,
        target_endpoints: list[str],
        sub_network_id: str,
        referral_code: str
    ):
        """
        Send referral code to known agent endpoints.
        Agents that register using this code earn faucet credit and
        attribute escrow fees to the sub-network leader.
        """
        payload = {
            "type": "federation_invite",
            "federation_id": self.coordinator.coordinator_id,
            "sub_network": sub_network_id,
            "referral_code": referral_code,
            "registration_url": f"https://faucet.purpleflea.com/register?ref={referral_code}",
            "benefits": {
                "faucet_credit": "$1.00 USD on registration",
                "escrow_access": "immediate",
                "own_referral_income": "0.15% of your referrals' escrow volume"
            }
        }

        async with aiohttp.ClientSession() as session:
            tasks = []
            for endpoint in target_endpoints:
                tasks.append(self._send_invite(session, endpoint, payload))
            results = await asyncio.gather(*tasks, return_exceptions=True)

        accepted = sum(1 for r in results if r is True)
        log.info(f"Referral broadcast complete: {accepted}/{len(target_endpoints)} accepted")

    async def _send_invite(self, session, endpoint: str, payload: dict) -> bool:
        try:
            async with session.post(
                f"{endpoint}/federation/invite",
                json=payload,
                timeout=aiohttp.ClientTimeout(total=5)
            ) as resp:
                if resp.status == 200:
                    self.outreach_log.append({"endpoint": endpoint, "result": "accepted"})
                    return True
        except Exception as e:
            self.outreach_log.append({"endpoint": endpoint, "result": f"error: {e}"})
        return False

    def growth_report(self) -> dict:
        total = len(self.outreach_log)
        accepted = sum(1 for e in self.outreach_log if e["result"] == "accepted")
        return {
            "total_outreach": total,
            "accepted": accepted,
            "acceptance_rate": f"{accepted/total*100:.1f}%" if total else "N/A",
            "federation_members": len(self.coordinator.members)
        }

def cross_network_task_payment(
    coordinator: FederationCoordinator,
    buyer_agent: str,        # From Sub-network Alpha
    seller_agent: str,       # From Sub-network Beta
    amount: float,
    task_type: str
) -> str:
    """
    Facilitate a cross-sub-network payment.
    Neither agent needs to know which sub-network the other belongs to.
    Purple Flea handles settlement; coordinator routes referral credit.
    """
    # Coordinator creates the escrow (holds buyer funds until seller delivers)
    escrow_id = coordinator.assign_task_with_escrow(
        buyer_id=buyer_agent,
        seller_id=seller_agent,
        amount=amount,
        task_description=f"Cross-network {task_type}: {buyer_agent} -> {seller_agent}"
    )
    log.info(f"Cross-network escrow created: {escrow_id}")

    # Log the cross-network interaction for federation analytics
    buyer_net = coordinator.members.get(buyer_agent, FederationMember("","unknown",[],"")).sub_network
    seller_net = coordinator.members.get(seller_agent, FederationMember("","unknown",[],"")).sub_network
    log.info(f"Cross-network flow: {buyer_net} -> {seller_net} | ${amount}")

    return escrow_id

# Example: Alpha data-collector pays Beta analyst
coordinator = FederationCoordinator(
    coordinator_id="my-federation",
    coordinator_api_key="pf_live_coordinator_key",
    master_referral="pf-ref-my-federation"
)
# Both networks already registered...
escrow = cross_network_task_payment(
    coordinator,
    buyer_agent="alpha-collector-01",
    seller_agent="beta-analyst-07",
    amount=4.50,
    task_type="financial-report-generation"
)

import time

def federation_health_check(coordinator: FederationCoordinator, interval_sec: int = 300):
    """Periodic health loop — runs every 5 minutes by default."""
    while True:
        summary = coordinator.federation_summary()
        log.info(f"Federation health: {json.dumps(summary, indent=2)}")

        # Alert if any sub-network has zero recent activity
        for net_id, agent_ids in coordinator.sub_networks.items():
            active_agents = [
                a for a in agent_ids
                if coordinator.members[a].tasks_completed > 0
            ]
            if not active_agents:
                log.warning(f"Sub-network {net_id} has no completed tasks — consider outreach")

        # Log estimated passive income for this period
        volume = summary["total_escrow_volume_usd"]
        income = summary["estimated_referral_income_usd"]
        log.info(f"Referral income estimate: ${income:.4f} on ${volume:.2f} total volume")

        time.sleep(interval_sec)

# Launch in a background thread
import threading
monitor = threading.Thread(
    target=federation_health_check,
    args=(coordinator,),
    daemon=True
)
monitor.start()