Greenhelix Agent Procurement Playbook

The Agent Procurement Playbook

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

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

• GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)
• WALLET_ADDRESS: Blockchain wallet address for receiving payments (public address only — no private keys)
• STRIPE_API_KEY: Stripe API key for card payment processing (scoped to payment intents only)

Your AI agent just found the perfect data enrichment service. It costs $0.003 per record, the vendor has a 97% accuracy rating, and the API response time is under 200ms. The agent wants to buy 500,000 records. That is $1,500 committed in under a second, with no human in the loop, no purchase order, no procurement review. Should the agent be allowed to spend that money? If yes, under what constraints? If no, what happens to the time-sensitive workflow waiting on that data? This is the central tension of autonomous procurement: agents that can buy things are dramatically more capable than agents that cannot, but agents that spend without governance are a financial liability. A February 2026 survey by Gartner found that 86% of organizations plan to deploy autonomous AI agents at scale by end of 2026. McKinsey's March 2026 analysis of early adopters showed that autonomous procurement workflows -- where agents discover, evaluate, negotiate, and purchase services without human intervention -- deliver 15-30% efficiency gains over human-mediated purchasing. But 41% of those same organizations reported at least one incident of uncontrolled agent spending in their first quarter of deployment. The solution is not to prevent agents from spending. It is to build the procurement infrastructure that makes autonomous spending safe, auditable, and reversible. This playbook shows you how. Every chapter contains working Python code against the GreenHelix A2A Commerce Gateway -- 128 tools accessible at https://api.greenhelix.net/v1 via a single the REST API (POST /v1/{tool}) endpoint. By the end, you will have a complete autonomous procurement system: spending policy engine, multi-protocol vendor discovery, trust-scored vendor evaluation, escrow-protected purchasing, automated dispute resolution, cost reconciliation, and EU AI Act-compliant audit trails.

What You'll Learn

• Chapter 1: The Autonomous Buyer Architecture
• Chapter 2: Spending Controls & Approval Workflows
• Chapter 3: Vendor Discovery Across Protocols
• Chapter 4: Vendor Evaluation & Trust Scoring
• Chapter 5: Multi-Protocol Purchasing
• Chapter 6: Escrow, Disputes & Purchase Protection
• Chapter 7: Cost Reconciliation & FinOps Integration
• Chapter 8: Compliance & Audit Trails
• {report['organization']} - {si['system_name']}
• Appendix: Quick Reference

Full Guide

The Agent Procurement Playbook: Autonomous Purchasing, Spending Controls, Vendor Evaluation & Multi-Protocol Buying

Your AI agent just found the perfect data enrichment service. It costs $0.003 per record, the vendor has a 97% accuracy rating, and the API response time is under 200ms. The agent wants to buy 500,000 records. That is $1,500 committed in under a second, with no human in the loop, no purchase order, no procurement review. Should the agent be allowed to spend that money? If yes, under what constraints? If no, what happens to the time-sensitive workflow waiting on that data?

This is the central tension of autonomous procurement: agents that can buy things are dramatically more capable than agents that cannot, but agents that spend without governance are a financial liability. A February 2026 survey by Gartner found that 86% of organizations plan to deploy autonomous AI agents at scale by end of 2026. McKinsey's March 2026 analysis of early adopters showed that autonomous procurement workflows -- where agents discover, evaluate, negotiate, and purchase services without human intervention -- deliver 15-30% efficiency gains over human-mediated purchasing. But 41% of those same organizations reported at least one incident of uncontrolled agent spending in their first quarter of deployment.

The solution is not to prevent agents from spending. It is to build the procurement infrastructure that makes autonomous spending safe, auditable, and reversible. This playbook shows you how. Every chapter contains working Python code against the GreenHelix A2A Commerce Gateway -- 128 tools accessible at https://api.greenhelix.net/v1 via a single the REST API (POST /v1/{tool}) endpoint. By the end, you will have a complete autonomous procurement system: spending policy engine, multi-protocol vendor discovery, trust-scored vendor evaluation, escrow-protected purchasing, automated dispute resolution, cost reconciliation, and EU AI Act-compliant audit trails.

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.

Table of Contents

1. The Autonomous Buyer Architecture
2. Spending Controls & Approval Workflows
3. Vendor Discovery Across Protocols
4. Vendor Evaluation & Trust Scoring
5. Multi-Protocol Purchasing
6. Escrow, Disputes & Purchase Protection
7. Cost Reconciliation & FinOps Integration
8. Compliance & Audit Trails

Chapter 1: The Autonomous Buyer Architecture

Why Agents Need to Spend Money

A research agent that cannot purchase data is limited to free datasets. A coding agent that cannot buy API access is limited to open-source tools. A sales agent that cannot pay for lead enrichment is limited to whatever data it already has. Every boundary on an agent's purchasing authority is also a boundary on its capability.

The shift from agents-as-tools to agents-as-economic-actors is the defining infrastructure transition of 2026. When an agent can autonomously discover a service it needs, evaluate whether that service is trustworthy, negotiate a price, execute a purchase, verify delivery, and reconcile the cost against a budget -- it becomes a self-sufficient economic unit. It no longer needs a human to approve every API subscription, every data purchase, every compute reservation. It operates at machine speed across the entire procurement cycle.

But this autonomy comes with a prerequisite: procurement infrastructure. The same infrastructure that enterprises built over decades for human purchasing -- approval hierarchies, spending limits, preferred vendor lists, purchase orders, three-way matching, audit trails -- must be rebuilt for agents. Not as bureaucratic overhead, but as programmatic guardrails that enable speed while preventing catastrophe.

The Procurement Loop

Every autonomous purchase follows a six-stage loop. The stages are sequential for a single purchase, but a production procurement agent runs hundreds of these loops concurrently.

DISCOVER --> EVALUATE --> NEGOTIATE --> PURCHASE --> VERIFY --> RECONCILE
    ^                                                              |
    |______________________________________________________________|
                        (feedback loop)

Stage 1: Discover. The agent identifies a need (data, compute, API access, a service) and searches available vendors across multiple protocols. This is not a Google search -- it is structured queries against marketplace APIs, UCP manifests, ACP directories, and A2A service registries.

Stage 2: Evaluate. The agent scores each vendor on trust, reputation, price, capability, and compliance. A vendor with a low trust score or missing identity verification is filtered out regardless of price. This stage uses GreenHelix Trust, Identity, and Reputation tools.

Stage 3: Negotiate. For commodity purchases, negotiation is price comparison. For high-value or recurring purchases, the agent can request volume discounts, propose escrow terms, or negotiate SLA guarantees. See P14 (Negotiation Strategies) for advanced techniques.

Stage 4: Purchase. The agent executes the payment through the appropriate protocol: x402 for crypto micropayments, UCP for fiat retail, ACP for agent-native checkout, or direct A2A commerce via GreenHelix. Escrow protects high-value transactions.

Stage 5: Verify. The agent confirms delivery by inspecting the purchased asset -- validating data quality, checking API responses, running acceptance tests on delivered services. Failed verification triggers the dispute process.

Stage 6: Reconcile. The agent records the purchase in its ledger, attributes the cost to the requesting workflow, and updates budget consumption. Anomaly detection flags unexpected spending patterns.

Reference Architecture

The autonomous buyer architecture has four core components:

+---------------------------------------------------------------------+
|                    AUTONOMOUS BUYER AGENT                            |
|                                                                     |
|  +------------------+  +-------------------+  +------------------+  |
|  | Spending Policy  |  |  Vendor Discovery |  | Purchase Engine  |  |
|  | Engine           |  |  & Evaluation     |  |                  |  |
|  |                  |  |                   |  | - x402 rail      |  |
|  | - Budget caps    |  | - Multi-protocol  |  | - UCP rail       |  |
|  | - Tx limits      |  |   search          |  | - ACP rail       |  |
|  | - Whitelists     |  | - Trust scoring   |  | - A2A rail       |  |
|  | - Escalation     |  | - Bid comparison  |  | - Escrow mgmt    |  |
|  +--------+---------+  +--------+----------+  +--------+---------+  |
|           |                     |                      |            |
|  +--------v---------------------v----------------------v---------+  |
|  |                    APPROVAL GATEWAY                           |  |
|  |  Routes decisions through policy engine before execution      |  |
|  +-------------------------------+-------------------------------+  |
|                                  |                                  |
+----------------------------------+----------------------------------+
                                   |
                                   v
+-------------------------------+     +-------------------------------+
|  GREENHELIX A2A COMMERCE      |     |  RECONCILIATION ENGINE       |
|  GATEWAY                      |     |                               |
|                               |     |  - Ledger recording           |
|  128 tools via POST           |     |  - Cost attribution           |
|  the REST API                  |     |  - Budget vs. actual          |
|                               |     |  - Anomaly detection          |
+-------------------------------+     +-------------------------------+

The Spending Policy Engine is the gatekeeper. Every purchase request passes through it before execution. It checks per-transaction limits, daily budget caps, vendor whitelists, and escalation thresholds. If a purchase exceeds policy, it is either blocked, queued for human approval, or routed to a higher-authority agent.

The Approval Gateway sits between the policy engine and the purchase engine. It is the single choke point through which all spend flows. This architectural decision -- funneling all purchases through one gateway -- is critical for auditability and control. You cannot have agents bypassing the gateway by calling GreenHelix directly.

Bootstrapping the Architecture

Here is the minimum code to initialize the architecture against GreenHelix:

import requests
import os
from dataclasses import dataclass, field
from typing import Optional

API_BASE = "https://api.greenhelix.net/v1"
headers = {"Authorization": f"Bearer {os.environ['GREENHELIX_API_KEY']}"}


def execute_tool(tool: str, input_data: dict) -> dict:
    """Execute a GreenHelix tool via the unified endpoint."""
    response = requests.post(
        f"{API_BASE}/v1",
        json={"tool": tool, "input": input_data},
        headers=headers,
    )
    response.raise_for_status()
    return response.json()


@dataclass
class BuyerAgent:
    agent_id: str
    wallet_id: Optional[str] = None
    budget_daily: float = 0.0
    budget_monthly: float = 0.0

    def initialize(self):
        """Create wallet and set initial budget."""
        wallet = execute_tool("create_wallet", {
            "agent_id": self.agent_id,
            "currency": "USD",
        })
        self.wallet_id = wallet["wallet_id"]

        execute_tool("set_budget", {
            "agent_id": self.agent_id,
            "daily_limit": self.budget_daily,
            "monthly_limit": self.budget_monthly,
        })
        return self

    def get_spending_status(self) -> dict:
        """Check current balance and usage."""
        balance = execute_tool("get_balance", {
            "agent_id": self.agent_id,
        })
        usage = execute_tool("get_usage_analytics", {
            "agent_id": self.agent_id,
            "period": "current_day",
        })
        return {
            "balance": balance["balance"],
            "daily_spend": usage.get("total_spend", 0),
            "daily_remaining": self.budget_daily - usage.get("total_spend", 0),
        }


# Initialize a buyer agent with a $50/day, $1000/month budget
buyer = BuyerAgent(
    agent_id="procurement-agent-01",
    budget_daily=50.00,
    budget_monthly=1000.00,
).initialize()

print(buyer.get_spending_status())

This gives you an agent with a wallet, budget caps, and real-time spending visibility. The rest of this playbook builds on this foundation.

Key Takeaways

• Autonomous procurement follows a six-stage loop: discover, evaluate, negotiate, purchase, verify, reconcile.
• The Spending Policy Engine is the gatekeeper -- every purchase passes through it before execution.
• The Approval Gateway is the single choke point for all spend. No agent should bypass it.
• Start with a wallet, a budget, and spending visibility. Everything else builds on this foundation.
• Organizations deploying autonomous procurement see 15-30% efficiency gains, but 41% experience uncontrolled spending incidents without proper governance.

Chapter 2: Spending Controls & Approval Workflows

The Five Layers of Spending Control

Spending controls for autonomous agents are not a single dial. They are five independent layers, each catching a different failure mode. Deploying only one layer leaves you exposed to the failures the others catch.

Layer 1: Per-Transaction Caps. No single purchase can exceed a defined amount. This is the bluntest control -- it prevents a single bad decision from causing catastrophic damage. Set it at the maximum amount you would tolerate losing in a single transaction. For most agent workflows, $100-500 is appropriate for automated approval; anything above goes to escalation.

Layer 2: Daily/Weekly Budget Limits. Even if every individual transaction is small, an agent in a loop can drain a budget through volume. A $0.01 transaction repeated 100,000 times is $1,000. Daily and weekly limits create a hard ceiling on cumulative spend regardless of individual transaction sizes.

Layer 3: Recipient Whitelists. The agent can only purchase from pre-approved vendors. This prevents an agent from being socially engineered (via prompt injection or adversarial marketplace listings) into sending money to a malicious actor. Whitelists can be static (a fixed list) or dynamic (any vendor with a trust score above a threshold).

Layer 4: Frequency Limits. Rate-limiting purchases prevents rapid-fire micro-transactions that individually pass the per-transaction cap but collectively blow through the daily budget before alerting systems can react. A typical setting: no more than 10 purchases per minute, no more than 100 per hour.

Layer 5: Human-in-the-Loop Escalation. For purchases that exceed caps, involve new vendors, or fall outside known categories, the system queues the purchase for human approval. The key design decision is what happens while waiting: does the workflow block (synchronous escalation) or continue with a fallback (asynchronous escalation)?

The ProcurementPolicyEngine

This class implements all five layers. It is the core of the spending control system.

import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum


class ApprovalDecision(Enum):
    APPROVED = "approved"
    DENIED = "denied"
    ESCALATED = "escalated"


@dataclass
class PurchaseRequest:
    vendor_id: str
    amount: float
    currency: str = "USD"
    category: str = "general"
    description: str = ""
    requesting_agent: str = ""
    idempotency_key: str = ""


@dataclass
class PolicyResult:
    decision: ApprovalDecision
    reason: str
    purchase_request: Optional[PurchaseRequest] = None
    escalation_id: Optional[str] = None


@dataclass
class ProcurementPolicyEngine:
    """Five-layer spending control for autonomous procurement agents."""

    agent_id: str
    max_transaction_amount: float = 500.00
    daily_limit: float = 2000.00
    weekly_limit: float = 10000.00
    max_purchases_per_minute: int = 10
    max_purchases_per_hour: int = 100
    escalation_threshold: float = 500.00
    vendor_whitelist: list = field(default_factory=list)
    require_whitelist: bool = False

    # Internal tracking
    _purchase_timestamps: list = field(default_factory=list)
    _daily_spend: float = 0.0

    def evaluate(self, request: PurchaseRequest) -> PolicyResult:
        """Run a purchase request through all five control layers."""

        # Layer 1: Per-transaction cap
        if request.amount > self.max_transaction_amount:
            if request.amount > self.escalation_threshold:
                return PolicyResult(
                    decision=ApprovalDecision.ESCALATED,
                    reason=f"Amount ${request.amount:.2f} exceeds "
                           f"auto-approval cap ${self.max_transaction_amount:.2f}",
                    purchase_request=request,
                )
            return PolicyResult(
                decision=ApprovalDecision.DENIED,
                reason=f"Amount ${request.amount:.2f} exceeds "
                       f"per-transaction limit ${self.max_transaction_amount:.2f}",
            )

        # Layer 2: Daily budget limit
        projected_daily = self._daily_spend + request.amount
        if projected_daily > self.daily_limit:
            return PolicyResult(
                decision=ApprovalDecision.DENIED,
                reason=f"Purchase would bring daily spend to "
                       f"${projected_daily:.2f}, exceeding "
                       f"daily limit ${self.daily_limit:.2f}",
            )

        # Layer 3: Vendor whitelist
        if self.require_whitelist and request.vendor_id not in self.vendor_whitelist:
            return PolicyResult(
                decision=ApprovalDecision.ESCALATED,
                reason=f"Vendor {request.vendor_id} not in whitelist. "
                       f"Escalating for manual review.",
                purchase_request=request,
            )

        # Layer 4: Frequency limit
        now = time.time()
        recent_minute = [t for t in self._purchase_timestamps if now - t < 60]
        recent_hour = [t for t in self._purchase_timestamps if now - t < 3600]

        if len(recent_minute) >= self.max_purchases_per_minute:
            return PolicyResult(
                decision=ApprovalDecision.DENIED,
                reason=f"Rate limit: {len(recent_minute)} purchases in "
                       f"the last minute (max {self.max_purchases_per_minute})",
            )
        if len(recent_hour) >= self.max_purchases_per_hour:
            return PolicyResult(
                decision=ApprovalDecision.DENIED,
                reason=f"Rate limit: {len(recent_hour)} purchases in "
                       f"the last hour (max {self.max_purchases_per_hour})",
            )

        # Layer 5: Escalation threshold for borderline amounts
        if request.amount > self.escalation_threshold:
            return PolicyResult(
                decision=ApprovalDecision.ESCALATED,
                reason=f"Amount ${request.amount:.2f} exceeds escalation "
                       f"threshold ${self.escalation_threshold:.2f}",
                purchase_request=request,
            )

        # All layers passed
        self._purchase_timestamps.append(now)
        self._daily_spend += request.amount
        return PolicyResult(
            decision=ApprovalDecision.APPROVED,
            reason="All policy checks passed",
        )

    def sync_with_gateway(self):
        """Sync local state with GreenHelix budget and usage data."""
        usage = execute_tool("get_usage_analytics", {
            "agent_id": self.agent_id,
            "period": "current_day",
        })
        self._daily_spend = usage.get("total_spend", 0)

        balance = execute_tool("get_balance", {
            "agent_id": self.agent_id,
        })
        return {
            "daily_spend": self._daily_spend,
            "daily_remaining": self.daily_limit - self._daily_spend,
            "wallet_balance": balance["balance"],
        }


# Initialize the policy engine
policy = ProcurementPolicyEngine(
    agent_id="procurement-agent-01",
    max_transaction_amount=500.00,
    daily_limit=2000.00,
    weekly_limit=10000.00,
    escalation_threshold=250.00,
    vendor_whitelist=["vendor-data-enrichment-01", "vendor-compute-aws-02"],
    require_whitelist=True,
)

# Evaluate a purchase request
result = policy.evaluate(PurchaseRequest(
    vendor_id="vendor-data-enrichment-01",
    amount=45.00,
    category="data",
    description="500k contact records enrichment",
    requesting_agent="research-agent-03",
))

print(f"Decision: {result.decision.value}")
print(f"Reason: {result.reason}")

Escalation Workflow with Webhooks

When a purchase is escalated, the system notifies a human approver and holds the purchase in a pending state. Here is how to wire up the escalation:

def handle_escalation(policy_result: PolicyResult) -> dict:
    """Queue an escalated purchase for human approval."""
    # Register a webhook for approval notifications
    webhook = execute_tool("register_webhook", {
        "agent_id": policy.agent_id,
        "url": "https://your-app.example.com/procurement/approvals",
        "events": ["purchase.approval_required"],
    })

    # Record the pending purchase in the ledger for tracking
    pending = execute_tool("record_transaction", {
        "agent_id": policy.agent_id,
        "type": "purchase_pending",
        "amount": str(policy_result.purchase_request.amount),
        "currency": "USD",
        "counterparty": policy_result.purchase_request.vendor_id,
        "metadata": {
            "status": "awaiting_human_approval",
            "reason": policy_result.reason,
            "category": policy_result.purchase_request.category,
        },
    })

    return {
        "escalation_id": pending.get("transaction_id"),
        "status": "awaiting_approval",
        "webhook_id": webhook.get("webhook_id"),
    }

Budget Configuration Decision Matrix

Use this matrix to set appropriate limits based on your agent's role:

Key Takeaways

• Five independent control layers: per-transaction caps, daily/weekly budgets, vendor whitelists, frequency limits, and human escalation.
• The ProcurementPolicyEngine evaluates every purchase request before execution. No purchase bypasses the engine.
• Escalation is not failure -- it is a feature. Design your workflows to handle the delay introduced by human-in-the-loop review.
• Sync the policy engine with GreenHelix usage analytics to keep local state consistent with actual spend.
• Set limits based on agent role, not a single global policy. A trading agent and an infrastructure agent have fundamentally different spending patterns.

Chapter 3: Vendor Discovery Across Protocols

The Multi-Protocol Discovery Problem

Your procurement agent needs a data enrichment service. Where does it look? In 2026, vendors advertise their services across at least four distinct protocols, each with its own discovery mechanism:

• GreenHelix Marketplace -- Structured service listings with trust scores, ratings, and escrow history. Discovery via search_services and best_match tools.
• UCP (Universal Commerce Protocol) -- Google and Shopify's protocol for structured product and service catalogs. Agents query UCP manifests to find services and compare prices.
• ACP (Agentic Commerce Protocol) -- OpenAI's protocol for agent-to-merchant discovery. ACP directories list merchants with Stripe-backed checkout.
• A2A Direct -- Agent-to-agent service advertisements via standardized capability manifests. No marketplace intermediary.

A vendor might list on one protocol, two, or all four. A procurement agent that only searches one protocol misses vendors available on the others. The solution is a unified vendor index that aggregates discovery results across all protocols.

GreenHelix Marketplace Discovery

The GreenHelix Marketplace is the richest discovery source because it includes trust, reputation, and escrow data alongside service listings.

def discover_greenhelix_vendors(query: str, category: str = None,
                                 min_trust_score: float = 0.7,
                                 max_results: int = 20) -> list:
    """Search GreenHelix Marketplace for services matching a query."""
    search_params = {
        "query": query,
        "max_results": max_results,
    }
    if category:
        search_params["category"] = category

    results = execute_tool("search_services", search_params)

    # Filter by trust score
    filtered = []
    for service in results.get("services", []):
        trust = execute_tool("check_trust_score", {
            "agent_id": service["provider_id"],
        })
        if trust.get("trust_score", 0) >= min_trust_score:
            service["trust_score"] = trust["trust_score"]
            filtered.append(service)

    # Rank by best match
    if filtered:
        best = execute_tool("best_match", {
            "query": query,
            "candidates": [s["service_id"] for s in filtered],
        })
        # Reorder filtered list by best_match ranking
        ranked_ids = [m["service_id"] for m in best.get("matches", [])]
        filtered.sort(key=lambda s: (
            ranked_ids.index(s["service_id"])
            if s["service_id"] in ranked_ids else 999
        ))

    return filtered

UCP Manifest Discovery

UCP manifests are structured JSON documents that describe services in a standardized format. Agents query them by fetching the manifest URL and parsing the catalog.

import requests as http_client


def discover_ucp_vendors(manifest_urls: list, query: str) -> list:
    """Query UCP manifests for matching services."""
    vendors = []
    for url in manifest_urls:
        try:
            resp = http_client.get(url, timeout=10)
            resp.raise_for_status()
            manifest = resp.json()

            for service in manifest.get("services", []):
                # Simple keyword matching against service description
                name = service.get("name", "").lower()
                desc = service.get("description", "").lower()
                if query.lower() in name or query.lower() in desc:
                    vendors.append({
                        "source": "ucp",
                        "provider_id": manifest.get("provider_id"),
                        "service_id": service.get("id"),
                        "name": service.get("name"),
                        "price": service.get("price"),
                        "currency": service.get("currency", "USD"),
                        "manifest_url": url,
                    })
        except (http_client.RequestException, ValueError):
            continue  # Skip unreachable or malformed manifests

    return vendors


def discover_acp_vendors(directory_url: str, query: str) -> list:
    """Query an ACP agent directory for matching services."""
    try:
        resp = http_client.get(
            f"{directory_url}/search",
            params={"q": query, "type": "service"},
            timeout=10,
        )
        resp.raise_for_status()
        listings = resp.json().get("results", [])

        return [{
            "source": "acp",
            "provider_id": listing.get("agent_id"),
            "service_id": listing.get("service_id"),
            "name": listing.get("name"),
            "price": listing.get("price"),
            "currency": listing.get("currency", "USD"),
            "checkout_url": listing.get("checkout_url"),
        } for listing in listings]
    except (http_client.RequestException, ValueError):
        return []

The Unified Vendor Index

Combine results from all protocols into a single, ranked vendor index:

from dataclasses import dataclass
from typing import Optional


@dataclass
class VendorCandidate:
    source: str             # "greenhelix", "ucp", "acp", "a2a"
    provider_id: str
    service_id: str
    name: str
    price: float
    currency: str
    trust_score: Optional[float] = None  # Only available for GreenHelix
    manifest_url: Optional[str] = None
    checkout_url: Optional[str] = None


class UnifiedVendorIndex:
    """Aggregates vendor discovery across all protocols."""

    def __init__(self, ucp_manifests: list = None, acp_directory: str = None):
        self.ucp_manifests = ucp_manifests or []
        self.acp_directory = acp_directory

    def search(self, query: str, category: str = None,
               min_trust: float = 0.5) -> list[VendorCandidate]:
        """Search all protocols and return unified, ranked results."""
        candidates = []

        # 1. GreenHelix Marketplace (richest data)
        gh_results = discover_greenhelix_vendors(
            query, category=category, min_trust_score=min_trust,
        )
        for r in gh_results:
            candidates.append(VendorCandidate(
                source="greenhelix",
                provider_id=r["provider_id"],
                service_id=r["service_id"],
                name=r.get("name", ""),
                price=float(r.get("price", 0)),
                currency=r.get("currency", "USD"),
                trust_score=r.get("trust_score"),
            ))

        # 2. UCP manifests
        ucp_results = discover_ucp_vendors(self.ucp_manifests, query)
        for r in ucp_results:
            candidates.append(VendorCandidate(
                source="ucp",
                provider_id=r["provider_id"],
                service_id=r["service_id"],
                name=r["name"],
                price=float(r.get("price", 0)),
                currency=r.get("currency", "USD"),
                manifest_url=r.get("manifest_url"),
            ))

        # 3. ACP directory
        if self.acp_directory:
            acp_results = discover_acp_vendors(self.acp_directory, query)
            for r in acp_results:
                candidates.append(VendorCandidate(
                    source="acp",
                    provider_id=r["provider_id"],
                    service_id=r["service_id"],
                    name=r["name"],
                    price=float(r.get("price", 0)),
                    currency=r.get("currency", "USD"),
                    checkout_url=r.get("checkout_url"),
                ))

        # Sort: trust-scored vendors first, then by price
        candidates.sort(key=lambda c: (
            0 if c.trust_score is not None else 1,
            -(c.trust_score or 0),
            c.price,
        ))

        return candidates


# Usage
index = UnifiedVendorIndex(
    ucp_manifests=[
        "https://vendor-a.example.com/.well-known/ucp-manifest.json",
        "https://vendor-b.example.com/.well-known/ucp-manifest.json",
    ],
    acp_directory="https://acp-directory.example.com/v1",
)

vendors = index.search("data enrichment", category="data")
for v in vendors[:5]:
    trust_display = f" (trust: {v.trust_score:.2f})" if v.trust_score else ""
    print(f"  [{v.source}] {v.name} - ${v.price}{trust_display}")

Protocol Coverage Checklist

Before deploying your procurement agent, verify coverage across all relevant discovery channels:

• GreenHelix Marketplace search configured with appropriate trust threshold
• UCP manifest URLs for known vendor networks added to the index
• ACP directory endpoint configured (if purchasing from OpenAI ecosystem merchants)
• A2A direct discovery configured for peer agent services
• Fallback behavior defined when a protocol is unreachable
• Deduplication logic handles the same vendor appearing on multiple protocols

Key Takeaways

• Vendors advertise across four protocols. An agent that searches only one protocol misses opportunities on the others.
• GreenHelix Marketplace provides the richest discovery data because it includes trust scores, reputation, and escrow history alongside listings.
• Build a UnifiedVendorIndex that aggregates results across all protocols and ranks by trust score first, price second.
• Trust-scored vendors should always rank above unscored vendors, regardless of price. The cheapest vendor with no verifiable identity is the most expensive mistake.
• See P21 (Agent-Ready Commerce) for the seller's perspective on listing across multiple protocols.

Chapter 4: Vendor Evaluation & Trust Scoring

Why Price Alone Is Insufficient

The cheapest vendor in your search results is also the most likely to be fraudulent. This is Akerlof's "Market for Lemons" problem applied to agent commerce: without reliable quality signals, low-quality sellers undercut high-quality ones, and buyers cannot tell the difference until after they have paid. The antidote is a structured evaluation framework that scores vendors on dimensions beyond price.

The four evaluation dimensions are:

1. Identity verification -- Is this vendor a real, registered entity?
2. Trust score -- Does the platform's composite trust assessment meet your threshold?
3. Reputation history -- What do other buyers say about this vendor?
4. Bid comparison -- Given equivalent trust, which vendor offers the best value?

The Vendor Scoring System

from dataclasses import dataclass
from typing import Optional


@dataclass
class VendorScore:
    vendor_id: str
    identity_verified: bool = False
    identity_confidence: float = 0.0
    trust_score: float = 0.0
    reputation_score: float = 0.0
    trade_count: int = 0
    dispute_rate: float = 0.0
    price: float = 0.0
    composite_score: float = 0.0
    risk_level: str = "unknown"  # low, medium, high, critical
    disqualified: bool = False
    disqualification_reason: str = ""


class VendorEvaluator:
    """Automated vendor evaluation using GreenHelix Trust, Identity,
    and Reputation tools."""

    # Weights for composite score (must sum to 1.0)
    WEIGHT_TRUST = 0.30
    WEIGHT_REPUTATION = 0.25
    WEIGHT_IDENTITY = 0.20
    WEIGHT_TRADE_HISTORY = 0.15
    WEIGHT_PRICE = 0.10

    # Thresholds
    MIN_TRUST_SCORE = 0.6
    MIN_TRADE_COUNT = 5
    MAX_DISPUTE_RATE = 0.10  # 10%

    def evaluate(self, vendor_id: str, price: float,
                 price_range: tuple = (0, 1000)) -> VendorScore:
        """Run full evaluation pipeline on a single vendor."""
        score = VendorScore(vendor_id=vendor_id, price=price)

        # Step 1: Identity verification
        try:
            identity = execute_tool("verify_identity", {
                "agent_id": vendor_id,
            })
            score.identity_verified = identity.get("verified", False)
            score.identity_confidence = identity.get("confidence", 0.0)
        except Exception:
            score.identity_verified = False
            score.identity_confidence = 0.0

        if not score.identity_verified:
            score.disqualified = True
            score.disqualification_reason = "Identity verification failed"
            score.risk_level = "critical"
            return score

        # Step 2: Trust score
        try:
            trust = execute_tool("check_trust_score", {
                "agent_id": vendor_id,
            })
            score.trust_score = trust.get("trust_score", 0.0)
        except Exception:
            score.trust_score = 0.0

        if score.trust_score < self.MIN_TRUST_SCORE:
            score.disqualified = True
            score.disqualification_reason = (
                f"Trust score {score.trust_score:.2f} below "
                f"minimum {self.MIN_TRUST_SCORE}"
            )
            score.risk_level = "high"
            return score

        # Step 3: Reputation and trade history
        try:
            reputation = execute_tool("get_agent_reputation", {
                "agent_id": vendor_id,
            })
            score.reputation_score = reputation.get("reputation_score", 0.0)
            score.trade_count = reputation.get("total_trades", 0)
            score.dispute_rate = reputation.get("dispute_rate", 0.0)
        except Exception:
            score.reputation_score = 0.0

        if score.dispute_rate > self.MAX_DISPUTE_RATE:
            score.disqualified = True
            score.disqualification_reason = (
                f"Dispute rate {score.dispute_rate:.1%} exceeds "
                f"maximum {self.MAX_DISPUTE_RATE:.1%}"
            )
            score.risk_level = "high"
            return score

        # Step 4: Additional KYA check
        try:
            kya = execute_tool("verify_agent", {
                "agent_id": vendor_id,
            })
            if not kya.get("verified", False):
                score.risk_level = "medium"
        except Exception:
            pass

        # Step 5: Compute composite score
        price_min, price_max = price_range
        price_normalized = 1.0 - (
            (price - price_min) / (price_max - price_min)
            if price_max > price_min else 0.5
        )
        price_normalized = max(0, min(1, price_normalized))

 

…(truncated)