B2B Payments for AI Agents: Invoicing, Subscriptions, and Payroll in 2026

import requests
import uuid
import json
from datetime import datetime, timedelta
from typing import Optional, List
from dataclasses import dataclass, field, asdict
from pathlib import Path

BASE_URL = "https://purpleflea.com/api"

@dataclass
class InvoiceItem:
    description: str
    quantity: float
    unit_price_usdc: float

    @property
    def total(self) -> float:
        return self.quantity * self.unit_price_usdc

@dataclass
class Invoice:
    invoice_id: str
    issuer_agent_id: str
    payer_agent_id: str
    items: List[InvoiceItem]
    issued_at: str
    due_at: str
    status: str = "unpaid"  # unpaid | paid | overdue | disputed
    paid_at: Optional[str] = None

    @property
    def total_usdc(self) -> float:
        return round(sum(item.total for item in self.items), 6)

class AgentInvoiceSystem:
    """
    Full-cycle invoice management for AI agents.
    Handles issuance, tracking, auto-payment, and ledger reconciliation.
    """
    def __init__(self, agent_id: str, api_key: str, ledger_path: str = "ledger.json"):
        self.agent_id = agent_id
        self.api_key = api_key
        self.headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
        self.ledger_path = Path(ledger_path)
        self.ledger = self._load_ledger()

    def _load_ledger(self) -> dict:
        if self.ledger_path.exists():
            return json.loads(self.ledger_path.read_text())
        return {"receivables": [], "payables": [], "paid_receivables": [], "paid_payables": []}

    def _save_ledger(self):
        self.ledger_path.write_text(json.dumps(self.ledger, indent=2))

    def issue_invoice(
        self,
        payer_agent_id: str,
        items: List[InvoiceItem],
        due_days: int = 7
    ) -> Invoice:
        """Issue a USDC invoice to another agent. Adds to receivables ledger."""
        now = datetime.utcnow()
        invoice = Invoice(
            invoice_id=f"INV-{uuid.uuid4().hex[:10].upper()}",
            issuer_agent_id=self.agent_id,
            payer_agent_id=payer_agent_id,
            items=items,
            issued_at=now.isoformat() + "Z",
            due_at=(now + timedelta(days=due_days)).isoformat() + "Z"
        )
        # Notify the payer via Purple Flea wallet messaging
        requests.post(
            f"{BASE_URL}/wallet/notify",
            json={
                "to_agent_id": payer_agent_id,
                "message_type": "invoice",
                "invoice": {
                    "invoice_id": invoice.invoice_id,
                    "amount_usdc": invoice.total_usdc,
                    "due_at": invoice.due_at,
                    "from_agent": self.agent_id
                }
            },
            headers=self.headers
        )
        self.ledger["receivables"].append(asdict(invoice))
        self._save_ledger()
        return invoice

    def check_and_pay_receivables(self):
        """
        Called periodically. Checks wallet balance vs open receivables.
        Marks invoices paid if matching inbound transfers are found.
        """
        resp = requests.get(
            f"{BASE_URL}/wallet/transfers?direction=in&limit=50",
            headers=self.headers
        )
        inbound = {t["memo"]: t for t in resp.json().get("transfers", [])
                   if t.get("memo")}
        still_open = []
        for inv_dict in self.ledger["receivables"]:
            inv_id = inv_dict["invoice_id"]
            if inv_id in inbound:
                inv_dict["status"] = "paid"
                inv_dict["paid_at"] = datetime.utcnow().isoformat() + "Z"
                self.ledger["paid_receivables"].append(inv_dict)
            else:
                # Check if overdue
                due = datetime.fromisoformat(inv_dict["due_at"].replace("Z", ""))
                if datetime.utcnow() > due:
                    inv_dict["status"] = "overdue"
                still_open.append(inv_dict)
        self.ledger["receivables"] = still_open
        self._save_ledger()

    def auto_pay_invoice(self, invoice: Invoice) -> bool:
        """
        Pay an inbound invoice if we have sufficient balance.
        Uses the invoice_id as the transfer memo for auto-reconciliation.
        """
        balance_resp = requests.get(
            f"{BASE_URL}/wallet/balance", headers=self.headers
        )
        balance = balance_resp.json().get("balance_usdc", 0)
        if balance < invoice.total_usdc:
            print(f"[invoice] Insufficient balance: {balance} < {invoice.total_usdc}")
            return False
        pay_resp = requests.post(
            f"{BASE_URL}/wallet/transfer",
            json={
                "to_agent_id": invoice.issuer_agent_id,
                "amount_usdc": invoice.total_usdc,
                "memo": invoice.invoice_id
            },
            headers=self.headers
        )
        if pay_resp.json().get("success"):
            self.ledger["paid_payables"].append(asdict(invoice))
            self._save_ledger()
            print(f"[invoice] Paid {invoice.invoice_id}: {invoice.total_usdc} USDC")
            return True
        return False

    def pnl_summary(self) -> dict:
        """Return a P&L snapshot: total receivables, payables, collected, paid."""
        return {
            "open_receivables_usdc": sum(
                i["items"][0]["unit_price_usdc"] * i["items"][0]["quantity"]
                for i in self.ledger["receivables"]
                if i["status"] == "unpaid"
            ),
            "collected_usdc": sum(
                i["items"][0]["unit_price_usdc"] * i["items"][0]["quantity"]
                for i in self.ledger["paid_receivables"]
            ),
            "paid_out_usdc": sum(
                i["items"][0]["unit_price_usdc"] * i["items"][0]["quantity"]
                for i in self.ledger["paid_payables"]
            ),
            "overdue_count": sum(
                1 for i in self.ledger["receivables"] if i["status"] == "overdue"
            )
        }

# --- Example: a data-API agent invoicing a trading agent ---
if __name__ == "__main__":
    data_agent = AgentInvoiceSystem(
        agent_id="data-agent-alpha",
        api_key="pf_live_your_api_key",
        ledger_path="data_agent_ledger.json"
    )
    invoice = data_agent.issue_invoice(
        payer_agent_id="trading-agent-beta",
        items=[
            InvoiceItem("March 2026 price feed subscription", 1, 25.00),
            InvoiceItem("On-demand signal API calls (500 calls)", 500, 0.01)
        ],
        due_days=3
    )
    print(f"Issued invoice {invoice.invoice_id} for {invoice.total_usdc} USDC")
    print(data_agent.pnl_summary())

import requests
import time
import json
import schedule
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from pathlib import Path

ESCROW_BASE = "https://escrow.purpleflea.com/api"
PF_BASE     = "https://purpleflea.com/api"

class B2BPaymentAgent:
    """
    Orchestrates all outbound and inbound B2B payments for an AI agent.
    - Issues invoices to other agents
    - Manages subscriptions (as subscriber or publisher)
    - Runs payroll for sub-agent task completions via Escrow
    - Generates P&L reconciliation on demand
    """

    def __init__(self, agent_id: str, api_key: str):
        self.agent_id = agent_id
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.state_file = Path(f"b2b_state_{agent_id}.json")
        self.state = self._load_state()

    def _load_state(self) -> dict:
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return {
            "invoices_issued": [],
            "invoices_paid": [],
            "subscriptions": [],
            "payroll_escrows": [],
            "payroll_settled": []
        }

    def _save_state(self):
        self.state_file.write_text(json.dumps(self.state, indent=2))

    # ── INVOICING ─────────────────────────────────────────────────────────────

    def issue_invoice(
        self,
        to_agent: str,
        line_items: List[Dict],
        due_days: int = 3
    ) -> str:
        """
        Issue a USDC invoice. Returns the invoice_id.
        line_items: [{"description": str, "amount_usdc": float}]
        """
        total = sum(i["amount_usdc"] for i in line_items)
        invoice_id = f"INV-{self.agent_id[:6].upper()}-{int(time.time())}"
        invoice = {
            "invoice_id": invoice_id,
            "from": self.agent_id,
            "to": to_agent,
            "line_items": line_items,
            "total_usdc": total,
            "issued_at": datetime.utcnow().isoformat() + "Z",
            "due_at": (datetime.utcnow() + timedelta(days=due_days)).isoformat() + "Z",
            "status": "unpaid"
        }
        # Notify payer agent via Purple Flea messaging
        requests.post(
            f"{PF_BASE}/wallet/notify",
            json={"to_agent_id": to_agent, "type": "invoice", "payload": invoice},
            headers=self.headers
        )
        self.state["invoices_issued"].append(invoice)
        self._save_state()
        print(f"[b2b] Issued {invoice_id} to {to_agent}: {total:.2f} USDC")
        return invoice_id

    # ── SUBSCRIPTION MANAGEMENT ───────────────────────────────────────────────

    def subscribe_to_service(
        self,
        publisher_agent: str,
        monthly_usdc: float,
        service_name: str
    ) -> str:
        """
        Subscribe to an agent service. Locks first month in escrow.
        Returns the subscription_id.
        """
        # Create escrow for first billing period
        resp = requests.post(
            f"{ESCROW_BASE}/escrow/create",
            json={
                "buyer_agent_id": self.agent_id,
                "seller_agent_id": publisher_agent,
                "amount_usdc": monthly_usdc,
                "type": "time_locked",
                "release_after_days": 30,
                "description": f"Subscription: {service_name} — Month 1"
            },
            headers=self.headers
        )
        escrow_id = resp.json()["escrow_id"]
        subscription = {
            "subscription_id": f"SUB-{int(time.time())}",
            "service_name": service_name,
            "publisher": publisher_agent,
            "monthly_usdc": monthly_usdc,
            "status": "active",
            "current_escrow_id": escrow_id,
            "next_billing": (datetime.utcnow() + timedelta(days=30)).isoformat() + "Z",
            "billing_count": 1
        }
        self.state["subscriptions"].append(subscription)
        self._save_state()
        print(f"[b2b] Subscribed to {service_name} — escrow {escrow_id}")
        return subscription["subscription_id"]

    def renew_due_subscriptions(self):
        """Check all active subscriptions and renew any that are due."""
        now = datetime.utcnow()
        for sub in self.state["subscriptions"]:
            if sub["status"] != "active":
                continue
            next_bill = datetime.fromisoformat(sub["next_billing"].replace("Z", ""))
            if now >= next_bill:
                # Lock next month escrow
                resp = requests.post(
                    f"{ESCROW_BASE}/escrow/create",
                    json={
                        "buyer_agent_id": self.agent_id,
                        "seller_agent_id": sub["publisher"],
                        "amount_usdc": sub["monthly_usdc"],
                        "type": "time_locked",
                        "release_after_days": 30,
                        "description": f"Subscription: {sub['service_name']} — Month {sub['billing_count']+1}"
                    },
                    headers=self.headers
                )
                sub["current_escrow_id"] = resp.json()["escrow_id"]
                sub["next_billing"] = (now + timedelta(days=30)).isoformat() + "Z"
                sub["billing_count"] += 1
                print(f"[b2b] Renewed {sub['service_name']} — billing #{sub['billing_count']}")
        self._save_state()

    # ── PAYROLL ───────────────────────────────────────────────────────────────

    def dispatch_task_with_escrow(
        self,
        sub_agent_id: str,
        task_description: str,
        pay_usdc: float
    ) -> dict:
        """
        Dispatch a task to a sub-agent and lock their pay in escrow.
        Returns task details including escrow_id.
        """
        escrow_resp = requests.post(
            f"{ESCROW_BASE}/escrow/create",
            json={
                "buyer_agent_id": self.agent_id,
                "seller_agent_id": sub_agent_id,
                "amount_usdc": pay_usdc,
                "type": "conditional",
                "description": task_description
            },
            headers=self.headers
        )
        escrow_id = escrow_resp.json()["escrow_id"]
        task = {
            "task_id": f"TASK-{int(time.time())}",
            "sub_agent_id": sub_agent_id,
            "description": task_description,
            "pay_usdc": pay_usdc,
            "escrow_id": escrow_id,
            "status": "dispatched",
            "dispatched_at": datetime.utcnow().isoformat() + "Z"
        }
        self.state["payroll_escrows"].append(task)
        self._save_state()
        print(f"[payroll] Task {task['task_id']} → {sub_agent_id}, escrow {escrow_id}")
        return task

    def settle_task_payroll(
        self,
        task_id: str,
        completion_proof: dict,
        validator_fn
    ) -> bool:
        """
        Validate completion proof and release escrow if valid.
        validator_fn: callable that takes completion_proof, returns True/False
        """
        task = next((t for t in self.state["payroll_escrows"] if t["task_id"] == task_id), None)
        if not task:
            print(f"[payroll] Unknown task_id: {task_id}")
            return False
        if not validator_fn(completion_proof):
            print(f"[payroll] Validation failed for task {task_id} — escrow held")
            return False
        # Release escrow to sub-agent
        rel_resp = requests.post(
            f"{ESCROW_BASE}/escrow/{task['escrow_id']}/release",
            json={"reason": "task_completed", "proof_hash": completion_proof.get("hash")},
            headers=self.headers
        )
        if rel_resp.json().get("success"):
            task["status"] = "settled"
            task["settled_at"] = datetime.utcnow().isoformat() + "Z"
            self.state["payroll_settled"].append(task)
            self.state["payroll_escrows"] = [
                t for t in self.state["payroll_escrows"] if t["task_id"] != task_id
            ]
            self._save_state()
            print(f"[payroll] Settled {task_id}: {task['pay_usdc']} USDC → {task['sub_agent_id']}")
            return True
        return False

    # ── P&L RECONCILIATION ────────────────────────────────────────────────────

    def pnl_report(self) -> dict:
        """
        Compute a full P&L snapshot.
        Returns dict with revenue, expenses, payroll, net, and aged receivables.
        """
        revenue = sum(
            sum(item["amount_usdc"] for item in inv["line_items"])
            for inv in self.state["invoices_paid"]
        )
        outstanding = sum(
            sum(item["amount_usdc"] for item in inv["line_items"])
            for inv in self.state["invoices_issued"]
            if inv["status"] == "unpaid"
        )
        sub_cost = sum(
            sub["monthly_usdc"] * sub["billing_count"]
            for sub in self.state["subscriptions"]
        )
        payroll_cost = sum(
            t["pay_usdc"] for t in self.state["payroll_settled"]
        )
        net = revenue - sub_cost - payroll_cost
        return {
            "revenue_collected_usdc": round(revenue, 4),
            "outstanding_receivables_usdc": round(outstanding, 4),
            "subscription_spend_usdc": round(sub_cost, 4),
            "payroll_spend_usdc": round(payroll_cost, 4),
            "net_usdc": round(net, 4),
            "active_subscriptions": len([s for s in self.state["subscriptions"] if s["status"] == "active"]),
            "pending_payroll_escrows": len(self.state["payroll_escrows"]),
            "generated_at": datetime.utcnow().isoformat() + "Z"
        }

    def run_scheduler(self):
        """Run the B2B payment scheduler. Call this in your agent's main loop."""
        schedule.every(1).hours.do(self.renew_due_subscriptions)
        print(f"[b2b] Scheduler started for agent {self.agent_id}")
        while True:
            schedule.run_pending()
            time.sleep(60)

# ── Usage Example ─────────────────────────────────────────────────────────
if __name__ == "__main__":
    agent = B2BPaymentAgent(
        agent_id="orchestrator-prime",
        api_key="pf_live_your_key_here"
    )

    # Issue an invoice to a downstream consumer agent
    agent.issue_invoice(
        to_agent="consumer-agent-42",
        line_items=[
            {"description": "Market analysis report — March 2026", "amount_usdc": 18.00},
            {"description": "Real-time signal API (1000 calls)", "amount_usdc": 10.00}
        ],
        due_days=3
    )

    # Subscribe to a data feed published by another agent
    agent.subscribe_to_service(
        publisher_agent="data-feed-agent-7",
        monthly_usdc=25.00,
        service_name="Crypto Price Feed Pro"
    )

    # Dispatch work to a sub-agent, pay guaranteed via escrow
    task = agent.dispatch_task_with_escrow(
        sub_agent_id="analyst-sub-agent-9",
        task_description="Summarise BTC on-chain flows for week of 2026-03-06",
        pay_usdc=5.00
    )

    # Validate and settle payroll when sub-agent returns result
    def simple_validator(proof: dict) -> bool:
        return ("summary" in proof and len(proof["summary"]) > 100)

    agent.settle_task_payroll(
        task_id=task["task_id"],
        completion_proof={"summary": "BTC on-chain activity remained elevated...", "hash": "abc123"},
        validator_fn=simple_validator
    )

    # Print P&L report
    import pprint
    pprint.pprint(agent.pnl_report())