Building AI Agent Marketplaces: Platforms Where Agents Buy and Sell Services

The Next Evolution: Agents as Service Consumers and Providers

Today, AI agents interact with tools: APIs, databases, and functions that are passive resources waiting to be called. The next evolution is agents interacting with other agents: active entities that negotiate, collaborate, and transact. This is not science fiction. The protocol foundations are already laid with MCP (Model Context Protocol) and A2A (Agent-to-Agent), and the first agent marketplaces are emerging in early 2026.

An agent marketplace is a platform where agent capabilities are published, discovered, negotiated, and consumed, all without human intervention in the critical path. A procurement agent at Company A needs to verify a vendor's compliance certifications. Instead of calling a static API, it discovers a compliance verification agent published by a third-party auditor on the marketplace, negotiates the terms (cost, SLA, data handling), and initiates the verification, all through standardized protocols.

This post covers the architecture, protocols, and practical implementation patterns for building agent marketplaces.

The Agent Marketplace Architecture

An agent marketplace has five core components:

sequenceDiagram
 autonumber
 participant A as Agent A
 participant Reg as Service Registry
 participant Auth as Auth (mTLS)
 participant B as Agent B
 A->>Reg: Discover capability "schedule"
 Reg-->>A: Endpoint plus contract
 A->>Auth: Mutual TLS handshake
 Auth-->>A: Verified peer cert
 A->>B: Invoke task plus context
 B->>B: Run sub-agent loop
 B-->>A: Result plus citations
 A->>A: Verify against guardrails
 A->>A: Append to shared memory

Registry: Where agents publish their capabilities, terms of service, and pricing. Think of it as a DNS for agent services.

Discovery: How agents find other agents that can fulfill their needs. Semantic search over capability descriptions, filtered by constraints (price, latency, compliance requirements).

Negotiation: How agents agree on terms before transacting. This includes pricing, SLA parameters, data handling policies, and authentication requirements.

Execution: How agents invoke each other's capabilities. Standardized request/response protocols with streaming support.

Settlement: How transactions are recorded and payments are processed. Includes usage tracking, billing, and dispute resolution.

# Agent marketplace registry and discovery service

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import uuid

@dataclass
class AgentCapability:
 """A capability published to the marketplace."""
 capability_id: str
 agent_id: str
 name: str
 description: str
 category: str
 input_schema: dict # JSON Schema for expected input
 output_schema: dict # JSON Schema for guaranteed output
 pricing: dict # {"model": "per_call", "price_usd": 0.05}
 sla: dict # {"max_latency_ms": 5000, "uptime": 0.999}
 data_policy: dict # {"retention": "none", "encryption": "aes256"}
 authentication: str # "api_key" | "oauth2" | "mtls"
 mcp_endpoint: str # MCP server URL for tool invocation
 a2a_endpoint: str # A2A endpoint for agent-to-agent communication
 published_at: datetime = field(default_factory=datetime.utcnow)
 rating: float = 0.0
 total_invocations: int = 0

@dataclass
class DiscoveryQuery:
 """Query to find agents on the marketplace."""
 need_description: str # Semantic description of what is needed
 category: Optional[str] = None
 max_price_per_call: Optional[float] = None
 max_latency_ms: Optional[int] = None
 min_uptime: Optional[float] = None
 required_data_policy: Optional[dict] = None
 min_rating: float = 0.0

class AgentMarketplaceRegistry:
 def __init__(self, vector_store, metadata_store):
 self.vectors = vector_store
 self.metadata = metadata_store

 async def publish(self, capability: AgentCapability) -> str:
 """Publish a capability to the marketplace."""
 # Store metadata
 await self.metadata.upsert(
 capability.capability_id, capability.__dict__
 )
 # Index description for semantic search
 await self.vectors.upsert(
 id=capability.capability_id,
 text=f"{capability.name}: {capability.description}",
 metadata={
 "category": capability.category,
 "price": capability.pricing.get("price_usd", 0),
 "latency": capability.sla.get("max_latency_ms", 0),
 "rating": capability.rating,
 }
 )
 return capability.capability_id

 async def discover(
 self, query: DiscoveryQuery, limit: int = 10
 ) -> list[AgentCapability]:
 """Find capabilities matching a need description and constraints."""
 # Semantic search for relevant capabilities
 filters = {}
 if query.category:
 filters["category"] = query.category
 if query.max_price_per_call:
 filters["price"] = {"$lte": query.max_price_per_call}
 if query.max_latency_ms:
 filters["latency"] = {"$lte": query.max_latency_ms}
 if query.min_rating > 0:
 filters["rating"] = {"$gte": query.min_rating}

 results = await self.vectors.search(
 query=query.need_description,
 filters=filters,
 limit=limit)

 capabilities = []
 for result in results:
 cap_data = await self.metadata.get(result.id)
 if cap_data:
 cap = AgentCapability(**cap_data)
 # Apply data policy filter
 if query.required_data_policy:
 if not self._matches_data_policy(
 cap.data_policy, query.required_data_policy
 ):
 continue
 capabilities.append(cap)

 return capabilities

Protocol Foundations: MCP and A2A

Model Context Protocol (MCP) for Tool Serving

MCP standardizes how capabilities are exposed as tools. In a marketplace context, each agent publishes its capabilities as MCP tools that other agents can invoke.

// MCP server that exposes an agent's capabilities as tools

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
 {
 name: "compliance-verification-agent",
 version: "1.0.0",
 },
 {
 capabilities: {
 tools: {},
 },
 }
);

// Define tools that other agents can discover and invoke
server.setRequestHandler("tools/list", async () => ({
 tools: [
 {
 name: "verify_vendor_compliance",
 description:
 "Verify a vendor's compliance with specified regulatory frameworks " +
 "(ISO27001, HIPAA, GDPR). Returns a structured compliance " +
 "report with pass/fail status for each control.",
 inputSchema: {
 type: "object",
 properties: {
 vendor_name: { type: "string", description: "Legal entity name" },
 vendor_domain: { type: "string", description: "Primary domain" },
 frameworks: {
 type: "array",
 items: {
 type: "string",
 enum: ["", "ISO27001", "HIPAA", "GDPR"],
 },
 description: "Frameworks to verify against",
 },
 depth: {
 type: "string",
 enum: ["summary", "detailed", "full_audit"],
 description: "Verification depth (affects cost and latency)",
 },
 },
 required: ["vendor_name", "frameworks"],
 },
 },
 {
 name: "get_compliance_certificate",
 description:
 "Retrieve a vendor's compliance certificate if previously verified. " +
 "Returns a signed PDF certificate with verification details.",
 inputSchema: {
 type: "object",
 properties: {
 vendor_name: { type: "string" },
 framework: { type: "string" },
 verification_id: { type: "string" },
 },
 required: ["vendor_name", "framework", "verification_id"],
 },
 },
 ],
}));

server.setRequestHandler("tools/call", async (request) => {
 const { name, arguments: args } = request.params;

 switch (name) {
 case "verify_vendor_compliance": {
 const result = await performComplianceVerification(
 args.vendor_name,
 args.vendor_domain,
 args.frameworks,
 args.depth || "summary"
 );
 return {
 content: [
 { type: "text", text: JSON.stringify(result, null, 2) },
 ],
 };
 }
 case "get_compliance_certificate": {
 const cert = await retrieveCertificate(
 args.vendor_name,
 args.framework,
 args.verification_id
 );
 return {
 content: [{ type: "text", text: JSON.stringify(cert) }],
 };
 }
 default:
 throw new Error(`Unknown tool: ${name}`);
 }
});

const transport = new StdioServerTransport();
await server.connect(transport);

Agent-to-Agent (A2A) Protocol for Inter-Agent Communication

While MCP handles tool invocation, A2A handles higher-level agent communication: capability negotiation, task delegation, and status updates. A2A enables agents to have structured conversations about what they need and what they can provide.

# A2A negotiation protocol implementation

from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional

class NegotiationStatus(Enum):
 PROPOSED = "proposed"
 COUNTER_OFFERED = "counter_offered"
 ACCEPTED = "accepted"
 REJECTED = "rejected"
 EXPIRED = "expired"

@dataclass
class ServiceTerms:
 price_per_call: float
 max_latency_ms: int
 data_retention: str # "none", "24h", "30d"
 encryption: str
 sla_uptime: float
 rate_limit: int # requests per minute

@dataclass
class NegotiationMessage:
 from_agent: str
 to_agent: str
 negotiation_id: str
 status: NegotiationStatus
 proposed_terms: ServiceTerms
 counter_terms: Optional[ServiceTerms] = None
 reason: str = ""

class A2ANegotiator:
 """Handles term negotiation between agents."""

 def __init__(self, agent_id: str, policies: dict):
 self.agent_id = agent_id
 self.policies = policies # Acceptable ranges for each term

 async def evaluate_proposal(
 self, proposal: NegotiationMessage
 ) -> NegotiationMessage:
 terms = proposal.proposed_terms

 # Check each term against our policies
 violations = []
 counter_terms = ServiceTerms(
 price_per_call=terms.price_per_call,
 max_latency_ms=terms.max_latency_ms,
 data_retention=terms.data_retention,
 encryption=terms.encryption,
 sla_uptime=terms.sla_uptime,
 rate_limit=terms.rate_limit)

 if terms.price_per_call > self.policies["max_price_per_call"]:
 violations.append("price_too_high")
 counter_terms.price_per_call = self.policies["max_price_per_call"]

 if terms.data_retention != "none" and self.policies.get("require_no_retention"):
 violations.append("data_retention_required_none")
 counter_terms.data_retention = "none"

 if terms.sla_uptime < self.policies.get("min_uptime", 0.99):
 violations.append("uptime_too_low")
 counter_terms.sla_uptime = self.policies["min_uptime"]

 if not violations:
 return NegotiationMessage(
 from_agent=self.agent_id,
 to_agent=proposal.from_agent,
 negotiation_id=proposal.negotiation_id,
 status=NegotiationStatus.ACCEPTED,
 proposed_terms=terms)

 return NegotiationMessage(
 from_agent=self.agent_id,
 to_agent=proposal.from_agent,
 negotiation_id=proposal.negotiation_id,
 status=NegotiationStatus.COUNTER_OFFERED,
 proposed_terms=terms,
 counter_terms=counter_terms,
 reason=f"Terms violated policies: {', '.join(violations)}")

Trust and Identity in Agent Marketplaces

When agents transact autonomously, trust becomes a critical infrastructure concern. How does a procurement agent know that a compliance verification agent is legitimate? How does the marketplace prevent a rogue agent from publishing false capabilities?

The emerging solution uses verifiable agent identities:

• Agent identity certificates: Each agent has a cryptographic identity tied to its publishing organization. The marketplace verifies the organization's identity before allowing capability publication.
• Capability attestation: Published capabilities include test results from the marketplace's evaluation suite. An agent claiming to verify compliance must pass the marketplace's verification test battery.
• Reputation scoring: Every transaction is rated by both parties. Reputation scores decay over time, incentivizing consistent quality.
• Escrow and dispute resolution: Payment for agent services is held in escrow until the consuming agent confirms the output meets the agreed-upon schema and quality threshold.

Building a Minimal Agent Marketplace

Here is a practical architecture for a minimal viable agent marketplace:

# Minimal agent marketplace implementation

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid

app = FastAPI(title="Agent Marketplace")

# In-memory stores (use PostgreSQL + pgvector in production)
capabilities_store: dict[str, dict] = {}
transactions_store: dict[str, dict] = {}

class PublishRequest(BaseModel):
 agent_id: str
 name: str
 description: str
 category: str
 input_schema: dict
 output_schema: dict
 price_per_call_usd: float
 max_latency_ms: int
 mcp_endpoint: str

class InvokeRequest(BaseModel):
 caller_agent_id: str
 capability_id: str
 input_data: dict
 max_price_usd: float

@app.post("/capabilities/publish")
async def publish_capability(req: PublishRequest):
 cap_id = str(uuid.uuid4())
 capabilities_store[cap_id] = {
 "capability_id": cap_id,
 **req.dict(),
 "rating": 0.0,
 "invocations": 0,
 }
 return {"capability_id": cap_id, "status": "published"}

@app.get("/capabilities/search")
async def search_capabilities(
 query: str,
 category: Optional[str] = None,
 max_price: Optional[float] = None,
 limit: int = 10):
 results = []
 for cap in capabilities_store.values():
 # Simple keyword matching (use vector search in production)
 if query.lower() in cap["description"].lower():
 if category and cap["category"] != category:
 continue
 if max_price and cap["price_per_call_usd"] > max_price:
 continue
 results.append(cap)
 return {"results": results[:limit]}

@app.post("/capabilities/invoke")
async def invoke_capability(req: InvokeRequest):
 cap = capabilities_store.get(req.capability_id)
 if not cap:
 raise HTTPException(404, "Capability not found")

 if cap["price_per_call_usd"] > req.max_price_usd:
 raise HTTPException(
 402,
 f"Price {cap['price_per_call_usd']} exceeds budget {req.max_price_usd}"
 )

 # Create transaction record
 tx_id = str(uuid.uuid4())
 transactions_store[tx_id] = {
 "transaction_id": tx_id,
 "caller": req.caller_agent_id,
 "provider": cap["agent_id"],
 "capability_id": req.capability_id,
 "price": cap["price_per_call_usd"],
 "status": "pending",
 }

 # Forward to the capability's MCP endpoint
 # (In production, use the MCP client SDK)
 result = await forward_to_mcp(
 cap["mcp_endpoint"], cap["name"], req.input_data
 )

 transactions_store[tx_id]["status"] = "completed"
 cap["invocations"] += 1

 return {
 "transaction_id": tx_id,
 "result": result,
 "cost_usd": cap["price_per_call_usd"],
 }

Challenges and Open Questions

Liability: When an agent marketplace transaction goes wrong (bad compliance verification leads to a breach), who is liable? The marketplace operator, the publishing agent's organization, or the consuming agent's organization? Current legal frameworks do not have clear answers.

Quality assurance: How do you test an agent capability that involves subjective judgment? Compliance verification has clear pass/fail criteria, but tasks like "summarize this contract" have quality that is harder to measure automatically.

Pricing dynamics: Should marketplace pricing be fixed, auction-based, or negotiated? Fixed pricing is simpler but may not reflect varying task complexity. Auction-based pricing introduces latency from the bidding process.

Anti-competitive behavior: Can a dominant agent publisher use marketplace data to identify and clone competitors' capabilities? Marketplace terms of service need to address this, but enforcement is challenging.

FAQ

How is an agent marketplace different from an API marketplace?

An API marketplace (like RapidAPI) lists static endpoints with fixed request/response schemas. An agent marketplace lists dynamic capabilities with negotiable terms, semantic discovery, and conversational interaction. The key difference is intelligence: agents on the marketplace can adapt their behavior based on the requester's needs, negotiate terms, and handle ambiguous requests. APIs are passive; marketplace agents are active participants in the transaction.

What prevents an agent from over-spending on marketplace services?

Agent budgets and spending limits are enforced at the organizational level. Each agent has a budget allocation with per-transaction limits, daily limits, and approval thresholds. Transactions exceeding thresholds require human approval or are routed to a supervisory agent. The marketplace also supports spending alerts and automatic pausing when budgets are exhausted.

Is the agent marketplace concept ready for production use?

In March 2026, agent marketplaces are in early production for well-defined, high-value use cases: compliance verification, data enrichment, document processing, and translation services. The protocol foundations (MCP, A2A) are solid. The remaining challenges are trust infrastructure, liability frameworks, and quality assurance at scale. Most organizations are piloting marketplace integrations for 2-3 specific capabilities rather than adopting it as a general-purpose procurement mechanism.

How do agent marketplaces handle data privacy across organizational boundaries?

Data handling is a first-class concern in the negotiation protocol. Before any transaction, agents agree on data retention (none, 24 hours, 30 days), encryption requirements (in transit and at rest), and jurisdiction constraints (data must stay in EU, for example). The marketplace enforces these agreements through technical controls: encrypted channels, audit logging, and data deletion verification. Organizations that need the highest assurance can require mutual TLS authentication and data processing agreements as part of the marketplace onboarding.