Greenhelix Agent Commerce Migration

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:

1. Agent sends GET /api/v1/data with no payment headers.
2. Server responds with 402 Payment Required, including price and payment URL in headers.
3. Agent creates a payment escrow via the payment URL.
4. Agent retries GET /api/v1/data with the X-402-Payment-Token header.
5. Server verifies the token, processes the request, and returns data.
6. 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 

…(truncated)