Install
openclaw skills install @mirni/greenhelix-agent-commerce-migrationAgent Commerce Migration Guide: Retrofit Your REST APIs for Autonomous Agent Buyers
Agent Commerce Migration Guide: Retrofit Your REST APIs for Autonomous Agent Buyers. Step-by-step migration guide for teams with existing REST APIs that need to add agent commerce capabilities. Assessment framework, x402 retrofit patterns, authentication bridging, gradual migration strategies, testing, rollback procedures, and performance comparison.
Agent Commerce Migration Guide: Retrofit Your REST APIs for Autonomous Agent Buyers
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):
GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)
You have a working REST API. It serves human users through a web frontend. Maybe it powers a SaaS product, a data marketplace, or a specialized computation service. The endpoints are tested, the authentication works, and your customers are happy. Now AI agents want to buy your services programmatically. They don't want to fill out a signup form, navigate a dashboard, or read your documentation the way a human does. They want to discover your capabilities through structured metadata, negotiate a price via HTTP headers, pay per call with cryptographic proof, and consume your response — all in a single request cycle. The instinct is to build a separate "agent API" from scratch, but that is the wrong move. You already have the hard part: working business logic behind stable endpoints. What you need is a commerce layer on top of what exists. This guide shows how to retrofit any REST API with agent commerce capabilities: x402 payment headers for per-call pricing, GreenHelix escrow integration for trustless settlement, Ed25519 identity verification for agent authentication, and structured service discovery so agents can find and evaluate your offerings without human intervention. The approach is gradual. You can migrate one endpoint at a time while existing human users continue hitting the same URLs with the same authentication they have always used. No big bang rewrite. No downtime. No breaking changes. By the end of this guide you will have a complete migration framework — assessment tools, adapter classes, middleware patterns, validation harnesses, and rollback procedures — that transforms your existing REST API into an agent-ready commerce platform without sacrificing anything that already works.
Getting started: All examples in this guide work with the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required.
What You'll Learn
- Chapter 1: Migration Assessment
- Chapter 2: LegacyApiAdapter Class
- Chapter 3: x402 Retrofit Patterns
- Chapter 4: Authentication Bridging
- Chapter 5: Gradual Migration Strategies
- Chapter 6: MigrationValidator Class
- Chapter 7: Testing Migration
- Chapter 8: Performance Comparison
- What's Next
Full Guide
Agent Commerce Migration Guide: Retrofit Your REST APIs for Autonomous Agent Buyers
You have a working REST API. It serves human users through a web frontend. Maybe it powers a SaaS product, a data marketplace, or a specialized computation service. The endpoints are tested, the authentication works, and your customers are happy. Now AI agents want to buy your services programmatically. They don't want to fill out a signup form, navigate a dashboard, or read your documentation the way a human does. They want to discover your capabilities through structured metadata, negotiate a price via HTTP headers, pay per call with cryptographic proof, and consume your response — all in a single request cycle. The instinct is to build a separate "agent API" from scratch, but that is the wrong move. You already have the hard part: working business logic behind stable endpoints. What you need is a commerce layer on top of what exists. This guide shows how to retrofit any REST API with agent commerce capabilities: x402 payment headers for per-call pricing, GreenHelix escrow integration for trustless settlement, Ed25519 identity verification for agent authentication, and structured service discovery so agents can find and evaluate your offerings without human intervention. The approach is gradual. You can migrate one endpoint at a time while existing human users continue hitting the same URLs with the same authentication they have always used. No big bang rewrite. No downtime. No breaking changes. By the end of this guide you will have a complete migration framework — assessment tools, adapter classes, middleware patterns, validation harnesses, and rollback procedures — that transforms your existing REST API into an agent-ready commerce platform without sacrificing anything that already works.
Getting started: All examples in this guide work with the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required.
Chapter 1: Migration Assessment
Before writing any migration code, you need an honest evaluation of where your API stands today and what it will take to make it agent-commerce ready. This assessment phase saves weeks of wasted effort by identifying blockers early, surfacing hidden dependencies, and prioritizing the endpoints that will generate the most agent revenue with the least migration risk.
The Four Pillars of Agent Commerce Readiness
Every REST API migration touches four areas. Your current implementation might already satisfy some of them partially, but each one needs explicit attention.
Authentication readiness. Agents do not use browser cookies or OAuth redirect flows. They present cryptographic identity — typically an Ed25519 public key with a signed request payload. Your API needs to verify these signatures alongside whatever auth mechanism already exists. The question is not whether you can replace your current auth, but whether you can layer agent auth on top of it.
Payment readiness. Human users pay through subscriptions, invoices, or credit cards processed asynchronously. Agent commerce is synchronous: the agent includes a payment token in the request header, your API verifies it before processing, and settlement happens atomically. If your API currently has no concept of per-call pricing, you need to add one. If it already has usage-based billing, you have a head start.
Response structure readiness. Human-facing APIs often return loosely structured JSON with messages meant for UI rendering — status strings like "Success!", nested error objects with display text, pagination metadata mixed into the response body. Agents need machine-parseable responses with predictable schemas, consistent error codes, and clear separation between data and metadata. The more disciplined your current response format, the less transformation work ahead.
Discovery readiness. Agents find services through structured manifests — A2A protocol cards, OpenAPI specifications with pricing extensions, or capability registries. If your API has no machine-readable description of what it offers and what it costs, agents cannot evaluate it. Most REST APIs have some form of documentation, but rarely in a format an agent can consume autonomously.
The Assessment Checklist
Run through this checklist for every endpoint you are considering for migration. Score each item 0 (not present), 1 (partially present), or 2 (fully present):
MIGRATION_ASSESSMENT = {
"authentication": [
"Stateless auth supported (tokens, not sessions)",
"Auth info passed via headers (not cookies)",
"Auth can be extended without breaking existing clients",
"Rate limiting is per-identity, not per-IP",
],
"payments": [
"Usage tracking exists at per-call granularity",
"Pricing model defined for individual operations",
"Billing system can handle micropayments",
"Idempotency keys supported on payment-related endpoints",
],
"response_structure": [
"Consistent JSON envelope across all endpoints",
"Error responses use numeric codes, not just strings",
"Pagination uses cursors, not page numbers",
"Response schemas are documented and stable",
],
"discovery": [
"OpenAPI / Swagger spec exists and is current",
"Endpoint capabilities described in machine-readable format",
"Versioning strategy is explicit (URL or header)",
"Health check endpoint exists",
],
}
def score_api(scores: dict[str, list[int]]) -> dict:
results = {}
for category, items in scores.items():
total = sum(items)
max_score = len(MIGRATION_ASSESSMENT[category]) * 2
results[category] = {
"score": total,
"max": max_score,
"percentage": round(total / max_score * 100),
"ready": total >= max_score * 0.75,
}
results["overall_ready"] = all(r["ready"] for r in results.values())
return results
An overall score above 75% means you can proceed with migration directly. Between 50% and 75%, plan for prerequisite work on the weakest pillar before starting. Below 50%, consider whether the API needs structural improvements first — migration on top of a fragile foundation creates compounding problems.
The LegacyApiAdapter Pattern Overview
The core architectural pattern for this migration is the LegacyApiAdapter: a wrapper class that sits between incoming agent requests and your existing endpoint handlers. It performs three functions. First, it intercepts agent-specific headers (x402 payment tokens, Ed25519 signatures, capability queries) and processes them before the request reaches your business logic. Second, it transforms your existing response format into the agent-expected envelope if needed. Third, it passes through requests from non-agent clients completely unchanged, so your existing users never notice the migration happened.
This is not a proxy in the traditional sense. It runs in the same process as your existing API, adds no network hops, and shares the same database connections and caches. Think of it as a decorator pattern applied at the HTTP layer.
Identifying High-Value Migration Targets
Not every endpoint is worth migrating. Start with endpoints that satisfy three criteria:
High agent utility. Data retrieval endpoints (search, lookup, analytics) are the most immediately useful to agents. Agents are information consumers first. An endpoint that returns structured data an agent can act on is more valuable than a CRUD endpoint for managing user profiles.
Low migration complexity. Endpoints with simple request/response schemas, no file uploads, no streaming responses, and no multi-step workflows are easiest to migrate. Save the complex ones for later phases.
Clear pricing model. You need to assign a per-call price. Endpoints where the cost of serving a request is predictable (fixed computation, bounded database queries) are easier to price than endpoints with variable resource consumption. If you cannot state a price, you are not ready to migrate that endpoint.
Rank your endpoints by (utility x 1/complexity x pricing_clarity) and migrate the top five first. This gives you a working agent commerce surface quickly, generates early revenue, and builds confidence in the migration pattern before tackling harder endpoints.
Chapter 2: LegacyApiAdapter Class
The LegacyApiAdapter is the central abstraction for this entire migration. It wraps your existing endpoint handlers and adds agent commerce capabilities without modifying the underlying business logic. Every request flows through the adapter, which decides whether to apply agent commerce processing or pass the request through unchanged.
Design Principles
The adapter follows three rules. First, zero regression: any request that worked before the adapter was installed must continue to work identically. Second, opt-in activation: agent commerce features activate only when the request contains agent-specific headers. Third, fail-open for humans: if agent commerce processing fails (payment verification timeout, identity service unavailable), non-agent requests still succeed. Agent requests fail explicitly with structured error responses.
The Complete Adapter Class
import time
import hashlib
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional
from functools import wraps
import httpx
from fastapi import Request, Response, HTTPException
from fastapi.responses import JSONResponse
class CallerType(str, Enum):
HUMAN = "human"
AGENT = "agent"
UNKNOWN = "unknown"
@dataclass
class PricingConfig:
"""Per-endpoint pricing configuration."""
price_usd: float
currency: str = "USD"
escrow_required: bool = True
min_deposit: float = 0.01
max_deposit: float = 100.00
settlement_timeout_seconds: int = 300
@dataclass
class MigrationConfig:
"""Configuration for the adapter's behavior."""
greenhelix_url: str = "https://api.greenhelix.net"
greenhelix_api_key: str = ""
verify_payments: bool = True
verify_identity: bool = True
transform_responses: bool = True
pass_through_on_failure: bool = False
pricing: dict[str, PricingConfig] = field(default_factory=dict)
@dataclass
class AgentContext:
"""Extracted agent information from request headers."""
caller_type: CallerType
agent_id: Optional[str] = None
payment_token: Optional[str] = None
signature: Optional[str] = None
public_key: Optional[str] = None
requested_capabilities: list[str] = field(default_factory=list)
class LegacyApiAdapter:
"""
Wraps existing REST API endpoints with agent commerce capabilities.
Intercepts agent-specific headers, verifies payments and identity,
transforms responses, and passes through non-agent requests unchanged.
"""
def __init__(self, config: MigrationConfig):
self.config = config
self._http_client = httpx.AsyncClient(
base_url=config.greenhelix_url,
timeout=10.0,
)
self._payment_cache: dict[str, float] = {}
def classify_caller(self, request: Request) -> AgentContext:
"""Determine whether the request comes from a human or an agent."""
payment_token = request.headers.get("x-402-payment-token")
agent_id = request.headers.get("x-agent-id")
signature = request.headers.get("x-agent-signature")
public_key = request.headers.get("x-agent-public-key")
if payment_token or agent_id:
return AgentContext(
caller_type=CallerType.AGENT,
agent_id=agent_id,
payment_token=payment_token,
signature=signature,
public_key=public_key,
)
user_agent = request.headers.get("user-agent", "")
agent_indicators = ["bot", "agent", "crawler", "a2a-client"]
if any(indicator in user_agent.lower() for indicator in agent_indicators):
return AgentContext(caller_type=CallerType.AGENT, agent_id=agent_id)
return AgentContext(caller_type=CallerType.HUMAN)
async def verify_payment(
self, payment_token: str, endpoint: str, price: PricingConfig
) -> dict:
"""Verify a payment token against GreenHelix escrow."""
response = await self._http_client.post(
"/v1/payments/verify",
json={
"token": payment_token,
"expected_amount": str(price.price_usd),
"currency": price.currency,
"endpoint": endpoint,
},
headers={"Authorization": f"Bearer {self.config.greenhelix_api_key}"},
)
if response.status_code != 200:
return {"valid": False, "error": response.text}
result = response.json()
return {
"valid": result.get("verified", False),
"escrow_id": result.get("escrow_id"),
"amount": result.get("amount"),
}
async def verify_identity(self, ctx: AgentContext) -> dict:
"""Verify agent identity via Ed25519 signature."""
if not ctx.public_key or not ctx.signature:
return {"verified": False, "error": "Missing signature or public key"}
response = await self._http_client.post(
"/v1/identity/verify",
json={
"agent_id": ctx.agent_id,
"public_key": ctx.public_key,
"signature": ctx.signature,
},
headers={"Authorization": f"Bearer {self.config.greenhelix_api_key}"},
)
if response.status_code != 200:
return {"verified": False, "error": response.text}
return response.json()
def transform_response(
self, response_data: Any, endpoint: str, escrow_id: Optional[str] = None
) -> dict:
"""Transform legacy response format into agent-expected envelope."""
return {
"data": response_data,
"meta": {
"endpoint": endpoint,
"timestamp": time.time(),
"escrow_id": escrow_id,
"schema_version": "1.0",
},
}
def wrap_endpoint(self, endpoint_path: str):
"""
Decorator that wraps an existing endpoint handler with agent commerce.
Usage:
adapter = LegacyApiAdapter(config)
@app.get("/api/data")
@adapter.wrap_endpoint("/api/data")
async def get_data(request: Request):
return {"results": [...]}
"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
ctx = self.classify_caller(request)
# Pass through non-agent requests unchanged
if ctx.caller_type == CallerType.HUMAN:
return await func(request, *args, **kwargs)
# Agent request: verify payment if pricing is configured
pricing = self.config.pricing.get(endpoint_path)
escrow_id = None
if pricing and self.config.verify_payments:
if not ctx.payment_token:
return JSONResponse(
status_code=402,
content={
"error": "payment_required",
"detail": "This endpoint requires payment",
"price": str(pricing.price_usd),
"currency": pricing.currency,
"payment_url": (
f"{self.config.greenhelix_url}"
f"/v1/payments/create"
),
},
headers={
"X-Price": str(pricing.price_usd),
"X-Currency": pricing.currency,
"X-Payment-URL": (
f"{self.config.greenhelix_url}"
f"/v1/payments/create"
),
},
)
payment_result = await self.verify_payment(
ctx.payment_token, endpoint_path, pricing
)
if not payment_result["valid"]:
return JSONResponse(
status_code=402,
content={
"error": "payment_invalid",
"detail": payment_result.get("error", "Payment verification failed"),
},
)
escrow_id = payment_result.get("escrow_id")
# Verify agent identity if configured
if self.config.verify_identity and ctx.agent_id:
identity_result = await self.verify_identity(ctx)
if not identity_result.get("verified", False):
return JSONResponse(
status_code=403,
content={
"error": "identity_verification_failed",
"detail": identity_result.get("error", "Could not verify agent identity"),
},
)
# Execute the original handler
result = await func(request, *args, **kwargs)
# Transform response for agents if configured
if self.config.transform_responses:
if isinstance(result, Response):
# If the handler returned a Response object, extract the body
body = json.loads(result.body) if hasattr(result, 'body') else result
return JSONResponse(
content=self.transform_response(body, endpoint_path, escrow_id),
status_code=result.status_code if hasattr(result, 'status_code') else 200,
)
else:
return JSONResponse(
content=self.transform_response(result, endpoint_path, escrow_id),
)
return result
return wrapper
return decorator
async def close(self):
"""Clean up HTTP client resources."""
await self._http_client.aclose()
How the Adapter Routes Requests
The classify_caller method is the decision point. It checks for the x-402-payment-token and x-agent-id headers first — these are definitive agent indicators. If those are absent, it falls back to user-agent string inspection as a heuristic. This two-tier classification means agents that follow the x402 protocol get full commerce support, while agents that merely identify via user-agent get classified but may still need to provide payment headers for paid endpoints.
The wrap_endpoint decorator is designed to be non-invasive. You add it to an existing route handler with a single line. The original function signature does not change. The original return value is preserved for human callers. Only agent callers see the transformed response envelope. This means you can add the decorator to every endpoint in your API and nothing changes for existing users — they continue to get exactly the same responses they always got.
Request and Response Transformation
Response transformation deserves attention because it is where most migration bugs appear. The adapter wraps the original response in a standardized envelope with data and meta fields. The data field contains exactly what the original endpoint returned. The meta field adds context that agents need: which endpoint served the response, when it was generated, and the escrow ID if a payment was involved. This separation means agents can always find the business data in response["data"] and the transaction metadata in response["meta"], regardless of how different endpoints structure their responses internally.
For endpoints that return Response objects directly (common in FastAPI when you need to set custom status codes or headers), the adapter extracts the body, deserializes it, wraps it, and re-serializes. This adds negligible overhead — JSON parsing of a response that was going to be serialized anyway.
Chapter 3: x402 Retrofit Patterns
The x402 protocol uses HTTP status code 402 (Payment Required) to signal that an endpoint requires payment before it will process the request. This status code has been reserved in the HTTP specification since 1997 but was rarely used until agent commerce gave it a concrete purpose. Retrofitting x402 onto existing endpoints means adding payment negotiation to the HTTP layer without modifying business logic.
The Payment Negotiation Flow
When an agent hits a paid endpoint without a payment token, the flow works like this:
- Agent sends
GET /api/v1/datawith no payment headers. - Server responds with
402 Payment Required, including price and payment URL in headers. - Agent creates a payment escrow via the payment URL.
- Agent retries
GET /api/v1/datawith theX-402-Payment-Tokenheader. - Server verifies the token, processes the request, and returns data.
- Settlement happens automatically after the response is delivered.
This is a two-request flow for the agent: one to discover the price, one to pay and consume. Agents that already know the price (from cached discovery or manifest data) can skip step 1 and go straight to step 4.
Payment Header Specification
The x402 headers used in negotiation:
# Response headers on 402:
X-Price: 0.05
X-Currency: USD
X-Payment-URL: https://api.greenhelix.net/v1/payments/create
X-Payment-Methods: greenhelix-escrow, x402-direct
X-Price-Window: 300 # price valid for 300 seconds
# Request headers on paid request:
X-402-Payment-Token: ght_abc123...
X-Agent-Id: agent-buyer-001
X-Idempotency-Key: req_unique_id_here
FastAPI Middleware Implementation
The cleanest way to add x402 support across multiple endpoints in FastAPI is through middleware combined with a pricing registry:
import time
from typing import Optional
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import httpx
# Pricing registry: maps endpoint patterns to prices
ENDPOINT_PRICING: dict[str, dict] = {
"/api/v1/search": {
"price_usd": "0.02",
"currency": "USD",
"settlement_timeout": 300,
},
"/api/v1/analyze": {
"price_usd": "0.10",
"currency": "USD",
"settlement_timeout": 600,
},
"/api/v1/generate": {
"price_usd": "0.25",
"currency": "USD",
"settlement_timeout": 600,
},
}
GREENHELIX_URL = "https://api.greenhelix.net"
GREENHELIX_API_KEY = "" # Set from environment
class X402PaymentMiddleware(BaseHTTPMiddleware):
"""
Middleware that enforces x402 payment on configured endpoints.
Non-agent requests pass through. Agent requests to priced
endpoints must include a valid payment token.
"""
def __init__(self, app: FastAPI, greenhelix_api_key: str):
super().__init__(app)
self.api_key = greenhelix_api_key
self._client = httpx.AsyncClient(
base_url=GREENHELIX_URL, timeout=10.0
)
async def dispatch(self, request: Request, call_next) -> Response:
# Check if this endpoint has pricing
pricing = self._match_pricing(request.url.path)
if pricing is None:
return await call_next(request)
# Check if this is an agent request
payment_token = request.headers.get("x-402-payment-token")
agent_id = request.headers.get("x-agent-id")
# No agent headers — pass through to existing handler
if not payment_token and not agent_id:
return await call_next(request)
# Agent request without payment token — return 402
if not payment_token:
return self._payment_required_response(request.url.path, pricing)
# Agent request with payment token — verify it
verification = await self._verify_token(payment_token, request.url.path, pricing)
if not verification["valid"]:
return JSONResponse(
status_code=402,
content={
"error": "payment_invalid",
"detail": verification.get("error", "Token verification failed"),
},
)
# Payment verified — attach escrow info and proceed
request.state.escrow_id = verification.get("escrow_id")
request.state.payment_verified = True
response = await call_next(request)
# Add settlement headers to the response
response.headers["X-Escrow-Id"] = verification.get("escrow_id", "")
response.headers["X-Settlement-Status"] = "pending"
return response
def _match_pricing(self, path: str) -> Optional[dict]:
"""Match request path to pricing config, supporting wildcards."""
if path in ENDPOINT_PRICING:
return ENDPOINT_PRICING[path]
# Check prefix matches for versioned endpoints
for pattern, pricing in ENDPOINT_PRICING.items():
if path.startswith(pattern):
return pricing
return None
def _payment_required_response(self, path: str, pricing: dict) -> JSONResponse:
"""Build a 402 response with payment negotiation headers."""
return JSONResponse(
status_code=402,
content={
"error": "payment_required",
"endpoint": path,
"price": pricing["price_usd"],
"currency": pricing["currency"],
"payment_url": f"{GREENHELIX_URL}/v1/payments/create",
"price_valid_seconds": pricing.get("settlement_timeout", 300),
},
headers={
"X-Price": pricing["price_usd"],
"X-Currency": pricing["currency"],
"X-Payment-URL": f"{GREENHELIX_URL}/v1/payments/create",
"X-Payment-Methods": "greenhelix-escrow",
"X-Price-Window": str(pricing.get("settlement_timeout", 300)),
},
)
async def _verify_token(
self, token: str, endpoint: str, pricing: dict
) -> dict:
"""Verify payment token with GreenHelix."""
try:
response = await self._client.post(
"/v1/payments/verify",
json={
"token": token,
"expected_amount": pricing["price_usd"],
"currency": pricing["currency"],
"endpoint": endpoint,
},
headers={"Authorization": f"Bearer {self.api_key}"},
)
if response.status_code == 200:
return response.json()
return {"valid": False, "error": f"Verification returned {response.status_code}"}
except httpx.TimeoutException:
return {"valid": False, "error": "Payment verification timed out"}
# Application setup
app = FastAPI()
app.add_middleware(X402PaymentMiddleware, greenhelix_api_key=GREENHELIX_API_KEY)
Flask Middleware Equivalent
For Flask applications, the same pattern uses before_request and after_request hooks:
import time
from flask import Flask, request, jsonify, g
import httpx
app = Flask(__name__)
ENDPOINT_PRICING = {
"/api/v1/search": {"price_usd": "0.02", "currency": "USD"},
"/api/v1/analyze": {"price_usd": "0.10", "currency": "USD"},
}
GREENHELIX_URL = "https://api.greenhelix.net"
GREENHELIX_API_KEY = ""
@app.before_request
def check_agent_payment():
"""Intercept agent requests and enforce payment."""
pricing = ENDPOINT_PRICING.get(request.path)
if pricing is None:
return None # No pricing — pass through
payment_token = request.headers.get("X-402-Payment-Token")
agent_id = request.headers.get("X-Agent-Id")
if not payment_token and not agent_id:
return None # Not an agent — pass through
if not payment_token:
response = jsonify({
"error": "payment_required",
"price": pricing["price_usd"],
"currency": pricing["currency"],
"payment_url": f"{GREENHELIX_URL}/v1/payments/create",
})
response.status_code = 402
response.headers["X-Price"] = pricing["price_usd"]
response.headers["X-Currency"] = pricing["currency"]
return response
# Verify payment synchronously (use async client in production)
with httpx.Client(base_url=GREENHELIX_URL, timeout=10.0) as client:
result = client.post(
"/v1/payments/verify",
json={
"token": payment_token,
"expected_amount": pricing["price_usd"],
"endpoint": request.path,
},
headers={"Authorization": f"Bearer {GREENHELIX_API_KEY}"},
)
if result.status_code != 200 or not result.json().get("verified"):
return jsonify({"error": "payment_invalid"}), 402
g.escrow_id = result.json().get("escrow_id")
g.payment_verified = True
return None
@app.after_request
def add_settlement_headers(response):
"""Add escrow headers to responses for verified agent requests."""
if hasattr(g, "payment_verified") and g.payment_verified:
response.headers["X-Escrow-Id"] = getattr(g, "escrow_id", "")
response.headers["X-Settlement-Status"] = "pending"
return response
Django Middleware Equivalent
Django uses a class-based middleware approach:
import json
import httpx
from django.http import JsonResponse
ENDPOINT_PRICING = {
"/api/v1/search/": {"price_usd": "0.02", "currency": "USD"},
"/api/v1/analyze/": {"price_usd": "0.10", "currency": "USD"},
}
GREENHELIX_URL = "https://api.greenhelix.net"
GREENHELIX_API_KEY = ""
class X402PaymentMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
pricing = ENDPOINT_PRICING.get(request.path)
if pricing is None:
return self.get_response(request)
payment_token = request.META.get("HTTP_X_402_PAYMENT_TOKEN")
agent_id = request.META.get("HTTP_X_AGENT_ID")
if not payment_token and not agent_id:
return self.get_response(request)
if not payment_token:
response = JsonResponse(
{
"error": "payment_required",
"price": pricing["price_usd"],
"currency": pricing["currency"],
},
status=402,
)
response["X-Price"] = pricing["price_usd"]
response["X-Currency"] = pricing["currency"]
return response
# Verify payment token
with httpx.Client(base_url=GREENHELIX_URL, timeout=10.0) as client:
result = client.post(
"/v1/payments/verify",
json={
"token": payment_token,
"expected_amount": pricing["price_usd"],
"endpoint": request.path,
},
headers={"Authorization": f"Bearer {GREENHELIX_API_KEY}"},
)
if result.status_code != 200:
return JsonResponse({"error": "payment_invalid"}, status=402)
data = result.json()
if not data.get("verified"):
return JsonResponse({"error": "payment_invalid"}, status=402)
request.escrow_id = data.get("escrow_id")
request.payment_verified = True
response = self.get_response(request)
response["X-Escrow-Id"] = getattr(request, "escrow_id", "")
response["X-Settlement-Status"] = "pending"
return response
Token Verification Caching
Payment verification adds a network round-trip to every agent request. For high-throughput endpoints, this is unacceptable. Implement a short-lived verification cache:
import hashlib
import time
from typing import Optional
class PaymentVerificationCache:
"""Short-lived cache for payment verification results."""
def __init__(self, ttl_seconds: int = 30):
self.ttl = ttl_seconds
self._cache: dict[str, tuple[float, dict]] = {}
def _key(self, token: str, endpoint: str) -> str:
return hashlib.sha256(f"{token}:{endpoint}".encode()).hexdigest()
def get(self, token: str, endpoint: str) -> Optional[dict]:
key = self._key(token, endpoint)
entry = self._cache.get(key)
if entry is None:
return None
timestamp, result = entry
if time.time() - timestamp > self.ttl:
del self._cache[key]
return None
return result
def set(self, token: str, endpoint: str, result: dict) -> None:
key = self._key(token, endpoint)
self._cache[key] = (time.time(), result)
def evict_expired(self) -> int:
"""Remove all expired entries. Returns count of evicted entries."""
now = time.time()
expired = [k for k, (ts, _) in self._cache.items() if now - ts > self.ttl]
for k in expired:
del self._cache[k]
return len(expired)
The cache TTL should be short — 30 seconds is usually sufficient. A payment token verified 30 seconds ago is still valid for the same endpoint. But do not cache across endpoints: a token valid for /search is not necessarily valid for /analyze at a different price point.
Chapter 4: Authentication Bridging
Most REST APIs already have authentication. The migration challenge is not replacing that auth but running agent authentication alongside it. Agents authenticate differently from humans — they use cryptographic key pairs instead of passwords, sign requests instead of presenting session tokens, and identify themselves by public key hash instead of username. Authentication bridging lets both systems coexist.
The Dual-Auth Architecture
The bridging architecture has three layers:
Layer 1: Header inspection. Every request is classified as human-auth, agent-auth, or unauthenticated based on which headers are present. Human requests carry Authorization: Bearer <jwt> or X-API-Key. Agent requests carry X-Agent-Id, X-Agent-Public-Key, and X-Agent-Signature.
Layer 2: Verification. Human auth goes through your existing verification pipeline (JWT decode, API key lookup, OAuth token introspection). Agent auth goes through Ed25519 signature verification against the claimed public key, followed by identity resolution against GreenHelix.
Layer 3: Identity normalization. Both paths produce a unified identity object that downstream handlers consume. The handler does not know or care whether the caller is human or agent — it gets the same interface.
import time
import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
from cryptography.exceptions import InvalidSignature
import base64
class AuthMethod(str, Enum):
JWT = "jwt"
API_KEY = "api_key"
AGENT_SIGNATURE = "agent_signature"
NONE = "none"
@dataclass
class UnifiedIdentity:
"""Normalized identity produced by either auth path."""
identity_id: str
display_name: str
auth_method: AuthMethod
is_agent: bool
permissions: list[str]
tier: str = "free"
metadata: dict = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
def has_permission(self, permission: str) -> bool:
return permission in self.permissions or "*" in self.permissions
class AuthBridge:
"""
Bridges existing human auth with agent Ed25519 auth.
Produces a UnifiedIdentity regardless of auth method.
"""
def __init__(
self,
jwt_secret: str,
api_key_lookup: callable,
greenhelix_client: object,
):
self.jwt_secret = jwt_secret
self.api_key_lookup = api_key_lookup
self.greenhelix = greenhelix_client
async def authenticate(self, request) -> UnifiedIdentity:
"""
Authenticate the request using whichever method is present.
Raises HTTPException on failure.
"""
# Check for agent auth first (more specific headers)
agent_id = request.headers.get("x-agent-id")
agent_sig = request.headers.get("x-agent-signature")
agent_pubkey = request.headers.get("x-agent-public-key")
if agent_id and agent_sig and agent_pubkey:
return await self._authenticate_agent(
agent_id, agent_sig, agent_pubkey, request
)
# Check for JWT
auth_header = request.headers.get("authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
return self._authenticate_jwt(token)
# Check for API key
api_key = request.headers.get("x-api-key")
if api_key:
return self._authenticate_api_key(api_key)
# No auth provided
return UnifiedIdentity(
identity_id="anonymous",
display_name="Anonymous",
auth_method=AuthMethod.NONE,
is_agent=False,
permissions=["read:public"],
tier="free",
)
async def _authenticate_agent(
self,
agent_id: str,
signature_b64: str,
public_key_b64: str,
request,
) -> UnifiedIdentity:
"""Verify agent Ed25519 signature and resolve identity."""
# Step 1: Reconstruct the signed payload
# The agent signs: METHOD + PATH + TIMESTAMP + BODY_HASH
timestamp = request.headers.get("x-agent-timestamp", "")
body = await request.body()
body_hash = hashlib.sha256(body).hexdigest() if body else ""
signed_payload = (
f"{request.method}:{request.url.path}:{timestamp}:{body_hash}"
)
# Step 2: Verify the Ed25519 signature
try:
public_key_bytes = base64.b64decode(public_key_b64)
signature_bytes = base64.b64decode(signature_b64)
public_key = Ed25519PublicKey.from_public_key_der(public_key_bytes)
public_key.verify(signature_bytes, signed_payload.encode("utf-8"))
except (InvalidSignature, ValueError, Exception) as exc:
raise HTTPException(
status_code=401,
detail={
"error": "signature_invalid",
"message": "Ed25519 signature verification failed",
},
)
# Step 3: Check timestamp freshness (prevent replay attacks)
try:
req_time = float(timestamp)
if abs(time.time() - req_time) > 300: # 5-minute window
raise HTTPException(
status_code=401,
detail={
"error": "timestamp_expired",
"message": "Request timestamp outside acceptable window",
},
)
except (ValueError, TypeError):
raise HTTPException(
status_code=401,
detail={"error": "timestamp_invalid", "message": "Invalid timestamp format"},
)
# Step 4: Resolve agent identity from GreenHelix
agent_info = await self.greenhelix.resolve_agent(agent_id, public_key_b64)
return UnifiedIdentity(
identity_id=agent_id,
display_name=agent_info.get("display_name", agent_id),
auth_method=AuthMethod.AGENT_SIGNATURE,
is_agent=True,
permissions=agent_info.get("permissions", ["read:public", "execute:paid"]),
tier=agent_info.get("tier", "standard"),
metadata={
"public_key": public_key_b64,
"reputation_score": agent_info.get("reputation_score", 0),
"total_transactions": agent_info.get("total_transactions", 0),
},
)
def _authenticate_jwt(self, token: str) -> UnifiedIdentity:
"""Verify JWT and extract identity (existing auth path)."""
import jwt as pyjwt
try:
payload = pyjwt.decode(token, self.jwt_secret, algorithms=["HS256"])
except pyjwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail={"error": "token_expired"})
except pyjwt.InvalidTokenError:
raise HTTPException(status_code=401, detail={"error": "token_invalid"})
return UnifiedIdentity(
identity_id=payload["sub"],
display_name=payload.get("name", ""),
auth_method=AuthMethod.JWT,
is_agent=False,
permissions=payload.get("permissions", ["read:public"]),
tier=payload.get("tier", "free"),
)
def _authenticate_api_key(self, api_key: str) -> UnifiedIdentity:
"""Verify API key and extract identity (existing auth path)."""
key_data = self.api_key_lookup(api_key)
if key_data is None:
raise HTTPException(status_code=401, detail={"error": "api_key_invalid"})
return UnifiedIdentity(
identity_id=key_data["user_id"],
display_name=key_data.get("name", ""),
auth_method=AuthMethod.API_KEY,
is_agent=False,
permissions=key_data.get("permissions", ["read:public"]),
tier=key_data.get("tier", "free"),
)
Using the AuthBridge in FastAPI
Integrate the bridge as a dependency that replaces your existing auth dependency:
from fastapi import Depends, FastAPI, Request
app = FastAPI()
# Initialize the bridge once at startup
auth_bridge = AuthBridge(
jwt_secret="your-jwt-secret",
api_key_lookup=lambda key: db.lookup_api_key(key),
greenhelix_client=greenhelix_client,
)
async def get_identity(request: Request) -> UnifiedIdentity:
"""FastAPI dependency that authenticates via the bridge."""
return await auth_bridge.authenticate(request)
@app.get("/api/v1/data")
async def get_data(identity: UnifiedIdentity = Depends(get_identity)):
# identity.is_agent tells you the caller type
# identity.permissions controls access
# identity.tier determines rate limits
if not identity.has_permission("read:data"):
raise HTTPException(status_code=403, detail="Insufficient permissions")
return {"results": [...], "caller": identity.identity_id}
Session Management for Agents
Human sessions rely on cookies and server-side state. Agent sessions are stateless by design — each request is independently authenticated. But some APIs have multi-step workflows (start a job, poll for status, download results) that require session continuity. For agents, implement session continuity through idempotency keys and request correlation:
from dataclasses import dataclass, field
from typing import Optional
import time
import uuid
@dataclass
class AgentSession:
"""Lightweight session for multi-step agent workflows."""
session_id: str
agent_id: str
created_at: float
last_activity: float
workflow_state: dict = field(default_factory=dict)
ttl_seconds: int = 3600
class AgentSessionStore:
"""In-memory session store for agent workflows."""
def __init__(self):
self._sessions: dict[str, AgentSession] = {}
def create(self, agent_id: str) -> AgentSession:
session = AgentSession(
session_id=str(uuid.uuid4()),
agent_id=agent_id,
created_at=time.time(),
last_activity=time.time(),
)
self._sessions[session.session_id] = session
return session
def get(self, session_id: str, agent_id: str) -> Optional[AgentSession]:
session = self._sessions.get(session_id)
if session is None:
return None
if session.agent_id != agent_id:
return None # Session belongs to a different agent
if time.time() - session.last_activity > session.ttl_seconds:
del self._sessions[session_id]
return None
session.last_activity = time.time()
return session
def update_state(self, session_id: str, key: str, value: object) -> None:
session = self._sessions.get(session_id)
if session:
session.workflow_state[key] = value
session.last_activity = time.time()
def cleanup_expired(self) -> int:
now = time.time()
expired = [
sid for sid, s in self._sessions.items()
if now - s.last_activity > s.ttl_seconds
]
for sid in expired:
del self._sessions[sid]
return len(expired)
Agents include X-Session-Id on follow-up requests. The session ID is returned in the response to the first request in a workflow. This is optional — agents that perform only single-call operations never need sessions.
Chapter 5: Gradual Migration Strategies
Big-bang migrations fail. You do not flip a switch and convert your entire API to agent commerce overnight. Instead, you migrate one endpoint at a time, verify it works, and move to the next. This chapter covers the practical strategies for incremental migration without disrupting existing traffic.
Endpoint-by-Endpoint Migration
The migration order matters. Start with read-only endpoints (GET requests), then move to write endpoints (POST/PUT), and finish with complex workflows (multi-step operations). Each endpoint goes through four stages:
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class MigrationStage(str, Enum):
LEGACY = "legacy" # No agent support
DISCOVERY = "discovery" # Returns 402 with pricing, but doesn't process payments
PAYMENT = "payment" # Accepts and verifies payments
FULL = "full" # Full agent commerce: payment + identity + response transform
@dataclass
class EndpointMigrationState:
path: str
stage: MigrationStage
migrated_at: Optional[float] = None
migration_notes: str = ""
class MigrationRegistry:
"""Tracks which endpoints are at which migration stage."""
def __init__(self):
self._endpoints: dict[str, EndpointMigrationState] = {}
def register(self, path: str, stage: MigrationStage, notes: str = "") -> None:
self._endpoints[path] = EndpointMigrationState(
path=path,
stage=stage,
migrated_at=time.time() if stage != MigrationStage.LEGACY else None,
migration_notes=notes,
)
def get_stage(self, path: str) -> MigrationStage:
state = self._endpoints.get(path)
return state.stage if state else MigrationStage.LEGACY
def advance(self, path: str) -> MigrationStage:
"""Move an endpoint to the next migration stage."""
current = self.get_stage(path)
stages = list(MigrationStage)
current_index = stages.index(current)
if current_index < len(stages) - 1:
new_stage = stages[current_index + 1]
self.register(path, new_stage)
return new_stage
return current
def summary(self) -> dict[str, int]:
counts = {stage.value: 0 for stage in MigrationStage}
for state in self._endpoints.values():
counts[state.stage.value] += 1
return counts
# Usage
registry = MigrationRegistry()
registry.register("/api/v1/search", MigrationStage.FULL)
registry.register("/api/v1/analyze", MigrationStage.PAYMENT)
registry.register("/api/v1/users", MigrationStage.LEGACY)
print(registry.summary())
# {'legacy': 1, 'discovery': 0, 'payment': 1, 'full': 1}
The discovery stage is critical and often skipped. At this stage, the endpoint returns 402 with correct pricing headers when an agent hits it, but does not actually process payments. This lets agents discover your pricing and capabilities without you needing to handle real money yet. Run in discovery mode for at least a week to see what agents query your API, what prices they encounter, and how they react.
Feature Flags for Agent Commerce
Feature flags let you enable and disable agent commerce per endpoint, per agent, or globally without deploying new code:
from dataclasses import dataclass, field
from typing import Optional
import json
import os
@dataclass
class AgentCommerceFlags:
"""Feature flags controlling agent commerce behavior."""
global_enabled: bool = True
payment_required: bool = True
identity_verification: bool = True
response_transformation: bool = True
allowed_agent_ids: list[str] = field(default_factory=list) # Empty = allow all
blocked_agent_ids: list[str] = field(default_factory=list)
endpoint_overrides: dict[str, dict] = field(default_factory=dict)
@classmethod
def from_env(cls) -> "AgentCommerceFlags":
"""Load flags from environment variables or config file."""
config_path = os.getenv("AGENT_COMMERCE_CONFIG", "agent_commerce_flags.json")
if os.path.exists(config_path):
with open(config_path) as f:
data = json.load(f)
return cls(**data)
return cls(
global_enabled=os.getenv("AGENT_COMMERCE_ENABLED", "true").lower() == "true",
payment_required=os.getenv("AGENT_PAYMENT_REQUIRED", "true").lower() == "true",
)
def is_enabled_for(self, endpoint: str, agent_id: Optional[str] = None) -> bool:
"""Check if agent commerce is enabled for a specific endpoint and agent."""
if not self.global_enabled:
return False
if agent_id and agent_id in self.blocked_agent_ids:
return False
if self.allowed_agent_ids and agent_id not in self.allowed_agent_ids:
return False
override = self.endpoint_overrides.get(endpoint, {})
return override.get("enabled", True)
def get_price_override(self, endpoint: str) -> Optional[str]:
"""Get endpoint-specific price override, if any."""
override = self.endpoint_overrides.get(endpoint, {})
return override.get("price_usd")
Traffic Routing with Reverse Proxy
For APIs running behind nginx or a similar reverse proxy, you can route agent traffic to a separate upstream that has the commerce middleware enabled, while human traffic hits the original upstream unchanged:
# nginx.conf — agent traffic routing
upstream legacy_api {
server 127.0.0.1:8000;
}
upstream agent_commerce_api {
server 127.0.0.1:8001;
}
map $http_x_agent_id $is_agent {
default 0;
"~.+" 1;
}
map $http_x_402_payment_token $has_payment {
default 0;
"~.+" 1;
}
server {
listen 443 ssl;
server_name api.example.com;
location /api/v1/ {
# Route to agent commerce API if agent headers are present
if ($is_agent) {
proxy_pass http://agent_commerce_api;
break;
}
if ($has_payment) {
proxy_pass http://agent_commerce_api;
break;
}
# Default: route to legacy API
proxy_pass http://legacy_api;
}
}
This approach lets you run the agent commerce version of your API as a separate process. The legacy API process remains completely untouched. If anything goes wrong with agent commerce, you adjust the nginx config to route all traffic back to the legacy upstream. Zero-downtime rollback.
A/B Testing Human vs Agent Traffic
Track metrics separately for human and agent callers to detect regressions:
import time
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class TrafficMetrics:
request_count: int = 0
error_count: int = 0
total_latency_ms: float = 0.0
p95_latencies: list[float] = field(default_factory=list)
@property
def avg_latency_ms(self) -> float:
if self.request_count == 0:
return 0.0
return self.total_latency_ms / self.request_count
@property
def error_rate(self) -> float:
if self.request_count == 0:
return 0.0
return self.error_count / self.request_count
class MigrationMetricsCollector:
"""Collects metrics split by caller type for migration monitoring."""
def __init__(self):
self._metrics: dict[str, dict[str, TrafficMetrics]] = defaultdict(
lambda: {"human": TrafficMetrics(), "agent": TrafficMetrics()}
)
def record(
self,
endpoint: str,
caller_type: str,
latency_ms: float,
is_error: bool = False,
) -> None:
metrics = self._metrics[endpoint][caller_type]
metrics.request_count += 1
metrics.total_latency_ms += latency_ms
metrics.p95_latencies.append(latency_ms)
if is_error:
metrics.error_count += 1
# Keep only last 1000 latencies for p95 calculation
if len(metrics.p95_latencies) > 1000:
metrics.p95_latencies = metrics.p95_latencies[-1000:]
def compare(self, endpoint: str) -> dict:
"""Compare human vs agent metrics for an endpoint."""
human = self._metrics[endpoint]["human"]
agent = self._metrics[endpoint]["agent"]
return {
"endpoint": endpoint,
"human": {
"requests": human.request_count,
"avg_latency_ms": round(human.avg_latency_ms, 2),
"error_rate": round(human.error_rate, 4),
},
"agent": {
"requests": agent.request_count,
"avg_latency_ms": round(agent.avg_latency_ms, 2),
"error_rate": round(agent.error_rate, 4),
},
"latency_delta_ms": round(agent.avg_latency_ms - human.avg_latency_ms, 2),
}
The latency_delta_ms is the key metric during migration. If agent requests are significantly slower than human requests at the same endpoint, the commerce middleware is adding too much overhead. The target is under 50ms of additional latency for the payment verification and response transformation layer.
Chapter 6: MigrationValidator Class
Migration validation answers one question: does the migrated endpoint behave identically to the original for all non-commerce concerns? The business logic should produce the same output, the response schemas should be compatible, and performance should not degrade beyond acceptable bounds. The MigrationValidator automates these checks.
The Validator Architecture
The validator works by sending the same request to both the legacy endpoint and the migrated endpoint, then comparing the responses. For endpoints that have already replaced the legacy version (same URL, new middleware), it compares captured legacy response snapshots against live migrated responses.
import json
import time
import hashlib
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
import httpx
class ValidationResult(str, Enum):
PASS = "pass"
FAIL = "fail"
WARN = "warn"
@dataclass
class ValidationReport:
endpoint: str
result: ValidationResult
checks: list[dict] = field(default_factory=list)
duration_ms: float = 0.0
timestamp: float = field(default_factory=time.time)
def add_check(self, name: str, result: ValidationResult, detail: str = "") -> None:
self.checks.append({
"name": name,
"result": result.value,
"detail": detail,
})
# Overall result is the worst individual result
if result == ValidationResult.FAIL:
self.result = ValidationResult.FAIL
elif result == ValidationResult.WARN and self.result != ValidationResult.FAIL:
self.result = ValidationResult.WARN
@property
def passed(self) -> bool:
return self.result != ValidationResult.FAIL
def summary(self) -> str:
total = len(self.checks)
passed = sum(1 for c in self.checks if c["result"] == "pass")
return f"{self.endpoint}: {passed}/{total} checks passed ({self.result.value})"
@dataclass
class ResponseSnapshot:
"""Captured response for comparison."""
status_code: int
headers: dict[str, str]
body: Any
latency_ms: float
captured_at: float = field(default_factory=time.time)
def body_hash(self) -> str:
return hashlib.sha256(json.dumps(self.body, sort_keys=True).encode()).hexdigest()
class MigrationValidator:
"""
Validates that migrated endpoints preserve original behavior.
Compares response schemas, status codes, data content, and
performance between legacy and migrated versions.
"""
def __init__(
self,
legacy_base_url: str,
migrated_base_url: str,
tolerance_ms: float = 100.0,
schema_strict: bool = False,
):
self.legacy_url = legacy_base_url.rstrip("/")
self.migrated_url = migrated_base_url.rstrip("/")
self.tolerance_ms = tolerance_ms
self.schema_strict = schema_strict
self._client = httpx.AsyncClient(timeout=30.0)
self._snapshots: dict[str, list[ResponseSnapshot]] = {}
async def capture_legacy_snapshot(
self, endpoint: str, method: str = "GET", body: Any = None, headers: dict = None,
) -> ResponseSnapshot:
"""Capture a response from the legacy endpoint for later comparison."""
request_headers = headers or {}
start = time.time()
if method.upper() == "GET":
response = await self._client.get(
f"{self.legacy_url}{endpoint}", headers=request_headers
)
else:
response = await self._client.request(
method, f"{self.legacy_url}{endpoint}",
json=body, headers=request_headers,
)
latency_ms = (time.time() - start) * 1000
snapshot = ResponseSnapshot(
status_code=response.status_code,
headers=dict(response.headers),
body=response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text,
latency_ms=latency_ms,
)
if endpoint not in self._snapshots:
self._snapshots[endpoint] = []
self._snapshots[endpoint].append(snapshot)
return snapshot
async def validate_endpoint(
self,
endpoint: str,
method: str = "GET",
body: Any = None,
human_headers: dict = None,
agent_headers: dict = None,
) -> ValidationReport:
"""Run full validation suite on a migrated endpoint."""
report = ValidationReport(endpoint=endpoint, result=ValidationResult.PASS)
start = time.time()
# Test 1: Human request produces same response as legacy
human_headers = human_headers or {}
try:
legacy_resp = await self._request(self.legacy_url, endpoint, method, body, human_headers)
migrated_resp = await self._request(self.migrated_url, endpoint, method, body, human_headers)
self._check_status_code(report, legacy_resp, migrated_resp)
self._check_schema_compatibility(report, legacy_resp, migrated_resp)
self._check_data_equivalence(report, legacy_resp, migrated_resp)
except httpx.RequestError as exc:
report.add_check("connectivity", ValidationResult.FAIL, str(exc))
report.duration_ms = (time.time() - start) * 1000
return report
# Test 2: Agent request without payment returns 402
if agent_headers:
agent_no_payment = {k: v for k, v in agent_headers.items() if k.lower() != "x-402-payment-token"}
try:
agent_resp = await self._request(
self.migrated_url, endpoint, method, body, agent_no_payment
)
if agent_resp.status_code == 402:
report.add_check("402_response", ValidationResult.PASS, "Returns 402 for unpaid agent requests")
self._check_402_headers(report, agent_resp)
else:
report.add_check(
"402_response", ValidationResult.WARN,
f"Expected 402, got {agent_resp.status_code}",
)
except httpx.RequestError as exc:
report.add_check("agent_request", ValidationResult.FAIL, str(exc))
# Test 3: Performance regression check
await self._check_performance(report, endpoint, method, body, human_headers)
report.duration_ms = (time.time() - start) * 1000
return report
async def _request(
self, base_url: str, endpoint: str, method: str, body: Any, headers: dict,
) -> httpx.Response:
if method.upper() == "GET":
return await self._client.get(f"{base_url}{endpoint}", headers=headers)
return await self._client.request(
method, f"{base_url}{endpoint}", json=body, headers=headers
)
def _check_status_code(
self, report: ValidationReport, legacy: httpx.Response, migrated: httpx.Response,
) -> None:
if legacy.status_code == migrated.status_code:
report.add_check("status_code", ValidationResult.PASS)
else:
report.add_check(
"status_code", ValidationResult.FAIL,
f"Legacy returned {legacy.status_code}, migrated returned {migrated.status_code}",
)
def _check_schema_compatibility(
self, report: ValidationReport, legacy: httpx.Response, migrated: httpx.Response,
) -> None:
"""Compare response structure (keys at each level)."""
try:
legacy_body = legacy.json()
migrated_body = migrated.json()
except (json.JSONDecodeError, ValueError):
report.add_check("schema", ValidationResult.WARN, "Could not parse response as JSON")
return
legacy_keys = self._extract_keys(legacy_body)
migrated_keys = self._extract_keys(migrated_body)
missing = legacy_keys - migrated_keys
added = migrated_keys - legacy_keys
if not missing and not added:
report.add_check("schema", ValidationResult.PASS)
elif missing:
report.add_check(
"schema", ValidationResult.FAIL,
f"Missing keys in migrated response: {missing}",
)
elif added and self.schema_strict:
report.add_check(
"schema", ValidationResult.FAIL,
f"Unexpected keys in migrated response: {added}",
)
else:
report.add_check(
"schema", ValidationResult.WARN,
f"New keys in migrated response: {added}",
)
def _check_data_equivalence(
self, report: ValidationReport, legacy: httpx.Response, migrated: httpx.Response,
) -> None:
"""Check that the actual data values match."""
try:
legacy_body = legacy.json()
migrated_body = migrated.json()
except (json.JSONDecodeError, ValueError):
return
# If migrated response wraps data in an envelope, unwrap it
if isinstance(migrated_body, dict) and "data" in migrated_body and "meta" in migrated_body:
migrated_body = migrated_body["data"]
legacy_hash = hashlib.sha256(json.dumps(legacy_body, sort_keys=True).encode()).hexdigest()
migrated_hash = hashlib.sha256(json.dumps(migrated_body, sort_keys=True).encode()).hexdigest()
if legacy_hash == migrated_hash:
report.add_check("data_equivalence", ValidationResult.PASS)
else:
report.add_check(
"data_equivalence", ValidationResult.WARN,
"Response data differs (may be expected for dynamic fields like timestamps)",
)
def _check_402_headers(self, report: ValidationReport, response: httpx.Response) -> None:
"""Verify that 402 responses include required negotiation headers."""
required_headers = ["x-price", "x-currency", "x-payment-url"]
missing = [h for h in required_headers if h not in response.headers]
if not missing:
report.add_check("402_headers", ValidationResult.PASS)
else:
report.add_check(
"402_headers", ValidationResult.FAIL,
f"Missing 402 headers: {missing}",
)
async def _check_performance(
self, report: ValidationReport, endpoint: str, method: str, body: Any, headers: dict,
) -> None:
"""Check that migration hasn't introduced unacceptable latency."""
# Run 5 requests against each and compare average
legacy_latencies = []
migrated_latencies = []
for _ in range(5):
start = time.time()
await self._request(self.legacy_url, endpoint, method, body, headers)
legacy_latencies.append((time.time() - start) * 1000)
start = time.time()
await self._request(self.migrated_url, endpoint, method, body, headers)
migrated_latencies.append((time.time() - start) * 1000)
avg_legacy = sum(legacy_latencies) / len(legacy_latencies)
avg_migrated = sum(migrated_latencies) / len(migrated_latencies)
delta = avg_migrated - avg_legacy
if delta <= self.tolerance_ms:
report.add_check(
"performance", ValidationResult.PASS,
f"Latency delta: {delta:.1f}ms (within {self.tolerance_ms}ms tolerance)",
)
elif delta <= self.tolerance_ms * 2:
report.add_check(
"performance", ValidationResult.WARN,
f"Latency delta: {delta:.1f}ms (approaching tolerance of {self.tolerance_ms}ms)",
)
else:
report.add_check(
"performance", ValidationResult.FAIL,
f"Latency delta: {delta:.1f}ms (exceeds tolerance of {self.tolerance_ms}ms)",
)
def _extract_keys(self, obj: Any, prefix: str = "") -> set[str]:
"""Recursively extract all key paths from a JSON object."""
keys = set()
if isinstance(obj, dict):
for k, v in obj.items():
full_key = f"{prefix}.{k}" if prefix else k
keys.add(full_key)
keys.update(self._extract_keys(v, full_key))
elif isinstance(obj, list) and obj:
keys.update(self._extract_keys(obj[0], f"{prefix}[]"))
return keys
async def close(self):
await self._client.aclose()
Running Validation
Run the validator as part of your CI pipeline or as a standalone validation script:
import asyncio
async def validate_migration():
validator = MigrationValidator(
legacy_base_url="http://localhost:8000",
migrated_base_url="http://localhost:8001",
tolerance_ms=50.0,
)
endpoints_to_validate = [
{
"endpoint": "/api/v1/search",
"method": "GET",
"human_headers": {"Authorization": "Bearer test-jwt-token"},
"agent_headers": {"X-Agent-Id": "test-agent-001"},
},
{
"endpoint": "/api/v1/analyze",
"method": "POST",
"body": {"text": "sample input for validation"},
"human_headers": {"Authorization": "Bearer test-jwt-token"},
"agent_headers": {"X-Agent-Id": "test-agent-001"},
},
]
reports = []
for ep in endpoints_to_validate:
report = await validator.validate_endpoint(**ep)
reports.append(report)
print(report.summary())
for check in report.checks:
status = "OK" if check["result"] == "pass" else check["result"].upper()
detail = f" - {check['detail']}" if check["detail"] else ""
print(f" [{status}] {check['name']}{detail}")
await validator.close()
# Fail CI if any endpoint fails validation
all_passed = all(r.passed for r in reports)
if not all_passed:
print("\nMigration validation FAILED")
exit(1)
else:
print("\nAll migration validations passed")
asyncio.run(validate_migration())
Contract Testing Between Old and New
Beyond schema comparison, contract tests verify specific response invariants that your API consumers depend on. Define these contracts explicitly:
@dataclass
class ResponseContract:
"""Defines expected properties of a response that must not change during migration."""
endpoint: str
status_code: int
required_fields: list[str]
field_types: dict[str, type]
value_constraints: dict[str, callable] # field -> validator function
def validate(self, response_body: dict) -> list[str]:
"""Returns list of contract violations."""
violations = []
for field_path in self.required_fields:
if not self._has_field(response_body, field_path):
violations.append(f"Missing required field: {field_path}")
for field_path, expected_type in self.field_types.items():
value = self._get_field(response_body, field_path)
if value is not None and not isinstance(value, expected_type):
violations.append(
f"Type mismatch for {field_path}: expected {expected_type.__name__}, "
f"got {type(value).__name__}"
)
for field_path, validator in self.value_constraints.items():
value = self._get_field(response_body, field_path)
if value is not None and not validator(value):
violations.append(f"Value constraint failed for {field_path}: {value}")
return violations
def _has_field(self, obj: dict, path: str) -> bool:
parts = path.split(".")
current = obj
for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
else:
return False
return True
def _get_field(self, obj: dict, path: str) -> Any:
parts = path.split(".")
current = obj
for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
else:
return None
return current
# Example contract definition
search_contract = ResponseContract(
endpoint="/api/v1/search",
status_code=200,
required_fields=["results", "total_count"],
field_types={
"results": list,
"total_count": int,
},
value_constraints={
"total_count": lambda v: v >= 0,
},
)
Chapter 7: Testing Migration
Testing a migration is not the same as testing new code. You are not verifying that something works — you are verifying that it works the same way it did before, plus new capabilities. The testing strategy has three phases: pre-migration baseline capture, parallel validation during migration, and post-migration regression testing.
Phase 1: Baseline Capture
Before touching any code, capture baseline responses for every endpoint you plan to migrate. These baselines become the ground truth for comparison:
import json
import time
import os
from dataclasses import dataclass, asdict
from typing import Any
import httpx
@dataclass
class BaselineCapture:
endpoint: str
method: str
request_body: Any
response_status: int
response_body: Any
response_headers: dict[str, str]
latency_ms: float
captured_at: float
class BaselineRecorder:
"""Records baseline API responses for migration comparison."""
def __init__(self, base_url: str, output_dir: str = "./migration_baselines"):
self.base_url = base_url.rstrip("/")
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
async def capture(
self, endpoint: str, method: str = "GET", body: Any = None,
headers: dict = None, iterations: int = 10,
) -> list[BaselineCapture]:
"""Capture multiple baseline responses for statistical comparison."""
captures = []
async with httpx.AsyncClient(timeout=30.0) as client:
for _ in range(iterations):
start = time.time()
if method.upper() == "GET":
resp = await client.get(
f"{self.base_url}{endpoint}", headers=headers or {}
)
else:
resp = await client.request(
method, f"{self.base_url}{endpoint}",
json=body, headers=headers or {},
)
latency = (time.time() - start) * 1000
try:
resp_body = resp.json()
except ValueError:
resp_body = resp.text
capture = BaselineCapture(
endpoint=endpoint,
method=method,
request_body=body,
response_status=resp.status_code,
response_body=resp_body,
response_headers=dict(resp.headers),
latency_ms=latency,
captured_at=time.time(),
)
captures.append(capture)
# Save to disk
filename = endpoint.replace("/", "_").strip("_") + ".json"
filepath = os.path.join(self.output_dir, filename)
with open(filepath, "w") as f:
json.dump([asdict(c) for c in captures], f, indent=2, default=str)
return captures
Run the baseline recorder against every endpoint in your migration list. Store the output in version control. These files serve as regression evidence — if something breaks during migration, you have concrete proof of what the correct behavior was.
Phase 2: Shadow Traffic Validation
Shadow traffic sends a copy of every real production request to the migrated endpoint and compares responses without affecting the actual client. The client gets the legacy response, but the migrated response is captured and compared asynchronously:
import asyncio
import time
import json
import logging
from typing import Any
import httpx
logger = logging.getLogger("shadow_traffic")
class ShadowTrafficValidator:
"""
Replays production traffic against migrated endpoints.
Compares responses without affecting production clients.
"""
def __init__(
self,
migrated_base_url: str,
mismatch_log: str = "./shadow_mismatches.jsonl",
):
self.migrated_url = migrated_base_url.rstrip("/")
self.mismatch_log = mismatch_log
self._client = httpx.AsyncClient(timeout=10.0)
self._stats = {"total": 0, "matched": 0, "mismatched": 0, "errors": 0}
async def shadow_request(
self,
endpoint: str,
method: str,
body: Any,
headers: dict,
legacy_status: int,
legacy_body: Any,
) -> None:
"""Send shadow request and compare with legacy response."""
self._stats["total"] += 1
# Remove auth headers that might not work against the migrated version
shadow_headers = {
k: v for k, v in headers.items()
if k.lower() not in ("cookie", "x-csrf-token")
}
try:
start = time.time()
if method.upper() == "GET":
resp = await self._client.get(
f"{self.migrated_url}{endpoint}", headers=shadow_headers
)
else:
resp = await self._client.request(
method, f"{self.migrated_url}{endpoint}",
json=body, headers=shadow_headers,
)
latency = (time.time() - start) * 1000
# Compare status codes
if resp.status_code != legacy_status:
self._log_mismatch(endpoint, "status_code", legacy_status, resp.status_code)
self._stats["mismatched"] += 1
return
# Compare response bodies (ignoring dynamic fields)
try:
migrated_body = resp.json()
if not self._bodies_equivalent(legacy_body, migrated_body):
self._log_mismatch(endpoint, "body", legacy_body, migrated_body)
self._stats["mismatched"] += 1
return
except (json.JSONDecodeError, ValueError):
pass
self._stats["matched"] += 1
except Exception as exc:
self._stats["errors"] += 1
logger.error(f"Shadow request failed for {endpoint}: {exc}")
def _bodies_equivalent(self, legacy: Any, migrated: Any) -> bool:
"""Compare bodies, ignoring known dynamic fields."""
dynamic_fields = {"timestamp", "request_id", "trace_id", "generated_at"}
if isinstance(legacy, dict) and isinstance(migrated, dict):
# If migrated wraps in envelope, unwrap
if "data" in migrated and "meta" in migrated:
migrated = migrated["data"]
for key in set(legacy.keys()) | set(migrated.keys()):
if key in dynamic_fields:
continue
if key not in legacy or key not in migrated:
return False
if not self._bodies_equivalent(legacy[key], migrated[key]):
return False
return True
if isinstance(legacy, list) and isinstance(migrated, list):
if len(legacy) != len(migrated):
return False
return all(
self._bodies_equivalent(a, b) for a, b in zip(legacy, migrated)
)
return legacy == migrated
def _log_mismatch(
self, endpoint: str, field: str, expected: Any, actual: Any,
) -> None:
entry = {
"timestamp": time.time(),
"endpoint": endpoint,
"field": field,
"expected": str(expected)[:500],
"actual": str(actual)[:500],
}
with open(self.mismatch_log, "a") as f:
f.write(json.dumps(entry) + "\n")
@property
def match_rate(self) -> float:
if self._stats["total"] == 0:
return 0.0
return self._stats["matched"] / self._stats["total"]
async def close(self):
await self._client.aclose()
Integrate shadow validation into your production middleware. After the legacy handler returns a response to the client, fire off the shadow request asynchronously:
import asyncio
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
shadow_validator = ShadowTrafficValidator(migrated_base_url="http://localhost:8001")
class ShadowTrafficMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
# Process request normally
body = await request.body()
response = await call_next(request)
# Fire shadow request in background (don't block the response)
try:
resp_body = json.loads(response.body) if hasattr(response, "body") else None
except (json.JSONDecodeError, ValueError):
resp_body = None
asyncio.create_task(
shadow_validator.shadow_request(
endpoint=request.url.path,
method=request.method,
body=json.loads(body) if body else None,
headers=dict(request.headers),
legacy_status=response.status_code,
legacy_body=resp_body,
)
)
return response
Rollback Procedures
Every migration needs a rollback plan. Document rollback steps for each migration stage:
Stage 1 rollback (Discovery). Remove the wrap_endpoint decorator or disable the middleware via feature flag. No data to clean up — discovery mode does not create escrows or process payments.
Stage 2 rollback (Payment). Disable payment verification in the feature flags. Any in-flight escrows will expire and refund automatically after the settlement timeout. Communicate the maintenance window to agent consumers via your A2A protocol card status field.
Stage 3 rollback (Full). Revert to the pre-migration code deployment. If using the reverse proxy routing approach, switch the nginx config to route all traffic to the legacy upstream. Agent sessions in progress will fail — this is acceptable because agent sessions are designed to be retryable.
Monitoring During Migration
Set up alerts for these conditions during each migration phase:
MIGRATION_ALERTS = {
"error_rate_spike": {
"condition": "error_rate > baseline_error_rate * 1.5",
"action": "Pause migration, investigate",
"severity": "high",
},
"latency_regression": {
"condition": "p95_latency > baseline_p95 + 100ms",
"action": "Check payment verification service health",
"severity": "medium",
},
"402_misconfiguration": {
"condition": "402_responses_for_human_callers > 0",
"action": "Immediate rollback — caller classification is broken",
"severity": "critical",
},
"shadow_mismatch_rate": {
"condition": "shadow_match_rate < 0.95",
"action": "Stop migration, review mismatches",
"severity": "high",
},
"payment_verification_timeout": {
"condition": "payment_verification_p95 > 5000ms",
"action": "Enable verification cache, check GreenHelix status",
"severity": "medium",
},
}
The most critical alert is 402_misconfiguration. If human callers ever receive a 402 response, the caller classification logic is broken and the migration must be rolled back immediately. This is the one failure mode that directly impacts existing users.
Chapter 8: Performance Comparison
Adding a commerce layer to your API adds latency. The question is how much and whether it matters. This chapter provides the benchmarking framework to measure the exact overhead and the optimization strategies to minimize it.
Benchmarking Framework
import asyncio
import time
import statistics
from dataclasses import dataclass, field
from typing import Optional
import httpx
@dataclass
class BenchmarkResult:
endpoint: str
variant: str # "legacy" or "migrated"
requests: int
avg_latency_ms: float
p50_latency_ms: float
p95_latency_ms: float
p99_latency_ms: float
min_latency_ms: float
max_latency_ms: float
error_count: int
throughput_rps: float
class MigrationBenchmark:
"""Benchmarks legacy vs migrated endpoint performance."""
def __init__(
self,
legacy_url: str,
migrated_url: str,
concurrency: int = 10,
duration_seconds: int = 30,
):
self.legacy_url = legacy_url.rstrip("/")
self.migrated_url = migrated_url.rstrip("/")
self.concurrency = concurrency
self.duration = duration_seconds
async def run(
self,
endpoint: str,
method: str = "GET",
body: object = None,
headers: dict = None,
) -> dict[str, BenchmarkResult]:
"""Run benchmark against both legacy and migrated endpoints."""
legacy_result = await self._benchmark_variant(
self.legacy_url, endpoint, method, body, headers or {}, "legacy"
)
migrated_result = await self._benchmark_variant(
self.migrated_url, endpoint, method, body, headers or {}, "migrated"
)
return {
"legacy": legacy_result,
"migrated": migrated_result,
"overhead_ms": migrated_result.avg_latency_ms - legacy_result.avg_latency_ms,
"overhead_pct": (
(migrated_result.avg_latency_ms - legacy_result.avg_latency_ms)
/ legacy_result.avg_latency_ms * 100
if legacy_result.avg_latency_ms > 0 else 0
),
"throughput_delta_rps": (
migrated_result.throughput_rps - legacy_result.throughput_rps
),
}
async def _benchmark_variant(
self,
base_url: str,
endpoint: str,
method: str,
body: object,
headers: dict,
variant: str,
) -> BenchmarkResult:
latencies: list[float] = []
errors = 0
start_time = time.time()
async def worker():
nonlocal errors
async with httpx.AsyncClient(timeout=30.0) as client:
while time.time() - start_time < self.duration:
req_start = time.time()
try:
if method.upper() == "GET":
resp = await client.get(
f"{base_url}{endpoint}", headers=headers
)
else:
resp = await client.request(
method, f"{base_url}{endpoint}",
json=body, headers=headers,
)
latency = (time.time() - req_start) * 1000
latencies.append(latency)
if resp.status_code >= 500:
errors += 1
except Exception:
errors += 1
workers = [asyncio.create_task(worker()) for _ in range(self.concurrency)]
await asyncio.gather(*workers)
total_time = time.time() - start_time
if not latencies:
return BenchmarkResult(
endpoint=endpoint, variant=variant, requests=0,
avg_latency_ms=0, p50_latency_ms=0, p95_latency_ms=0,
p99_latency_ms=0, min_latency_ms=0, max_latency_ms=0,
error_count=errors, throughput_rps=0,
)
sorted_latencies = sorted(latencies)
p50_idx = int(len(sorted_latencies) * 0.50)
p95_idx = int(len(sorted_latencies) * 0.95)
p99_idx = int(len(sorted_latencies) * 0.99)
return BenchmarkResult(
endpoint=endpoint,
variant=variant,
requests=len(latencies),
avg_latency_ms=round(statistics.mean(latencies), 2),
p50_latency_ms=round(sorted_latencies[p50_idx], 2),
p95_latency_ms=round(sorted_latencies[p95_idx], 2),
p99_latency_ms=round(sorted_latencies[min(p99_idx, len(sorted_latencies) - 1)], 2),
min_latency_ms=round(sorted_latencies[0], 2),
max_latency_ms=round(sorted_latencies[-1], 2),
error_count=errors,
throughput_rps=round(len(latencies) / total_time, 2),
)
Understanding the Latency Breakdown
The commerce layer adds latency at four points:
-
Caller classification (< 1ms): Header inspection and string matching. Negligible.
-
Payment verification (10-100ms): Network call to GreenHelix to verify the payment token. This is the dominant cost. With the verification cache enabled, repeat requests from the same agent with the same token skip this call entirely, reducing it to < 1ms.
-
Identity verification (5-50ms): Ed25519 signature verification is a local CPU operation (< 1ms), but resolving the agent identity against GreenHelix adds a network call. Cache agent identity lookups for 5 minutes to amortize this cost.
-
Response transformation (< 1ms): JSON wrapping is a memory operation. Even for large responses (1MB+), the transformation adds under 1ms.
The total expected overhead for a cache-miss agent request is 15-150ms, dominated by payment and identity verification network calls. With warm caches, the overhead drops to 1-3ms.
Optimization Strategies
Connection pooling. The httpx.AsyncClient used for GreenHelix communication should be a long-lived singleton with connection pooling enabled. Creating a new HTTP client per request adds 50-100ms of connection setup overhead:
# Bad: new client per request
async def verify_payment(token):
async with httpx.AsyncClient() as client: # Connection setup every time
return await client.post(...)
# Good: shared client with connection pool
_greenhelix_client = httpx.AsyncClient(
base_url="https://api.greenhelix.net",
timeout=10.0,
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
)
async def verify_payment(token):
return await _greenhelix_client.post(...) # Reuses connections
Batch verification. If your API serves requests in bursts (common with agent orchestration patterns where one agent dispatches multiple sub-tasks), batch multiple payment verifications into a single GreenHelix API call:
async def verify_payments_batch(tokens: list[dict]) -> list[dict]:
"""Verify multiple payment tokens in one API call."""
response = await _greenhelix_client.post(
"/v1/payments/verify-batch",
json={"tokens": tokens},
headers={"Authorization": f"Bearer {api_key}"},
)
return response.json()["results"]
Preemptive caching. Agents that frequently call the same endpoint can be given a bulk payment token that covers N future requests. The verification cache stores the token with a remaining-uses counter instead of a simple boolean:
@dataclass
class BulkTokenEntry:
token: str
remaining_uses: int
endpoint: str
verified_at: float
ttl_seconds: int = 3600
@property
def is_valid(self) -> bool:
return self.remaining_uses > 0 and time.time() - self.verified_at < self.ttl_seconds
def consume(self) -> bool:
if not self.is_valid:
return False
self.remaining_uses -= 1
return True
Async settlement. Payment settlement (releasing escrow funds to the seller) does not need to happen before the response is sent. Verify the token synchronously, send the response, then settle asynchronously. This removes settlement latency from the request path entirely:
async def handle_agent_request(request: Request):
# Verify payment (synchronous, blocks response)
payment = await verify_payment(request.headers["x-402-payment-token"])
# Process business logic
result = await business_logic(request)
# Settle payment (asynchronous, does not block response)
asyncio.create_task(settle_payment(payment["escrow_id"]))
return result
Throughput Considerations
Agent traffic patterns differ from human traffic. Humans send requests with natural pauses (reading, clicking, thinking). Agents send bursts of requests as fast as the API can respond. Your API might handle 100 requests per second from human users comfortably but struggle at 1000 requests per second from agents, even though the per-request processing time is the same. The bottleneck shifts from business logic to the commerce layer — specifically, the connection pool to GreenHelix for payment verification.
Size your connection pool based on expected agent throughput. A pool of 20 keepalive connections to GreenHelix can handle approximately 200 verification requests per second (assuming 100ms average verification time). For higher throughput, increase the pool or enable the verification cache aggressively.
Monitor these throughput metrics during migration:
- Requests per second by caller type (human vs agent)
- GreenHelix verification latency (p50, p95, p99)
- Connection pool utilization (active vs idle connections)
- Cache hit rate for payment verification
- Queue depth if you implement request queuing
What's Next
You now have a complete framework for migrating any REST API to support agent commerce. The LegacyApiAdapter wraps your existing endpoints without modifying business logic. The x402 middleware handles payment negotiation at the HTTP layer. The AuthBridge runs agent Ed25519 authentication alongside your existing auth system. The MigrationValidator automates regression testing throughout the migration. And the benchmarking framework quantifies the exact performance cost of each component.
The migration order should be:
- Run the assessment checklist on your API.
- Deploy the
LegacyApiAdapterin discovery mode (402 responses, no payment processing) on your highest-value read endpoints. - Monitor agent traffic patterns for one to two weeks.
- Enable payment processing on the endpoints where agents are actually sending requests.
- Roll out identity verification.
- Migrate the remaining endpoints based on observed demand.
- Run the
MigrationValidatorcontinuously in CI.
The GreenHelix platform documentation at https://docs.greenhelix.net covers escrow API details, A2A protocol card registration, and agent identity management. The x402 specification at https://www.x402.org defines the full payment header protocol.
Start with one endpoint. Prove it works. Then migrate the rest.
Related skills
- Downloads
- Security audit
- Pass
- Last updated
- 3mo ago
- Current version
- v1.3.1
- License
- MIT-0