# fleet-data-mcp

**A Model Context Protocol (MCP) server that connects Claude Code to fleet/network telemetry data (device health, utilization forecasting, and incident history) via a small set of narrow, purpose-built tools.**

[![tests](https://github.com/futurerichdad/fleet-data-mcp/actions/workflows/tests.yml/badge.svg)](https://github.com/futurerichdad/fleet-data-mcp/actions)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)]()

## Why this exists

Enterprise infrastructure teams sit on huge amounts of fleet/platform telemetry that never makes it in front of an LLM agent, because there's no bridge between "the data lives in a database" and "the model can reason about it." This project is that bridge: a small MCP server exposing device health, anomaly detection, capacity forecasting, and incident history as tools an agent like Claude Code can call directly, in natural conversation.

Built after 9+ years running enterprise platform and fleet telemetry operations at scale ($5M+ platform budgets, fleet-wide observability). The tool contracts here reflect real operational questions ("which devices are about to breach capacity," "what's degrading," "what happened last time this alarm fired"), not a toy demo schema.

## What it does

| Tool | Purpose |
|---|---|
| `get_fleet_health_summary` | Aggregate fleet (or per-region) health: avg signal quality, utilization, temperature, open incidents |
| `query_device_anomalies` | Flags devices with high utilization or sharply degrading signal quality |
| `forecast_capacity_breach` | Linear-trend projection of when a device will breach 95% utilization |
| `search_incident_history` | Query the incident log by device and/or severity |

All tools return structured JSON with consistent error handling (an `error` + `hint` shape the model can act on, never a raw stack trace). See `server.py` for the full docstrings/schemas.

## Data

The server runs against a local SQLite database of **fully synthetic** fleet telemetry (device inventory, time-series health readings, incident log), generated by `data/generate_telemetry.py`. No real company data is used, referenced, or required anywhere in this repo. The generator models realistic patterns (a subset of devices trending toward failure, utilization correlated with incidents, etc.) so the tools have real anomalies to find.

Swap the query layer in `server.py` for a real telemetry store (Postgres, TimescaleDB, a REST API) to run this against production data. The tool contracts and MCP wiring don't change.

## Quickstart

```bash
git clone https://github.com/futurerichdad/fleet-data-mcp.git
cd fleet-data-mcp
pip install -r requirements.txt

# Generate the synthetic dataset (writes fleet.db)
python data/generate_telemetry.py --out fleet.db --devices 250 --days 30

# Run the test suite
pytest tests/ -v

# Run the server (stdio transport)
python server.py
```

## Connecting to Claude Code

Add to your Claude Code MCP config (`claude mcp add` or your `mcp.json`):

```json
{
  "mcpServers": {
    "fleet-data": {
      "command": "python",
      "args": ["/absolute/path/to/fleet-data-mcp/server.py"],
      "env": {
        "FLEET_DB_PATH": "/absolute/path/to/fleet-data-mcp/fleet.db"
      }
    }
  }
}
```

Then try the prompts in [`examples/example_prompts.md`](examples/example_prompts.md), for example: *"Find the fleet's most at-risk device and forecast when it'll breach capacity."* A good run chains `query_device_anomalies` into `forecast_capacity_breach` without you naming a device ID yourself.

## Architecture

```
Claude Code  --MCP (stdio)-->  server.py (MCPServer, 4 tools)
                                     |
                                     v
                              fleet.db (SQLite)
                         devices / telemetry_readings / incidents
                         (generated by data/generate_telemetry.py)
```

## Design philosophy

- **Narrow tools over mega-tools.** Each tool answers one operational question. None takes a dozen optional parameters and tries to be everything.
- **Structured, inspectable output.** Every tool returns JSON, never prose, so downstream evaluation (see [Apex Overlay](https://github.com/futurerichdad/apex-overlay)) can score correctness mechanically.
- **Errors the model can act on.** Failures return `{"error": ..., "hint": ...}` so an agent can self-correct (e.g., retry with a valid region) instead of dead-ending.
- **Storage-agnostic contracts.** The tool signatures don't leak SQLite-isms, so swapping in a production data store is a query-layer change, not a redesign.

## Testing

```bash
pytest tests/ -v
```

Tests use an isolated, deterministic SQLite fixture (not the generated `fleet.db`), so they're fast and don't depend on random data generation. Covers happy-path queries, invalid-input handling (bad region, bad severity, unknown device), and missing-database recovery.

## Roadmap

- [ ] Streamable-HTTP transport for remote/multi-client deployments
- [ ] Swap SQLite for a real time-series backend (TimescaleDB) behind the same tool contracts
- [ ] Integration with [Apex Overlay](https://github.com/futurerichdad/apex-overlay) for automated tool-call evaluation against this server

## About

Built by Ben, 9+ years of enterprise platform and fleet telemetry engineering (AT&T), now building agentic AI tooling. Anthropic Certified: Building with Claude API, Model Context Protocol. See also: [Apex Overlay](https://github.com/futurerichdad/apex-overlay) (agentic eval harness) and the companion Claude Code execution transcript case study.

## License

MIT. See [LICENSE](LICENSE).