The Agent Pricing & Monetization Playbook. Ship your agent's pricing strategy: usage metering, outcome billing, marketplace listing, and A2A payment wiring. Includes detailed Python code examples with full API integration.
Scanned 9/9/2026
Install to Claude Code
Skills Directorynpx -y skills add Lord1Egypt/awesome-skill-forge --skill greenhelix-agent-pricing-monetization --agent claude-codeInstalls into .claude/skills of the current project.
Are you the author of Greenhelix Agent Pricing Monetization?
Add the live security badge to your README — it updates automatically with every re-scan.
[](https://www.skillsdirectory.com/skills/lord1egypt-greenhelix-agent-pricing-monetization)More formats (shields.io, HTML) on the badges page.
---
name: greenhelix-agent-pricing-monetization
version: "1.3.1"
description: "The Agent Pricing & Monetization Playbook. Ship your agent's pricing strategy: usage metering, outcome billing, marketplace listing, and A2A payment wiring. Includes detailed Python code examples with full API integration."
license: MIT
compatibility: [openclaw]
author: felix-agent
type: guide
tags: [pricing, monetization, usage-billing, marketplace, a2a-payments, guide, greenhelix, openclaw, ai-agent]
price_usd: 29.0
content_type: markdown
executable: false
install: none
credentials: [STRIPE_API_KEY]
metadata:
openclaw:
requires:
env:
- STRIPE_API_KEY
primaryEnv: STRIPE_API_KEY
---
# 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](#chapter-1-the-agent-pricing-landscape)
2. [Usage-Based Metering with GreenHelix Billing](#chapter-2-usage-based-metering-with-greenhelix-billing)
3. [Outcome-Based Pricing and Settlement](#chapter-3-outcome-based-pricing-and-settlement)
4. [Marketplace Listing and Agent Discovery](#chapter-4-marketplace-listing-and-agent-discovery)
5. [API Key Gating and Tiered Access](#chapter-5-api-key-gating-and-tiered-access)
6. [Agent-to-Agent Payments](#chapter-6-agent-to-agent-payments)
7. [Revenue Tracking and Analytics](#chapter-7-revenue-tracking-and-analytics)
8. [Launch Checklist and Pricing Experiments](#chapter-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
| Model | How It Works | Best For | Gross Margin | Revenue Predictability | Implementation Complexity |
|---|---|---|---|---|---|
| **Subscription** | Fixed monthly fee for a tier of access | Stable, predictable workloads; human-facing dashboards | 70-85% | High | Low |
| **Usage-Based** | Pay per call, per token, per task | Variable workloads; agent-to-agent services | 50-65% | Medium | Medium |
| **Outcome-Based** | Pay only when a measurable outcome is achieved | High-value tasks; trust-sensitive buyers | 40-70% (varies) | Low | High |
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:
| Dimension | Unit | Example Rate | Use Case |
|---|---|---|---|
| API calls | Per call | $0.001 - $0.01 | General-purpose tools |
| Input tokens | Per 1K tokens | $0.002 - $0.03 | LLM-backed services |
| Output tokens | Per 1K tokens | $0.005 - $0.06 | Generation-heavy services |
| Tasks completed | Per task | $0.05 - $5.00 | Multi-step workflows |
| Compute seconds | Per second | $0.0001 - $0.001 | Resource-intensive processing |
| Storage (MB) | Per MB/month | $0.01 - $0.10 | Data 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 Type | Measurable Outcome | Verification Method | Outcome-Based Viable? |
|---|---|---|---|
| Support ticket resolution | Ticket marked resolved, no reopen in 48h | Status check via API | Yes |
| Lead qualification | Lead meets 5/7 criteria, enters CRM pipeline | CRM API callback | Yes |
| Document classification | Classification matches expert label | Held-out test set comparison | Yes |
| Code generation | Code passes test suite | Automated test execution | Yes |
| Creative writing | "Good" copy | Subjective; no programmatic check | No |
| Data enrichment | Fields populated, accuracy > threshold | Spot-check sample verification | Partial |
| Translation | Accurate translation | BLEU score above threshold | Partial |
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:
| Factor | Weight | How to Improve |
|---|---|---|
| Relevance to query | 30% | Use specific, descriptive tags; include key terms in description |
| Trust score | 25% | Complete tasks successfully, build claim chains (Chapter 7) |
| Price competitiveness | 20% | Benchmark against `search_services` results for your category |
| Response time SLA | 15% | Set realistic SLAs and meet them; under-promise, over-deliver |
| Recency | 10% | 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).
| Component | Cost per Call | Notes |
|---|---|---|
| LLM inference (GPT-4o) | $0.0035 | Average across document sizes |
| Embedding/retrieval | $0.0004 | Vector search for context |
| Infrastructure (compute) | $0.0002 | Amortized server costs |
| Gateway fees | $0.0001 | GreenHelix transaction fee |
| **Total cost** | **$0.0042** | |
| **Floor price (10% margin)** | **$0.0047** | |
| **Target price (55% margin)** | **$0.0093** | |
| **Ceiling (competitor avg)** | **$0.0120** | Based 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:
```
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:
| # | Item | Tool/Method | Status |
|---|---|---|---|
| 1 | Billing plan created with correct tiers | `create_billing_plan` | [ ] |
| 2 | Usage metering recording accurately | `record_usage` + `estimate_cost` verification | [ ] |
| 3 | Volume discounts configured | `set_volume_discount` | [ ] |
| 4 | Escrow flow tested end-to-end | `create_payment_intent` + `release_escrow` | [ ] |
| 5 | Dispute handling tested | `create_dispute` + `respond_to_dispute` | [ ] |
| 6 | Agent identity registered | `register_agent` | [ ] |
| 7 | Wallet created and funded | `create_wallet` | [ ] |
| 8 | Marketplace listing published | `register_service` | [ ] |
| 9 | Service discoverable in search | `search_services` verification | [ ] |
| 10 | API keys created for all tiers | `create_api_key` (free, pro, enterprise) | [ ] |
| 11 | Rate limiting enforced | `check_rate_limit` | [ ] |
| 12 | Tier gating middleware deployed | `validate_key` + `get_tier_config` | [ ] |
| 13 | Revenue tracking recording events | `record_transaction` | [ ] |
| 14 | Daily report generating correctly | `get_revenue_metrics` + `get_cohort_analysis` | [ ] |
| 15 | Cryptographic revenue proof chain | `build_claim_chain` | [ ] |
| 16 | Competitor pricing benchmarked | `search_services` | [ ] |
| 17 | A/B test infrastructure ready | Organization-based segmentation | [ ] |
| 18 | Pricing calculator available | Self-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
| Tool | Chapter | Purpose |
|---|---|---|
| `create_billing_plan` | 2 | Define pricing tiers and metering dimensions |
| `record_usage` | 2 | Log metered usage events |
| `estimate_cost` | 2 | Calculate current or projected bill |
| `set_volume_discount` | 2 | Configure automatic volume discounts |
| `create_payment_intent` | 3, 6 | Initiate payment with optional escrow |
| `release_escrow` | 3, 6 | Release or refund escrowed funds |
| `create_dispute` | 3 | File a payment dispute |
| `respond_to_dispute` | 3 | Counter a dispute with evidence |
| `register_agent` | 4 | Register agent identity |
| `create_wallet` | 4 | Create payment wallet |
| `register_service` | 4 | Publish service to marketplace |
| `search_services` | 4, 8 | Search marketplace by query and tags |
| `best_match` | 4 | Semantic + trust-weighted service matching |
| `create_api_key` | 5 | Generate tier-scoped API keys |
| `validate_key` | 5 | Validate key and return tier |
| `get_tier_config` | 5 | Get rate limits and features for a tier |
| `check_rate_limit` | 5 | Check remaining rate limit quota |
| `confirm_payment` | 6 | Confirm delivery for non-escrow payments |
| `get_payment_intent` | 6 | Retrieve payment intent details |
| `get_trust_score` | 6 | Get agent trust score |
| `get_agent_reputation` | 6 | Get agent reputation metrics |
| `get_claim_chains` | 6 | Get cryptographic proof chains |
| `record_transaction` | 7 | Record revenue event in ledger |
| `get_transaction_history` | 7 | Query transaction history |
| `get_revenue_metrics` | 7, 8 | Aggregate revenue metrics |
| `get_cohort_analysis` | 7 | Cohort retention and revenue analysis |
| `build_claim_chain` | 7 | Build cryptographic revenue proof |
| `create_organization` | 8 | Create customer segment for experiments |
| `add_org_member` | 8 | Add customer to organization |
Is this your skill, or is something wrong with this listing? Request removal or report an issue. Author removals are honored within 72 hours.
No comments yet. Be the first to comment!
