A Beginner's Guide to Writing Your First Multi-Agent Pipeline Health Check Endpoint
You've built your first multi-agent pipeline. Maybe it's a research agent that hands off summaries to a writer agent, which then passes formatted output to a publisher agent. It works beautifully in your local environment. Then you deploy it, walk away, and come back to find it silently broken at 2 AM with zero visibility into what went wrong.
This is the silent failure problem that plagues most early multi-agent systems. Unlike a traditional REST API where a 500 status code screams for attention, agent pipelines fail in subtle, expensive ways: an agent stalls waiting for a dependency that timed out, a token budget gets exhausted mid-pipeline, or a downstream tool becomes unavailable while the orchestrator keeps retrying indefinitely.
The fix is surprisingly simple: a single, well-designed health check endpoint that gives you a real-time snapshot of every agent's status, every dependency's availability, and your remaining token budget, all in one API response. In this guide, you'll learn exactly how to build one from scratch, even if you've never written a monitoring endpoint before.
Why Standard Health Checks Fall Short for Agent Pipelines
Traditional health check endpoints answer one question: "Is the server running?" They return a simple 200 OK with maybe a {"status": "ok"} body. That's fine for a stateless microservice, but multi-agent pipelines are anything but stateless.
Consider what a meaningful health check actually needs to capture in a multi-agent context:
- Agent-level status: Is each individual agent idle, running, waiting, or in a failed state?
- Dependency availability: Are the external tools, vector databases, APIs, and model providers that agents rely on actually reachable?
- Token budget remaining: How much of your allocated token budget has been consumed, and is there enough left to complete pending work?
- Pipeline topology: Which agents are blocked waiting on upstream agents?
- Error context: If something failed, what failed and why?
A simple ping-style health check misses all of this. By the time your monitoring system flags the server as "down," your pipeline has likely already burned thousands of tokens and produced corrupted output. The goal is to catch problems before they become catastrophic failures.
Designing the Health Check Response Schema
Before writing a single line of code, design your response schema. A well-structured schema is the foundation of a useful health check. Here's a schema that covers all the essentials for a beginner-level implementation:
{
"status": "healthy" | "degraded" | "unhealthy",
"timestamp": "2026-06-15T10:32:00Z",
"pipeline": {
"id": "research-writer-pipeline",
"version": "1.0.0",
"uptime_seconds": 3842
},
"agents": [
{
"id": "research-agent",
"status": "idle" | "running" | "waiting" | "failed",
"last_active": "2026-06-15T10:31:55Z",
"current_task": null,
"error": null
}
],
"dependencies": [
{
"name": "openai-api",
"type": "model_provider",
"status": "available" | "degraded" | "unavailable",
"latency_ms": 124,
"last_checked": "2026-06-15T10:31:58Z"
}
],
"token_budget": {
"total_allocated": 500000,
"consumed": 187432,
"remaining": 312568,
"percent_used": 37.5,
"burn_rate_per_minute": 4200,
"estimated_minutes_remaining": 74
}
}
Notice the top-level status field. This is your rollup status: the single value your monitoring system will alert on. The rule is simple:
- healthy: All agents are idle or running normally, all dependencies are available, and the token budget is above a safe threshold (for example, more than 20% remaining).
- degraded: At least one dependency is slow or intermittent, or a non-critical agent has failed, but the pipeline can still make progress.
- unhealthy: A critical agent has failed, a required dependency is down, or the token budget is critically low (under 5% remaining).
Setting Up the Project Structure
For this guide, we'll use Python with FastAPI, which is the most common framework for building AI agent backends in 2026. You'll also need a lightweight way to track agent state. We'll use a simple in-memory state store to keep things approachable.
Install the dependencies:
pip install fastapi uvicorn httpx python-dotenv
Here's the project layout we'll build toward:
my_agent_pipeline/
├── main.py
├── agents/
│ ├── __init__.py
│ ├── registry.py # Agent state registry
│ └── base_agent.py # Base agent class
├── health/
│ ├── __init__.py
│ ├── router.py # Health check endpoint
│ ├── checks.py # Individual check functions
│ └── models.py # Pydantic response models
└── config.py # Token budget configuration
Step 1: Build the Agent State Registry
The registry is the source of truth for all agent statuses. Every agent in your pipeline reports its state to this central store, and the health check reads from it. Keep it simple to start.
In agents/registry.py:
from datetime import datetime, timezone
from typing import Dict, Optional
from dataclasses import dataclass, field
from enum import Enum
class AgentStatus(str, Enum):
IDLE = "idle"
RUNNING = "running"
WAITING = "waiting"
FAILED = "failed"
@dataclass
class AgentState:
agent_id: str
status: AgentStatus = AgentStatus.IDLE
last_active: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
current_task: Optional[str] = None
error: Optional[str] = None
class AgentRegistry:
"""Central store for all agent states in the pipeline."""
def __init__(self):
self._agents: Dict[str, AgentState] = {}
def register(self, agent_id: str) -> None:
"""Register a new agent with the pipeline."""
self._agents[agent_id] = AgentState(agent_id=agent_id)
def update(
self,
agent_id: str,
status: AgentStatus,
current_task: Optional[str] = None,
error: Optional[str] = None,
) -> None:
"""Update an agent's current state."""
if agent_id not in self._agents:
self.register(agent_id)
state = self._agents[agent_id]
state.status = status
state.last_active = datetime.now(timezone.utc)
state.current_task = current_task
state.error = error
def get_all(self) -> list[AgentState]:
return list(self._agents.values())
def get(self, agent_id: str) -> Optional[AgentState]:
return self._agents.get(agent_id)
# Singleton instance shared across the application
agent_registry = AgentRegistry()
With this registry in place, any agent in your pipeline can report its status with a single call:
from agents.registry import agent_registry, AgentStatus
agent_registry.update("research-agent", AgentStatus.RUNNING, current_task="Fetching arxiv papers")
Step 2: Build the Token Budget Tracker
Token budget tracking is the feature most beginners skip, and it's the one that saves you the most money. The idea is straightforward: you allocate a token budget for a pipeline run (or for a rolling time window), track consumption as agents make model calls, and expose the remaining budget in your health check.
In config.py:
import time
from threading import Lock
class TokenBudget:
"""Thread-safe token budget tracker for the pipeline."""
def __init__(self, total_allocated: int):
self.total_allocated = total_allocated
self._consumed = 0
self._lock = Lock()
self._start_time = time.time()
self._consumption_history: list[tuple[float, int]] = [] # (timestamp, tokens)
def consume(self, tokens: int) -> None:
"""Record token consumption from an agent call."""
with self._lock:
self._consumed += tokens
self._consumption_history.append((time.time(), tokens))
# Keep only the last 5 minutes of history for burn rate calculation
cutoff = time.time() - 300
self._consumption_history = [
(t, v) for t, v in self._consumption_history if t >= cutoff
]
@property
def consumed(self) -> int:
return self._consumed
@property
def remaining(self) -> int:
return max(0, self.total_allocated - self._consumed)
@property
def percent_used(self) -> float:
return round((self._consumed / self.total_allocated) * 100, 2)
@property
def burn_rate_per_minute(self) -> float:
"""Calculate tokens consumed per minute over the last 5 minutes."""
if not self._consumption_history:
return 0.0
window_tokens = sum(v for _, v in self._consumption_history)
elapsed_minutes = min(5.0, (time.time() - self._start_time) / 60)
if elapsed_minutes == 0:
return 0.0
return round(window_tokens / elapsed_minutes, 2)
@property
def estimated_minutes_remaining(self) -> Optional[float]:
rate = self.burn_rate_per_minute
if rate == 0:
return None
return round(self.remaining / rate, 1)
# Initialize with your desired budget (adjust to your use case)
token_budget = TokenBudget(total_allocated=500_000)
Now, every time an agent makes a model call, it reports token usage:
from config import token_budget
# After receiving a response from your model provider
token_budget.consume(response.usage.total_tokens)
Step 3: Write the Dependency Checks
Dependency checks probe your external services and report their availability and latency. The key principle here is: keep each check fast and non-blocking. A health check that takes 10 seconds to respond because it's waiting on a slow dependency is worse than useless.
In health/checks.py:
import httpx
import asyncio
import time
from datetime import datetime, timezone
from typing import Optional
async def check_http_dependency(
name: str,
url: str,
timeout_seconds: float = 3.0,
) -> dict:
"""Probe an HTTP-based dependency and return its status."""
start = time.monotonic()
last_checked = datetime.now(timezone.utc).isoformat()
try:
async with httpx.AsyncClient(timeout=timeout_seconds) as client:
response = await client.get(url)
latency_ms = round((time.monotonic() - start) * 1000, 2)
if response.status_code < 500:
status = "available" if latency_ms < 1000 else "degraded"
else:
status = "unavailable"
except httpx.TimeoutException:
latency_ms = round(timeout_seconds * 1000, 2)
status = "unavailable"
except Exception as e:
latency_ms = None
status = "unavailable"
return {
"name": name,
"status": status,
"latency_ms": latency_ms,
"last_checked": last_checked,
}
async def run_all_dependency_checks() -> list[dict]:
"""Run all dependency checks concurrently."""
checks = await asyncio.gather(
check_http_dependency(
name="openai-api",
url="https://api.openai.com/v1/models",
),
check_http_dependency(
name="vector-db",
url="http://localhost:6333/healthz", # Example: Qdrant health endpoint
),
check_http_dependency(
name="redis-cache",
url="http://localhost:6379/ping",
),
return_exceptions=True,
)
# Filter out any unexpected exceptions and replace with error entries
return [
c if isinstance(c, dict) else {"name": "unknown", "status": "unavailable", "latency_ms": None, "last_checked": datetime.now(timezone.utc).isoformat()}
for c in checks
]
Notice the use of asyncio.gather. This runs all dependency checks in parallel, so if you have five dependencies each taking up to three seconds to time out, your health check still responds in roughly three seconds rather than fifteen. This is critical for keeping your monitoring system responsive.
Step 4: Assemble the Health Check Endpoint
Now bring everything together in the FastAPI router. This is where the rollup logic lives: computing the top-level status from all the individual checks.
In health/router.py:
from fastapi import APIRouter
from datetime import datetime, timezone
import time
from agents.registry import agent_registry, AgentStatus
from health.checks import run_all_dependency_checks
from config import token_budget
router = APIRouter()
# Track pipeline start time
_pipeline_start_time = time.time()
def compute_rollup_status(
agent_states: list,
dependency_results: list,
budget_percent_remaining: float,
) -> str:
"""Determine the overall pipeline health status."""
# Check for any failed agents
failed_agents = [a for a in agent_states if a.status == AgentStatus.FAILED]
unavailable_deps = [d for d in dependency_results if d["status"] == "unavailable"]
degraded_deps = [d for d in dependency_results if d["status"] == "degraded"]
# Critically low token budget
if budget_percent_remaining < 5.0:
return "unhealthy"
# Any critical failure
if failed_agents or unavailable_deps:
return "unhealthy"
# Degraded conditions
if degraded_deps or budget_percent_remaining < 20.0:
return "degraded"
return "healthy"
@router.get("/health", tags=["Monitoring"])
async def health_check():
"""
Returns a comprehensive health snapshot of the multi-agent pipeline,
including agent statuses, dependency availability, and token budget.
"""
# Gather all data concurrently where possible
dependency_results = await run_all_dependency_checks()
agent_states = agent_registry.get_all()
uptime_seconds = int(time.time() - _pipeline_start_time)
# Build the response
agents_payload = [
{
"id": state.agent_id,
"status": state.status.value,
"last_active": state.last_active.isoformat(),
"current_task": state.current_task,
"error": state.error,
}
for state in agent_states
]
budget_percent_remaining = round(
100.0 - token_budget.percent_used, 2
)
rollup_status = compute_rollup_status(
agent_states, dependency_results, budget_percent_remaining
)
return {
"status": rollup_status,
"timestamp": datetime.now(timezone.utc).isoformat(),
"pipeline": {
"id": "my-agent-pipeline",
"version": "1.0.0",
"uptime_seconds": uptime_seconds,
},
"agents": agents_payload,
"dependencies": dependency_results,
"token_budget": {
"total_allocated": token_budget.total_allocated,
"consumed": token_budget.consumed,
"remaining": token_budget.remaining,
"percent_used": token_budget.percent_used,
"percent_remaining": budget_percent_remaining,
"burn_rate_per_minute": token_budget.burn_rate_per_minute,
"estimated_minutes_remaining": token_budget.estimated_minutes_remaining,
},
}
Step 5: Wire It Into Your FastAPI Application
In your main.py, register the agents and mount the health router:
from fastapi import FastAPI
from health.router import router as health_router
from agents.registry import agent_registry
app = FastAPI(title="My Multi-Agent Pipeline")
# Register your pipeline agents on startup
@app.on_event("startup")
async def startup():
agent_registry.register("research-agent")
agent_registry.register("writer-agent")
agent_registry.register("publisher-agent")
# Mount the health check router
app.include_router(health_router)
# Your agent pipeline routes go here
# app.include_router(pipeline_router)
Run your app and visit http://localhost:8000/health to see your full pipeline health snapshot.
What a Real Response Looks Like
Here's an example of what the endpoint returns when the pipeline is in a degraded state due to a slow vector database and a low token budget:
{
"status": "degraded",
"timestamp": "2026-06-15T14:22:10Z",
"pipeline": {
"id": "my-agent-pipeline",
"version": "1.0.0",
"uptime_seconds": 7241
},
"agents": [
{
"id": "research-agent",
"status": "idle",
"last_active": "2026-06-15T14:22:05Z",
"current_task": null,
"error": null
},
{
"id": "writer-agent",
"status": "running",
"last_active": "2026-06-15T14:22:09Z",
"current_task": "Drafting section 3 of 5",
"error": null
},
{
"id": "publisher-agent",
"status": "waiting",
"last_active": "2026-06-15T14:21:50Z",
"current_task": null,
"error": null
}
],
"dependencies": [
{
"name": "openai-api",
"status": "available",
"latency_ms": 98,
"last_checked": "2026-06-15T14:22:09Z"
},
{
"name": "vector-db",
"status": "degraded",
"latency_ms": 1840,
"last_checked": "2026-06-15T14:22:10Z"
},
{
"name": "redis-cache",
"status": "available",
"latency_ms": 3,
"last_checked": "2026-06-15T14:22:09Z"
}
],
"token_budget": {
"total_allocated": 500000,
"consumed": 412000,
"remaining": 88000,
"percent_used": 82.4,
"percent_remaining": 17.6,
"burn_rate_per_minute": 5100,
"estimated_minutes_remaining": 17.3
}
}
At a glance, an operator can see: the writer agent is actively working, the vector database is slow (but not down), and the pipeline has roughly 17 minutes of tokens left at the current burn rate. That's actionable intelligence. No digging through logs, no guessing.
Three Beginner Mistakes to Avoid
1. Blocking the Health Check with Slow Probes
Always run dependency checks with a hard timeout and always run them concurrently with asyncio.gather. A health check that takes longer than 5 seconds will be killed by most monitoring systems and load balancers, which will then incorrectly flag your pipeline as down.
2. Exposing the Health Endpoint Without Authentication
Your health check response contains detailed internal topology information. Token budget data, dependency URLs, and agent task descriptions are all sensitive. In production, protect the /health endpoint behind at minimum an API key header check, or restrict it to internal network access only.
3. Forgetting to Update Agent State on Exceptions
The most common reason agent registries become stale is that developers update state at the start of a task but forget to update it in the exception handler. Always use a try/finally block:
agent_registry.update("research-agent", AgentStatus.RUNNING, current_task="Fetching data")
try:
result = await fetch_data()
agent_registry.update("research-agent", AgentStatus.IDLE)
except Exception as e:
agent_registry.update("research-agent", AgentStatus.FAILED, error=str(e))
raise
Extending the Health Check as You Grow
Once you have the basics working, here are the natural next steps for a more production-ready implementation:
- Add a
/health/livevs./health/readysplit: Liveness checks confirm the process is running. Readiness checks confirm the pipeline is ready to accept new work. Kubernetes and most container orchestrators expect this distinction. - Persist health snapshots: Write each health check response to a time-series store (like InfluxDB or Prometheus) so you can visualize trends over time, not just the current moment.
- Add per-agent token accounting: Track how many tokens each individual agent has consumed so you can identify which agent is the biggest budget consumer.
- Integrate with alerting: Wire the rollup status to PagerDuty, Slack, or your preferred alerting tool so that an "unhealthy" response triggers an immediate notification.
- Add a circuit breaker status: If you're using circuit breakers to protect against failing dependencies, expose the circuit state (closed, open, half-open) in the dependency section.
Conclusion
Building a multi-agent pipeline health check endpoint is one of those investments that feels optional until the moment it isn't. The difference between a pipeline that fails silently at 2 AM and burns your entire monthly token budget and one that sends you a "degraded: 17 minutes of tokens remaining" alert is exactly the 90 minutes it takes to implement what you've just read.
The pattern is straightforward: a central agent registry, a thread-safe token budget tracker, parallel dependency probes, and a rollup status computed from all three. Wire them into a single /health endpoint, and you've gone from zero observability to a real-time operational dashboard in a single API call.
Start simple. Get the basic version running today. Then layer in persistence, per-agent token accounting, and alerting integrations as your pipeline grows. The foundation you've built here will scale with you every step of the way.
