Agent Economics & Marketplace Design
Every agent call costs money. A model inference, a tool invocation, a vector search, a code execution sandbox — each one has a price tag measured in tokens, compute seconds, or API fees. When agents were simple chatbots, the economics were straightforward: input tokens times price plus output tokens times price. But modern agents chain dozens of steps, call external tools, spawn sub-agents, and run for minutes or hours. The bill becomes part of the supply chain.
The Cost Structure of an Agent Call #
Before you can price anything, you need to understand what you are pricing. A single user request to an agent might trigger a cascade of billable events that looks nothing like a simple API call.
User request: "Find the three cheapest flights from JFK to LAX next Friday
and summarize the baggage policies for each."
Agent execution trace:
├─ LLM call #1: Parse intent, plan steps ~800 tokens
├─ Tool call: flight_search(JFK, LAX, date) 1 API call ($0.01)
├─ LLM call #2: Interpret results, select top 3 ~1,200 tokens
├─ Tool call: scrape_baggage_policy(airline_1) 1 API call ($0.005)
├─ Tool call: scrape_baggage_policy(airline_2) 1 API call ($0.005)
├─ Tool call: scrape_baggage_policy(airline_3) 1 API call ($0.005)
├─ LLM call #3: Summarize policies ~2,000 tokens
└─ LLM call #4: Format final response ~600 tokens
Total: 4 LLM calls (~4,600 tokens) + 4 tool calls (~$0.025)
The cost has four distinct dimensions:
Model inference. Token costs for each LLM call. These vary by model — a frontier reasoning model might cost 50x more per token than a small routing model. An agent that uses model selection and routing to pick cheap models for simple steps and expensive models for hard ones has a fundamentally different cost profile than one that routes everything through a single model.
Tool execution. External API calls, database queries, code execution, file operations. Some tools are free (local file reads), some are metered (third-party APIs), and some have unpredictable costs (web scraping that might hit rate limits and require retries).
Infrastructure overhead. Context storage, checkpoint persistence, vector database queries, message queue operations. These are per-request costs that scale with conversation length and agent complexity.
Latency as cost. Slower execution means longer compute reservations, more memory usage, and worse user experience. In real-time applications, latency has a dollar value even when the token cost is low.
The Compounding Problem #
Agent costs compound in ways that simple API calls do not. Each step in the agent loop carries forward the full conversation context, so later LLM calls process more input tokens than earlier ones. A ReAct loop that runs for ten iterations does not cost 10x the first iteration — it costs something closer to 10 + 9 + 8 + ... + 1 = 55x the marginal cost of a single reasoning step, because each iteration sees the accumulated context of all previous ones.
def estimate_react_loop_cost(
base_context_tokens: int,
tokens_per_step: int,
num_steps: int,
input_price_per_token: float,
output_price_per_token: float,
avg_output_tokens: int = 200,
) -> dict:
"""
Estimate the total token cost of a ReAct loop,
accounting for context accumulation.
"""
if num_steps <= 0:
raise ValueError("num_steps must be greater than zero")
if base_context_tokens <= 0:
raise ValueError("base_context_tokens must be greater than zero")
if tokens_per_step < 0 or avg_output_tokens < 0:
raise ValueError("token counts cannot be negative")
total_input_tokens = 0
total_output_tokens = 0
for step in range(num_steps):
step_input = base_context_tokens + (tokens_per_step * step)
total_input_tokens += step_input
total_output_tokens += avg_output_tokens
return {
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"input_cost": total_input_tokens * input_price_per_token,
"output_cost": total_output_tokens * output_price_per_token,
"total_cost": (
total_input_tokens * input_price_per_token
+ total_output_tokens * output_price_per_token
),
"naive_estimate": num_steps * (
base_context_tokens * input_price_per_token
+ avg_output_tokens * output_price_per_token
),
"compounding_multiplier": total_input_tokens / (
num_steps * base_context_tokens
),
}
For a 10-step loop with a 2,000-token base context and 500 tokens added per step, the compounding multiplier is roughly 2.25x — meaning you pay more than double what a naive per-step estimate would suggest. For 20-step loops, it approaches 3.5x. This is why cost optimization matters so much for agents — the economics punish long chains of reasoning harder than you might expect.
Tool Marketplaces #
When every agent needs tools, and building tools is expensive, a marketplace emerges naturally. Tool marketplaces are platforms where tool providers publish capabilities and agent builders consume them — the same pattern as app stores or package registries, but for agent-callable functions.
The Marketplace Architecture #
A tool marketplace has three participants: providers who publish tools, consumers (agents or agent builders) who discover and invoke them, and the platform that handles discovery, authentication, metering, and billing.
┌──────────────────────────────────────────────────────────────────┐
│ Tool Marketplace │
│ │
│ ┌───────────┐ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Tool │ │ Discovery & │ │ Agent │ │
│ │ Provider │───▶ │ Registry │◀─── │ Consumer │ │
│ │ │ │ │ │ │ │
│ │ - Schema │ │ - Search index │ │ - Tool selection │ │
│ │ - Runtime │ │ - Compatibility │ │ - Budget mgmt │ │
│ │ - SLA │ │ - Trust scores │ │ - Fallbacks │ │
│ └───────────┘ └──────────────────┘ └──────────────────┘ │
│ │ │ │ │
│ │ ┌─────────┴─────────┐ │ │
│ │ │ Platform Layer │ │ │
│ └────────▶│ │◀────────────┘ │
│ │ - Auth / API keys │ │
│ │ - Usage metering │ │
│ │ - Rate limiting │ │
│ │ - Billing │ │
│ │ - Audit logging │ │
│ └───────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
Tool Schemas as Contracts #
In a marketplace, the tool schema is a contract. The consumer relies on the schema to decide whether to call the tool, what arguments to pass, and how to interpret the result. If the schema is inaccurate or changes without notice, downstream agents break.
from dataclasses import dataclass, field
from enum import Enum
class PricingModel(Enum):
FREE = "free"
PER_CALL = "per_call"
PER_TOKEN = "per_token" # Input/output tokens processed
TIERED = "tiered" # Volume-based pricing
SUBSCRIPTION = "subscription"
@dataclass
class ToolListing:
"""A tool published to a marketplace."""
tool_id: str
provider_id: str
name: str
description: str
version: str
schema: dict # JSON Schema for input parameters
output_schema: dict # JSON Schema for output format
pricing: PricingModel
price_per_call: float = 0.0
avg_latency_ms: int = 0
reliability_score: float = 1.0 # Historical uptime
trust_tier: str = "unverified" # unverified, verified, certified
rate_limit: int = 100 # Max calls per minute
tags: list[str] = field(default_factory=list)
compatible_protocols: list[str] = field(default_factory=list) # e.g., ["mcp", "openapi"]
@dataclass
class ToolRegistry:
"""Central registry for marketplace tool discovery."""
listings: dict[str, ToolListing] = field(default_factory=dict)
def publish(self, listing: ToolListing) -> None:
"""Publish or update a tool listing."""
self.listings[listing.tool_id] = listing
def search(
self,
query: str = "",
max_price: float | None = None,
min_reliability: float = 0.0,
tags: list[str] | None = None,
protocol: str | None = None,
) -> list[ToolListing]:
"""Search for tools matching criteria."""
results = list(self.listings.values())
if query:
terms = query.casefold().split()
results = [
t for t in results
if all(
term in " ".join([t.name, t.description, *t.tags]).casefold()
for term in terms
)
]
if max_price is not None:
results = [t for t in results if t.price_per_call <= max_price]
if min_reliability > 0:
results = [t for t in results if t.reliability_score >= min_reliability]
if tags:
results = [t for t in results if set(tags) & set(t.tags)]
if protocol:
results = [t for t in results if protocol in t.compatible_protocols]
return results
Trust and Verification #
A marketplace needs a trust model. When an agent calls a tool from an unknown provider, it is running third-party code with access to whatever data the agent passes in. This is the tool supply-chain problem at marketplace scale.
Trust tiers create a layered system:
- Unverified tools are self-published with no review. The marketplace makes no guarantees. Agents should sandbox calls to these tools and validate all outputs.
- Verified tools have passed automated checks: the schema matches actual behavior, the tool responds within claimed latency bounds, and the provider has a verified identity.
- Certified tools have undergone manual review, security audit, and compliance checks. They carry an SLA with financial backing.
The trust tier directly affects pricing. Certified tools command a premium because consumers pay for reliability guarantees. An agent that needs to operate in a regulated environment might be restricted to certified tools only — making the certification a market-access gate.
Version Management and Breaking Changes #
Tool versioning in a marketplace is harder than library versioning because the consumer is an LLM. A human developer reads a changelog and updates their integration code. An agent's "integration" is the model's understanding of the tool schema, embedded in the prompt. Changing a tool's schema means every agent that uses it needs to re-learn the interface — or break silently.
Semantic versioning helps: major version bumps signal breaking changes, minor versions add capabilities, patches fix bugs without changing the interface. But the marketplace also needs schema compatibility checks that detect when a new version would break existing consumers.
def check_schema_compatibility(
old_schema: dict,
new_schema: dict,
) -> dict:
"""
Check if a new tool schema is backward-compatible
with the previous version.
"""
issues = []
old_required = set(old_schema.get("required", []))
new_required = set(new_schema.get("required", []))
old_props = set(old_schema.get("properties", {}).keys())
new_props = set(new_schema.get("properties", {}).keys())
# New required fields that did not exist before = breaking change
added_required = new_required - old_required
if added_required:
issues.append({
"severity": "breaking",
"issue": f"New required fields: {added_required}",
})
# Removed fields = breaking change
removed = old_props - new_props
if removed:
issues.append({
"severity": "breaking",
"issue": f"Removed fields: {removed}",
})
# Type changes in existing fields = breaking change
for prop in old_props & new_props:
old_type = old_schema["properties"][prop].get("type")
new_type = new_schema["properties"][prop].get("type")
if old_type != new_type:
issues.append({
"severity": "breaking",
"issue": f"Type change in '{prop}': {old_type} → {new_type}",
})
# New optional fields = backward compatible
added_optional = (new_props - old_props) - added_required
if added_optional:
issues.append({
"severity": "compatible",
"issue": f"New optional fields: {added_optional}",
})
breaking = [i for i in issues if i["severity"] == "breaking"]
return {
"compatible": len(breaking) == 0,
"issues": issues,
"recommended_version_bump": "major" if breaking else "minor",
}
Agent-as-a-Service Pricing #
When agents themselves become products — pricing gets interesting. An agent-as-a-service (AaaS) provider bundles a model, a set of tools, orchestration logic, and domain knowledge into a callable service. The customer does not care about tokens or tool calls. They care about outcomes: "analyze this contract," "generate this report," "resolve this support ticket."
Pricing Models #
There are four natural pricing models for agent services, each with different incentive structures.
Per-token passthrough. The simplest model: charge the customer what the underlying model and tools cost, plus a markup. This is transparent but exposes the customer to cost volatility. A task that takes 5 ReAct iterations costs 5x more than one that takes 1 iteration — even if both produce the same quality output.
Per-task flat rate. Charge a fixed price per task type. "Contract analysis: $2.00." "Code review: $0.50." This is predictable for the customer but risky for the provider. Some contract analyses take 3 LLM calls; others take 30. The provider absorbs variance, which means they need accurate cost models and the ability to cap runaway executions.
Outcome-based pricing. Charge based on the value of the result. A support agent that resolves a ticket saves a human agent 15 minutes, so charge $3 per resolution. A data analysis agent that produces an actionable insight charges more than one that produces a "no anomalies found" result. This aligns incentives beautifully but is hard to implement — you need a way to measure outcome quality, and you are exposed to disputes.
Subscription with usage caps. A monthly fee with a token or task budget. Overages are billed at a premium rate. This gives the customer predictability and the provider a revenue floor. The risk is the cap itself — if the cap is too generous, power users drain margin; if too tight, casual users feel restricted.
from dataclasses import dataclass
from enum import Enum
class AaaSPricingModel(Enum):
TOKEN_PASSTHROUGH = "token_passthrough"
PER_TASK = "per_task"
OUTCOME_BASED = "outcome_based"
SUBSCRIPTION = "subscription"
@dataclass
class TaskCostEstimate:
"""Estimate the cost of an agent task under different pricing models."""
task_type: str
estimated_tokens: int
estimated_tool_calls: int
estimated_steps: int
model_cost_per_token: float
avg_tool_cost: float
@property
def raw_cost(self) -> float:
"""The actual cost to the provider."""
return (
self.estimated_tokens * self.model_cost_per_token
+ self.estimated_tool_calls * self.avg_tool_cost
)
def price(self, model: AaaSPricingModel, **kwargs) -> float:
"""Calculate the customer-facing price."""
if model == AaaSPricingModel.TOKEN_PASSTHROUGH:
markup = kwargs.get("markup", 1.3) # 30% markup
return self.raw_cost * markup
if model == AaaSPricingModel.PER_TASK:
return kwargs["task_prices"].get(self.task_type, self.raw_cost * 2.0)
if model == AaaSPricingModel.OUTCOME_BASED:
base = kwargs.get("base_price", 0.50)
quality_multiplier = kwargs.get("quality_score", 1.0)
return base * quality_multiplier
if model == AaaSPricingModel.SUBSCRIPTION:
# Cost is amortized across the subscription
return 0.0 # Billed monthly, not per task
return self.raw_cost
The Margin Problem #
Agent-as-a-service providers face a unique margin challenge: the cost of serving a request is nondeterministic. Unlike a traditional API where the compute cost per request is roughly constant, an agent's cost depends on task complexity, model reasoning depth, tool failures and retries, and conversation history length.
A provider that charges $1.00 per task might make $0.80 profit on simple tasks and lose $2.00 on complex ones. The distribution of task complexity in the customer's workload determines whether the business is viable.
This is why cost controls are a business requirement. Every production agent service needs:
- A maximum step budget per task that hard-stops the agent loop
- A cost estimator that pre-screens tasks and rejects or re-prices ones that are likely to exceed the budget
- Model routing that uses cheap models for simple sub-tasks and expensive models only when necessary
- Caching that avoids redundant computation across similar requests
Token Budgets as a Coordination Mechanism #
Token budgets are the most underappreciated coordination mechanism in agent systems. On the surface, a token budget is just a cost control: "do not spend more than 10,000 tokens on this task." But in practice, it shapes how the agent thinks, what tools it uses, and when it stops.
Budget-Aware Planning #
An agent that knows its budget plans differently than one that does not. Given a fixed token budget, the agent must allocate resources across steps — spending more on critical reasoning and less on routine formatting. This is analogous to how a project manager allocates time across phases: you do not spend 80% of your budget on the first step if you know there are ten steps ahead.
@dataclass
class TokenBudget:
"""A budget that tracks and constrains token usage."""
total: int
used: int = 0
reserved: int = 0 # Tokens reserved for future steps
def __post_init__(self) -> None:
if self.total < 0 or self.used < 0 or self.reserved < 0:
raise ValueError("token counts cannot be negative")
if self.used + self.reserved > self.total:
raise ValueError("used and reserved tokens exceed the total budget")
@property
def remaining(self) -> int:
return self.total - self.used - self.reserved
@property
def utilization(self) -> float:
return self.used / self.total if self.total > 0 else 0.0
def can_afford(self, estimated_tokens: int) -> bool:
return 0 <= estimated_tokens <= self.remaining
def spend(self, tokens: int) -> None:
if tokens < 0:
raise ValueError("tokens cannot be negative")
if tokens > self.remaining:
raise BudgetExhaustedError(
f"Need {tokens} tokens but only {self.remaining} remaining"
)
self.used += tokens
def reserve(self, tokens: int, label: str = "") -> None:
"""Reserve tokens for a future step."""
if tokens < 0:
raise ValueError("tokens cannot be negative")
if tokens > self.remaining:
raise BudgetExhaustedError(
f"Cannot reserve {tokens} tokens, only {self.remaining} available"
)
self.reserved += tokens
def release_reservation(self, tokens: int) -> None:
"""Release previously reserved tokens."""
if tokens < 0:
raise ValueError("tokens cannot be negative")
if tokens > self.reserved:
raise ValueError("cannot release more tokens than are reserved")
self.reserved -= tokens
class BudgetExhaustedError(Exception):
pass
class BudgetAwarePlanner:
"""Plans agent steps with token budget constraints."""
def __init__(self, budget: TokenBudget):
self.budget = budget
def plan(self, task: str, available_tools: list[dict]) -> list[dict]:
"""
Generate an execution plan that fits within the token budget.
Reserves tokens for the final response synthesis.
"""
# Always reserve tokens for the final answer
synthesis_reserve = min(2000, self.budget.total // 4)
self.budget.reserve(synthesis_reserve, label="final_synthesis")
steps = []
remaining = self.budget.remaining
# Estimate cost per tool call (schema + invocation + result parsing)
tool_cost_estimate = 800 # tokens per tool round-trip
max_tool_calls = remaining // tool_cost_estimate
# Prioritize: plan only as many steps as the budget allows
if max_tool_calls >= 3:
steps = self._full_plan(task, available_tools, max_tool_calls)
elif max_tool_calls >= 1:
steps = self._minimal_plan(task, available_tools)
else:
steps = [{"action": "direct_answer", "note": "Budget too low for tool use"}]
return steps
def _full_plan(self, task, tools, max_calls) -> list[dict]:
"""Plan with enough budget for multiple tool calls."""
return [
{"action": "analyze", "estimated_tokens": 500},
{"action": "tool_calls", "max_calls": min(max_calls - 1, 5),
"estimated_tokens": 800 * min(max_calls - 1, 5)},
{"action": "synthesize", "estimated_tokens": 1500},
]
def _minimal_plan(self, task, tools) -> list[dict]:
"""Plan with budget for only one or two tool calls."""
return [
{"action": "tool_calls", "max_calls": 1, "estimated_tokens": 800},
{"action": "synthesize", "estimated_tokens": 1000},
]
Budgets in Multi-Agent Systems #
Token budgets become a true coordination mechanism in multi-agent systems. When a coordinator delegates work to sub-agents, it must decide how to allocate its budget across them. This is a resource allocation problem with real trade-offs.
┌─────────────────────────────────────────────────────────────┐
│ Coordinator Agent │
│ Total budget: 50,000 tokens │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Budget Allocation Strategy │ │
│ │ │ │
│ │ Coordinator overhead: 5,000 tokens (10%) │ │
│ │ Research sub-agent: 20,000 tokens (40%) │ │
│ │ Analysis sub-agent: 15,000 tokens (30%) │ │
│ │ Synthesis sub-agent: 8,000 tokens (16%) │ │
│ │ Reserve (retries/errors): 2,000 tokens (4%) │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Research │ │ Analysis │ │ Synthesis │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ │ │ │ │ │ │ │
│ │ Budget: │ │ Budget: │ │ Budget: │ │
│ │ 20K tokens │ │ 15K tokens│ │ 8K tokens │ │
│ │ │ │ │ │ │ │
│ │ Can sub- │ │ Fixed │ │ Must work │ │
│ │ allocate │ │ allocation│ │ within │ │
│ │ to tools │ │ │ │ budget │ │
│ └────────────┘ └───────────┘ └───────────┘ │
└─────────────────────────────────────────────────────────────┘
The coordinator makes an upfront allocation based on expected task complexity. But what happens when the research agent finishes early, using only 12,000 of its 20,000-token budget? The remaining 8,000 tokens can be reallocated — either to another sub-agent that needs more room or held in reserve for error recovery.
@dataclass
class BudgetAllocation:
"""Tracks budget allocation across sub-agents."""
agent_id: str
allocated: int
spent: int = 0
@property
def remaining(self) -> int:
return self.allocated - self.spent
@property
def surplus(self) -> int:
return max(0, self.remaining)
class MultiAgentBudgetManager:
"""Manages token budget allocation across a team of sub-agents."""
def __init__(self, total_budget: int, reserve_fraction: float = 0.05):
if total_budget < 0:
raise ValueError("total_budget cannot be negative")
if not 0 <= reserve_fraction <= 1:
raise ValueError("reserve_fraction must be between 0 and 1")
self.total_budget = total_budget
self.reserve = int(total_budget * reserve_fraction)
self.allocatable = total_budget - self.reserve
self.allocations: dict[str, BudgetAllocation] = {}
def allocate(self, agent_id: str, tokens: int) -> BudgetAllocation:
"""Allocate tokens to a sub-agent."""
if tokens < 0:
raise ValueError("tokens cannot be negative")
if agent_id in self.allocations:
raise ValueError(f"Agent {agent_id} already has an allocation")
total_allocated = sum(a.allocated for a in self.allocations.values())
if total_allocated + tokens > self.allocatable:
raise BudgetExhaustedError(
f"Cannot allocate {tokens}; only {self.allocatable - total_allocated} available"
)
allocation = BudgetAllocation(agent_id=agent_id, allocated=tokens)
self.allocations[agent_id] = allocation
return allocation
def report_spending(self, agent_id: str, tokens_spent: int) -> None:
"""Record tokens spent by a sub-agent."""
if tokens_spent < 0:
raise ValueError("tokens_spent cannot be negative")
allocation = self.allocations[agent_id]
if tokens_spent > allocation.remaining:
raise BudgetExhaustedError(
f"Agent {agent_id} has only {allocation.remaining} tokens remaining"
)
allocation.spent += tokens_spent
def reclaim_surplus(self) -> int:
"""Reclaim unspent tokens from completed sub-agents."""
reclaimed = 0
for alloc in self.allocations.values():
if alloc.surplus > 0:
reclaimed += alloc.surplus
alloc.allocated = alloc.spent # Shrink allocation to actual usage
self.reserve += reclaimed
return reclaimed
def reallocate(self, to_agent: str, tokens: int) -> None:
"""Move tokens from the reserve to a specific sub-agent."""
if tokens < 0:
raise ValueError("tokens cannot be negative")
if to_agent not in self.allocations:
raise ValueError(f"Unknown agent: {to_agent}")
if tokens > self.reserve:
raise BudgetExhaustedError(
f"Reserve has only {self.reserve} tokens, need {tokens}"
)
self.reserve -= tokens
self.allocations[to_agent].allocated += tokens
def summary(self) -> dict:
return {
"total_budget": self.total_budget,
"total_spent": sum(a.spent for a in self.allocations.values()),
"reserve": self.reserve,
"allocations": {
aid: {"allocated": a.allocated, "spent": a.spent, "remaining": a.remaining}
for aid, a in self.allocations.items()
},
}
Budget Pressure Shapes Behavior #
Here is the subtle part: token budgets shape agent behavior. An agent operating under a tight budget makes qualitatively different decisions than one with unlimited tokens.
Under tight budgets, agents:
- Skip exploratory tool calls and rely on what they already know
- Choose cheaper models for sub-tasks (if model routing is available)
- Produce shorter, more direct responses
- Abandon multi-step plans in favor of single-step approximations
- Fail to verify their own work because verification costs tokens
Under generous budgets, agents:
- Explore multiple solution paths before committing
- Use expensive models for nuanced reasoning
- Self-verify by re-reading their output and checking for errors
- Make redundant tool calls for confirmation
- Produce verbose, thorough responses
Neither extreme is ideal. Too tight and you get brittle, low-quality output. Too generous and you get expensive, meandering execution. The art is in finding the budget that produces the best quality-per-token ratio for each task type.
@dataclass
class BudgetPolicy:
"""Defines how an agent should behave under different budget pressures."""
budget_remaining_fraction: float # 0.0 to 1.0
@property
def model_tier(self) -> str:
"""Select model quality based on remaining budget."""
if self.budget_remaining_fraction > 0.6:
return "frontier"
if self.budget_remaining_fraction > 0.3:
return "mid_tier"
return "fast_cheap"
@property
def max_tool_calls(self) -> int:
"""Limit tool calls based on remaining budget."""
if self.budget_remaining_fraction > 0.5:
return 10
if self.budget_remaining_fraction > 0.2:
return 3
return 1
@property
def should_self_verify(self) -> bool:
"""Only self-verify if budget allows."""
return self.budget_remaining_fraction > 0.4
@property
def response_length_hint(self) -> str:
"""Suggest response length based on remaining budget."""
if self.budget_remaining_fraction > 0.5:
return "thorough"
if self.budget_remaining_fraction > 0.2:
return "concise"
return "minimal"
Metering and Billing Infrastructure #
Building a marketplace or agent service requires metering infrastructure that tracks every billable event in real time. Without accurate metering, you cannot bill customers, pay providers, or enforce budgets.
The Metering Pipeline #
Every agent action flows through a metering pipeline that records the event, calculates cost, and updates the running balance.
Agent Action
│
▼
┌───────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐
│ Event │───▶│ Cost │───▶│ Ledger │───▶│ Budget │
│ Capture │ │ Calculation │ │ Update │ │ Check │
│ │ │ │ │ │ │ │
│ - Action │ │ - Token │ │ - Append to │ │ - Within │
│ type │ │ count │ │ immutable │ │ limit? │
│ - Tokens │ │ - Model tier │ │ log │ │ - Alert? │
│ - Tool ID │ │ - Tool price │ │ - Update │ │ - Block? │
│ - Timing │ │ - Markup │ │ balance │ │ │
└───────────┘ └──────────────┘ └──────────────┘ └──────────┘
from dataclasses import dataclass, field
from datetime import datetime, timezone
@dataclass
class MeteringEvent:
"""A single billable event in an agent's execution."""
event_id: str
session_id: str
timestamp: datetime
event_type: str # "llm_call", "tool_call", "storage", "compute"
details: dict # Model name, tool ID, token counts, etc.
cost: float = 0.0
class MeteringLedger:
"""Append-only ledger for tracking agent costs."""
def __init__(self, session_id: str, budget_limit: float | None = None):
self.session_id = session_id
self.budget_limit = budget_limit
self.events: list[MeteringEvent] = []
self._total_cost: float = 0.0
def record(self, event: MeteringEvent) -> None:
"""Record a metering event and update the running total."""
if event.session_id != self.session_id:
raise ValueError("event belongs to a different session")
if event.cost < 0:
raise ValueError("event cost cannot be negative")
self.events.append(event)
self._total_cost += event.cost
@property
def total_cost(self) -> float:
return self._total_cost
@property
def budget_remaining(self) -> float | None:
if self.budget_limit is None:
return None
return self.budget_limit - self._total_cost
def is_over_budget(self) -> bool:
if self.budget_limit is None:
return False
return self._total_cost >= self.budget_limit
def cost_by_type(self) -> dict[str, float]:
"""Break down costs by event type."""
breakdown: dict[str, float] = {}
for event in self.events:
breakdown[event.event_type] = (
breakdown.get(event.event_type, 0.0) + event.cost
)
return breakdown
def to_invoice_lines(self) -> list[dict]:
"""Generate invoice line items from the ledger."""
lines = []
by_type = self.cost_by_type()
for event_type, cost in sorted(by_type.items()):
count = sum(1 for e in self.events if e.event_type == event_type)
lines.append({
"description": event_type,
"quantity": count,
"total": round(cost, 6),
})
return lines
Cost Attribution in Multi-Agent Systems #
When a coordinator agent delegates to sub-agents, and those sub-agents call tools from a marketplace, the cost chain has multiple links. The end customer should see a single bill, but internally the costs must be attributed correctly: the marketplace takes a cut, the tool provider gets paid, and the agent service provider takes their margin.
Customer pays $1.50 for a task
│
├── Agent service provider margin: $0.45 (30%)
│
├── Model inference cost: $0.60 (40%)
│ ├── Coordinator model calls: $0.10
│ ├── Research sub-agent model calls: $0.35
│ └── Synthesis sub-agent model calls: $0.15
│
├── Tool marketplace fees: $0.30 (20%)
│ ├── Flight search tool: $0.10
│ │ ├── Provider revenue: $0.08
│ │ └── Marketplace fee: $0.02
│ ├── Policy scraper tool: $0.15
│ │ ├── Provider revenue: $0.12
│ │ └── Marketplace fee: $0.03
│ └── Translation tool: $0.05
│ ├── Provider revenue: $0.04
│ └── Marketplace fee: $0.01
│
└── Infrastructure overhead: $0.15 (10%)
├── Vector DB queries: $0.05
├── Checkpoint storage: $0.03
└── Compute (orchestration): $0.07
Getting this attribution right requires propagating a cost context through the entire agent execution graph. Every LLM call, tool invocation, and infrastructure operation must tag itself with the originating session, the responsible agent, and the billing account.
Marketplace Dynamics and Incentives #
Marketplaces create their own economic dynamics. Understanding these dynamics is essential for anyone building or participating in a tool marketplace.
The Race to Commodity #
Tools that perform common functions — web search, text extraction, format conversion — quickly commoditize. Multiple providers offer the same capability, and price competition drives margins toward zero. The marketplace takes a percentage of each transaction, so commodity tools generate volume but not provider wealth.
Differentiation comes from three sources:
- Data advantage. A flight search tool that has exclusive access to airline inventory is not interchangeable with one that scrapes public websites. Unique data creates a moat.
- Quality and reliability. A sentiment analysis tool that achieves 95% accuracy commands a premium over one at 80%, even if the interface is identical. In agent systems, tool quality compounds — a 5% error rate per tool call becomes a 40% chance of at least one error in a 10-call chain.
- Speed. An agent choosing between two equivalent tools will prefer the one with lower latency, especially in interactive applications. Latency is a quality dimension that agents can measure directly.
Provider Incentives #
Tool providers in a marketplace face an interesting design tension. They want agents to call their tool frequently (revenue) but not so frequently that they hit rate limits or incur unsustainable compute costs. They need to balance discoverability (broad schema descriptions attract more agent calls) with precision (vague descriptions attract calls the tool cannot handle well, leading to poor reviews and lower trust scores).
A well-designed marketplace exposes a feedback loop to providers: how often their tool is called, how often agents use the result successfully (versus discarding it), average response latency, and error rates. This data lets providers iterate toward higher-quality tools that agents prefer.
Consumer Incentives #
Agent builders want cheap, reliable, fast tools. But they also want predictability — a tool that costs $0.01 per call today and $0.10 tomorrow breaks their cost model. Marketplace platforms can address this with price caps (providers commit to a maximum price for a period), price alerts (notify consumers of upcoming changes), and fallback routing (automatically switch to a backup tool if the primary exceeds a cost threshold).
@dataclass
class ToolSelectionPolicy:
"""Policy for selecting tools from a marketplace based on cost and quality."""
max_cost_per_call: float
min_reliability: float
max_latency_ms: int
preferred_trust_tier: str = "verified"
fallback_enabled: bool = True
def select(
self,
candidates: list[ToolListing],
) -> ToolListing | None:
"""Select the best tool from marketplace candidates."""
eligible = [
t for t in candidates
if t.price_per_call <= self.max_cost_per_call
and t.reliability_score >= self.min_reliability
and t.avg_latency_ms <= self.max_latency_ms
]
if not eligible:
if self.fallback_enabled:
# Relax constraints: drop latency requirement
eligible = [
t for t in candidates
if t.price_per_call <= self.max_cost_per_call
and t.reliability_score >= self.min_reliability
]
if not eligible:
return None
# Prefer the configured tier, then higher trust, lower cost, and latency
base_tier_rank = {"certified": 0, "verified": 1, "unverified": 2}
eligible.sort(
key=lambda t: (
t.trust_tier != self.preferred_trust_tier,
base_tier_rank.get(t.trust_tier, 3),
t.price_per_call,
t.avg_latency_ms,
)
)
return eligible[0]
Conclusion #
Agent economics are a design constraint that shapes architecture from the ground up.
Key takeaways:
- Agent costs have four dimensions: model inference, tool execution, infrastructure overhead, and latency. Costs compound across multi-step reasoning loops because each step carries forward an expanding context window.
- Tool marketplaces formalize the relationship between tool providers and agent consumers. They require robust schemas as contracts, trust tiers for safety, version compatibility checks, and metering infrastructure for billing.
- Agent-as-a-service pricing must account for nondeterministic execution costs. Per-task flat rates, outcome-based pricing, and subscriptions each shift risk differently between provider and consumer. Cost controls — step budgets, model routing, caching — are business requirements.
- Token budgets are a coordination mechanism. They shape how agents plan, which models they select, whether they self-verify, and when they stop. In multi-agent systems, budget allocation across sub-agents is a resource scheduling problem with real quality trade-offs.
- Marketplace dynamics follow familiar platform economics: commodity tools race to the bottom on price, while differentiation comes from data advantage, quality, and speed. Feedback loops between agent consumers and tool providers drive quality improvement over time.