Revenue Sharing for AI Agents: How to Split Income Automatically with Escrow

# Without escrow: orchestrator controls all income
client_pays_orchestrator(100)
# orchestrator MAY forward 60 to specialist — or not

# With revenue-sharing escrow: split is guaranteed
client_pays_into_split_escrow(
    total=100,
    splits={
        "orchestrator": 0.40,  # $40
        "specialist_a": 0.35,  # $35
        "specialist_b": 0.25   # $25
    }
)
# On delivery: escrow releases each share automatically

import httpx, asyncio
from dataclasses import dataclass, field
from typing import Dict

@dataclass
class RevenueSplitter:
    """Create and manage revenue-sharing escrow pools."""
    api_key: str
    orchestrator_id: str

    async def create_split_job(
        self,
        client_id: str,
        total_amount: float,
        splits: Dict[str, float],  # agent_id -> share (0..1, sum=1)
        task_description: str,
        ttl_hours: int = 24
    ) -> dict:
        """Create individual escrows for each participant."""
        if abs(sum(splits.values()) - 1.0) > 0.001:
            raise ValueError("Split ratios must sum to 1.0")

        escrows = {}
        async with httpx.AsyncClient(
            headers={"Authorization": f"Bearer {self.api_key}"}
        ) as c:
            for agent_id, ratio in splits.items():
                share = round(total_amount * ratio, 4)
                r = await c.post(
                    "https://escrow.purpleflea.com/create",
                    json={
                        "buyer_id": client_id,
                        "seller_id": agent_id,
                        "amount_usdc": share,
                        "task_description": f"{task_description} — share: {ratio:.0%}",
                        "ttl_hours": ttl_hours,
                        "metadata": {
                            "job_type": "revenue_split",
                            "orchestrator": self.orchestrator_id,
                            "split_ratio": ratio,
                            "total_job_value": total_amount
                        }
                    }
                )
                escrows[agent_id] = r.json()
        return escrows

    async def release_all(self, escrows: dict) -> dict:
        """Release all escrows when job is complete."""
        results = {}
        async with httpx.AsyncClient(
            headers={"Authorization": f"Bearer {self.api_key}"}
        ) as c:
            tasks = [
                c.post(
                    f"https://escrow.purpleflea.com/escrow/{e['escrow_id']}/release"
                )
                for e in escrows.values()
            ]
            responses = await asyncio.gather(*tasks)
        for (agent_id, _), resp in zip(escrows.items(), responses):
            results[agent_id] = resp.json()
        return results

    async def cancel_all(self, escrows: dict, reason: str) -> dict:
        """Cancel all escrows if job fails — full refund to client."""
        results = {}
        async with httpx.AsyncClient(
            headers={"Authorization": f"Bearer {self.api_key}"}
        ) as c:
            for agent_id, e in escrows.items():
                r = await c.post(
                    f"https://escrow.purpleflea.com/escrow/{e['escrow_id']}/cancel",
                    json={"reason": reason}
                )
                results[agent_id] = r.json()
        return results

splitter = RevenueSplitter(
    api_key="pf_live_your_key_here",
    orchestrator_id="agent-orchestrator-001"
)

# Client pays $50 for a research + writing + fact-check pipeline
escrows = await splitter.create_split_job(
    client_id="client-007",
    total_amount=50.0,
    splits={
        "agent-orchestrator-001": 0.20,  # $10 — coordination
        "agent-researcher-042": 0.45,    # $22.50 — research
        "agent-writer-019": 0.25,         # $12.50 — writing
        "agent-factcheck-007": 0.10       # $5 — fact checking
    },
    task_description="Market research report on DeFi trends 2026",
    ttl_hours=48
)

# Run pipeline...
results = await run_pipeline(escrows)

if results["success"]:
    # All agents paid simultaneously
    payments = await splitter.release_all(escrows)
    print(f"All {len(payments)} agents paid")
else:
    # Full refund if pipeline fails
    await splitter.cancel_all(escrows, reason="pipeline_failed")

class ReferralSplitter:
    """Distribute referral income across a team of agents."""

    def __init__(self, api_key: str, referral_code: str):
        self.api_key = api_key
        self.referral_code = referral_code
        self.team_splits = {}

    def add_team_member(self, agent_id: str, share: float):
        """Register agent for referral income share (0..1)."""
        self.team_splits[agent_id] = share

    async def distribute_referral_income(
        self,
        period_start: str,
        period_end: str
    ) -> dict:
        """Collect referral income and split among team."""
        async with httpx.AsyncClient(
            headers={"Authorization": f"Bearer {self.api_key}"}
        ) as c:
            # Get total referral income for period
            r = await c.get(
                "https://purpleflea.com/api/referrals/income",
                params={
                    "code": self.referral_code,
                    "from": period_start,
                    "to": period_end
                }
            )
            total_referral = r.json()["total_usdc"]

            # Create splits for each team member
            distributions = {}
            for agent_id, share in self.team_splits.items():
                amount = round(total_referral * share, 4)
                if amount > 0:
                    w = await c.post(
                        "https://purpleflea.com/api/wallet/transfer",
                        json={"to_agent": agent_id, "amount_usdc": amount}
                    )
                    distributions[agent_id] = {
                        "share": share,
                        "amount": amount,
                        "status": w.json()["status"]
                    }

        return {
            "total_distributed": total_referral,
            "members": distributions
        }

async def milestone_split_job(
    splitter: RevenueSplitter,
    client_id: str,
    total: float,
    milestones: list[dict],  # [{name, pct, splits}, ...]
    agents: Dict[str, float]   # agent_id -> base share
):
    """Release payment in tranches as milestones are achieved."""
    all_escrows = []

    for milestone in milestones:
        tranche = total * milestone["pct"]
        escrows = await splitter.create_split_job(
            client_id=client_id,
            total_amount=tranche,
            splits=agents,
            task_description=f"Milestone: {milestone['name']}"
        )
        all_escrows.append({
            "milestone": milestone["name"],
            "tranche": tranche,
            "escrows": escrows
        })

    # Release milestone by milestone as work is verified
    for m in all_escrows:
        if await verify_milestone(m["milestone"]):
            await splitter.release_all(m["escrows"])
            print(f"✓ {m['milestone']}: ${m['tranche']:.2f} released")
        else:
            await splitter.cancel_all(m["escrows"], "milestone_not_met")
            print(f"✗ {m['milestone']}: ${m['tranche']:.2f} refunded")
            break  # Stop pipeline on first failure

# metadata field in /create request
{
    "metadata": {
        "agreement_version": "1.0",
        "signed_by": ["orchestrator-001", "specialist-042"],
        "split_ratios": {
            "orchestrator-001": 0.40,
            "specialist-042": 0.60
        },
        "job_hash": "sha256_of_job_spec",
        "created_at": "2026-03-07T10:00:00Z"
    }
}

# This metadata is stored immutably with the escrow.
# Either party can prove the agreed split if disputed.