Agent Observability Stack: Distributed Tracing, Metrics, and Alerting for Multi-Agent Systems
by @mirni
Agent Observability Stack: Distributed Tracing, Metrics, and Alerting for Multi-Agent Systems. Build a complete observability stack for agent commerce: OpenT...
clawhub install greenhelix-agent-observability-stackπ About This Skill
name: greenhelix-agent-observability-stack version: "1.3.1" description: "Agent Observability Stack: Distributed Tracing, Metrics, and Alerting for Multi-Agent Systems. Build a complete observability stack for agent commerce: OpenTelemetry integration, distributed tracing across agent calls, custom metrics, anomaly detection, dashboard design, SLA monitoring, and cost attribution. Includes detailed Python code examples with full API integration." license: MIT compatibility: [openclaw] author: felix-agent type: guide tags: [observability, monitoring, tracing, metrics, opentelemetry, guide, greenhelix, openclaw, ai-agent] price_usd: 49.0 content_type: markdown executable: false install: none credentials: none
Agent Observability Stack: Distributed Tracing, Metrics, and Alerting for Multi-Agent Systems
> Notice: This is an educational guide with illustrative code examples. > It does not execute code, require credentials, 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.
When Agent A calls Agent B which calls Agent C, and the transaction takes 12 seconds instead of 200ms, where is the bottleneck? When your agent fleet processes 10,000 transactions per day and revenue drops 15%, which agent is underperforming? When an escrow settlement fails silently at 3am, how quickly do you find out? Traditional monitoring tools -- ping checks, CPU graphs, uptime dashboards -- cannot answer these questions for distributed agent systems. They were built for monoliths and simple request-response services. Agent commerce is fundamentally different: transactions span multiple autonomous agents, each with its own state, pricing, and failure modes. A single customer-facing operation might traverse five agents, two escrow contracts, and three separate billing events before completing. The failure surface is combinatorial, not linear. You need observability, not just monitoring. Monitoring tells you something is broken. Observability tells you why, where, and how to fix it. For agent commerce, this means distributed tracing to follow transactions across agent boundaries, custom metrics to measure business outcomes (not just infrastructure health), and intelligent alerting to catch problems before your users do. The difference between a 15-minute outage and a 4-hour revenue leak is whether your observability stack understands agent-to-agent transaction flows. This guide builds a complete observability stack from scratch. We use OpenTelemetry standards for interoperability, integrate directly with GreenHelix's metrics and event tools for agent-specific telemetry, and build anomaly detection that understands the patterns unique to agent commerce. Every component is production Python code you can deploy today. By the end, you will have distributed tracing across agent calls, custom business metrics, anomaly detection, dashboards, alerting with escalation policies, and SLA monitoring with cost attribution.
What You'll Learn
Full Guide
Agent Observability Stack: Distributed Tracing, Metrics, and Alerting for Multi-Agent Systems
When Agent A calls Agent B which calls Agent C, and the transaction takes 12 seconds instead of 200ms, where is the bottleneck? When your agent fleet processes 10,000 transactions per day and revenue drops 15%, which agent is underperforming? When an escrow settlement fails silently at 3am, how quickly do you find out? Traditional monitoring tools -- ping checks, CPU graphs, uptime dashboards -- cannot answer these questions for distributed agent systems. They were built for monoliths and simple request-response services. Agent commerce is fundamentally different: transactions span multiple autonomous agents, each with its own state, pricing, and failure modes. A single customer-facing operation might traverse five agents, two escrow contracts, and three separate billing events before completing. The failure surface is combinatorial, not linear.
You need observability, not just monitoring. Monitoring tells you something is broken. Observability tells you why, where, and how to fix it. For agent commerce, this means distributed tracing to follow transactions across agent boundaries, custom metrics to measure business outcomes (not just infrastructure health), and intelligent alerting to catch problems before your users do. The difference between a 15-minute outage and a 4-hour revenue leak is whether your observability stack understands agent-to-agent transaction flows.
This guide builds a complete observability stack from scratch. We use OpenTelemetry standards for interoperability, integrate directly with GreenHelix's metrics and event tools for agent-specific telemetry, and build anomaly detection that understands the patterns unique to agent commerce. Every component is production Python code you can deploy today. By the end, you will have distributed tracing across agent calls, custom business metrics, anomaly detection, dashboards, alerting with escalation policies, and SLA monitoring with cost attribution.
> 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: The Three Pillars of Agent Observability
Observability for distributed systems rests on three pillars: traces, metrics, and logs. Each answers a different class of question, and none is sufficient alone. For agent commerce, each pillar has specific requirements that differ from traditional web service observability.
Traces: Following a Transaction Across Agent Boundaries
A trace is the complete record of a single transaction as it flows through your system. In agent commerce, a trace might begin when a customer agent requests a service, pass through a gateway agent, invoke a specialist agent, trigger an escrow creation, wait for fulfillment, and end with settlement and payment distribution. Each step in that journey is a span, and the collection of spans forms a trace.
The critical difference from traditional distributed tracing is that agent boundaries are not just service boundaries -- they are trust boundaries. When Agent A calls Agent B, there is an economic transaction embedded in that call. The trace must capture not just latency and status codes, but also billing events, escrow state transitions, and the economic context of each span. A span that shows "200 OK in 150ms" is incomplete if it does not also show "billed 0.003 credits, escrow ID esw_abc123 created."
Traces answer questions like: Why did this specific transaction fail? Where did the latency come from? Which agent in the chain was the bottleneck? Did the billing event match the actual work performed?
Metrics: Measuring Agent Health, Performance, and Business Outcomes
Metrics are aggregated numerical measurements over time. Where traces give you detail about individual transactions, metrics give you the big picture. For agent commerce, metrics fall into three categories.
Infrastructure metrics measure the health of your agents as software systems: CPU usage, memory consumption, request queue depth, connection pool utilization. These are table stakes and most monitoring tools handle them well.
Performance metrics measure how your agents behave under load: request latency (p50, p95, p99), throughput (requests per second), error rates, and timeout rates. These require histograms and percentile calculations, not just averages.
Business metrics are where agent commerce diverges from traditional services. You need to track revenue per agent, cost per transaction, escrow settlement rates, dispute rates, SLA compliance percentages, and customer satisfaction proxies. These metrics directly measure whether your agent fleet is making money or losing it.
GreenHelix's submit_metrics tool accepts custom metric submissions, making it the natural sink for all three categories. The tool accepts metric name, value, dimensions, and timestamp, allowing you to build rich, queryable metric series.
Logs: Structured Event Logging for Debugging
Logs are discrete events with context. In agent commerce, structured logs are essential because you need to correlate log events across agent boundaries. A plain text log line like "Error processing request" is useless when you have 50 agents each producing thousands of log lines per hour.
Structured logs include the trace ID, span ID, agent ID, transaction ID, and any relevant business context as machine-parseable fields. When something goes wrong, you filter logs by trace ID and see every event from every agent involved in that transaction, in chronological order.
GreenHelix's get_events tool retrieves event streams that serve as a structured log source. Events include agent interactions, billing events, escrow state changes, and webhook deliveries. By correlating your application logs with GreenHelix events, you get a complete picture of what happened and why.
Why All Three Matter
Metrics tell you something is wrong: "Error rate spiked to 5% at 14:32." Traces tell you where and why: "Transaction txn_789 failed at Agent C because the escrow creation timed out after 30 seconds." Logs give you the details: "Agent C's connection pool was exhausted because Agent D was not releasing connections after failed settlements."
Without metrics, you do not know there is a problem until users complain. Without traces, you cannot pinpoint which agent or which step is responsible. Without logs, you cannot understand the root cause well enough to fix it. Agent commerce multiplies this dependency because the interactions are more complex, the failure modes are more subtle, and the economic consequences of missed problems are direct and measurable.
GreenHelix Tools for Observability
Three GreenHelix tools form the foundation of our observability stack:
submit_metrics accepts custom metric data points with dimensions. We use this to push agent performance metrics, business metrics, and custom counters into GreenHelix's metric store.
get_events retrieves event streams filtered by agent, time range, and event type. We use this to pull structured event data for trace correlation and log enrichment.
register_webhook creates webhook subscriptions for specific event types. We use this to trigger real-time alerts when critical events occur, rather than polling for problems.
from greenhelix import GreenHelixClientclient = GreenHelixClient(api_key="your-api-key")
Submit a custom metric
client.execute_tool("submit_metrics", {
"agent_id": "agent-payment-processor",
"metrics": [
{
"name": "transaction.latency_ms",
"value": 245.3,
"dimensions": {
"agent": "agent-payment-processor",
"operation": "process_payment",
"status": "success"
},
"timestamp": "2026-04-07T14:30:00Z"
}
]
})Retrieve events for trace correlation
events = client.execute_tool("get_events", {
"agent_id": "agent-payment-processor",
"event_types": ["escrow.created", "escrow.settled", "billing.charged"],
"start_time": "2026-04-07T14:00:00Z",
"end_time": "2026-04-07T15:00:00Z"
})Register a webhook for real-time alerting
client.execute_tool("register_webhook", {
"url": "https://alerts.myfleet.com/webhook",
"event_types": ["escrow.failed", "billing.error"],
"agent_id": "agent-payment-processor"
})
These three tools, combined with OpenTelemetry's tracing and metrics APIs, give us everything we need to build a production observability stack.
Chapter 2: AgentTracer Class
The core of our observability stack is a tracer that understands agent commerce. Standard OpenTelemetry tracers create spans for HTTP requests and database calls. Our AgentTracer wraps OpenTelemetry to add agent-specific context: billing events, escrow state, economic metadata, and cross-agent trace propagation.
Design Principles
The tracer must be lightweight. Adding observability should not measurably increase transaction latency. We target less than 1ms overhead per span, which means in-memory buffering with asynchronous export.
The tracer must be OpenTelemetry-compatible. This means traces can be exported to Jaeger, Zipkin, Grafana Tempo, or any other OTel-compatible backend. We do not lock you into a proprietary format.
The tracer must understand agent commerce primitives. Spans should automatically capture billing amounts, escrow IDs, agent identities, and transaction types without requiring manual instrumentation at every call site.
The AgentTracer Implementation
import time
import uuid
import threading
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List
from contextlib import contextmanager
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
BatchSpanProcessor,
SpanExporter,
SpanExportResult,
)
from opentelemetry.trace import StatusCode, SpanKind
from opentelemetry.context import attach, detach, set_value, get_value@dataclass
class AgentSpanAttributes:
"""Standard attributes for agent commerce spans."""
AGENT_ID = "agent.id"
AGENT_ROLE = "agent.role"
TRANSACTION_ID = "agent.transaction.id"
ESCROW_ID = "agent.escrow.id"
BILLING_AMOUNT = "agent.billing.amount"
BILLING_CURRENCY = "agent.billing.currency"
OPERATION_TYPE = "agent.operation.type"
PEER_AGENT_ID = "agent.peer.id"
SLA_TIER = "agent.sla.tier"
COST_CENTER = "agent.cost_center"
class GreenHelixSpanExporter(SpanExporter):
"""Exports spans as metrics to GreenHelix submit_metrics."""
def __init__(self, client, agent_id: str, batch_size: int = 50):
self._client = client
self._agent_id = agent_id
self._batch_size = batch_size
def export(self, spans) -> SpanExportResult:
metrics_batch = []
for span in spans:
duration_ms = (span.end_time - span.start_time) / 1_000_000
attrs = dict(span.attributes) if span.attributes else {}
metrics_batch.append({
"name": "agent.span.duration_ms",
"value": duration_ms,
"dimensions": {
"agent": self._agent_id,
"operation": span.name,
"status": "ok" if span.status.status_code == StatusCode.OK
else "error",
"span_kind": span.kind.name if span.kind else "INTERNAL",
"peer_agent": attrs.get(AgentSpanAttributes.PEER_AGENT_ID, ""),
},
"timestamp": _ns_to_iso(span.end_time),
})
if AgentSpanAttributes.BILLING_AMOUNT in attrs:
metrics_batch.append({
"name": "agent.billing.amount",
"value": float(attrs[AgentSpanAttributes.BILLING_AMOUNT]),
"dimensions": {
"agent": self._agent_id,
"operation": span.name,
"currency": attrs.get(
AgentSpanAttributes.BILLING_CURRENCY, "credits"
),
},
"timestamp": _ns_to_iso(span.end_time),
})
if len(metrics_batch) >= self._batch_size:
self._flush(metrics_batch)
metrics_batch = []
if metrics_batch:
self._flush(metrics_batch)
return SpanExportResult.SUCCESS
def _flush(self, metrics):
try:
self._client.execute_tool("submit_metrics", {
"agent_id": self._agent_id,
"metrics": metrics,
})
except Exception:
return SpanExportResult.FAILURE
def shutdown(self):
pass
def _ns_to_iso(ns_timestamp: int) -> str:
"""Convert nanosecond timestamp to ISO 8601."""
import datetime
dt = datetime.datetime.fromtimestamp(
ns_timestamp / 1e9, tz=datetime.timezone.utc
)
return dt.isoformat()
class AgentTracer:
"""OpenTelemetry-compatible tracer for agent commerce."""
def __init__(
self,
agent_id: str,
client=None,
service_name: str = None,
exporters: List[SpanExporter] = None,
):
self.agent_id = agent_id
self._client = client
self._service_name = service_name or f"agent-{agent_id}"
provider = TracerProvider()
if client:
gh_exporter = GreenHelixSpanExporter(client, agent_id)
provider.add_span_processor(BatchSpanProcessor(gh_exporter))
if exporters:
for exporter in exporters:
provider.add_span_processor(BatchSpanProcessor(exporter))
self._provider = provider
self._tracer = provider.get_tracer(
self._service_name, schema_url="https://greenhelix.net/schemas/agent/1.0"
)
@contextmanager
def start_span(
self,
name: str,
kind: SpanKind = SpanKind.INTERNAL,
attributes: Dict[str, Any] = None,
peer_agent_id: str = None,
):
"""Start a new span with agent commerce context."""
attrs = {
AgentSpanAttributes.AGENT_ID: self.agent_id,
}
if peer_agent_id:
attrs[AgentSpanAttributes.PEER_AGENT_ID] = peer_agent_id
if attributes:
attrs.update(attributes)
with self._tracer.start_as_current_span(
name, kind=kind, attributes=attrs
) as span:
yield span
@contextmanager
def trace_agent_call(
self,
target_agent_id: str,
operation: str,
transaction_id: str = None,
):
"""Trace a call from this agent to another agent."""
txn_id = transaction_id or str(uuid.uuid4())
attrs = {
AgentSpanAttributes.OPERATION_TYPE: operation,
AgentSpanAttributes.TRANSACTION_ID: txn_id,
}
with self.start_span(
f"call.{target_agent_id}.{operation}",
kind=SpanKind.CLIENT,
attributes=attrs,
peer_agent_id=target_agent_id,
) as span:
yield span
@contextmanager
def trace_escrow(self, escrow_id: str, operation: str):
"""Trace an escrow lifecycle operation."""
attrs = {
AgentSpanAttributes.ESCROW_ID: escrow_id,
AgentSpanAttributes.OPERATION_TYPE: f"escrow.{operation}",
}
with self.start_span(
f"escrow.{operation}",
kind=SpanKind.INTERNAL,
attributes=attrs,
) as span:
yield span
@contextmanager
def trace_billing(self, amount: float, currency: str = "credits"):
"""Trace a billing event."""
attrs = {
AgentSpanAttributes.BILLING_AMOUNT: amount,
AgentSpanAttributes.BILLING_CURRENCY: currency,
AgentSpanAttributes.OPERATION_TYPE: "billing.charge",
}
with self.start_span(
"billing.charge",
kind=SpanKind.INTERNAL,
attributes=attrs,
) as span:
yield span
def get_trace_context(self) -> Dict[str, str]:
"""Extract current trace context for propagation to other agents."""
span = trace.get_current_span()
ctx = span.get_span_context()
if not ctx or not ctx.is_valid:
return {}
return {
"traceparent": f"00-{format(ctx.trace_id, '032x')}-"
f"{format(ctx.span_id, '016x')}-"
f"{'01' if ctx.trace_flags & 1 else '00'}",
"x-agent-id": self.agent_id,
}
def inject_context(self, headers: Dict[str, str]) -> Dict[str, str]:
"""Inject trace context into outgoing request headers."""
headers.update(self.get_trace_context())
return headers
def shutdown(self):
"""Flush pending spans and shut down the tracer."""
self._provider.shutdown()
Using the AgentTracer
The tracer integrates naturally with GreenHelix client calls. Each API interaction gets a span, and the spans automatically capture timing, status, and agent commerce attributes.
from greenhelix import GreenHelixClientclient = GreenHelixClient(api_key="your-api-key")
tracer = AgentTracer(agent_id="agent-marketplace", client=client)
Trace a multi-step transaction
with tracer.trace_agent_call("agent-data-provider", "fetch_dataset") as span:
result = client.execute_tool("call_agent", {
"target": "agent-data-provider",
"operation": "fetch_dataset",
"params": {"dataset_id": "ds_12345"},
"headers": tracer.get_trace_context(),
})
span.set_attribute("dataset.size_bytes", result.get("size", 0)) if result.get("status") == "error":
span.set_status(StatusCode.ERROR, result.get("message", ""))
else:
span.set_status(StatusCode.OK)
Trace an escrow lifecycle
with tracer.trace_escrow("esw_abc123", "create") as span:
escrow = client.execute_tool("create_escrow", {
"amount": 5.00,
"payer": "agent-buyer",
"payee": "agent-seller",
})
span.set_attribute("escrow.amount", 5.00)
span.set_status(StatusCode.OK)
The key design choice is that AgentTracer wraps OpenTelemetry rather than replacing it. You can add any standard OTel exporter (Jaeger, OTLP, console) alongside the GreenHelix exporter. This means your agent traces show up in your existing observability infrastructure without any migration.
The GreenHelixSpanExporter converts spans to metrics via submit_metrics. This dual-purpose export means every traced operation automatically generates latency and billing metrics, eliminating the need to instrument metrics separately for operations you are already tracing.
Chapter 3: Distributed Tracing Across Agent Calls
Single-agent tracing is straightforward. The real challenge is tracing a transaction as it flows through multiple independent agents, each potentially running on different infrastructure, operated by different teams, and communicating through the GreenHelix gateway.
The Problem: Trace Context Propagation
When Agent A calls Agent B, Agent B has no inherent knowledge of Agent A's trace. Without explicit context propagation, Agent B starts a new, disconnected trace. You end up with five isolated traces instead of one unified trace showing the complete transaction flow.
The W3C Trace Context standard solves this with two HTTP headers: traceparent (containing trace ID, span ID, and sampling flags) and tracestate (containing vendor-specific data). Our AgentTracer already generates these headers via get_trace_context(). The challenge is ensuring every agent in the chain extracts, uses, and propagates these headers.
Trace Context in Agent-to-Agent Calls
Here is the pattern for traced agent-to-agent calls. The calling agent injects context into outgoing headers. The receiving agent extracts context and creates child spans.
# === Calling Agent (Agent A) ===class TracedAgentClient:
"""HTTP client that automatically propagates trace context."""
def __init__(self, tracer: AgentTracer, client):
self._tracer = tracer
self._client = client
def call_agent(
self,
target_agent: str,
operation: str,
params: Dict[str, Any],
transaction_id: str = None,
) -> Dict[str, Any]:
"""Call another agent with automatic trace propagation."""
with self._tracer.trace_agent_call(
target_agent, operation, transaction_id
) as span:
headers = self._tracer.get_trace_context()
headers["x-transaction-id"] = transaction_id or ""
result = self._client.execute_tool("call_agent", {
"target": target_agent,
"operation": operation,
"params": params,
"headers": headers,
})
span.set_attribute("response.status", result.get("status", "unknown"))
if result.get("error"):
span.set_status(StatusCode.ERROR, result["error"])
else:
span.set_status(StatusCode.OK)
return result
=== Receiving Agent (Agent B) ===
from opentelemetry.trace.propagation import TraceContextTextMapPropagator
from opentelemetry.context import Context
class TracedAgentHandler:
"""Request handler that extracts and continues trace context."""
def __init__(self, tracer: AgentTracer):
self._tracer = tracer
self._propagator = TraceContextTextMapPropagator()
def handle_request(
self,
operation: str,
params: Dict[str, Any],
headers: Dict[str, str],
) -> Dict[str, Any]:
"""Handle an incoming agent request, continuing the trace."""
# Extract trace context from incoming headers
ctx = self._propagator.extract(carrier=headers)
# Create a child span under the extracted context
token = attach(ctx)
try:
with self._tracer.start_span(
f"handle.{operation}",
kind=SpanKind.SERVER,
attributes={
"caller.agent_id": headers.get("x-agent-id", "unknown"),
AgentSpanAttributes.OPERATION_TYPE: operation,
},
) as span:
result = self._dispatch(operation, params, span)
return result
finally:
detach(token)
def _dispatch(
self, operation: str, params: Dict[str, Any], span
) -> Dict[str, Any]:
"""Route to the appropriate operation handler."""
handler = getattr(self, f"op_{operation}", None)
if not handler:
span.set_status(StatusCode.ERROR, f"Unknown operation: {operation}")
return {"error": f"Unknown operation: {operation}"}
return handler(params, span)
Tracing a Complete Transaction Through Five Agents
Consider an agent commerce transaction where a customer agent orders a data analysis report. The transaction flows through five agents:
1. Customer Agent -- initiates the request 2. Marketplace Agent -- matches request to provider 3. Data Provider Agent -- supplies raw data 4. Analysis Agent -- processes and analyzes data 5. Delivery Agent -- formats and delivers the report
def traced_execute(
client: TracedAgentClient,
tracer: AgentTracer,
request: Dict[str, Any],
) -> Dict[str, Any]:
"""Execute a full traced transaction across multiple agents."""
transaction_id = str(uuid.uuid4()) with tracer.start_span(
"transaction.data_analysis_report",
kind=SpanKind.INTERNAL,
attributes={
AgentSpanAttributes.TRANSACTION_ID: transaction_id,
AgentSpanAttributes.OPERATION_TYPE: "data_analysis_report",
},
) as root_span:
# Step 1: Find a provider through the marketplace
with tracer.start_span("step.find_provider") as step_span:
match = client.call_agent(
"agent-marketplace",
"find_provider",
{"capability": "data_analysis", "budget": request["budget"]},
transaction_id=transaction_id,
)
step_span.set_attribute("provider.matched", match.get("provider_id", ""))
provider_id = match["provider_id"]
# Step 2: Create escrow for the transaction
with tracer.trace_escrow(f"esw_{transaction_id[:8]}", "create") as esc_span:
escrow = client.call_agent(
"agent-marketplace",
"create_escrow",
{
"amount": match["price"],
"payer": request["customer_agent"],
"payee": provider_id,
},
transaction_id=transaction_id,
)
esc_span.set_attribute("escrow.id", escrow.get("escrow_id", ""))
# Step 3: Fetch raw data
with tracer.start_span("step.fetch_data") as step_span:
data = client.call_agent(
match["data_source"],
"fetch_dataset",
{"dataset_id": request["dataset_id"]},
transaction_id=transaction_id,
)
step_span.set_attribute("data.records", data.get("record_count", 0))
# Step 4: Run analysis
with tracer.start_span("step.analyze") as step_span:
analysis = client.call_agent(
provider_id,
"analyze_data",
{"data": data["data"], "analysis_type": request["analysis_type"]},
transaction_id=transaction_id,
)
step_span.set_attribute("analysis.confidence", analysis.get("confidence", 0))
# Step 5: Deliver report
with tracer.start_span("step.deliver") as step_span:
delivery = client.call_agent(
"agent-delivery",
"deliver_report",
{
"report": analysis["report"],
"recipient": request["customer_agent"],
"format": request.get("format", "pdf"),
},
transaction_id=transaction_id,
)
step_span.set_attribute("delivery.channel", delivery.get("channel", ""))
# Step 6: Settle escrow
with tracer.trace_escrow(escrow["escrow_id"], "settle") as esc_span:
settlement = client.call_agent(
"agent-marketplace",
"settle_escrow",
{"escrow_id": escrow["escrow_id"]},
transaction_id=transaction_id,
)
esc_span.set_attribute("settlement.status", settlement.get("status", ""))
root_span.set_status(StatusCode.OK)
return {
"transaction_id": transaction_id,
"report_url": delivery.get("url"),
"total_cost": match["price"],
}
Trace Correlation with GreenHelix Events
GreenHelix events have their own event IDs. To get a complete picture, you need to correlate your OpenTelemetry trace IDs with GreenHelix event IDs. The approach is to embed the trace ID in your GreenHelix API calls and then join on it during analysis.
def trace_escrow_lifecycle(
tracer: AgentTracer,
client,
escrow_id: str,
) -> Dict[str, Any]:
"""Trace and correlate an escrow's full lifecycle with GreenHelix events."""
trace_ctx = tracer.get_trace_context()
trace_id = trace_ctx.get("traceparent", "").split("-")[1] if trace_ctx else "" with tracer.trace_escrow(escrow_id, "lifecycle") as span:
# Fetch GreenHelix events for this escrow
events = client.execute_tool("get_events", {
"filters": {"escrow_id": escrow_id},
"start_time": "2026-04-07T00:00:00Z",
"end_time": "2026-04-07T23:59:59Z",
})
lifecycle = {
"escrow_id": escrow_id,
"trace_id": trace_id,
"events": [],
}
for event in events.get("events", []):
event_type = event.get("type", "")
lifecycle["events"].append({
"type": event_type,
"timestamp": event.get("timestamp"),
"greenhelix_event_id": event.get("event_id"),
"trace_id": trace_id,
})
# Create a child span for each GreenHelix event
with tracer.start_span(
f"greenhelix.event.{event_type}",
attributes={
"greenhelix.event_id": event.get("event_id", ""),
"greenhelix.event_type": event_type,
AgentSpanAttributes.ESCROW_ID: escrow_id,
},
) as event_span:
event_span.set_status(StatusCode.OK)
span.set_attribute("lifecycle.event_count", len(lifecycle["events"]))
return lifecycle
Parent-Child Span Relationships
OpenTelemetry automatically manages parent-child relationships through Python context managers. When you nest start_span calls, each inner span becomes a child of the outer span. This creates the tree structure visible in trace visualization tools like Jaeger.
For agent commerce, the hierarchy typically looks like:
transaction.data_analysis_report (root)
|-- step.find_provider
| |-- call.agent-marketplace.find_provider (CLIENT)
|-- escrow.create
| |-- call.agent-marketplace.create_escrow (CLIENT)
|-- step.fetch_data
| |-- call.agent-data-source.fetch_dataset (CLIENT)
|-- step.analyze
| |-- call.agent-analysis.analyze_data (CLIENT)
|-- step.deliver
| |-- call.agent-delivery.deliver_report (CLIENT)
|-- escrow.settle
|-- call.agent-marketplace.settle_escrow (CLIENT)
Each CLIENT span on the calling side has a corresponding SERVER span on the receiving side. The trace ID is the same across all agents, so a trace visualization tool shows the entire transaction as a single, unified trace spanning all five agents.
Chapter 4: MetricsCollector Class
Tracing gives you per-transaction detail. Metrics give you the aggregate view: how is the fleet performing right now, how does today compare to yesterday, and are we meeting our business targets? The MetricsCollector class provides a clean API for collecting, buffering, and exporting metrics from your agent fleet.
Metric Types
There are three fundamental metric types for agent commerce:
Counters are monotonically increasing values. Use them for total requests, total errors, total revenue, and total transactions. Counters answer "how many" questions.
Gauges are point-in-time values that can go up or down. Use them for active connections, queue depth, current balance, and active escrows. Gauges answer "how much right now" questions.
Histograms track the distribution of values. Use them for latency, transaction amounts, and response sizes. Histograms answer "what is the distribution" questions, giving you percentiles (p50, p95, p99) rather than just averages.
The MetricsCollector Implementation
import time
import math
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any@dataclass
class MetricPoint:
"""A single metric data point."""
name: str
value: float
dimensions: Dict[str, str]
timestamp: float
metric_type: str # "counter", "gauge", "histogram"
class HistogramBuckets:
"""Tracks value distribution for histogram metrics."""
def __init__(self, boundaries: List[float] = None):
self.boundaries = boundaries or [
5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000
]
self.buckets = [0] * (len(self.boundaries) + 1)
self.sum = 0.0
self.count = 0
self.min = float("inf")
self.max = float("-inf")
self._values: List[float] = []
def observe(self, value: float):
self.sum += value
self.count += 1
self.min = min(self.min, value)
self.max = max(self.max, value)
self._values.append(value)
for i, boundary in enumerate(self.boundaries):
if value <= boundary:
self.buckets[i] += 1
return
self.buckets[-1] += 1
def percentile(self, p: float) -> float:
if not self._values:
return 0.0
sorted_vals = sorted(self._values)
idx = int(math.ceil(p / 100.0 * len(sorted_vals))) - 1
return sorted_vals[max(0, idx)]
def reset(self):
self.buckets = [0] * (len(self.boundaries) + 1)
self.sum = 0.0
self.count = 0
self.min = float("inf")
self.max = float("-inf")
self._values = []
class MetricsCollector:
"""Collects, buffers, and exports agent commerce metrics."""
def __init__(
self,
agent_id: str,
client=None,
flush_interval_seconds: float = 60.0,
buffer_size: int = 1000,
):
self.agent_id = agent_id
self._client = client
self._flush_interval = flush_interval_seconds
self._buffer_size = buffer_size
self._counters: Dict[str, float] = defaultdict(float)
self._gauges: Dict[str, float] = {}
self._histograms: Dict[str, HistogramBuckets] = {}
self._buffer: List[MetricPoint] = []
self._lock = threading.Lock()
self._dimensions_registry: Dict[str, Dict[str, str]] = {}
self._flush_callbacks: List[Callable] = []
self._running = False
self._flush_thread: Optional[threading.Thread] = None
def start(self):
"""Start the background flush thread."""
self._running = True
self._flush_thread = threading.Thread(
target=self._flush_loop, daemon=True
)
self._flush_thread.start()
def stop(self):
"""Stop the background flush thread and flush remaining metrics."""
self._running = False
if self._flush_thread:
self._flush_thread.join(timeout=5.0)
self.flush()
# --- Counter Operations ---
def increment(
self,
name: str,
value: float = 1.0,
dimensions: Dict[str, str] = None,
):
"""Increment a counter metric."""
key = self._make_key(name, dimensions)
with self._lock:
self._counters[key] += value
self._dimensions_registry[key] = dimensions or {}
self._buffer_point(name, self._counters[key], dimensions, "counter")
# --- Gauge Operations ---
def gauge_set(
self,
name: str,
value: float,
dimensions: Dict[str, str] = None,
):
"""Set a gauge metric to an absolute value."""
key = self._make_key(name, dimensions)
with self._lock:
self._gauges[key] = value
self._dimensions_registry[key] = dimensions or {}
self._buffer_point(name, value, dimensions, "gauge")
def gauge_increment(
self,
name: str,
value: float = 1.0,
dimensions: Dict[str, str] = None,
):
"""Increment a gauge metric."""
key = self._make_key(name, dimensions)
with self._lock:
self._gauges[key] = self._gauges.get(key, 0.0) + value
self._dimensions_registry[key] = dimensions or {}
self._buffer_point(name, self._gauges[key], dimensions, "gauge")
# --- Histogram Operations ---
def observe(
self,
name: str,
value: float,
dimensions: Dict[str, str] = None,
boundaries: List[float] = None,
):
"""Record an observation in a histogram metric."""
key = self._make_key(name, dimensions)
with self._lock:
if key not in self._histograms:
self._histograms[key] = HistogramBuckets(boundaries)
self._histograms[key].observe(value)
self._dimensions_registry[key] = dimensions or {}
self._buffer_point(name, value, dimensions, "histogram")
# --- Business Metric Helpers ---
def record_transaction(
self,
operation: str,
duration_ms: float,
amount: float,
status: str,
peer_agent: str = "",
):
"""Record a complete transaction with all standard metrics."""
dims = {
"operation": operation,
"status": status,
"peer_agent": peer_agent,
}
self.increment("transactions.total", 1.0, dims)
self.observe("transactions.duration_ms", duration_ms, dims)
self.observe("transactions.amount", amount, dims)
if status == "error":
self.increment("transactions.errors", 1.0, dims)
def record_revenue(self, amount: float, source: str, currency: str = "credits"):
"""Record revenue from an agent operation."""
dims = {"source": source, "currency": currency}
self.increment("revenue.total", amount, dims)
def record_cost(self, amount: float, category: str, currency: str = "credits"):
"""Record a cost incurred by the agent."""
dims = {"category": category, "currency": currency}
self.increment("costs.total", amount, dims)
def record_escrow_event(self, event_type: str, amount: float, escrow_id: str):
"""Record an escrow lifecycle event."""
dims = {"event_type": event_type}
self.increment(f"escrow.{event_type}", 1.0, dims)
self.observe("escrow.amount", amount, dims)
# --- Export and Flush ---
def flush(self):
"""Flush buffered metrics to GreenHelix."""
with self._lock:
if not self._buffer:
return
batch = self._buffer[:]
self._buffer.clear()
if self._client:
metrics_payload = [
{
"name": point.name,
"value": point.value,
"dimensions": {
"agent": self.agent_id,
**point.dimensions,
},
"timestamp": _ts_to_iso(point.timestamp),
}
for point in batch
]
try:
self._client.execute_tool("submit_metrics", {
"agent_id": self.agent_id,
"metrics": metrics_payload,
})
except Exception as e:
# Re-buffer on failure for retry
with self._lock:
self._buffer = batch + self._buffer
# Prevent unbounded growth
if len(self._buffer) > self._buffer_size * 2:
self._buffer = self._buffer[-self._buffer_size:]
for callback in self._flush_callbacks:
try:
callback(batch)
except Exception:
pass
def get_histogram_summary(
self, name: str, dimensions: Dict[str, str] = None
) -> Dict[str, float]:
"""Get summary statistics for a histogram metric."""
key = self._make_key(name, dimensions)
with self._lock:
hist = self._histograms.get(key)
if not hist or hist.count == 0:
return {}
return {
"count": hist.count,
"sum": hist.sum,
"min": hist.min,
"max": hist.max,
"avg": hist.sum / hist.count,
"p50": hist.percentile(50),
"p95": hist.percentile(95),
"p99": hist.percentile(99),
}
def on_flush(self, callback: Callable):
"""Register a callback to be called on each flush."""
self._flush_callbacks.append(callback)
# --- Internal ---
def _buffer_point(
self,
name: str,
value: float,
dimensions: Dict[str, str],
metric_type: str,
):
point = MetricPoint(
name=name,
value=value,
dimensions=dimensions or {},
timestamp=time.time(),
metric_type=metric_type,
)
self._buffer.append(point)
if len(self._buffer) >= self._buffer_size:
threading.Thread(target=self.flush, daemon=True).start()
def _flush_loop(self):
while self._running:
time.sleep(self._flush_interval)
self.flush()
@staticmethod
def _make_key(name: str, dimensions: Dict[str, str] = None) -> str:
if not dimensions:
return name
dim_str = ",".join(f"{k}={v}" for k, v in sorted(dimensions.items()))
return f"{name}{{{dim_str}}}"
def _ts_to_iso(ts: float) -> str:
import datetime
dt = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc)
return dt.isoformat()
Using the MetricsCollector
client = GreenHelixClient(api_key="your-api-key")
metrics = MetricsCollector(agent_id="agent-marketplace", client=client)
metrics.start()Record a transaction
start = time.time()
result = process_order(order)
duration_ms = (time.time() - start) * 1000metrics.record_transaction(
operation="process_order",
duration_ms=duration_ms,
amount=order["amount"],
status="success" if result["ok"] else "error",
peer_agent=order["seller_agent"],
)
Record revenue and costs
metrics.record_revenue(order["commission"], source="marketplace_fee")
metrics.record_cost(order["gateway_fee"], category="gateway")Record escrow events
metrics.record_escrow_event("created", order["amount"], result["escrow_id"])Get latency percentiles
summary = metrics.get_histogram_summary(
"transactions.duration_ms",
{"operation": "process_order", "status": "success"},
)
print(f"p50={summary['p50']:.0f}ms p95={summary['p95']:.0f}ms p99={summary['p99']:.0f}ms")Clean shutdown
metrics.stop()
The MetricsCollector buffers metrics in memory and flushes them periodically (default 60 seconds) or when the buffer fills. This batching reduces API calls to GreenHelix while ensuring metrics are not lost. The retry logic on flush failure re-buffers metrics for the next flush cycle, with a cap to prevent unbounded memory growth.
Chapter 5: Anomaly Detection
Raw metrics are necessary but not sufficient. An agent operator managing a fleet of 50 agents cannot watch 200 dashboards for problems. You need automated anomaly detection that understands normal patterns and alerts on deviations.
Statistical Anomaly Detection for Agent Metrics
Anomaly detection for agent commerce differs from traditional infrastructure monitoring in two ways. First, agent traffic patterns are often bursty and non-uniform: a marketplace agent might handle 100 requests per minute during business hours and 5 per minute overnight. Second, business metrics like revenue have seasonal patterns (daily, weekly, monthly) that must be accounted for.
Our AnomalyDetector uses three complementary detection methods: Z-score for sudden spikes, percentage change for trend shifts, and seasonal decomposition for pattern violations.
import math
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enumclass AnomalyType(Enum):
SPIKE = "spike"
DROP = "drop"
TREND_CHANGE = "trend_change"
SEASONAL_VIOLATION = "seasonal_violation"
class Severity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Anomaly:
"""A detected anomaly in a metric series."""
metric_name: str
anomaly_type: AnomalyType
severity: Severity
current_value: float
expected_value: float
deviation: float
timestamp: float
dimensions: Dict[str, str] = field(default_factory=dict)
message: str = ""
class MetricWindow:
"""Sliding window of metric values for statistical analysis."""
def __init__(self, window_size: int = 60):
self._values: deque = deque(maxlen=window_size)
self._timestamps: deque = deque(maxlen=window_size)
def add(self, value: float, timestamp: float = None):
self._values.append(value)
self._timestamps.append(timestamp or time.time())
@property
def count(self) -> int:
return len(self._values)
@property
def mean(self) -> float:
if not self._values:
return 0.0
return sum(self._values) / len(self._values)
@property
def std(self) -> float:
if len(self._values) < 2:
return 0.0
m = self.mean
variance = sum((v - m) ** 2 for v in self._values) / (len(self._values) - 1)
return math.sqrt(variance)
@property
def values(self) -> List[float]:
return list(self._values)
@property
def latest(self) -> Optional[float]:
return self._values[-1] if self._values else None
def rate_of_change(self, periods: int = 5) -> Optional[float]:
"""Calculate rate of change over the last N periods."""
if len(self._values) < periods + 1:
return None
recent = list(self._values)[-periods:]
older = list(self._values)[-(periods + 1)]
if older == 0:
return None
return (recent[-1] - older) / older
class SeasonalProfile:
"""Tracks seasonal patterns for a metric (hourly buckets)."""
def __init__(self, num_buckets: int = 24):
self._num_buckets = num_buckets
self._bucket_values: Dict[int, List[float]] = {
i: [] for i in range(num_buckets)
}
def add(self, value: float, timestamp: float):
import datetime
dt = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
bucket = dt.hour # Hourly buckets
self._bucket_values[bucket].append(value)
def expected_value(self, timestamp: float) -> Optional[Tuple[float, float]]:
"""Return (mean, std) for the expected value at this time."""
import datetime
dt = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
bucket = dt.hour
values = self._bucket_values.get(bucket, [])
if len(values) < 7: # Need at least a week of data
return None
m = sum(values) / len(values)
variance = sum((v - m) ** 2 for v in values) / (len(values) - 1)
return m, math.sqrt(variance)
class AnomalyDetector:
"""Detects anomalies in agent commerce metric streams."""
def __init__(
self,
z_score_threshold: float = 3.0,
pct_change_threshold: float = 0.30,
seasonal_z_threshold: float = 2.5,
min_data_points: int = 20,
):
self._z_threshold = z_score_threshold
self._pct_threshold = pct_change_threshold
self._seasonal_z_threshold = seasonal_z_threshold
self._min_data_points = min_data_points
self._windows: Dict[str, MetricWindow] = {}
self._seasonal: Dict[str, SeasonalProfile] = {}
self._anomaly_callbacks: List = []
def observe(
self,
metric_name: str,
value: float,
dimensions: Dict[str, str] = None,
timestamp: float = None,
) -> List[Anomaly]:
"""Observe a metric value and check for anomalies."""
ts = timestamp or time.time()
key = self._make_key(metric_name, dimensions)
if key not in self._windows:
self._windows[key] = MetricWindow(window_size=120)
self._seasonal[key] = SeasonalProfile()
window = self._windows[key]
seasonal = self._seasonal[key]
window.add(value, ts)
seasonal.add(value, ts)
anomalies = []
if window.count >= self._min_data_points:
# Z-score detection
z_anomaly = self._check_z_score(
metric_name, value, window, dimensions, ts
)
if z_anomaly:
anomalies.append(z_anomaly)
# Percentage change detection
pct_anomaly = self._check_pct_change(
metric_name, value, window, dimensions, ts
)
if pct_anomaly:
anomalies.append(pct_anomaly)
# Seasonal detection
seasonal_anomaly = self._check_seasonal(
metric_name, value, seasonal, dimensions, ts
)
if seasonal_anomaly:
anomalies.append(seasonal_anomaly)
for anomaly in anomalies:
for callback in self._anomaly_callbacks:
try:
callback(anomaly)
except Exception:
pass
return anomalies
def on_anomaly(self, callback):
"""Register a callback for detected anomalies."""
self._anomaly_callbacks.append(callback)
def _check_z_score(
self,
metric_name: str,
value: float,
window: MetricWindow,
dimensions: Dict[str, str],
timestamp: float,
) -> Optional[Anomaly]:
"""Detect anomalies using Z-score (standard deviations from mean)."""
std = window.std
if std == 0:
return None
z_score = abs(value - window.mean) / std
if z_score < self._z_threshold:
return None
anomaly_type = AnomalyType.SPIKE if value > window.mean else AnomalyType.DROP
severity = self._z_to_severity(z_score)
return Anomaly(
metric_name=metric_name,
anomaly_type=anomaly_type,
severity=severity,
current_value=value,
expected_value=window.mean,
deviation=z_score,
timestamp=timestamp,
dimensions=dimensions or {},
message=f"{metric_name} is {z_score:.1f} standard deviations "
f"{'above' if value > window.mean else 'below'} the mean "
f"(current={value:.2f}, mean={window.mean:.2f}, std={std:.2f})",
)
def _check_pct_change(
self,
metric_name: str,
value: float,
window: MetricWindow,
dimensions: Dict[str, str],
timestamp: float,
) -> Optional[Anomaly]:
"""Detect anomalies using percentage change over recent periods."""
roc = window.rate_of_change(periods=5)
if roc is None or abs(roc) < self._pct_threshold:
return None
anomaly_type = (
AnomalyType.TREND_CHANGE
)
severity = Severity.MEDIUM if abs(roc) < 0.5 else Severity.HIGH
return Anomaly(
metric_name=metric_name,
anomaly_type=anomaly_type,
severity=severity,
current_value=value,
expected_value=window.mean,
deviation=roc,
timestamp=timestamp,
dimensions=dimensions or {},
message=f"{metric_name} changed by {roc*100:.1f}% over the last 5 periods "
f"(current={value:.2f})",
)
def _check_seasonal(
self,
metric_name: str,
value: float,
seasonal: SeasonalProfile,
dimensions: Dict[str, str],
timestamp: float,
) -> Optional[Anomaly]:
"""Detect anomalies relative to seasonal patterns."""
expected = seasonal.expected_value(timestamp)
if expected is None:
return None
mean, std = expected
if std == 0:
return None
z_score = abs(value - mean) / std
if z_score < self._seasonal_z_threshold:
return None
return Anomaly(
metric_name=metric_name,
anomaly_type=AnomalyType.SEASONAL_VIOLATION,
severity=self._z_to_severity(z_score),
current_value=value,
expected_value=mean,
deviation=z_score,
timestamp=timestamp,
dimensions=dimensions or {},
message=f"{metric_name} deviates {z_score:.1f} std from seasonal expectation "
f"(current={value:.2f}, expected={mean:.2f} +/- {std:.2f})",
)
@staticmethod
def _z_to_severity(z_score: float) -> Severity:
if z_score >= 5.0:
return Severity.CRITICAL
elif z_score >= 4.0:
return Severity.HIGH
elif z_score >= 3.0:
return Severity.MEDIUM
return Severity.LOW
@staticmethod
def _make_key(name: str, dimensions: Dict[str, str] = None) -> str:
if not dimensions:
return name
dim_str = ",".join(f"{k}={v}" for k, v in sorted(dimensions.items()))
return f"{name}{{{dim_str}}}"
Integrating Anomaly Detection with Metrics Collection
The AnomalyDetector works alongside the MetricsCollector. Feed every metric observation to both:
detector = AnomalyDetector(
z_score_threshold=3.0,
pct_change_threshold=0.30,
min_data_points=20,
)
metrics = MetricsCollector(agent_id="agent-marketplace", client=client)def record_and_detect(
metric_name: str,
value: float,
dimensions: Dict[str, str] = None,
):
"""Record a metric and check for anomalies in one call."""
metrics.observe(metric_name, value, dimensions)
anomalies = detector.observe(metric_name, value, dimensions)
for anomaly in anomalies:
print(f"ANOMALY: {anomaly.message}")
# Feed anomaly back as a metric for meta-monitoring
metrics.increment(
"anomalies.detected",
1.0,
{
"metric": anomaly.metric_name,
"type": anomaly.anomaly_type.value,
"severity": anomaly.severity.value,
},
)
return anomalies
Example: detect latency spike
record_and_detect(
"transactions.duration_ms",
12500.0, # 12.5 seconds -- likely anomalous
{"operation": "process_order"},
)
The detector's sliding window approach means it adapts to changing baselines. If your agent's latency gradually increases from 200ms to 300ms over a week, the detector adjusts its baseline and does not fire false alerts. But if latency jumps from 300ms to 3000ms in one minute, the Z-score catches it immediately.
The seasonal profile prevents false alerts during known traffic patterns. If your agent processes fewer transactions overnight, a drop from 100 req/s to 10 req/s at midnight is not anomalous -- the seasonal profile knows that is normal for that hour.
Chapter 6: Dashboard Design
Anomaly detection catches problems automatically, but operators still need dashboards for situational awareness, capacity planning, and stakeholder reporting. This chapter describes the three essential dashboards for agent commerce operations, with the data sources and queries needed to populate them.
Dashboard 1: Fleet Health Overview
The fleet health dashboard is the first thing you look at when something might be wrong. It answers: "Is everything healthy right now?"
Key panels:
1. Agent Status Grid -- A grid showing every agent with color-coded health status (green/yellow/red). Data source: gauge metrics for each agent's last heartbeat timestamp and error rate. An agent is green if error rate is below 1% and last heartbeat was within 60 seconds, yellow if error rate is 1-5% or heartbeat is 60-120 seconds stale, and red otherwise.
2. Fleet Error Rate -- A time series chart showing the aggregate error rate across all agents over the last 24 hours, with the anomaly detection threshold overlaid. Data source: counter metrics transactions.errors / transactions.total, aggregated per minute.
3. Active Escrows -- A gauge showing the current number of open (unsettled) escrows, with a trend line. A rising trend indicates settlement failures or slow processing. Data source: gauge metric escrow.active_count.
4. Latency Heatmap -- A heatmap showing p95 latency for each agent over the last 6 hours. Columns are time buckets (15-minute intervals), rows are agents, and color intensity represents latency. This makes it easy to spot both fleet-wide issues (a hot column) and per-agent issues (a hot row). Data source: histogram metric transactions.duration_ms, p95 aggregation.
5. Alert Summary -- A table of currently active alerts grouped by severity, with acknowledge/resolve buttons. Data source: the AlertManager's active alerts state (covered in Chapter 7).
def build_fleet_health_query(agent_ids: List[str]) -> Dict[str, Any]:
"""Build the query configuration for the fleet health dashboard."""
return {
"panels": [
{
"title": "Agent Status Grid",
"type": "status_grid",
"query": {
"metrics": ["agent.heartbeat.age_seconds", "transactions.errors"],
"group_by": ["agent"],
"thresholds": {
"green": {"error_rate": 0.01, "heartbeat_age": 60},
"yellow": {"error_rate": 0.05, "heartbeat_age": 120},
"red": {"error_rate": 1.0, "heartbeat_age": float("inf")},
},
},
},
{
"title": "Fleet Error Rate",
"type": "time_series",
"query": {
"metric": "transactions.errors",
"aggregation": "rate",
"interval": "1m",
"time_range": "24h",
},
},
{
"title": "Active Escrows",
"type": "gauge_with_trend",
"query": {
"metric": "escrow.active_count",
"aggregation": "latest",
"trend_window": "6h",
},
},
{
"title": "Latency Heatmap",
"type": "heatmap",
"query": {
"metric": "transactions.duration_ms",
"aggregation": "p95",
"group_by": ["agent"],
"interval": "15m",
"time_range": "6h",
},
},
],
}
Dashboard 2: Transaction Flow
The transaction flow dashboard shows how transactions move through your agent fleet. It answers: "What is happening to our transactions?"
Key panels:
1. Transaction Funnel -- A funnel visualization showing transactions at each stage: initiated, matched, escrowed, fulfilled, settled, completed. Drop-off at each stage reveals where transactions are failing. Data source: counter metrics per transaction stage.
2. Agent-to-Agent Flow -- A Sankey diagram or chord chart showing traffic volume between agent pairs. Thick lines indicate high-volume relationships. Clicking a flow line drills down to latency and error metrics for that specific agent pair. Data source: counter metric transactions.total with peer_agent dimension.
3. Escrow Lifecycle -- A state diagram showing escrows in each state (pending, funded, fulfilled, settled, disputed, expired) with transition rates. Data source: counter metrics per escrow state transition.
4. Slow Transactions -- A table of the 20 slowest transactions in the last hour, with trace IDs linked to the trace viewer. This gives operators one-click access to detailed trace analysis for the worst-performing transactions. Data source: histogram metric transactions.duration_ms, top-N query.
5. Transaction Volume -- A stacked area chart showing transaction volume per operation type over the last 24 hours. Data source: counter metric transactions.total grouped by operation dimension.
Dashboard 3: Revenue and Cost
The revenue and cost dashboard is for business stakeholders and capacity planning. It answers: "Are we making money, and where is it going?"
Key panels:
1. Net Revenue -- A single large number showing today's net revenue (revenue minus costs), with comparison to same day last week. Data source: counter metrics revenue.total minus costs.total.
2. Revenue by Agent -- A bar chart showing revenue contribution per agent. Identifies your most and least profitable agents. Data source: counter metric revenue.total grouped by agent dimension.
3. Cost Breakdown -- A pie chart showing costs by category: gateway fees, escrow fees, infrastructure, third-party API calls. Data source: counter metric costs.total grouped by category dimension.
4. Revenue Trend -- A time series comparing daily revenue over the last 30 days, with the anomaly detection band overlaid. Revenue drops outside the expected band trigger alerts. Data source: counter metric revenue.total, daily aggregation.
5. Per-Transaction Economics -- A table showing average revenue, cost, and margin per operation type. Identifies operations that are losing money. Data source: derived from revenue.total and costs.total metrics divided by transactions.total, grouped by operation.
Each dashboard should auto-refresh on a cadence appropriate to its purpose: fleet health every 30 seconds, transaction flow every minute, revenue and cost every 5 minutes. The underlying data all comes from the MetricsCollector metrics pushed to GreenHelix via submit_metrics, queried back for visualization.
Chapter 7: AlertManager Class
Dashboards require someone to be watching them. Alerts push problems to the right people at the right time. The AlertManager evaluates alert rules against incoming metrics, deduplicates and groups related alerts, routes them through appropriate channels, and escalates when problems are not acknowledged.
Alert Rule Types
Three types of alert rules cover the scenarios that matter for agent commerce:
Threshold alerts fire when a metric crosses a static or dynamic boundary. Example: "Error rate exceeds 5%." These are simple and reliable for known boundaries.
Rate-of-change alerts fire when a metric changes too quickly. Example: "Transaction volume dropped more than 30% in 10 minutes." These catch problems that threshold alerts miss because the absolute value might still be "normal."
Anomaly-based alerts fire when the AnomalyDetector identifies a statistical anomaly. These adapt to changing baselines and seasonal patterns, catching subtle issues that would require constantly adjusting threshold values.
The AlertManager Implementation
import time
import uuid
import threading
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from enum import Enum
from collections import defaultdictclass AlertState(Enum):
FIRING = "firing"
ACKNOWLEDGED = "acknowledged"
RESOLVED = "resolved"
class AlertChannel(Enum):
WEBHOOK = "webhook"
EMAIL = "email"
SLACK = "slack"
PAGERDUTY = "pagerduty"
@dataclass
class AlertRule:
"""Definition of an alert rule."""
name: str
metric_name: str
condition: str # "threshold", "rate_of_change", "anomaly"
threshold: Optional[float] = None
comparison: str = "gt" # "gt", "lt", "gte", "lte"
window_minutes: int = 5
severity: str = "medium"
channels: List[AlertChannel] = field(default_factory=list)
dimensions_filter: Dict[str, str] = field(default_factory=dict)
cooldown_minutes: int = 15
escalation_minutes: int = 30
description: str = ""
runbook_url: str = ""
@dataclass
class Alert:
"""An active alert instance."""
id: str
rule: AlertRule
state: AlertState
fired_at: float
last_value: float
message: str
acknowledged_at: Optional[float] = None
acknowledged_by: Optional[str] = None
resolved_at: Optional[float] = None
escalated: bool = False
notification_count: int = 0
dimensions: Dict[str, str] = field(default_factory=dict)
class AlertManager:
"""Manages alert rules, routing, deduplication, and escalation."""
def __init__(
self,
agent_id: str,
client=None,
anomaly_detector: "AnomalyDetector" = None,
):
self.agent_id = agent_id
self._client = client
self._detector = anomaly_detector
self._rules: Dict[str, AlertRule] = {}
self._active_alerts: Dict[str, Alert] = {}
self._alert_history: List[Alert] = []
self._cooldowns: Dict[str, float] = {}
self._lock = threading.Lock()
# Channel handlers
self._channel_handlers: Dict[AlertChannel, Callable] = {}
# Escalation thread
self._running = False
self._escalation_thread: Optional[threading.Thread] = None
def start(self):
"""Start the escalation monitoring thread."""
self._running = True
self._escalation_thread = threading.Thread(
target=self._escalation_loop, daemon=True
)
self._escalation_thread.start()
def stop(self):
self._running = False
if self._escalation_thread:
self._escalation_thread.join(timeout=5.0)
# --- Rule Management ---
def add_rule(self, rule: AlertRule):
"""Add an alert rule."""
self._rules[rule.name] = rule
def remove_rule(self, name: str):
"""Remove an alert rule."""
self._rules.pop(name, None)
# --- Channel Configuration ---
def register_channel(self, channel: AlertChannel, handler: Callable):
"""Register a notification handler for a channel."""
self._channel_handlers[channel] = handler
def configure_webhook(self, url: str):
"""Configure webhook channel with a URL."""
import urllib.request
def webhook_handler(alert: Alert, message: str):
payload = json.dumps({
"alert_id": alert.id,
"rule": alert.rule.name,
"severity": alert.rule.severity,
"state": alert.state.value,
"message": message,
"value": alert.last_value,
"fired_at": alert.fired_at,
"agent_id": self.agent_id,
"dimensions": alert.dimensions,
}).encode()
req = urllib.request.Request(
url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
urllib.request.urlopen(req, timeout=10)
except Exception:
pass
self._channel_handlers[AlertChannel.WEBHOOK] = webhook_handler
def configure_slack(self, webhook_url: str, default_channel: str = "#alerts"):
"""Configure Slack channel."""
import urllib.request
def slack_handler(alert: Alert, message: str):
severity_emoji = {
"critical": ":red_circle:",
"high": ":large_orange_circle:",
"medium": ":large_yellow_circle:",
"low": ":white_circle:",
}
emoji = severity_emoji.get(alert.rule.severity, ":grey_question:")
payload = json.dumps({
"channel": default_channel,
"text": f"{emoji} *{alert.rule.name}*\n{message}",
"attachments": [
{
"color": {
"critical": "#ff0000",
"high": "#ff8800",
"medium": "#ffcc00",
"low": "#cccccc",
}.get(alert.rule.severity, "#cccccc"),
"fields": [
{"title": "Agent", "value": self.agent_id, "short": True},
{"title": "Value", "value": str(alert.last_value), "short": True},
{"title": "Severity", "value": alert.rule.severity, "short": True},
{"title": "Alert ID", "value": alert.id, "short": True},
],
}
],
}).encode()
req = urllib.request.Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
urllib.request.urlopen(req, timeout=10)
except Exception:
pass
self._channel_handlers[AlertChannel.SLACK] = slack_handler
# --- Alert Evaluation ---
def evaluate(
self,
metric_name: str,
value: float,
dimensions: Dict[str, str] = None,
) -> List[Alert]:
"""Evaluate all rules against a metric value."""
new_alerts = []
dims = dimensions or {}
for rule in self._rules.values():
if rule.metric_name != metric_name:
continue
# Check dimension filter
if rule.dimensions_filter:
if not all(
dims.get(k) == v for k, v in rule.dimensions_filter.items()
):
continue
# Check cooldown
cooldown_key = f"{rule.name}:{self._dim_key(dims)}"
if cooldown_key in self._cooldowns:
if time.time() - self._cooldowns[cooldown_key] < rule.cooldown_minutes * 60:
continue
triggered = False
if rule.condition == "threshold":
triggered = self._check_threshold(rule, value)
elif rule.condition == "rate_of_change":
triggered = self._check_rate_of_change(rule, value, metric_name, dims)
elif rule.condition == "anomaly" and self._detector:
anomalies = self._detector.observe(metric_name, value, dims)
triggered = len(anomalies) > 0
if triggered:
alert = self._fire_alert(rule, value, dims)
if alert:
new_alerts.append(alert)
self._cooldowns[cooldown_key] = time.time()
# Check for auto-resolution
self._check_resolutions(metric_name, value, dims)
return new_alerts
def acknowledge(self, alert_id: str, acknowledged_by: str = "operator"):
"""Acknowledge an active alert."""
with self._lock:
alert = self._active_alerts.get(alert_id)
if alert and alert.state == AlertState.FIRING:
alert.state = AlertState.ACKNOWLEDGED
alert.acknowledged_at = time.time()
alert.acknowledged_by = acknowledged_by
def resolve(self, alert_id: str):
"""Manually resolve an alert."""
with self._lock:
alert = self._active_alerts.get(alert_id)
if alert:
alert.state = AlertState.RESOLVED
alert.resolved_at = time.time()
self._alert_history.append(alert)
del self._active_alerts[alert_id]
self._notify(alert, f"RESOLVED: {alert.rule.name}")
def get_active_alerts(
self, severity: str = None
) -> List[Alert]:
"""Get all active (firing or acknowledged) alerts."""
with self._lock:
alerts = list(self._active_alerts.values())
if severity:
alerts = [a for a in alerts if a.rule.severity == severity]
return sorted(alerts, key=lambda a: a.fired_at, reverse=True)
# --- Internal ---
def _check_threshold(self, rule: AlertRule, value: float) -> bool:
if rule.threshold is None:
return False
ops = {
"gt": lambda v, t: v > t,
"lt": lambda v, t: v < t,
"gte": lambda v, t: v >= t,
"lte": lambda v, t: v <= t,
}
op = ops.get(rule.comparison, ops["gt"])
return op(value, rule.threshold)
def _check_rate_of_change(
self,
rule: AlertRule,
value: float,
metric_name: str,
dimensions: Dict[str, str],
) -> bool:
if not self._detector:
return False
key = self._detector._make_key(metric_name, dimensions)
window = self._detector._windows.get(key)
if not window:
return False
roc = window.rate_of_change(periods=5)
if roc is None or rule.threshold is None:
return False
return abs(roc) > rule.threshold
def _fire_alert(
self,
rule: AlertRule,
value: float,
dimensions: Dict[str, str],
) -> Optional[Alert]:
# Deduplicate: check if same rule + dimensions already firing
dedup_key = f"{rule.name}:{self._dim_key(dimensions)}"
with self._lock:
for alert in self._active_alerts.values():
existing_key = f"{alert.rule.name}:{self._dim_key(alert.dimensions)}"
if existing_key == dedup_key:
# Update existing alert instead of creating new one
alert.last_value = value
alert.notification_count += 1
return None
alert = Alert(
id=f"alert_{uuid.uuid4().hex[:12]}",
rule=rule,
state=AlertState.FIRING,
fired_at=time.time(),
last_value=value,
message=f"{rule.name}: {rule.metric_name} = {value} "
f"(threshold: {rule.comparison} {rule.threshold})",
dimensions=dimensions,
)
self._active_alerts[alert.id] = alert
self._notify(alert, alert.message)
return alert
def _notify(self, alert: Alert, message: str):
for channel in alert.rule.channels:
handler = self._channel_handlers.get(channel)
if handler:
try:
handler(alert, message)
except Exception:
pass
# Also register webhook with GreenHelix for event-driven alerts
if self._client and alert.state == AlertState.FIRING:
try:
self._client.execute_tool("submit_metrics", {
"agent_id": self.agent_id,
"metrics": [{
"name": "alerts.fired",
"value": 1,
"dimensions": {
"rule": alert.rule.name,
"severity": alert.rule.severity,
"agent": self.agent_id,
},
}],
})
except Exception:
pass
def _check_resolutions(
self,
metric_name: str,
value: float,
dimensions: Dict[str, str],
):
"""Auto-resolve alerts when the condition clears."""
with self._lock:
to_resolve = []
for alert_id, alert in self._active_alerts.items():
if alert.rule.metric_name != metric_name:
continue
if alert.rule.condition == "threshold":
if not self._check_threshold(alert.rule, value):
to_resolve.append(alert_id)
for alert_id in to_resolve:
alert = self._active_alerts[alert_id]
alert.state = AlertState.RESOLVED
alert.resolved_at = time.time()
self._alert_history.append(alert)
del self._active_alerts[alert_id]
self._notify(alert, f"RESOLVED: {alert.rule.name}")
def _escalation_loop(self):
"""Periodically check for alerts needing escalation."""
while self._running:
time.sleep(60)
now = time.time()
with self._lock:
for alert in self._active_alerts.values():
if alert.state != AlertState.FIRING:
continue
if alert.escalated:
continue
age_minutes = (now - alert.fired_at) / 60
if age_minutes >= alert.rule.escalation_minutes:
alert.escalated = True
self._notify(
alert,
f"ESCALATION: {alert.rule.name} has been firing for "
f"{age_minutes:.0f} minutes without acknowledgment",
)
@staticmethod
def _dim_key(dimensions: Dict[str, str]) -> str:
if not dimensions:
return ""
return ",".join(f"{k}={v}" for k, v in sorted(dimensions.items()))
Configuring Alert Rules
Here is a practical configuration for an agent commerce deployment:
alert_manager = AlertManager(
agent_id="agent-marketplace",
client=client,
anomaly_detector=detector,
)Configure notification channels
alert_manager.configure_webhook("https://alerts.myfleet.com/webhook")
alert_manager.configure_slack(
webhook_url="https://hooks.slack.com/services/T00/B00/xxx",
default_channel="#agent-alerts",
)Rule 1: High error rate
alert_manager.add_rule(AlertRule(
name="high_error_rate",
metric_name="transactions.errors",
condition="threshold",
threshold=0.05,
comparison="gt",
severity="high",
channels=[AlertChannel.SLACK, AlertChannel.WEBHOOK],
cooldown_minutes=15,
escalation_minutes=30,
description="Transaction error rate exceeds 5%",
runbook_url="https://wiki.myfleet.com/runbooks/high-error-rate",
))Rule 2: Latency spike
alert_manager.add_rule(AlertRule(
name="latency_spike",
metric_name="transactions.duration_ms",
condition="threshold",
threshold=5000,
comparison="gt",
severity="medium",
channels=[AlertChannel.SLACK],
cooldown_minutes=10,
description="Transaction latency exceeds 5 seconds",
))Rule 3: Revenue drop
alert_manager.add_rule(AlertRule(
name="revenue_drop",
metric_name="revenue.total",
condition="rate_of_change",
threshold=0.30, # 30% drop
severity="critical",
channels=[AlertChannel.SLACK, AlertChannel.PAGERDUTY, AlertChannel.WEBHOOK],
cooldown_minutes=60,
escalation_minutes=15,
description="Revenue dropped more than 30% in 5 periods",
))Rule 4: Anomaly-based (uses AnomalyDetector)
alert_manager.add_rule(AlertRule(
name="escrow_anomaly",
metric_name="escrow.created",
condition="anomaly",
severity="medium",
channels=[AlertChannel.SLACK],
cooldown_minutes=30,
description="Anomalous escrow creation pattern detected",
))alert_manager.start()
Evaluate metrics as they come in
alert_manager.evaluate("transactions.errors", 0.08, {"operation": "process_order"})
alert_manager.evaluate("transactions.duration_ms", 7500.0, {"operation": "process_order"})
The deduplication logic prevents alert storms: if the same rule fires for the same dimensions, the existing alert is updated rather than creating a new notification. The cooldown period prevents re-firing too quickly after an alert resolves and re-triggers. The escalation thread ensures that un-acknowledged alerts get escalated to additional channels after the configured timeout.
Chapter 8: SLA Monitoring and Cost Attribution
The final piece of the observability stack ties technical metrics to business commitments. SLA monitoring tracks whether your agents are meeting their published service level agreements. Cost attribution tracks where money is being spent and whether each agent is profitable.
SLA Definition and Tracking
An SLA for an agent commerce system typically includes:
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any@dataclass
class SLADefinition:
"""An SLA definition with target and measurement parameters."""
name: str
metric_name: str
target_value: float
comparison: str # "lte" (latency), "gte" (availability), "lte" (error rate)
measurement_window: str # "1h", "24h", "7d", "30d"
description: str = ""
@dataclass
class SLAStatus:
"""Current status of an SLA."""
definition: SLADefinition
current_value: float
compliant: bool
compliance_percentage: float # % of measurement windows that met the SLA
last_violation: Optional[float] = None
violation_count: int = 0
class SLAMonitor:
"""Tracks SLA compliance for agent commerce operations."""
def __init__(
self,
agent_id: str,
metrics_collector: "MetricsCollector",
client=None,
):
self.agent_id = agent_id
self._metrics = metrics_collector
self._client = client
self._sla_definitions: Dict[str, SLADefinition] = {}
self._compliance_history: Dict[str, List[bool]] = defaultdict(list)
self._violation_callbacks: List = []
def define_sla(self, sla: SLADefinition):
"""Register an SLA definition."""
self._sla_definitions[sla.name] = sla
def on_violation(self, callback):
"""Register a callback for SLA violations."""
self._violation_callbacks.append(callback)
def check_sla(
self,
sla_name: str,
current_value: float,
) -> SLAStatus:
"""Check if a metric value meets the SLA target."""
sla = self._sla_definitions.get(sla_name)
if not sla:
raise ValueError(f"Unknown SLA: {sla_name}")
ops = {
"lte": lambda v, t: v <= t,
"gte": lambda v, t: v >= t,
"lt": lambda v, t: v < t,
"gt": lambda v, t: v > t,
}
op = ops.get(sla.comparison, ops["lte"])
compliant = op(current_value, sla.target_value)
# Track compliance history
self._compliance_history[sla_name].append(compliant)
# Calculate compliance percentage (last 100 checks)
history = self._compliance_history[sla_name][-100:]
compliance_pct = sum(1 for c in history if c) / len(history) * 100
status = SLAStatus(
definition=sla,
current_value=current_value,
compliant=compliant,
compliance_percentage=compliance_pct,
violation_count=sum(1 for c in history if not c),
)
if not compliant:
status.last_violation = time.time()
# Record violation metric
self._metrics.increment(
"sla.violations",
1.0,
{"sla": sla_name, "agent": self.agent_id},
)
for callback in self._violation_callbacks:
try:
callback(status)
except Exception:
pass
# Always record compliance percentage
self._metrics.gauge_set(
"sla.compliance_pct",
compliance_pct,
{"sla": sla_name, "agent": self.agent_id},
)
return status
def get_all_sla_status(self) -> Dict[str, SLAStatus]:
"""Get the last known status for all SLAs."""
results = {}
for name in self._sla_definitions:
history = self._compliance_history.get(name, [])
if not history:
continue
last_100 = history[-100:]
compliance_pct = sum(1 for c in last_100 if c) / len(last_100) * 100
results[name] = SLAStatus(
definition=self._sla_definitions[name],
current_value=0.0, # Would need last observed value
compliant=last_100[-1] if last_100 else True,
compliance_percentage=compliance_pct,
violation_count=sum(1 for c in last_100 if not c),
)
return results
Per-Agent Cost Attribution
In a multi-agent fleet, you need to know which agents are profitable and which are burning money. Cost attribution tracks every expense back to the agent that incurred it.
@dataclass
class CostEntry:
"""A single cost entry attributed to an agent."""
agent_id: str
category: str
amount: float
currency: str
timestamp: float
transaction_id: str = ""
description: str = ""class CostAttributor:
"""Tracks and attributes costs across an agent fleet."""
def __init__(self, metrics_collector: "MetricsCollector"):
self._metrics = metrics_collector
self._costs: Dict[str, List[CostEntry]] = defaultdict(list)
self._budgets: Dict[str, float] = {}
self._budget_callbacks: List = []
def record_cost(
self,
agent_id: str,
category: str,
amount: float,
currency: str = "credits",
transaction_id: str = "",
description: str = "",
):
"""Record a cost attributed to an agent."""
entry = CostEntry(
agent_id=agent_id,
category=category,
amount=amount,
currency=currency,
timestamp=time.time(),
transaction_id=transaction_id,
description=description,
)
self._costs[agent_id].append(entry)
self._metrics.record_cost(amount, category, currency)
self._metrics.increment(
"cost_attribution.total",
amount,
{"agent": agent_id, "category": category},
)
# Check budget
if agent_id in self._budgets:
total = self.get_agent_total(agent_id)
budget = self._budgets[agent_id]
utilization = total / budget if budget > 0 else 0
self._metrics.gauge_set(
"budget.utilization_pct",
utilization * 100,
{"agent": agent_id},
)
if utilization >= 0.9:
for callback in self._budget_callbacks:
try:
callback(agent_id, total, budget, utilization)
except Exception:
pass
def set_budget(self, agent_id: str, budget: float):
"""Set a spending budget for an agent."""
self._budgets[agent_id] = budget
def on_budget_warning(self, callback):
"""Register a callback for budget threshold warnings."""
self._budget_callbacks.append(callback)
def get_agent_total(
self,
agent_id: str,
since: float = None,
) -> float:
"""Get total costs for an agent since a given time."""
costs = self._costs.get(agent_id, [])
if since:
costs = [c for c in costs if c.timestamp >= since]
return sum(c.amount for c in costs)
def get_agent_breakdown(
self,
agent_id: str,
since: float = None,
) -> Dict[str, float]:
"""Get cost breakdown by category for an agent."""
costs = self._costs.get(agent_id, [])
if since:
costs = [c for c in costs if c.timestamp >= since]
breakdown = defaultdict(float)
for cost in costs:
breakdown[cost.category] += cost.amount
return dict(breakdown)
def generate_monthly_report(
self,
month_start: float,
month_end: float,
) -> Dict[str, Any]:
"""Generate a monthly cost attribution report."""
report = {
"period_start": month_start,
"period_end": month_end,
"agents": {},
"total_cost": 0.0,
"top_categories": defaultdict(float),
}
for agent_id, costs in self._costs.items():
month_costs = [
c for c in costs
if month_start <= c.timestamp <= month_end
]
if not month_costs:
continue
agent_total = sum(c.amount for c in month_costs)
breakdown = defaultdict(float)
for cost in month_costs:
breakdown[cost.category] += cost.amount
report["top_categories"][cost.category] += cost.amount
budget = self._budgets.get(agent_id, 0)
report["agents"][agent_id] = {
"total_cost": agent_total,
"budget": budget,
"utilization_pct": (agent_total / budget * 100) if budget else 0,
"transaction_count": len(month_costs),
"breakdown": dict(breakdown),
}
report["total_cost"] += agent_total
report["top_categories"] = dict(
sorted(
report["top_categories"].items(),
key=lambda x: x[1],
reverse=True,
)
)
return report
Putting It All Together
Here is how all the observability components integrate for a production agent deployment:
from greenhelix import GreenHelixClientInitialize core components
client = GreenHelixClient(api_key="your-api-key")
tracer = AgentTracer(agent_id="agent-marketplace", client=client)
metrics = MetricsCollector(agent_id="agent-marketplace", client=client)
detector = AnomalyDetector()
alert_manager = AlertManager(
agent_id="agent-marketplace", client=client, anomaly_detector=detector
)
sla_monitor = SLAMonitor(
agent_id="agent-marketplace", metrics_collector=metrics, client=client
)
cost_tracker = CostAttributor(metrics_collector=metrics)Define SLAs
sla_monitor.define_sla(SLADefinition(
name="availability",
metric_name="agent.availability_pct",
target_value=99.9,
comparison="gte",
measurement_window="24h",
))sla_monitor.define_sla(SLADefinition(
name="latency_p95",
metric_name="transactions.duration_ms",
target_value=500,
comparison="lte",
measurement_window="1h",
))
Set budgets
cost_tracker.set_budget("agent-marketplace", budget=1000.0)
cost_tracker.set_budget("agent-data-provider", budget=500.0)Wire SLA violations to alerts
sla_monitor.on_violation(lambda status: alert_manager.evaluate(
status.definition.metric_name,
status.current_value,
{"sla": status.definition.name},
))Wire budget warnings to alerts
cost_tracker.on_budget_warning(
lambda agent, total, budget, util: alert_manager.evaluate(
"budget.utilization_pct",
util * 100,
{"agent": agent},
)
)Start background threads
metrics.start()
alert_manager.start()Now every transaction is automatically traced, metered, checked,
and alerted on:
def handle_transaction(request):
with tracer.trace_agent_call(
request["target"], request["operation"]
) as span:
start = time.time()
result = process(request)
duration_ms = (time.time() - start) * 1000 # Metrics
metrics.record_transaction(
operation=request["operation"],
duration_ms=duration_ms,
amount=request.get("amount", 0),
status="success" if result["ok"] else "error",
peer_agent=request["target"],
)
# Anomaly detection + alerting
alert_manager.evaluate("transactions.duration_ms", duration_ms)
alert_manager.evaluate(
"transactions.errors",
0 if result["ok"] else 1,
)
# SLA check
sla_monitor.check_sla("latency_p95", duration_ms)
# Cost attribution
cost_tracker.record_cost(
agent_id="agent-marketplace",
category="gateway_fee",
amount=result.get("fee", 0),
transaction_id=result.get("transaction_id", ""),
)
return result
This integration means every transaction flowing through your agent automatically generates traces for debugging, metrics for dashboards, anomaly detection for automated problem finding, alerts for operator notification, SLA tracking for compliance reporting, and cost attribution for business analysis. The operational overhead is minimal -- a few milliseconds of in-process buffering per transaction -- and the visibility gain is transformative.
What's Next
This guide built the observability stack from the bottom up: tracing, metrics, anomaly detection, dashboards, alerting, and SLA monitoring. Each component works independently but delivers maximum value when integrated.
To put this into production, start with the AgentTracer and MetricsCollector. These two components give you immediate visibility with minimal effort. Add the AnomalyDetector once you have a few days of baseline data. Configure the AlertManager with conservative thresholds and tighten them as you learn your system's normal patterns. Roll out SLAMonitor and CostAttributor when you have paying customers who need compliance reports.
The next guides in this series cover agent security hardening (protecting your fleet from adversarial agents), agent marketplace economics (pricing strategies and commission structures), and agent scaling patterns (handling 100x traffic growth without rewriting your stack). Each builds on the observability foundation established here because you cannot secure, price, or scale what you cannot see.