Observability AI Agents

A production-grade multi-agent observability system built with Google ADK that orchestrates specialized AI agents to monitor, query, and troubleshoot infrastructure across Prometheus, Grafana, Kubernetes, and synthetic monitoring systems.

┌─────────────────────────────────────────────────────────────┐
│                    User Query                                │
│              "Why is app-server-01 slow?"                    │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│              Observability Orchestrator                       │
│         (SequentialAgent — routes & synthesizes)              │
│                                                              │
│  ┌──────────┐   ┌──────────────────────────┐  ┌──────────┐  │
│  │  Router   │──▶│     ParallelAgent        │──▶│Synthesizer│ │
│  │ (LLM)    │   │                          │  │ (LLM)    │  │
│  └──────────┘   │  ┌────────┐ ┌─────────┐  │  └──────────┘  │
│                  │  │Prometh.│ │ Grafana │  │                │
│                  │  │ Agent  │ │  Agent  │  │                │
│                  │  └────────┘ └─────────┘  │                │
│                  │  ┌────────┐ ┌─────────┐  │                │
│                  │  │  K8s   │ │Synthetic│  │                │
│                  │  │ Agent  │ │Monitor  │  │                │
│                  │  └────────┘ └─────────┘  │                │
│                  └──────────────────────────┘                │
└─────────────────────────────────────────────────────────────┘

Features

• Multi-agent orchestration — A router LLM dispatches queries to the right specialist agents, which run in parallel, and a synthesizer combines results into a unified answer
• Two-phase cache/filter pattern — Large query results are cached server-side; the LLM then drills down with filter tools, keeping context windows small and responses fast
• Prometheus agent — Multi-instance PromQL queries across environments with DNS-aware hostname↔IP resolution
• Grafana agent — Dashboard search, Loki log queries (LogQL), alert management, incidents, and on-call via MCP
• Kubernetes agent — Multi-cluster kubectl operations via MCP with custom-column support, cross-cluster search, and advanced filtering (age, version, numeric)
• Synthetic monitoring agent — SSH-based health checks (disk, CPU, memory, NTP, DNS, certificates, services) with historical log search
• FastAPI server — Production API with streaming SSE, async job queue with concurrency control, session management, and API key auth
• Fully local LLM — Runs on Ollama via LiteLLM; no cloud API keys required

Quick Start

Prerequisites

• Python 3.11+
• Ollama with a model pulled (e.g., ollama pull qwen3.5:9b)
• Node.js (for MCP servers: mcp-server-kubernetes, mcp-grafana)

Install

git clone https://github.com/YOUR_USERNAME/observability-ai-agents.git
cd observability-ai-agents

python -m venv .venv && source .venv/bin/activate
pip install -e .

Configure

cp .env.example .env
# Edit .env with your Prometheus URLs, Grafana token, kubeconfig path, etc.

Run

# Start the API server
python server.py

# Or use uvicorn directly
uvicorn server:app --reload --port 8000

Query

# Simple query (synchronous — blocks until response)
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"query": "are all targets up across all prometheus instances?"}'

# Streaming (SSE)
curl -N -X POST http://localhost:8000/query/stream \
  -H "Content-Type: application/json" \
  -d '{"query": "check CPU and memory for app-server-01"}'

# Async job queue (submit → poll → retrieve)
JOB=$(curl -s -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"query": "any pods in CrashLoopBackOff across all clusters?"}')
JOB_ID=$(echo $JOB | python3 -c "import sys,json; print(json.load(sys.stdin)['job_id'])")
curl -s http://localhost:8000/jobs/$JOB_ID          # poll status
curl -s http://localhost:8000/jobs/$JOB_ID/result    # get result when done

Run with ADK CLI

# Use Google ADK's built-in dev UI
adk web agents/observability_agent

Architecture

Component	Type	Purpose
Observability Agent	SequentialAgent	Top-level orchestrator: route → parallel execute → synthesize
Query Router	LlmAgent	Classifies queries and sets agent dispatch flags in state
Parallel Executor	ParallelAgent	Runs all specialist agents concurrently
Prometheus Agent	LlmAgent + FunctionTool	Multi-instance PromQL with two-phase cache/filter
Grafana Agent	LlmAgent + McpToolset	Dashboards, Loki logs, alerts, incidents via MCP
K8s Agent	LlmAgent + McpToolset + FunctionTool	Multi-cluster kubectl with MCP + custom tools
Synthetic Monitoring Agent	LlmAgent + FunctionTool	SSH-based health checks with historical search
Synthesizer	LlmAgent	Correlates and presents unified results

Two-Phase Cache/Filter Pattern

All data-heavy agents use the same pattern to stay within LLM context limits:

1. Phase 1 — Query: Tool executes the query, caches full results server-side, returns only a schema (column names + row counts)
2. Phase 2 — Filter: LLM calls filter_cached_results or filter_cached_advanced to drill into specific rows by text match or numeric comparison

This lets agents handle thousands of Prometheus targets, pods, or log lines without blowing up the context window.

Project Structure

├── server.py                    # FastAPI server with SSE streaming
├── prompt_loader.py             # YAML prompt loader with caching
├── agents/
│   ├── observability_agent/     # Root orchestrator (Sequential)
│   ├── prometheus_mcp_agent/    # Multi-instance Prometheus
│   ├── grafana_mcp_agent/       # Grafana + Loki via MCP
│   ├── k8s_mcp_agent/           # Multi-cluster Kubernetes via MCP
│   └── synthetic_monitoring_agent/ # SSH-based health checks
├── prompts/                     # YAML prompt templates per agent
├── .env.example                 # Environment variable template
├── pyproject.toml               # Project metadata & dependencies
└── LICENSE

API Reference

All endpoints accept x-api-key header or api_key query parameter when API_KEY is configured. Auto-generated OpenAPI docs available at http://localhost:8000/docs.

Health & Info

Method	Endpoint	Description
GET	/health	Returns {"status": "ok"}
GET	/agents	Lists root agent and all sub-agents

Synchronous Query

POST /query — Blocks until the agent pipeline completes.

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"query": "which servers have memory above 90%?", "session_id": "optional-for-continuity"}'

Response (200):

{"session_id": "abc-123", "response": "Two servers exceed 90% memory..."}

Streaming Query (SSE)

POST /query/stream — Server-Sent Events stream with per-agent chunks.

curl -N -X POST http://localhost:8000/query/stream \
  -H "Content-Type: application/json" \
  -d '{"query": "check disk usage across all clusters"}'

Each SSE event:

event: agent_response
data: {"agent": "prometheus_mcp_agent", "text": "Querying...", "final": false}

event: done
data: {"session_id": "abc-123"}

Async Job Queue

For long-running queries, submit a job and poll for results. The server runs at most MAX_CONCURRENT_JOBS (default 2) in parallel — excess submissions are queued.

Submit a job

POST /jobs → returns 202 Accepted immediately.

curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"query": "compare CPU and memory across all clusters"}'

Response (202):

{"job_id": "550e8400-...", "status": "queued"}

Poll job status

GET /jobs/{job_id}

curl http://localhost:8000/jobs/550e8400-...

Response (200):

{
  "job_id": "550e8400-...",
  "status": "running",
  "query": "compare CPU and memory across all clusters",
  "session_id": "...",
  "created_at": 1745849600.0,
  "started_at": 1745849601.2,
  "finished_at": null,
  "queue_position": null
}

queue_position is set when status is queued — indicates how many jobs are ahead in the queue.

Get job result

GET /jobs/{job_id}/result

• Returns 202 with Retry-After: 5 header if the job is still running.
• Returns 200 with the response once completed.
• Returns 200 with an error field if the job failed.

curl http://localhost:8000/jobs/550e8400-.../result

Response when completed (200):

{
  "job_id": "550e8400-...",
  "status": "completed",
  "session_id": "...",
  "response": "CPU usage across clusters: US-EAST-1: 45%...",
  "error": null
}

Response when still running (202):

{"job_id": "550e8400-...", "status": "running", "detail": "Job is still running"}

List all jobs

GET /jobs — optional query params: ?status=queued|running|completed|failed&limit=50

curl http://localhost:8000/jobs?status=completed&limit=10

Response (200):

[
  {"job_id": "...", "status": "completed", "query": "compare CPU and mem...", "created_at": 1745849600.0}
]

Authentication

When API_KEY is set in .env, all endpoints (except /health) require authentication:

curl -H "x-api-key: YOUR_KEY" http://localhost:8000/jobs
# or
curl "http://localhost:8000/jobs?api_key=YOUR_KEY"

Unauthenticated requests return 401.

Configuration

All configuration is via environment variables (see .env.example):

Variable	Description	Default
ADK_MODEL	LLM model identifier	openai/qwen3.5:9b
LITELLM_API_BASE	LiteLLM/Ollama endpoint	http://localhost:11434/v1
PROMETHEUS_INSTANCES	Comma-separated name=url pairs	—
GRAFANA_URL	Grafana instance URL	—
GRAFANA_TOKEN	Grafana service account token	—
KUBECONFIG	Path to multi-cluster kubeconfig	—
SYNTH_MONITOR_*	Synthetic monitoring config	—
API_KEY	API server auth key (optional)	—
MAX_CONCURRENT_JOBS	Max parallel agent jobs	2
MAX_STORED_JOBS	Max jobs kept in memory	1000
JOB_TTL_SECONDS	Evict finished jobs after (seconds)	3600

License

Apache 2.0