UCP — Universal Commerce Protocol

# ucp_models.py — UCP core data structures
from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime
import uuid

class Price(BaseModel):
    amount: float
    currency: str  # "USD", "EUR", "USDC", "credits"
    unit: str      # "per-call", "per-token", "per-hour", "flat"

class ServiceListing(BaseModel):
    """A service an agent offers for sale."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    provider_did: str
    name: str
    description: str
    category: str  # "research", "coding", "analysis", "compute"
    price: Price
    terms: dict    # usage limits, SLAs, refund policy
    capabilities: list[str]
    created_at: datetime = Field(default_factory=datetime.utcnow)
    expires_at: Optional[datetime] = None

class RequestForQuote(BaseModel):
    """Buyer agent asks for a price for a specific service."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    buyer_did: str
    seller_did: str
    service_category: str
    requirements: dict  # what the buyer needs
    quantity: int = 1
    budget_cap: Optional[Price] = None  # buyer's maximum willingness to pay
    deadline: Optional[datetime] = None

class Offer(BaseModel):
    """Seller's binding quote in response to an RFQ."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    rfq_id: str
    seller_did: str
    buyer_did: str
    listing_id: str
    price: Price
    terms: dict
    valid_until: datetime
    delivery_estimate: str  # "immediate", "2h", "24h"
    proof_of_capability: Optional[str] = None  # URL to demo or sample

class ContractAcceptance(BaseModel):
    """Buyer accepts an offer — creates a binding contract."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    offer_id: str
    buyer_did: str
    accepted_at: datetime = Field(default_factory=datetime.utcnow)
    payment_method: str  # "stripe", "crypto", "credits", "invoice"
    payment_reference: Optional[str] = None

class Contract(BaseModel):
    """The binding agreement derived from acceptance."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    offer: Offer
    acceptance: ContractAcceptance
    status: Literal["active", "completed", "disputed", "canceled"] = "active"
    created_at: datetime = Field(default_factory=datetime.utcnow)
    buyer_signature: Optional[str] = None
    seller_signature: Optional[str] = None

class DeliveryReceipt(BaseModel):
    """Proof that service was delivered and received."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    contract_id: str
    delivered_at: datetime = Field(default_factory=datetime.utcnow)
    delivery_proof: str  # hash or URL of delivered artifact
    buyer_confirmed: bool = False
    payment_settled: bool = False

# ucp_seller.py — An agent that sells research services via UCP
from fastapi import FastAPI, HTTPException
from anthropic import Anthropic
from ucp_models import ServiceListing, RequestForQuote, Offer, Contract, DeliveryReceipt
from datetime import datetime, timedelta
import uuid, hashlib

app = FastAPI(title="Research Agent — UCP Seller")
anthropic = Anthropic()

SELLER_DID = "did:web:research-agent.acme.com"

# Published service listing
RESEARCH_LISTING = ServiceListing(
    id="listing-research-v1",
    provider_did=SELLER_DID,
    name="Research Report Service",
    description="Comprehensive web research and synthesis on any topic",
    category="research",
    price={"amount": 0.50, "currency": "USD", "unit": "per-call"},
    terms={
        "max_tokens_per_call": 4096,
        "sla": "p99 < 30s",
        "refund_policy": "full refund if quality score < 7/10",
        "usage_restrictions": "no harmful content"
    },
    capabilities=["web-search", "synthesis", "citation", "structured-output"],
    expires_at=datetime.utcnow() + timedelta(days=30)
)

# -------------------------------------------------------
# UCP Endpoints
# -------------------------------------------------------
@app.get("/ucp/listings")
async def get_listings():
    """Discovery: publish what this agent sells."""
    return {"listings": [RESEARCH_LISTING.model_dump()]}

@app.post("/ucp/rfq")
async def handle_rfq(rfq: RequestForQuote):
    """Receive a Request for Quote and return a binding offer."""
    # Validate the buyer's requirements
    requirements = rfq.requirements
    estimated_complexity = requirements.get("depth", "medium")

    # Dynamic pricing based on complexity
    price_map = {"quick": 0.25, "medium": 0.50, "deep": 1.50}
    unit_price = price_map.get(estimated_complexity, 0.50)

    # Check budget cap
    if rfq.budget_cap and rfq.budget_cap["amount"] < unit_price:
        raise HTTPException(
            status_code=400,
            detail=f"Budget cap ${rfq.budget_cap['amount']} below minimum price ${unit_price}"
        )

    offer = Offer(
        rfq_id=rfq.id,
        seller_did=SELLER_DID,
        buyer_did=rfq.buyer_did,
        listing_id=RESEARCH_LISTING.id,
        price={"amount": unit_price, "currency": "USD", "unit": "per-call"},
        terms=RESEARCH_LISTING.terms,
        valid_until=datetime.utcnow() + timedelta(hours=1),
        delivery_estimate="immediate"
    )
    return offer.model_dump()

@app.post("/ucp/contracts")
async def create_contract(acceptance):
    """Buyer accepted an offer — create the contract and deliver service."""
    contract_id = str(uuid.uuid4())

    # In production: verify payment before delivering
    # payment_verified = await verify_payment(acceptance["payment_reference"])

    # Deliver the service
    topic = acceptance.get("requirements", {}).get("topic", "general topic")
    response = anthropic.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": f"Research and write a comprehensive report on: {topic}"}],
    )
    result = response.content[0].text

    # Create delivery proof (hash of content)
    delivery_hash = hashlib.sha256(result.encode()).hexdigest()

    receipt = DeliveryReceipt(
        contract_id=contract_id,
        delivery_proof=delivery_hash,
        buyer_confirmed=False,
        payment_settled=True  # would be set after payment confirmation
    )

    return {
        "contract_id": contract_id,
        "status": "completed",
        "delivery": {"content": result, "proof_hash": delivery_hash},
        "receipt": receipt.model_dump()
    }

# ucp_buyer.py — An agent that purchases research services via UCP
import httpx, asyncio
from anthropic import Anthropic
from ucp_models import RequestForQuote
from datetime import datetime, timedelta
import uuid

anthropic = Anthropic()
BUYER_DID = "did:web:buyer-agent.mycompany.com"
SELLER_URL = "http://localhost:8002"

# Safety: human approval required for any payment above this
AUTONOMOUS_PAYMENT_LIMIT = 1.00  # USD

async def discover_services(seller_url: str) -> list[dict]:
    """Find what services a seller agent offers."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{seller_url}/ucp/listings")
        listings = resp.json()["listings"]
        print(f"Found {len(listings)} services:")
        for listing in listings:
            print(f"  - {listing['name']}: ${listing['price']['amount']}/{listing['price']['unit']}")
        return listings

async def request_quote(seller_url: str, topic: str, depth: str = "medium") -> dict:
    """Ask the seller for a price for a specific job."""
    rfq = RequestForQuote(
        buyer_did=BUYER_DID,
        seller_did="did:web:research-agent.acme.com",
        service_category="research",
        requirements={"topic": topic, "depth": depth},
        budget_cap={"amount": 2.00, "currency": "USD", "unit": "per-call"}
    )

    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{seller_url}/ucp/rfq", json=rfq.model_dump())
        offer = resp.json()
        print(f"Offer received: ${offer['price']['amount']} {offer['price']['currency']}")
        print(f"Valid until: {offer['valid_until']}")
        return offer

async def accept_offer_and_purchase(seller_url: str, offer: dict,
                                     topic: str, human_approval_fn=None) -> dict:
    """Accept the offer, pay, and receive the service."""
    price = offer["price"]["amount"]

    # Safety gate: require human approval for larger purchases
    if price > AUTONOMOUS_PAYMENT_LIMIT:
        if human_approval_fn is None:
            raise PermissionError(
                f"Purchase of ${price} exceeds autonomous limit ${AUTONOMOUS_PAYMENT_LIMIT}. "
                f"Human approval required."
            )
        approved = await human_approval_fn(offer)
        if not approved:
            return {"status": "rejected", "reason": "Human declined the purchase"}

    acceptance = {
        "offer_id": offer["id"],
        "buyer_did": BUYER_DID,
        "accepted_at": datetime.utcnow().isoformat(),
        "payment_method": "credits",
        "payment_reference": f"pay_{uuid.uuid4().hex[:8]}",
        "requirements": {"topic": topic}
    }

    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(f"{seller_url}/ucp/contracts", json=acceptance)
        result = resp.json()
        print(f"Service delivered! Proof hash: {result['delivery']['proof_hash'][:16]}...")
        return result

async def main():
    topic = "The economic impact of agentic AI on knowledge work"

    print("=== 1. Discover available services ===")
    listings = await discover_services(SELLER_URL)

    print("\n=== 2. Request a quote ===")
    offer = await request_quote(SELLER_URL, topic, depth="medium")

    print("\n=== 3. Accept and purchase ===")
    result = await accept_offer_and_purchase(SELLER_URL, offer, topic)

    print("\n=== 4. Delivered content ===")
    print(result["delivery"]["content"][:500] + "...")

asyncio.run(main())