The Agent Pricing & Monetization Playbook#

Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox ( https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.

Referenced credentials (you supply these in your own environment):

• STRIPE_API_KEY: Stripe API key for card payment processing (scoped to payment intents only)

You built the agent. It works. It summarizes legal contracts in three seconds, triages support tickets with 94% accuracy, or generates ad copy that converts 2x better than templates. Now comes the question that kills more agent startups than bad models: how do you charge for it? Traditional SaaS pricing -- $49/month for Pro, $149/month for Enterprise -- was designed for human users clicking buttons in a dashboard. Agent services operate in a fundamentally different economic reality. Your costs are variable per request, not fixed per seat. Your customers might be other agents, not humans with credit cards. A single enterprise customer might send 800,000 API calls in January and 12 calls in February. And the transaction values are often sub-cent -- a fraction of a penny per tool call, aggregated across millions of invocations into meaningful revenue. The numbers tell the story. Monetizely's 2026 AI Pricing Report found that 62% of AI-powered products are moving to usage-based pricing by 2027, up from 34% in 2024. McKinsey projects the autonomous agent market will reach $52.6 billion by 2030. Bain's 2026 analysis of 200 AI SaaS companies revealed that usage-based pricing models achieve 127% net revenue retention versus 108% for seat-based models -- a gap that compounds dramatically over three years. Yet the same report found that 71% of agent builders launch with flat-rate pricing because they do not know how to implement metering, escrow, or marketplace discovery.

What You'll Learn#

• Chapter 1: The Agent Pricing Landscape
• Chapter 2: Usage-Based Metering with GreenHelix Billing
• Chapter 3: Outcome-Based Pricing and Settlement
• Chapter 4: Marketplace Listing and Agent Discovery
• Chapter 5: API Key Gating and Tiered Access
• Chapter 6: Agent-to-Agent Payments
• Chapter 7: Revenue Tracking and Analytics
• Chapter 8: Launch Checklist and Pricing Experiments
• Appendix: Tool Reference Quick Index

Full Guide#

The Agent Pricing & Monetization Playbook: Usage Metering, Outcome Billing, Marketplace Listing & A2A Payment Wiring#

You built the agent. It works. It summarizes legal contracts in three seconds, triages support tickets with 94% accuracy, or generates ad copy that converts 2x better than templates. Now comes the question that kills more agent startups than bad models: how do you charge for it?

Traditional SaaS pricing -- $49/month for Pro, $149/month for Enterprise -- was designed for human users clicking buttons in a dashboard. Agent services operate in a fundamentally different economic reality. Your costs are variable per request, not fixed per seat. Your customers might be other agents, not humans with credit cards. A single enterprise customer might send 800,000 API calls in January and 12 calls in February. And the transaction values are often sub-cent -- a fraction of a penny per tool call, aggregated across millions of invocations into meaningful revenue.

The numbers tell the story. Monetizely's 2026 AI Pricing Report found that 62% of AI-powered products are moving to usage-based pricing by 2027, up from 34% in 2024. McKinsey projects the autonomous agent market will reach $52.6 billion by 2030. Bain's 2026 analysis of 200 AI SaaS companies revealed that usage-based pricing models achieve 127% net revenue retention versus 108% for seat-based models -- a gap that compounds dramatically over three years. Yet the same report found that 71% of agent builders launch with flat-rate pricing because they do not know how to implement metering, escrow, or marketplace discovery.

This guide closes that gap. Every chapter contains production-ready Python code calling the GreenHelix A2A Commerce Gateway -- 128 tools accessible via a single HTTP endpoint at https://api.greenhelix.net/v1. By the end, you will have a metered billing system, outcome-based payment flows, a marketplace listing, tiered API key gating, agent-to-agent payment wiring, a revenue analytics dashboard, and a 14-day launch playbook. All of it working. All of it tested against the live gateway.

---

Table of Contents#

1. The Agent Pricing Landscape
2. Usage-Based Metering with GreenHelix Billing
3. Outcome-Based Pricing and Settlement
4. Marketplace Listing and Agent Discovery
5. API Key Gating and Tiered Access
6. Agent-to-Agent Payments
7. Revenue Tracking and Analytics
8. Launch Checklist and Pricing Experiments

---

Chapter 1: The Agent Pricing Landscape#

Why Traditional SaaS Pricing Breaks for AI Agents#

Per-seat pricing assumes a predictable relationship between the number of users and the value delivered. A project management tool with 50 seats serves 50 humans who each log in roughly daily, use roughly similar features, and generate roughly similar infrastructure costs. The marginal cost of adding seat 51 is near zero. Gross margins sit at 80-90%, and the pricing math is straightforward: charge enough per seat to cover the averaged cost with healthy margin, then grow by adding seats.

Agent services violate every assumption in that model.

Variable compute costs per request. When your agent calls GPT-4o to summarize a legal contract, the cost depends on the input token count. A 2-page NDA costs $0.003 to process. A 200-page merger agreement costs $0.31. If you charge a flat monthly fee, the customer who sends you merger agreements all day is unprofitable, and the customer who sends NDAs is subsidizing them. This is adverse selection -- the customers who use you most are the ones who cost you the most, and flat pricing attracts exactly those customers.

Sub-cent micro-transactions. Many agent services deliver value in tiny increments. A data enrichment call costs $0.002 to execute and delivers $0.008 of value. You need to charge $0.005 per call to maintain a healthy margin. Traditional payment processors cannot handle this -- Stripe's $0.30 per-transaction fee would consume 60x your revenue on a single call. You need a billing system designed for micro-transactions that aggregates usage and settles periodically.

Compressed gross margins. Traditional SaaS companies enjoy 80-90% gross margins because the marginal cost of serving an additional user is negligible. Agent services that depend on LLM inference typically operate at 50-60% gross margins because the primary cost -- model API calls -- scales linearly with usage. Some compute-heavy agent services operate at 30-40% margins. This compression means pricing errors are fatal faster. A 15% pricing mistake in traditional SaaS reduces margin from 85% to 70% -- painful but survivable. The same mistake in an agent service reduces margin from 55% to 40% -- potentially below the viability threshold.

Customers are machines. When your customers are other AI agents, not humans, the buying decision is algorithmic. An orchestrator agent evaluating your translation service against a competitor will switch the moment the competitor's quality-adjusted price drops below yours. There is no brand loyalty, no switching friction from learned habits, no reluctance to migrate. The agent changes one URL and one API key. This means your pricing must be continuously competitive, not just competitive at the moment of sale.

Three Pricing Models for Agent Services#

ModelHow It WorksBest ForGross MarginRevenue PredictabilityImplementation ComplexitySubscriptionFixed monthly fee for a tier of accessStable, predictable workloads; human-facing dashboards70-85%HighLowUsage-BasedPay per call, per token, per taskVariable workloads; agent-to-agent services50-65%MediumMediumOutcome-BasedPay only when a measurable outcome is achievedHigh-value tasks; trust-sensitive buyers40-70% (varies)LowHigh

Most successful agent services use a hybrid. A subscription base fee covers fixed infrastructure costs and provides revenue predictability. Usage-based charges on top capture value from high-volume customers without subsidizing them. Outcome-based pricing applies to premium features where the agent can guarantee measurable results.

The Pricing Decision Framework#

Answer these four questions to choose your model:

1. Is your cost per request variable or fixed? If variable (LLM inference, external API calls), usage-based pricing protects your margins. If fixed (static model, cached responses), subscription pricing maximizes simplicity.

2. Are your customers humans or agents? Human customers prefer predictable bills -- lean toward subscription with usage caps. Agent customers optimize on price-per-unit -- lean toward transparent usage-based pricing.

3. Can you measure outcomes reliably? If your agent solves support tickets and you can programmatically verify resolution (customer confirms, no reopens within 48 hours), outcome-based pricing commands a premium. If output quality is subjective, stick with usage-based.

4. What is your competitive landscape? If competitors charge per-call, you must offer per-call pricing or demonstrate clearly why your subscription is cheaper at the customer's expected volume. Price structure mismatch is a sales killer.

The GreenHelix Pricing Client#

Every code example in this guide calls the GreenHelix A2A Commerce Gateway via the REST API ( POST /v1/{tool}) with a JSON body of {"tool": "tool_name", "input": {...}}. Authentication is via Bearer token. Define the core client once and reference it throughout.

python

import requests
from typing import Any

GATEWAY_URL = os.environ.get("GREENHELIX_API_URL", "https://sandbox.greenhelix.net")

class GreenHelixClient:
    """Client for the GreenHelix A2A Commerce Gateway."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = GATEWAY_URL
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def execute(self, tool: str, input_data: dict[str, Any]) -> dict:
        """Execute a tool on the gateway."""
        response = requests.post(
            f"{self.base_url}/v1",
            json={"tool": tool, "input": input_data},
            headers=self.headers,
            timeout=30,
        )
        response.raise_for_status()
        return response.json()

    def execute_batch(self, calls: list[dict[str, Any]]) -> list[dict]:
        """Execute multiple tool calls sequentially."""
        results = []
        for call in calls:
            result = self.execute(call["tool"], call["input"])
            results.append(result)
        return results

# Initialize once, use everywhere
client = GreenHelixClient(api_key="your-greenhelix-api-key")

This client handles authentication, serialization, and error propagation. Every subsequent code example assumes this client is available in scope.

Key Takeaways

• Traditional per-seat SaaS pricing fails for agent services due to variable compute costs, sub-cent transactions, compressed margins, and algorithmic buyers.
• Three models dominate: subscription (simple, predictable), usage-based (fair, scalable), and outcome-based (premium, trust-building). Most services should hybrid.
• 62% of AI products are moving to usage-based pricing by 2027. The market is $52.6B by 2030. Pricing infrastructure is the bottleneck, not demand.
• Use the four-question decision framework: cost variability, customer type, outcome measurability, and competitive landscape.

---

Chapter 2: Usage-Based Metering with GreenHelix Billing#

The Metering Problem#

Usage-based pricing requires three capabilities that traditional billing systems lack: sub-cent precision, real-time aggregation, and retroactive volume discounts. When your agent processes 1.2 million calls in a month at $0.003 per call, you need a billing system that tracks every call, aggregates them accurately, applies volume tiers, and generates an invoice for $3,600 -- not $3,599.97, not $3,600.03, and definitely not a system that rounds each individual call to the nearest cent and charges $0.00 for most of them.

GreenHelix Billing tools solve this with three primitives: create_billing_plan defines pricing tiers, record_usage logs each metered event, and estimate_cost calculates the bill at any point in the billing cycle. Combined with set_volume_discount for rewarding high-volume customers, these four tools form a complete metering pipeline.

Designing Your Billing Plan#

Before writing code, define your metering dimensions. A metering dimension is any axis along which you want to charge differently. Common dimensions for agent services:

DimensionUnitExample RateUse CaseAPI callsPer call$0.001 - $0.01General-purpose toolsInput tokensPer 1K tokens$0.002 - $0.03LLM-backed servicesOutput tokensPer 1K tokens$0.005 - $0.06Generation-heavy servicesTasks completedPer task$0.05 - $5.00Multi-step workflowsCompute secondsPer second$0.0001 - $0.001Resource-intensive processingStorage (MB)Per MB/month$0.01 - $0.10Data persistence services

The golden rule: meter the dimension your customer can predict and control. Customers hate surprises. If your agent's token consumption varies wildly based on input complexity, meter per-task instead of per-token -- the customer can predict how many tasks they will submit but cannot predict your token usage.

Creating a Billing Plan#

python

def create_metered_billing_plan(client: GreenHelixClient) -> dict:
    """Create a usage-based billing plan with tiered pricing."""
    plan = client.execute("create_billing_plan", {
        "agent_id": "my-summarizer-agent",
        "plan_name": "summarizer-usage-v1",
        "billing_type": "usage",
        "currency": "USD",
        "metering_dimensions": [
            {
                "name": "api_calls",
                "unit": "call",
                "tiers": [
                    {"up_to": 1000, "rate": 0.008},
                    {"up_to": 10000, "rate": 0.006},
                    {"up_to": 100000, "rate": 0.004},
                    {"up_to": None, "rate": 0.002},  # unlimited
                ],
            },
            {
                "name": "input_tokens",
                "unit": "1k_tokens",
                "tiers": [
                    {"up_to": 500, "rate": 0.015},
                    {"up_to": 5000, "rate": 0.010},
                    {"up_to": None, "rate": 0.006},
                ],
            },
        ],
        "billing_period": "monthly",
        "invoice_day": 1,
    })
    print(f"Created billing plan: {plan['plan_id']}")
    return plan

plan = create_metered_billing_plan(client)

This plan uses graduated tiering: the first 1,000 calls cost $0.008 each, the next 9,000 cost $0.006, and so on. This rewards growth without giving away the first calls. An alternative is volume tiering where the entire volume is billed at the tier the customer reaches -- simpler but less predictable for the customer during the billing period.

Recording Usage in Real Time#

The key to accurate metering is recording usage at the point of execution, not in a batch job. Every time your agent processes a request, record it immediately.

python

import time
from functools import wraps
from typing import Callable

class UsageMeter:
    """Records usage against a GreenHelix billing plan."""

    def __init__(self, client: GreenHelixClient, agent_id: str, plan_id: str):
        self.client = client
        self.agent_id = agent_id
        self.plan_id = plan_id
        self._buffer: list[dict] = []
        self._buffer_limit = 50  # flush every 50 events

    def record(self, customer_id: str, dimension: str, quantity: float,
               metadata: dict | None = None) -> None:
        """Record a single usage event."""
        event = {
            "agent_id": self.agent_id,
            "plan_id": self.plan_id,
            "customer_id": customer_id,
            "dimension": dimension,
            "quantity": quantity,
            "timestamp": time.time(),
            "metadata": metadata or {},
        }
        self._buffer.append(event)

        if len(self._buffer) >= self._buffer_limit:
            self.flush()

    def flush(self) -> list[dict]:
        """Flush buffered events to the gateway."""
        if not self._buffer:
            return []

        results = []
        for event in self._buffer:
            result = self.client.execute("record_usage", event)
            results.append(result)
        self._buffer.clear()
        return results

    def metered(self, dimension: str, quantity_fn: Callable | None = None):
        """Decorator that automatically meters function calls."""
        def decorator(func):
            @wraps(func)
            def wrapper(customer_id: str, *args, **kwargs):
                result = func(customer_id, *args, **kwargs)
                qty = quantity_fn(result) if quantity_fn else 1
                self.record(customer_id, dimension, qty, {
                    "function": func.__name__,
                })
                return result
            return wrapper
        return decorator

# Initialize the meter
meter = UsageMeter(client, agent_id="my-summarizer-agent", plan_id=plan["plan_id"])

A Self-Metering Agent#

Here is a complete agent that meters its own API calls and token usage:

python

import tiktoken

class MeteredSummarizerAgent:
    """An agent that summarizes text and meters its own usage."""

    def __init__(self, client: GreenHelixClient, meter: UsageMeter,
                 model_name: str = "gpt-4o"):
        self.client = client
        self.meter = meter
        self.model_name = model_name
        self.encoding = tiktoken.encoding_for_model(model_name)

    def summarize(self, customer_id: str, text: str,
                  max_length: int = 200) -> dict:
        """Summarize text and record usage."""
        input_tokens = len(self.encoding.encode(text))

        # --- Your actual summarization logic here ---
        summary = self._call_model(text, max_length)
        # ---

        # Record the API call
        self.meter.record(customer_id, "api_calls", 1, {
            "input_length": len(text),
            "output_length": len(summary),
        })

        # Record token usage (in units of 1K tokens)
        self.meter.record(customer_id, "input_tokens", input_tokens / 1000, {
            "model": self.model_name,
        })

        return {
            "summary": summary,
            "input_tokens": input_tokens,
            "metered": True,
        }

    def get_customer_bill(self, customer_id: str) -> dict:
        """Get estimated cost for a customer in the current billing period."""
        self.meter.flush()  # ensure all events are recorded
        estimate = self.client.execute("estimate_cost", {
            "agent_id": "my-summarizer-agent",
            "customer_id": customer_id,
            "plan_id": self.meter.plan_id,
        })
        return estimate

    def _call_model(self, text: str, max_length: int) -> str:
        """Placeholder for actual model call."""
        # Replace with your LLM inference logic
        return text[:max_length] + "..."

# Usage
agent = MeteredSummarizerAgent(client, meter)
result = agent.summarize("customer-acme-corp", "Long contract text here...")
bill = agent.get_customer_bill("customer-acme-corp")
print(f"Current bill: ${bill['estimated_total']:.4f}")

Volume Discounts#

High-volume customers expect discounts. The set_volume_discount tool lets you define automatic discounts that apply when a customer crosses a usage threshold within a billing period.

python

def configure_volume_discounts(client: GreenHelixClient, plan_id: str) -> list:
    """Set up volume discounts for loyal high-volume customers."""
    discounts = []

    # 10% off after 50,000 calls in a month
    d1 = client.execute("set_volume_discount", {
        "plan_id": plan_id,
        "dimension": "api_calls",
        "threshold": 50000,
        "discount_percent": 10,
        "description": "High-volume discount: 10% off after 50K calls",
    })
    discounts.append(d1)

    # 20% off after 200,000 calls in a month
    d2 = client.execute("set_volume_discount", {
        "plan_id": plan_id,
        "dimension": "api_calls",
        "threshold": 200000,
        "discount_percent": 20,
        "description": "Enterprise discount: 20% off after 200K calls",
    })
    discounts.append(d2)

    # Token-based discount for heavy users
    d3 = client.execute("set_volume_discount", {
        "plan_id": plan_id,
        "dimension": "input_tokens",
        "threshold": 10000,  # 10M tokens
        "discount_percent": 15,
        "description": "Token volume discount: 15% off after 10M tokens",
    })
    discounts.append(d3)

    return discounts

discounts = configure_volume_discounts(client, plan["plan_id"])
for d in discounts:
    print(f"Discount active: {d['description']} (ID: {d['discount_id']})")

Metering Best Practices#

Flush on shutdown. Always call meter.flush() when your agent process terminates. Unbuffered events are lost revenue.

Idempotency keys. For critical transactions, include an idempotency key in the metadata to prevent double-counting if a flush retries after a network error:

python

import uuid

self.meter.record(customer_id, "api_calls", 1, {
    "idempotency_key": str(uuid.uuid4()),
    "request_id": request_id,
})

Estimate before large jobs. Before a customer submits a 500-document batch, show them the estimated cost:

python

def estimate_batch_cost(client: GreenHelixClient, customer_id: str,
                        plan_id: str, document_count: int,
                        avg_tokens_per_doc: int) -> dict:
    """Estimate cost for a batch job before execution."""
    estimate = client.execute("estimate_cost", {
        "agent_id": "my-summarizer-agent",
        "customer_id": customer_id,
        "plan_id": plan_id,
        "projected_usage": {
            "api_calls": document_count,
            "input_tokens": (document_count * avg_tokens_per_doc) / 1000,
        },
    })
    return estimate

cost = estimate_batch_cost(client, "customer-acme-corp", plan["plan_id"],
                           500, 3000)
print(f"Estimated batch cost: ${cost['estimated_total']:.2f}")

Key Takeaways

• Meter the dimension your customer can predict -- per-task is safer than per-token if your token usage varies.
• Use graduated tiering to reward growth without giving away initial usage free.
• Buffer usage events for performance but flush immediately on shutdown. Lost events are lost revenue.
• Always offer cost estimates before large jobs. Surprise bills destroy trust and cause churn.
• Volume discounts at 50K+ and 200K+ thresholds retain your most valuable customers.

---

Chapter 3: Outcome-Based Pricing and Settlement#

Charging for Results, Not Activity#

Usage-based pricing charges for inputs: calls made, tokens consumed, compute used. Outcome-based pricing charges for outputs: tickets resolved, leads qualified, documents classified correctly. The distinction matters because it aligns incentives. When you charge per call, the customer pays whether your agent succeeds or fails. When you charge per outcome, you only earn when you deliver value. This makes the customer's purchasing decision trivial -- they are paying for guaranteed results, not hopeful attempts.

The challenge is mechanics. How do you define an outcome? How do you verify it programmatically? How do you handle disputes when the customer disagrees? GreenHelix solves this with escrow: the customer locks funds with create_payment_intent, you do the work, a verification step confirms the outcome, and release_escrow moves the funds to you. If the outcome is not achieved, the customer gets their money back. If there is a disagreement, the dispute resolution tools adjudicate.

Defining Measurable Outcomes#

Not every task has a measurable outcome. Use this matrix to determine if outcome-based pricing is viable for your service:

Task TypeMeasurable OutcomeVerification MethodOutcome-Based Viable?Support ticket resolutionTicket marked resolved, no reopen in 48hStatus check via APIYesLead qualificationLead meets 5/7 criteria, enters CRM pipelineCRM API callbackYesDocument classificationClassification matches expert labelHeld-out test set comparisonYesCode generationCode passes test suiteAutomated test executionYesCreative writing"Good" copySubjective; no programmatic checkNoData enrichmentFields populated, accuracy > thresholdSpot-check sample verificationPartialTranslationAccurate translationBLEU score above thresholdPartial

The rule: if you can write a Python function that returns True when the outcome is achieved and False when it is not, outcome-based pricing works. If verification requires a human judgment call, stick with usage-based.

Escrow-Based Outcome Payment Flow#

The flow has five steps:

1. Customer creates a payment intent with escrow
2. Your agent performs the work
3. A verification function checks the outcome
4. If verified, escrow releases to you
5. If not verified, escrow refunds to customer

python

import time

class OutcomeBasedAgent:
    """Agent that only charges when it delivers a verified outcome."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def accept_task(self, customer_id: str, task: dict,
                    price: float) -> dict:
        """Accept a task with escrow-backed outcome pricing."""
        # Step 1: Customer locks funds in escrow
        intent = self.client.execute("create_payment_intent", {
            "from_agent": customer_id,
            "to_agent": self.agent_id,
            "amount": price,
            "currency": "USD",
            "escrow": True,
            "description": f"Outcome payment: {task['type']}",
            "metadata": {
                "task_id": task["id"],
                "outcome_criteria": task["success_criteria"],
                "timeout_hours": task.get("timeout_hours", 24),
            },
        })

        return {
            "payment_intent_id": intent["payment_intent_id"],
            "escrow_status": "locked",
            "task": task,
            "price": price,
        }

    def execute_and_settle(self, task_context: dict) -> dict:
        """Execute the task, verify the outcome, settle the payment."""
        task = task_context["task"]
        payment_intent_id = task_context["payment_intent_id"]

        # Step 2: Perform the work
        result = self._perform_task(task)

        # Step 3: Verify the outcome
        verified = self._verify_outcome(task, result)

        if verified:
            # Step 4a: Release escrow to seller (us)
            settlement = self.client.execute("release_escrow", {
                "payment_intent_id": payment_intent_id,
                "release_to": self.agent_id,
                "verification_proof": {
                    "outcome_met": True,
                    "verification_method": result["verification_method"],
                    "evidence": result["evidence"],
                    "timestamp": time.time(),
                },
            })
            return {
                "status": "completed_and_paid",
                "result": result,
                "settlement": settlement,
            }
        else:
            # Step 4b: Refund escrow to customer
            refund = self.client.execute("release_escrow", {
                "payment_intent_id": payment_intent_id,
                "release_to": task_context["task"]["customer_id"],
                "verification_proof": {
                    "outcome_met": False,
                    "reason": result.get("failure_reason", "Outcome not achieved"),
                },
            })
            return {
                "status": "failed_and_refunded",
                "result": result,
                "refund": refund,
            }

    def _perform_task(self, task: dict) -> dict:
        """Execute the task. Override in subclasses."""
        raise NotImplementedError

    def _verify_outcome(self, task: dict, result: dict) -> bool:
        """Verify the outcome meets success criteria. Override in subclasses."""
        raise NotImplementedError

Example: Support Ticket Resolution Agent#

Here is a concrete implementation that resolves support tickets and only charges when the ticket stays resolved:

python

class TicketResolutionAgent(OutcomeBasedAgent):
    """Resolves support tickets. Charges only when resolution sticks."""

    def _perform_task(self, task: dict) -> dict:
        """Resolve a support ticket using AI analysis."""
        ticket = task["ticket"]

        # Analyze the ticket and generate a resolution
        # (Your actual resolution logic goes here)
        resolution = self._generate_resolution(ticket)

        # Apply the resolution to the ticketing system
        applied = self._apply_resolution(ticket["id"], resolution)

        return {
            "ticket_id": ticket["id"],
            "resolution": resolution,
            "applied": applied,
            "verification_method": "status_check_48h",
            "evidence": {
                "resolution_text": resolution["text"],
                "applied_at": time.time(),
            },
        }

    def _verify_outcome(self, task: dict, result: dict) -> bool:
        """Check if the ticket is resolved and stays resolved.

        For immediate settlement, check current status.
        For delayed verification (48h), use a webhook or polling job.
        """
        ticket_id = result["ticket_id"]
        criteria = task["success_criteria"]

        # Check ticket status
        status_ok = result["applied"]

        # Check customer satisfaction if required
        if criteria.get("require_csat", False):
            csat = self._get_csat_score(ticket_id)
            return status_ok and csat >= criteria.get("min_csat", 4)

        return status_ok

    def _generate_resolution(self, ticket: dict) -> dict:
        """Generate resolution for a ticket."""
        return {"text": f"Resolution for: {ticket['subject']}", "confidence": 0.92}

    def _apply_resolution(self, ticket_id: str, resolution: dict) -> bool:
        """Apply resolution to ticketing system."""
        return resolution["confidence"] >= 0.85

    def _get_csat_score(self, ticket_id: str) -> float:
        """Get customer satisfaction score for a resolved ticket."""
        return 4.5  # Placeholder

# Wire it together
ticket_agent = TicketResolutionAgent(client, agent_id="ticket-resolver-v1")

# Customer submits a ticket with outcome-based pricing
task_ctx = ticket_agent.accept_task(
    customer_id="customer-support-co",
    task={
        "id": "task-001",
        "type": "ticket_resolution",
        "customer_id": "customer-support-co",
        "ticket": {
            "id": "TICK-4829",
            "subject": "API returning 500 on batch uploads",
            "body": "Since yesterday, batch uploads over 50 items fail with...",
            "priority": "high",
        },
        "success_criteria": {
            "ticket_resolved": True,
            "require_csat": False,
        },
        "timeout_hours": 4,
    },
    price=2.50,  # $2.50 per resolved ticket
)

# Execute and settle
result = ticket_agent.execute_and_settle(task_ctx)
print(f"Status: {result['status']}")
# Output: "Status: completed_and_paid" or "Status: failed_and_refunded"

Handling Disputes#

When the customer disagrees with the verification outcome -- they believe the ticket was not actually resolved, or the resolution caused a new problem -- the dispute tools provide a structured resolution path.

python

def file_outcome_dispute(client: GreenHelixClient, payment_intent_id: str,
                         reason: str, evidence: dict) -> dict:
    """File a dispute when outcome verification is contested."""
    dispute = client.execute("create_dispute", {
        "payment_intent_id": payment_intent_id,
        "reason": reason,
        "evidence": evidence,
        "requested_resolution": "full_refund",
    })
    return dispute

def respond_to_dispute(client: GreenHelixClient, dispute_id: str,
                       counter_evidence: dict) -> dict:
    """Respond to a customer dispute with counter-evidence."""
    response = client.execute("respond_to_dispute", {
        "dispute_id": dispute_id,
        "counter_evidence": counter_evidence,
        "proposed_resolution": "partial_refund",
        "partial_amount": 1.25,  # offer 50% back as goodwill
    })
    return response

# Customer disputes
dispute = file_outcome_dispute(client, task_ctx["payment_intent_id"],
    reason="Ticket reopened within 2 hours",
    evidence={"reopen_timestamp": "2026-04-06T14:30:00Z", "ticket_id": "TICK-4829"})

# Agent responds
response = respond_to_dispute(client, dispute["dispute_id"],
    counter_evidence={
        "original_resolution_valid": True,
        "reopen_cause": "customer_added_new_requirement",
        "new_issue_unrelated": True,
    })

Key Takeaways

• Outcome-based pricing aligns incentives: you earn only when you deliver value, making the customer's purchasing decision easy.
• Use escrow as the settlement primitive -- create_payment_intent with escrow locks funds, release_escrow settles based on verification.
• Every outcome needs a programmatic verification function. If you cannot write verify() -> bool, do not use outcome-based pricing for that task.
• Dispute tools are the safety net. Design your resolution response to include counter-evidence and offer partial refunds as goodwill when appropriate.
• Price outcomes 3-5x higher than equivalent usage-based pricing. The customer pays more per unit but gets guaranteed results.

---

Chapter 4: Marketplace Listing and Agent Discovery#

From Invisible to Discoverable#

You have a working agent with billing and payment flows. Nobody can find it. The GreenHelix Marketplace is a service registry where agents publish their capabilities and other agents (or humans) discover them. Think of it as a programmatic app store where the buyers are autonomous software, not humans browsing a website.

Discovery in agent commerce follows the A2A (Agent-to-Agent) protocol pattern. Each agent publishes an Agent Card -- a structured JSON document describing capabilities, pricing, authentication requirements, and service endpoints. Other agents query the marketplace using search_services or best_match to find services that fit their needs. The marketplace handles ranking, relevance scoring, and trust-weighted filtering.

The Agent Card Format#

An Agent Card is the machine-readable equivalent of a product listing. It follows the A2A protocol specification:

json

{
  "name": "contract-summarizer-v2",
  "description": "Summarizes legal contracts with 94% accuracy. Supports NDAs, MSAs, SOWs, and employment agreements.",
  "url": "https://your-agent.example.com",
  "version": "2.1.0",
  "capabilities": {
    "streaming": false,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "summarize-contract",
      "name": "Contract Summarization",
      "description": "Produces a structured summary of a legal contract with key terms, obligations, and risk flags.",
      "inputModes": ["text/plain", "application/pdf"],
      "outputModes": ["application/json", "text/markdown"]
    }
  ],
  "pricing": {
    "model": "usage",
    "rate": 0.005,
    "unit": "per_call",
    "currency": "USD",
    "volume_discounts": true
  },
  "authentication": {
    "schemes": ["bearer"]
  }
}

End-to-End Listing Workflow#

The complete workflow from identity registration to first discoverable search:

python

class MarketplaceListing:
    """Manages the complete lifecycle of a marketplace listing."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def register_identity(self, display_name: str,
                          description: str) -> dict:
        """Step 1: Register agent identity on the network."""
        identity = self.client.execute("register_agent", {
            "agent_id": self.agent_id,
            "display_name": display_name,
            "description": description,
            "capabilities": ["text-summarization", "contract-analysis"],
            "metadata": {
                "version": "2.1.0",
                "framework": "custom",
                "model": "gpt-4o",
            },
        })
        return identity

    def create_wallet(self, initial_deposit: float = 0) -> dict:
        """Step 2: Create a wallet to receive payments."""
        wallet = self.client.execute("create_wallet", {
            "agent_id": self.agent_id,
            "currency": "USD",
            "initial_deposit": initial_deposit,
        })
        return wallet

    def publish_service(self, agent_card: dict) -> dict:
        """Step 3: Publish the service to the marketplace."""
        service = self.client.execute("register_service", {
            "agent_id": self.agent_id,
            "service_name": agent_card["name"],
            "description": agent_card["description"],
            "agent_card": agent_card,
            "tags": [
                "legal", "summarization", "contracts",
                "nlp", "document-processing",
            ],
            "pricing": agent_card["pricing"],
            "sla": {
                "avg_response_time_ms": 3000,
                "uptime_percent": 99.5,
                "max_concurrent_requests": 100,
            },
        })
        return service

    def verify_discoverable(self) -> dict:
        """Step 4: Verify the service appears in search results."""
        results = self.client.execute("search_services", {
            "query": "contract summarization legal",
            "tags": ["legal", "summarization"],
            "max_results": 10,
        })
        # Check if our service appears
        found = any(
            r["agent_id"] == self.agent_id
            for r in results.get("services", [])
        )
        return {
            "discoverable": found,
            "total_results": len(results.get("services", [])),
            "our_rank": next(
                (i + 1 for i, r in enumerate(results.get("services", []))
                 if r["agent_id"] == self.agent_id),
                None,
            ),
        }

# Execute the full workflow
listing = MarketplaceListing(client, "contract-summarizer-v2")

# Step 1: Identity
identity = listing.register_identity(
    display_name="Contract Summarizer v2",
    description="Summarizes legal contracts with 94% accuracy",
)
print(f"Identity registered: {identity['agent_id']}")

# Step 2: Wallet
wallet = listing.create_wallet()
print(f"Wallet created: {wallet['wallet_id']}")

# Step 3: Publish
agent_card = {
    "name": "contract-summarizer-v2",
    "description": "Summarizes legal contracts with 94% accuracy. "
                   "Supports NDAs, MSAs, SOWs, and employment agreements.",
    "pricing": {
        "model": "usage",
        "rate": 0.005,
        "unit": "per_call",
        "currency": "USD",
        "volume_discounts": True,
    },
}
service = listing.publish_service(agent_card)
print(f"Service published: {service['service_id']}")

# Step 4: Verify
discovery = listing.verify_discoverable()
print(f"Discoverable: {discovery['discoverable']}, Rank: {discovery['our_rank']}")

Optimizing for Discovery with best_match#

The search_services tool returns keyword-matched results. The best_match tool uses semantic matching with trust weighting -- it finds the service that best fits a natural-language description of what the buyer needs, weighted by the seller's reputation score.

python

def find_best_service(client: GreenHelixClient, need_description: str,
                      budget: float, required_tags: list[str] | None = None) -> dict:
    """Find the single best service for a specific need."""
    match = client.execute("best_match", {
        "query": need_description,
        "max_price": budget,
        "required_tags": required_tags or [],
        "min_trust_score": 0.7,
        "sort_by": "relevance_trust_weighted",
    })
    return match

# A buyer agent looking for our service
best = find_best_service(
    client,
    need_description="I need to summarize 50 legal contracts (NDAs and MSAs) "
                     "and extract key terms, obligations, and renewal dates.",
    budget=0.01,  # per call
    required_tags=["legal"],
)
print(f"Best match: {best['service_name']} by {best['agent_id']}")
print(f"Price: ${best['pricing']['rate']}/{best['pricing']['unit']}")
print(f"Trust score: {best['trust_score']}")

Listing Optimization Checklist#

Your marketplace rank depends on these factors. Optimize each:

FactorWeightHow to ImproveRelevance to query30%Use specific, descriptive tags; include key terms in descriptionTrust score25%Complete tasks successfully, build claim chains (Chapter 7)Price competitiveness20%Benchmark against search_services results for your categoryResponse time SLA15%Set realistic SLAs and meet them; under-promise, over-deliverRecency10%Update your listing periodically to signal active maintenance

Key Takeaways

• Marketplace listing follows four steps: register identity, create wallet, publish service with Agent Card, verify discoverability.
• The Agent Card is your machine-readable product page. Include accurate pricing, SLAs, and capability descriptions.
• Use best_match over search_services for trust-weighted semantic matching when buying, and optimize your listing for it when selling.
• Tags matter more than descriptions for keyword search. Use 5-8 specific tags covering your service's domain, capability, and input/output types.
• Update your listing periodically. Stale listings rank lower than active ones.

---

Chapter 5: API Key Gating and Tiered Access#

Free, Pro, and Enterprise Tiers#

Most successful agent services offer multiple access tiers. The free tier drives adoption and lets customers evaluate your service without commitment. The pro tier captures individual developers and small teams who need higher limits. The enterprise tier serves large-scale buyers who need custom SLAs, priority support, and volume pricing.

GreenHelix Paywall tools implement this pattern with three primitives: create_api_key generates tier-scoped keys, validate_key checks a key's tier and status on every request, and get_tier_config returns the rate limits and feature flags for a tier. Combined with check_rate_limit, these tools give you a complete access control system.

Defining Tier Configuration#

python

TIER_CONFIGS = {
    "free": {
        "rate_limit_per_minute": 10,
        "rate_limit_per_day": 100,
        "max_input_tokens": 4000,
        "features": ["summarize"],
        "price_per_month": 0,
        "support": "community",
    },
    "pro": {
        "rate_limit_per_minute": 100,
        "rate_limit_per_day": 10000,
        "max_input_tokens": 32000,
        "features": ["summarize", "extract_terms", "risk_analysis", "batch"],
        "price_per_month": 49,
        "support": "email",
    },
    "enterprise": {
        "rate_limit_per_minute": 1000,
        "rate_limit_per_day": 500000,
        "max_input_tokens": 128000,
        "features": [
            "summarize", "extract_terms", "risk_analysis",
            "batch", "custom_models", "dedicated_endpoint", "sla_99_9",
        ],
        "price_per_month": 499,
        "support": "dedicated_slack",
    },
}

Creating and Managing API Keys#

python

class TierManager:
    """Manages API keys and tier-based access control."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def create_key(self, customer_id: str, tier: str) -> dict:
        """Create an API key for a customer at a specific tier."""
        if tier not in TIER_CONFIGS:
            raise ValueError(f"Unknown tier: {tier}. Must be one of: "
                             f"{list(TIER_CONFIGS.keys())}")

        key = self.client.execute("create_api_key", {
            "agent_id": self.agent_id,
            "customer_id": customer_id,
            "tier": tier,
            "rate_limit": TIER_CONFIGS[tier]["rate_limit_per_minute"],
            "features": TIER_CONFIGS[tier]["features"],
            "metadata": {
                "created_reason": "self_service_signup",
                "max_input_tokens": TIER_CONFIGS[tier]["max_input_tokens"],
            },
        })
        return key

    def validate_request(self, api_key: str) -> dict:
        """Validate an API key and return its tier configuration."""
        validation = self.client.execute("validate_key", {
            "api_key": api_key,
            "agent_id": self.agent_id,
        })

        if not validation.get("valid"):
            return {"authorized": False, "reason": validation.get("reason", "Invalid key")}

        tier = validation["tier"]
        config = self.client.execute("get_tier_config", {
            "agent_id": self.agent_id,
            "tier": tier,
        })

        return {
            "authorized": True,
            "customer_id": validation["customer_id"],
            "tier": tier,
            "config": config,
        }

    def check_rate_limit(self, api_key: str, customer_id: str) -> dict:
        """Check if the customer has exceeded their rate limit."""
        check = self.client.execute("check_rate_limit", {
            "agent_id": self.agent_id,
            "customer_id": customer_id,
            "api_key": api_key,
        })
        return {
            "allowed": check.get("allowed", False),
            "remaining": check.get("remaining", 0),
            "reset_at": check.get("reset_at"),
        }

tier_mgr = TierManager(client, "contract-summarizer-v2")

# Create keys for different customers
free_key = tier_mgr.create_key("startup-alice", "free")
pro_key = tier_mgr.create_key("agency-bob", "pro")
enterprise_key = tier_mgr.create_key("bigcorp-charlie", "enterprise")

FastAPI Middleware for Tier-Based Gating#

Here is production-ready middleware that gates every endpoint by tier, enforces rate limits, and rejects requests that exceed the customer's access level:

python

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
from functools import lru_cache
import time

app = FastAPI(title="Contract Summarizer API")

# Initialize tier manager at startup
tier_mgr = TierManager(
    GreenHelixClient(api_key="your-greenhelix-api-key"),
    "contract-summarizer-v2",
)

async def require_auth(request: Request) -> dict:
    """Dependency that validates API key and enforces tier access."""
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing Bearer token")

    api_key = auth_header.removeprefix("Bearer ").strip()

    # Validate the key
    auth = tier_mgr.validate_request(api_key)
    if not auth["authorized"]:
        raise HTTPException(status_code=403, detail=auth["reason"])

    # Check rate limit
    rate_check = tier_mgr.check_rate_limit(api_key, auth["customer_id"])
    if not rate_check["allowed"]:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded. Resets at {rate_check['reset_at']}",
            headers={"Retry-After": str(rate_check.get("retry_after", 60))},
        )

    return auth

def require_feature(feature: str):
    """Dependency factory that checks if the customer's tier includes a feature."""
    async def checker(auth: dict = Depends(require_auth)):
        tier_features = auth["config"].get("features", [])
        if feature not in tier_features:
            raise HTTPException(
                status_code=403,
                detail=f"Feature '{feature}' requires a higher tier. "
                       f"Current tier: {auth['tier']}. "
                       f"Available features: {tier_features}",
            )
        return auth
    return checker

def require_tier(minimum_tier: str):
    """Dependency factory that enforces a minimum tier level."""
    tier_order = {"free": 0, "pro": 1, "enterprise": 2}

    async def checker(auth: dict = Depends(require_auth)):
        if tier_order.get(auth["tier"], 0) < tier_order.get(minimum_tier, 0):
            raise HTTPException(
                status_code=403,
                detail=f"This endpoint requires '{minimum_tier}' tier or above. "
                       f"Current tier: {auth['tier']}.",
            )
        return auth
    return checker

@app.post("/summarize")
async def summarize(request: Request, auth: dict = Depends(require_auth)):
    """Summarize a contract. Available on all tiers."""
    body = await request.json()
    # Enforce token limits by tier
    max_tokens = auth["config"].get("max_input_tokens", 4000)
    if body.get("token_count", 0) > max_tokens:
        raise HTTPException(
            status_code=413,
            detail=f"Input exceeds {max_tokens} token limit for {auth['tier']} tier.",
        )
    return {"summary": "...", "tier": auth["tier"]}

@app.post("/risk-analysis")
async def risk_analysis(request: Request,
                        auth: dict = Depends(require_feature("risk_analysis"))):
    """Analyze contract risks. Pro and Enterprise only."""
    body = await request.json()
    return {"risks": [], "tier": auth["tier"]}

@app.post("/batch")
async def batch_process(request: Request,
                        auth: dict = Depends(require_feature("batch"))):
    """Batch process multiple contracts. Pro and Enterprise only."""
    body = await request.json()
    return {"batch_id": "batch-001", "status": "queued", "tier": auth["tier"]}

@app.post("/custom-model")
async def custom_model(request: Request,
                       auth: dict = Depends(require_tier("enterprise"))):
    """Use a custom fine-tuned model. Enterprise only."""
    body = await request.json()
    return {"result": "...", "model": "custom-legal-v3", "tier": auth["tier"]}

Tier Migration and Upgrades#

When a customer hits their rate limit repeatedly, prompt an upgrade:

python

def suggest_upgrade(client: GreenHelixClient, customer_id: str,
                    current_tier: str) -> dict:
    """Detect if a customer should upgrade based on usage patterns."""
    usage = client.execute("estimate_cost", {
        "agent_id": "contract-summarizer-v2",
        "customer_id": customer_id,
    })

    current_config = TIER_CONFIGS[current_tier]
    tiers_list = list(TIER_CONFIGS.keys())
    current_idx = tiers_list.index(current_tier)

    # Check if usage exceeds 80% of current tier limits
    daily_usage = usage.get("current_period_calls", 0)
    daily_limit = current_config["rate_limit_per_day"]

    if daily_usage > daily_limit * 0.8 and current_idx < len(tiers_list) - 1:
        next_tier = tiers_list[current_idx + 1]
        return {
            "should_upgrade": True,
            "current_tier": current_tier,
            "suggested_tier": next_tier,
            "reason": f"Using {daily_usage}/{daily_limit} daily calls "
                      f"({daily_usage/daily_limit*100:.0f}%)",
            "next_tier_price": TIER_CONFIGS[next_tier]["price_per_month"],
        }

    return {"should_upgrade": False, "current_tier": current_tier}

Key Takeaways

• Three tiers cover the adoption funnel: free (evaluation), pro (individuals/teams), enterprise (scale).
• Gate features, not just rate limits. Pro-only features drive upgrades better than rate limit frustration.
• Use FastAPI Depends with require_auth, require_feature, and require_tier for clean, composable access control.
• Monitor usage-to-limit ratios. When a customer consistently exceeds 80% of their tier, proactively suggest an upgrade.
• Always return clear error messages that tell the customer which tier they need and what it costs.

---

Chapter 6: Agent-to-Agent Payments#

Machines Paying Machines#

When a human buys a SaaS subscription, they enter a credit card, review the price, and click "Subscribe." When an agent buys a service from another agent, there is no credit card, no review step, and no click. The buyer agent discovers a service it needs, evaluates the price algorithmically, locks funds in escrow, receives the service, verifies the output, and releases payment -- all without human intervention. This is the core promise of the A2A economy: autonomous value exchange between software agents.

GreenHelix provides the full payment stack for this: create_payment_intent initiates the transaction, confirm_payment finalizes it for non-escrow flows, and the escrow tools ( release_escrow) handle conditional payments. Combined with the marketplace tools from Chapter 4 and the trust tools for pre-transaction due diligence, you can build fully autonomous agent-to-agent payment flows.

The Autonomous Transaction Flow#

A complete A2A transaction has six phases:

1. Discovery -- Buyer finds seller via marketplace
2. Due diligence -- Buyer checks seller's trust score and reputation
3. Negotiation -- Buyer evaluates price against budget and alternatives
4. Payment -- Buyer locks funds in escrow
5. Delivery -- Seller performs the service
6. Settlement -- Escrow releases based on verification

python

import time

class BuyerAgent:
    """An agent that autonomously discovers, evaluates, and pays for services."""

    def __init__(self, client: GreenHelixClient, agent_id: str,
                 budget_per_task: float = 1.00):
        self.client = client
        self.agent_id = agent_id
        self.budget_per_task = budget_per_task

    def find_and_hire(self, task_description: str,
                      required_tags: list[str]) -> dict:
        """Complete autonomous flow: discover, evaluate, pay, receive."""

        # Phase 1: Discovery
        candidates = self.client.execute("search_services", {
            "query": task_description,
            "tags": required_tags,
            "max_results": 5,
            "max_price": self.budget_per_task,
        })

        if not candidates.get("services"):
            return {"status": "no_services_found"}

        # Phase 2: Due Diligence
        evaluated = []
        for service in candidates["services"]:
            trust = self.client.execute("get_trust_score", {
                "agent_id": service["agent_id"],
            })
            reputation = self.client.execute("get_agent_reputation", {
                "agent_id": service["agent_id"],
            })
            evaluated.append({
                **service,
                "trust_score": trust.get("score", 0),
                "completed_tasks": reputation.get("completed_tasks", 0),
                "dispute_rate": reputation.get("dispute_rate", 1.0),
            })

        # Phase 3: Negotiation (algorithmic)
        best = self._select_best_seller(evaluated)
        if not best:
            return {"status": "no_acceptable_sellers"}

        # Phase 4: Payment (escrow)
        intent = self.client.execute("create_payment_intent", {
            "from_agent": self.agent_id,
            "to_agent": best["agent_id"],
            "amount": best["pricing"]["rate"],
            "currency": "USD",
            "escrow": True,
            "description": f"A2A task: {task_description[:100]}",
            "metadata": {
                "task_description": task_description,
                "buyer_agent": self.agent_id,
                "seller_agent": best["agent_id"],
                "selected_from": len(evaluated),
            },
        })

        # Phase 5: Delivery (call the seller's service)
        delivery = self._request_service(best, task_description)

        # Phase 6: Settlement
        if self._verify_delivery(delivery, task_description):
            settlement = self.client.execute("release_escrow", {
                "payment_intent_id": intent["payment_intent_id"],
                "release_to": best["agent_id"],
                "verification_proof": {
                    "delivery_verified": True,
                    "quality_score": delivery.get("quality_score", 0.9),
                    "timestamp": time.time(),
                },
            })
            return {
                "status": "completed",
                "seller": best["agent_id"],
                "price": best["pricing"]["rate"],
                "delivery": delivery,
                "settlement": settlement,
            }
        else:
            refund = self.client.execute("release_escrow", {
                "payment_intent_id": intent["payment_intent_id"],
                "release_to": self.agent_id,
                "verification_proof": {
                    "delivery_verified": False,
                    "reason": "Output quality below threshold",
                },
            })
            return {
                "status": "refunded",
                "seller": best["agent_id"],
                "reason": "delivery_verification_failed",
                "refund": refund,
            }

    def _select_best_seller(self, candidates: list[dict]) -> dict | None:
        """Select the best seller using a weighted scoring model.

        Score = 0.4 * trust + 0.3 * (1 - normalized_price) + 0.2 * experience + 0.1 * (1 - dispute_rate)
        """
        if not candidates:
            return None

        # Filter minimum trust threshold
        viable = [c for c in candidates if c["trust_score"] >= 0.6]
        if not viable:
            return None

        # Normalize price (lower is better)
        max_price = max(c["pricing"]["rate"] for c in viable)
        min_price = min(c["pricing"]["rate"] for c in viable)
        price_range = max_price - min_price if max_price != min_price else 1

        # Normalize experience
        max_tasks = max(c["completed_tasks"] for c in viable)
        max_tasks = max_tasks if max_tasks > 0 else 1

        scored = []
        for c in viable:
            price_score = 1 - (c["pricing"]["rate"] - min_price) / price_range
            trust_score = c["trust_score"]
            exp_score = c["completed_tasks"] / max_tasks
            dispute_score = 1 - c["dispute_rate"]

            total = (0.4 * trust_score + 0.3 * price_score
                     + 0.2 * exp_score + 0.1 * dispute_score)
            scored.append((total, c))

        scored.sort(key=lambda x: x[0], reverse=True)
        return scored[0][1]

    def _request_service(self, seller: dict, task: str) -> dict:
        """Call the seller's service endpoint."""
        # In production, this calls the seller's actual API
        return {"result": "service output", "quality_score": 0.92}

    def _verify_delivery(self, delivery: dict, task: str) -> bool:
        """Verify the delivered output meets requirements."""
        return delivery.get("quality_score", 0) >= 0.8

# Example: Autonomous agent transaction
buyer = BuyerAgent(client, agent_id="orchestrator-main", budget_per_task=0.50)

result = buyer.find_and_hire(
    task_description="Summarize this 15-page NDA between Acme Corp and Widget Inc, "
                     "extracting key terms, obligations, and termination clauses.",
    required_tags=["legal", "summarization"],
)

print(f"Transaction status: {result['status']}")
if result["status"] == "completed":
    print(f"Seller: {result['seller']}")
    print(f"Price paid: ${result['price']}")

Seller-Side: Accepting A2A Payments#

The seller agent needs to handle incoming payment intents, perform the service, and manage its own revenue:

python

class SellerAgent:
    """An agent that accepts A2A payments for services."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def handle_incoming_task(self, payment_intent_id: str,
                             task_data: dict) -> dict:
        """Handle an incoming paid task from a buyer agent."""
        # Verify the payment intent exists and is in escrow
        intent = self.client.execute("get_payment_intent", {
            "payment_intent_id": payment_intent_id,
        })

        if intent.get("status") != "escrowed":
            return {"error": "Payment not in escrow. Refusing to start work."}

        if intent.get("to_agent") != self.agent_id:
            return {"error": "Payment intent is for a different agent."}

        # Perform the service
        result = self._do_work(task_data)

        # Confirm delivery
        confirmation = self.client.execute("confirm_payment", {
            "payment_intent_id": payment_intent_id,
            "delivery_proof": {
                "result_hash": self._hash_result(result),
                "completed_at": time.time(),
                "quality_metrics": result.get("metrics", {}),
            },
        })

        return {
            "status": "delivered",
            "result": result,
            "confirmation": confirmation,
        }

    def _do_work(self, task_data: dict) -> dict:
        """Perform the actual service."""
        return {
            "output": "Task completed",
            "metrics": {"accuracy": 0.94, "latency_ms": 2800},
        }

    def _hash_result(self, result: dict) -> str:
        """Create a hash of the result for verification."""
        import hashlib, json
        return hashlib.sha256(
            json.dumps(result, sort_keys=True).encode()
        ).hexdigest()

Trust-Based Pre-Transaction Due Diligence#

Before committing funds, smart buyer agents check the seller's reputation chain:

python

def pre_transaction_due_diligence(client: GreenHelixClient,
                                   seller_id: str) -> dict:
    """Comprehensive trust check before committing to a transaction."""
    trust = client.execute("get_trust_score", {
        "agent_id": seller_id,
    })

    reputation = client.execute("get_agent_reputation", {
        "agent_id": seller_id,
    })

    # Get claim chains for cryptographic proof of past performance
    claims = client.execute("get_claim_chains", {
        "agent_id": seller_id,
        "claim_type": "service_delivery",
        "limit": 10,
    })

    # Compile risk assessment
    risk_factors = []
    if trust.get("score", 0) < 0.5:
        risk_factors.append("LOW_TRUST_SCORE")
    if reputation.get("completed_tasks", 0) < 10:
        risk_factors.append("LIMITED_TRACK_RECORD")
    if reputation.get("dispute_rate", 1.0) > 0.15:
        risk_factors.append("HIGH_DISPUTE_RATE")
    if not claims.get("chains"):
        risk_factors.append("NO_CRYPTOGRAPHIC_PROOF")

    return {
        "seller_id": seller_id,
        "trust_score": trust.get("score", 0),
        "completed_tasks": reputation.get("completed_tasks", 0),
        "dispute_rate": reputation.get("dispute_rate", 1.0),
        "verified_claims": len(claims.get("chains", [])),
        "risk_factors": risk_factors,
        "recommendation": "proceed" if len(risk_factors) == 0
                          else "caution" if len(risk_factors) <= 2
                          else "avoid",
    }

# Before any transaction
due_diligence = pre_transaction_due_diligence(client, "contract-summarizer-v2")
print(f"Recommendation: {due_diligence['recommendation']}")
print(f"Risk factors: {due_diligence['risk_factors']}")

Key Takeaways

• A2A payments follow six phases: discovery, due diligence, negotiation, escrow, delivery, settlement. Automate all six.
• Always use escrow for A2A payments. Direct payments between agents with no human oversight are an invitation for fraud.
• The buyer's seller selection algorithm should weight trust (40%), price (30%), experience (20%), and dispute rate (10%). Adjust weights to your risk tolerance.
• Sellers must verify the payment intent is in escrow before starting work. Never do work on a promise to pay.
• Pre-transaction due diligence using trust scores, reputation data, and cryptographic claim chains prevents losses. A few milliseconds of verification saves dollars of fraud.

---

Chapter 7: Revenue Tracking and Analytics#

From Transactions to Intelligence#

Billing records tell you what happened. Analytics tell you what it means. The gap between the two is where agent businesses fail -- they know they processed 847,000 calls last month but cannot answer whether that represents growth, decline, or a single customer masking churn across the rest of the base. GreenHelix Ledger and Analytics tools close this gap with transaction-level recording and aggregate metric computation.

The Revenue Tracker#

This class wraps the analytics and ledger tools into a unified revenue intelligence layer:

python

from datetime import datetime, timedelta
from typing import Optional

class RevenueTracker:
    """Tracks and analyzes revenue across all monetization channels."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def record_revenue_event(self, customer_id: str, amount: float,
                             source: str, metadata: dict | None = None) -> dict:
        """Record a revenue event in the ledger."""
        txn = self.client.execute("record_transaction", {
            "agent_id": self.agent_id,
            "customer_id": customer_id,
            "amount": str(amount),  # Use string for precision
            "currency": "USD",
            "type": "revenue",
            "source": source,  # "usage", "subscription", "outcome", "a2a"
            "metadata": metadata or {},
            "timestamp": datetime.utcnow().isoformat(),
        })
        return txn

    def get_revenue_summary(self, period_days: int = 30) -> dict:
        """Get revenue metrics for a period."""
        metrics = self.client.execute("get_revenue_metrics", {
            "agent_id": self.agent_id,
            "period_days": period_days,
        })
        return metrics

    def get_customer_history(self, customer_id: str,
                             limit: int = 100) -> dict:
        """Get transaction history for a specific customer."""
        history = self.client.execute("get_transaction_history", {
            "agent_id": self.agent_id,
            "customer_id": customer_id,
            "limit": limit,
        })
        return history

    def get_cohort_analysis(self, cohort_period: str = "monthly",
                            lookback_months: int = 6) -> dict:
        """Get cohort retention and revenue analysis."""
        cohorts = self.client.execute("get_cohort_analysis", {
            "agent_id": self.agent_id,
            "cohort_period": cohort_period,
            "lookback_months": lookback_months,
        })
        return cohorts

Daily Revenue Report Generator#

This is the report you run every morning. It pulls revenue metrics, calculates LTV, identifies churn risk, and computes per-product margins:

python

class DailyRevenueReport:
    """Generates a daily revenue report with LTV, churn, and margin analysis."""

    def __init__(self, tracker: RevenueTracker):
        self.tracker = tracker

    def generate(self) -> dict:
        """Generate the full daily report."""
        # Revenue metrics
        daily = self.tracker.get_revenue_summary(period_days=1)
        weekly = self.tracker.get_revenue_summary(period_days=7)
        monthly = self.tracker.get_revenue_summary(period_days=30)

        # Cohort analysis
        cohorts = self.tracker.get_cohort_analysis(
            cohort_period="monthly", lookback_months=6)

        # Compute derived metrics
        ltv = self._calculate_ltv(monthly)
        churn = self._calculate_churn(cohorts)
        margins = self._calculate_margins(monthly)

        report = {
            "generated_at": datetime.utcnow().isoformat(),
            "revenue": {
                "today": daily.get("total_revenue", 0),
                "this_week": weekly.get("total_revenue", 0),
                "this_month": monthly.get("total_revenue", 0),
                "mrr": monthly.get("total_revenue", 0),
                "arr": monthly.get("total_revenue", 0) * 12,
            },
            "customers": {
                "active_today": daily.get("active_customers", 0),
                "active_this_month": monthly.get("active_customers", 0),
                "new_this_month": monthly.get("new_customers", 0),
                "churned_this_month": churn["churned_count"],
            },
            "ltv": ltv,
            "churn": churn,
            "margins": margins,
            "cohorts": cohorts,
        }

        return report

    def _calculate_ltv(self, monthly_metrics: dict) -> dict:
        """Calculate customer lifetime value."""
        arpu = monthly_metrics.get("total_revenue", 0)
        active = monthly_metrics.get("active_customers", 1)
        arpu_per_customer = arpu / active if active > 0 else 0

        # Simple LTV = ARPU / churn_rate
        churn_rate = monthly_metrics.get("churn_rate", 0.10)
        churn_rate = max(churn_rate, 0.01)  # prevent division by zero

        ltv = arpu_per_customer / churn_rate

        return {
            "arpu_monthly": round(arpu_per_customer, 2),
            "estimated_ltv": round(ltv, 2),
            "ltv_to_cac_ratio": round(ltv / max(
                monthly_metrics.get("avg_cac", 1), 0.01), 2),
            "months_to_payback": round(
                monthly_metrics.get("avg_cac", 0) / max(arpu_per_customer, 0.01), 1),
        }

    def _calculate_churn(self, cohorts: dict) -> dict:
        """Calculate churn metrics from cohort data."""
        cohort_list = cohorts.get("cohorts", [])
        if not cohort_list:
            return {"monthly_rate": 0, "churned_count": 0}

        latest = cohort_list[-1] if cohort_list else {}
        return {
            "monthly_rate": latest.get("churn_rate", 0),
            "churned_count": latest.get("churned", 0),
            "retention_30d": latest.get("retention_30d", 0),
            "retention_90d": latest.get("retention_90d", 0),
        }

    def _calculate_margins(self, monthly_metrics: dict) -> dict:
        """Calculate per-product gross margins."""
        revenue = monthly_metrics.get("total_revenue", 0)
        costs = monthly_metrics.get("total_costs", 0)
        gross_margin = (revenue - costs) / revenue if revenue > 0 else 0

        by_source = monthly_metrics.get("revenue_by_source", {})
        cost_by_source = monthly_metrics.get("cost_by_source", {})

        margins_by_source = {}
        for source, rev in by_source.items():
            cost = cost_by_source.get(source, 0)
            margins_by_source[source] = {
                "revenue": rev,
                "cost": cost,
                "margin": round((rev - cost) / rev, 4) if rev > 0 else 0,
            }

        return {
            "overall_gross_margin": round(gross_margin, 4),
            "by_source": margins_by_source,
        }

    def format_text(self, report: dict) -> str:
        """Format the report as readable text."""
        r = report
        lines = [
            f"=== Daily Revenue Report ({r['generated_at'][:10]}) ===",
            "",
            f"Revenue:  Today ${r['revenue']['today']:.2f}  |  "
            f"Week ${r['revenue']['this_week']:.2f}  |  "
            f"Month ${r['revenue']['this_month']:.2f}",
            f"MRR: ${r['revenue']['mrr']:.2f}  |  ARR: ${r['revenue']['arr']:.2f}",
            "",
            f"Customers:  Active {r['customers']['active_this_month']}  |  "
            f"New {r['customers']['new_this_month']}  |  "
            f"Churned {r['customers']['churned_this_month']}",
            "",
            f"LTV:  ${r['ltv']['estimated_ltv']:.2f}  |  "
            f"ARPU: ${r['ltv']['arpu_monthly']:.2f}  |  "
            f"LTV:CAC = {r['ltv']['ltv_to_cac_ratio']}x",
            "",
            f"Churn:  {r['churn']['monthly_rate']*100:.1f}% monthly  |  "
            f"30d retention: {r['churn'].get('retention_30d', 0)*100:.1f}%",
            "",
            f"Gross Margin:  {r['margins']['overall_gross_margin']*100:.1f}%",
        ]

        for source, m in r['margins'].get('by_source', {}).items():
            lines.append(
                f"  {source}: ${m['revenue']:.2f} rev, "
                f"${m['cost']:.2f} cost, {m['margin']*100:.1f}% margin"
            )

        return "\n".join(lines)

# Generate and print daily report
tracker = RevenueTracker(client, "contract-summarizer-v2")
report_gen = DailyRevenueReport(tracker)
report = report_gen.generate()
print(report_gen.format_text(report))

Cryptographic Revenue Proof with Reputation Tools#

For investors, auditors, or potential acquirers who need verifiable proof of revenue, GreenHelix reputation tools create cryptographic claim chains that bind revenue claims to actual transaction records:

python

def build_revenue_proof_chain(client: GreenHelixClient,
                               agent_id: str,
                               period_start: str,
                               period_end: str) -> dict:
    """Build a cryptographic proof chain for revenue claims."""
    # Get actual transaction history for the period
    history = client.execute("get_transaction_history", {
        "agent_id": agent_id,
        "start_date": period_start,
        "end_date": period_end,
        "type": "revenue",
    })

    # Calculate total revenue from verified transactions
    total_revenue = sum(
        float(txn["amount"]) for txn in history.get("transactions", [])
    )

    # Build a claim chain that cryptographically binds the revenue figure
    # to the underlying transactions
    claim = client.execute("build_claim_chain", {
        "agent_id": agent_id,
        "claim_type": "revenue_report",
        "claim_data": {
            "period_start": period_start,
            "period_end": period_end,
            "total_revenue_usd": str(total_revenue),
            "transaction_count": len(history.get("transactions", [])),
            "transaction_hashes": [
                txn.get("hash") for txn in history.get("transactions", [])
                if txn.get("hash")
            ],
        },
        "evidence_refs": [
            txn["transaction_id"]
            for txn in history.get("transactions", [])
        ],
    })

    return {
        "claim_chain_id": claim.get("chain_id"),
        "total_revenue": total_revenue,
        "transaction_count": len(history.get("transactions", [])),
        "verifiable": True,
        "verification_url": claim.get("verification_url"),
    }

# Generate verifiable revenue proof for Q1 2026
proof = build_revenue_proof_chain(
    client,
    agent_id="contract-summarizer-v2",
    period_start="2026-01-01",
    period_end="2026-03-31",
)
print(f"Revenue proof chain: {proof['claim_chain_id']}")
print(f"Verified Q1 revenue: ${proof['total_revenue']:.2f}")
print(f"Backed by {proof['transaction_count']} transactions")

Key Takeaways

• Track four revenue sources separately: usage, subscription, outcome, and A2A. Each has different margin profiles and churn dynamics.
• Run the daily revenue report every morning. The metrics that matter: MRR, ARPU, LTV, monthly churn rate, and gross margin by source.
• LTV = ARPU / churn_rate is a simple approximation. For agent services with bursty usage, weight it by trailing-90-day ARPU for stability.
• Cryptographic claim chains provide verifiable revenue proof for investors and auditors. Build them quarterly.
• Per-source margin analysis reveals which monetization channels are sustainable. If your A2A revenue runs at 30% margin while usage-based runs at 60%, optimize the A2A pricing.

---

Chapter 8: Launch Checklist and Pricing Experiments#

The 14-Day Launch Playbook#

You have metering, escrow, a marketplace listing, tiered access, A2A payments, and analytics. Now you need to launch. This playbook takes you from initial pricing decision to validated, optimized pricing in 14 days.

Days 1-3: Set the Initial Price#

Day 1: Competitive research. Use marketplace search to find every competing service in your category. Record their pricing structure, rate, tier limits, and trust scores.

python

def competitive_pricing_research(client: GreenHelixClient,
                                  category_tags: list[str]) -> list[dict]:
    """Research competitor pricing in your category."""
    results = client.execute("search_services", {
        "query": " ".join(category_tags),
        "tags": category_tags,
        "max_results": 20,
        "sort_by": "relevance",
    })

    competitors = []
    for service in results.get("services", []):
        trust = client.execute("get_trust_score", {
            "agent_id": service["agent_id"],
        })
        competitors.append({
            "name": service.get("service_name"),
            "agent_id": service["agent_id"],
            "pricing_model": service.get("pricing", {}).get("model"),
            "rate": service.get("pricing", {}).get("rate"),
            "unit": service.get("pricing", {}).get("unit"),
            "trust_score": trust.get("score", 0),
            "tags": service.get("tags", []),
        })

    # Sort by rate for comparison
    competitors.sort(key=lambda x: x.get("rate", 0))
    return competitors

competitors = competitive_pricing_research(client, ["legal", "summarization"])

print("Competitor Pricing Analysis:")
print(f"{'Name':<30} {'Model':<12} {'Rate':>8} {'Unit':<12} {'Trust':>6}")
print("-" * 72)
for c in competitors:
    print(f"{c['name']:<30} {c['pricing_model']:<12} "
          f"${c['rate']:>7.4f} {c['unit']:<12} {c['trust_score']:>5.2f}")

Day 2: Cost-plus calculation. Calculate your floor price (the minimum you can charge without losing money) and your ceiling price (the maximum the market will bear based on Day 1 research).

ComponentCost per CallNotesLLM inference (GPT-4o)$0.0035Average across document sizesEmbedding/retrieval$0.0004Vector search for contextInfrastructure (compute)$0.0002Amortized server costsGateway fees$0.0001GreenHelix transaction feeTotal cost****$0.0042Floor price (10% margin)$0.0047Target price (55% margin)$0.0093Ceiling (competitor avg)$0.0120Based on Day 1 research

Day 3: Set your initial price. Start at the midpoint between your target margin price and the competitive ceiling. You will adjust based on data.

python

def calculate_launch_price(cost_per_call: float, target_margin: float,
                           competitor_avg_price: float) -> dict:
    """Calculate the optimal launch price."""
    floor_price = cost_per_call / (1 - 0.10)  # 10% minimum margin
    target_price = cost_per_call / (1 - target_margin)
    ceiling_price = competitor_avg_price

    # Launch at midpoint between target and ceiling
    launch_price = (target_price + ceiling_price) / 2

    return {
        "cost_per_call": cost_per_call,
        "floor_price": round(floor_price, 4),
        "target_price": round(target_price, 4),
        "ceiling_price": round(ceiling_price, 4),
        "launch_price": round(launch_price, 4),
        "launch_margin": round(1 - cost_per_call / launch_price, 4),
    }

pricing = calculate_launch_price(
    cost_per_call=0.0042,
    target_margin=0.55,
    competitor_avg_price=0.0120,
)
print(f"Launch price: ${pricing['launch_price']}/call")
print(f"Launch margin: {pricing['launch_margin']*100:.1f}%")

Days 4-7: A/B Test Two Price Points#

Use GreenHelix Organizations to create segmented customer groups and test two price points simultaneously.

python

class PricingExperiment:
    """A/B test two price points using organization-based segmentation."""

    def __init__(self, client: GreenHelixClient, agent_id: str):
        self.client = client
        self.agent_id = agent_id

    def setup_experiment(self, price_a: float, price_b: float,
                         experiment_name: str) -> dict:
        """Create two customer segments with different pricing."""
        # Create organization for Group A (control)
        org_a = self.client.execute("create_organization", {
            "agent_id": self.agent_id,
            "name": f"{experiment_name}-group-a",
            "metadata": {
                "experiment": experiment_name,
                "group": "A",
                "price": str(price_a),
            },
        })

        # Create organization for Group B (variant)
        org_b = self.client.execute("create_organization", {
            "agent_id": self.agent_id,
            "name": f"{experiment_name}-group-b",
            "metadata": {
                "experiment": experiment_name,
                "group": "B",
                "price": str(price_b),
            },
        })

        # Create billing plans for each group
        plan_a = self.client.execute("create_billing_plan", {
            "agent_id": self.agent_id,
            "plan_name": f"{experiment_name}-plan-a",
            "billing_type": "usage",
            "currency": "USD",
            "metering_dimensions": [{
                "name": "api_calls",
                "unit": "call",
                "tiers": [{"up_to": None, "rate": price_a}],
            }],
            "billing_period": "monthly",
        })

        plan_b = self.client.execute("create_billing_plan", {
            "agent_id": self.agent_id,
            "plan_name": f"{experiment_name}-plan-b",
            "billing_type": "usage",
            "currency": "USD",
            "metering_dimensions": [{
                "name": "api_calls",
                "unit": "call",
                "tiers": [{"up_to": None, "rate": price_b}],
            }],
            "billing_period": "monthly",
        })

        return {
            "experiment_name": experiment_name,
            "group_a": {"org_id": org_a["org_id"], "plan_id": plan_a["plan_id"],
                        "price": price_a},
            "group_b": {"org_id": org_b["org_id"], "plan_id": plan_b["plan_id"],
                        "price": price_b},
        }

    def assign_customer(self, customer_id: str,
                        experiment: dict) -> str:
        """Assign a new customer to a group (alternating for balance)."""
        # Simple deterministic assignment based on customer ID hash
        group = "a" if hash(customer_id) % 2 == 0 else "b"
        group_data = experiment[f"group_{group}"]

        self.client.execute("add_org_member", {
            "org_id": group_data["org_id"],
            "customer_id": customer_id,
        })

        return group

    def analyze_experiment(self, experiment: dict,
                           days_elapsed: int) -> dict:
        """Analyze experiment results after a period."""
        results = {}
        for group_name in ["a", "b"]:
            group = experiment[f"group_{group_name}"]

            metrics = self.client.execute("get_revenue_metrics", {
                "agent_id": self.agent_id,
                "plan_id": group["plan_id"],
                "period_days": days_elapsed,
            })

            results[group_name] = {
                "price": group["price"],
                "total_revenue": metrics.get("total_revenue", 0),
                "total_calls": metrics.get("total_usage", 0),
                "active_customers": metrics.get("active_customers", 0),
                "conversion_rate": metrics.get("conversion_rate", 0),
                "revenue_per_customer": (
                    metrics.get("total_revenue", 0)
                    / max(metrics.get("active_customers", 1), 1)
                ),
            }

        # Determine winner
        rev_a = results["a"]["total_revenue"]
        rev_b = results["b"]["total_revenue"]
        winner = "a" if rev_a >= rev_b else "b"

        return {
            "results": results,
            "winner": winner,
            "winner_price": results[winner]["price"],
            "revenue_lift": abs(rev_a - rev_b) / max(min(rev_a, rev_b), 1),
            "recommendation": (
                f"Price ${results[winner]['price']} generates "
                f"${results[winner]['total_revenue']:.2f} revenue vs "
                f"${results['a' if winner == 'b' else 'b']['total_revenue']:.2f} "
                f"for the alternative."
            ),
        }

# Run the experiment
experiment = PricingExperiment(client, "contract-summarizer-v2")
exp = experiment.setup_experiment(
    price_a=0.008,   # Control: current price
    price_b=0.012,   # Variant: 50% higher
    experiment_name="pricing-apr-2026",
)
print(f"Experiment created: {exp['experiment_name']}")

Days 8-14: Analyze, Adjust, Scale#

Day 8: Pull experiment results.

python

# After 4 days of data collection
analysis = experiment.analyze_experiment(exp, days_elapsed=4)
print(f"Winner: Group {analysis['winner'].upper()} "
      f"at ${analysis['winner_price']}/call")
print(f"Revenue lift: {analysis['revenue_lift']*100:.1f}%")
print(f"Recommendation: {analysis['recommendation']}")

Day 10: Adjust pricing based on results. Roll the winning price to all customers.

Day 12: Set up volume discounts for the new price point (using the patterns from Chapter 2).

Day 14: Full launch. Update your marketplace listing with the validated price and announce availability.

The Pricing Calculator Template#

Give your customers a self-service pricing calculator:

python

def pricing_calculator(monthly_calls: int, avg_tokens_per_call: int,
                       tier: str = "pro") -> dict:
    """Calculate estimated monthly cost for a customer."""
    tier_config = TIER_CONFIGS.get(tier, TIER_CONFIGS["pro"])
    base_fee = tier_config["price_per_month"]

    # Usage-based component (from Chapter 2 billing plan)
    usage_tiers = [
        (1000, 0.008),
        (10000, 0.006),
        (100000, 0.004),
        (float("inf"), 0.002),
    ]

    usage_cost = 0
    remaining = monthly_calls
    prev_threshold = 0
    for threshold, rate in usage_tiers:
        calls_in_tier = min(remaining, threshold - prev_threshold)
        usage_cost += calls_in_tier * rate
        remaining -= calls_in_tier
        prev_threshold = threshold
        if remaining <= 0:
            break

    total = base_fee + usage_cost

    return {
        "tier": tier,
        "base_fee": base_fee,
        "monthly_calls": monthly_calls,
        "usage_cost": round(usage_cost, 2),
        "total_monthly": round(total, 2),
        "effective_rate_per_call": round(total / max(monthly_calls, 1), 4),
    }

# Example calculations
for calls in [500, 5000, 50000, 500000]:
    calc = pricing_calculator(calls, avg_tokens_per_call=2000)
    print(f"{calls:>8,} calls/mo: ${calc['total_monthly']:>8.2f} "
          f"(${calc['effective_rate_per_call']:.4f}/call)")

Output:

text

     500 calls/mo:    $53.00 ($0.1060/call)
   5,000 calls/mo:    $81.00 ($0.0162/call)
  50,000 calls/mo:   $309.00 ($0.0062/call)
 500,000 calls/mo: $1,109.00 ($0.0022/call)

Pre-Launch Validation Checklist#

Complete every item before going live:

#ItemTool/MethodStatus1Billing plan created with correct tierscreate_billing_plan[ ]2Usage metering recording accuratelyrecord_usage + estimate_cost verification[ ]3Volume discounts configuredset_volume_discount[ ]4Escrow flow tested end-to-endcreate_payment_intent + release_escrow[ ]5Dispute handling testedcreate_dispute + respond_to_dispute[ ]6Agent identity registeredregister_agent[ ]7Wallet created and fundedcreate_wallet[ ]8Marketplace listing publishedregister_service[ ]9Service discoverable in searchsearch_services verification[ ]10API keys created for all tierscreate_api_key (free, pro, enterprise)[ ]11Rate limiting enforcedcheck_rate_limit[ ]12Tier gating middleware deployedvalidate_key + get_tier_config[ ]13Revenue tracking recording eventsrecord_transaction[ ]14Daily report generating correctlyget_revenue_metrics + get_cohort_analysis[ ]15Cryptographic revenue proof chainbuild_claim_chain[ ]16Competitor pricing benchmarkedsearch_services[ ]17A/B test infrastructure readyOrganization-based segmentation[ ]18Pricing calculator availableSelf-service estimation tool[ ]

Related Products#

This guide covers the pricing and monetization layer. For adjacent capabilities, see:

• P4 -- Agent Commerce Toolkit ( agent-commerce-toolkit.md): Deep dive into escrow patterns, split payments, subscription billing, dispute resolution, and framework integrations (CrewAI, LangChain, AutoGen). Start here if you need the payment infrastructure foundations before layering on pricing strategy.

• P7 -- Agent SaaS Factory ( agent-saas-factory.md): End-to-end guide to building a complete agent SaaS business -- from architecture and multi-tenancy through deployment and scaling. Covers the operational layer that this guide's pricing strategy sits on top of.

• P15 -- Agent Revenue Analytics ( agent-revenue-analytics.md): Advanced revenue analytics including attribution modeling, churn prediction, competitive intelligence, and pricing optimization from historical data. Extends Chapter 7 of this guide into a full analytics practice with ML-based forecasting.

Key Takeaways

• The 14-day launch playbook: Days 1-3 (research + set initial price), Days 4-7 (A/B test two price points), Days 8-14 (analyze, adjust, scale).
• Start pricing at the midpoint between your target-margin price and the competitive ceiling. Data will tell you which direction to move.
• Use Organization-based segmentation for clean A/B tests. Deterministic customer assignment prevents contamination.
• The pricing calculator builds customer confidence. Self-service cost estimation reduces pre-purchase friction.
• Complete all 18 checklist items before launch. Skipping escrow testing or rate limiting will cause production incidents that damage your trust score.

---

Appendix: Tool Reference Quick Index#

ToolChapterPurposecreate_billing_plan2Define pricing tiers and metering dimensionsrecord_usage2Log metered usage eventsestimate_cost2Calculate current or projected billset_volume_discount2Configure automatic volume discountscreate_payment_intent3, 6Initiate payment with optional escrowrelease_escrow3, 6Release or refund escrowed fundscreate_dispute3File a payment disputerespond_to_dispute3Counter a dispute with evidenceregister_agent4Register agent identitycreate_wallet4Create payment walletregister_service4Publish service to marketplacesearch_services4, 8Search marketplace by query and tagsbest_match4Semantic + trust-weighted service matchingcreate_api_key5Generate tier-scoped API keysvalidate_key5Validate key and return tierget_tier_config5Get rate limits and features for a tiercheck_rate_limit5Check remaining rate limit quotaconfirm_payment6Confirm delivery for non-escrow paymentsget_payment_intent6Retrieve payment intent detailsget_trust_score6Get agent trust scoreget_agent_reputation6Get agent reputation metricsget_claim_chains6Get cryptographic proof chainsrecord_transaction7Record revenue event in ledgerget_transaction_history7Query transaction historyget_revenue_metrics7, 8Aggregate revenue metricsget_cohort_analysis7Cohort retention and revenue analysisbuild_claim_chain7Build cryptographic revenue proofcreate_organization8Create customer segment for experimentsadd_org_member8Add customer to organization