connected-car-mcp

An MCP (Model Context Protocol) server over a synthetic connected-vehicle fleet: telemetry, rule-based anomaly detection, and maintenance recommendations, exposed as five narrow tools instead of one open-ended query interface.

Built as a small, self-contained illustration of a specific design habit: deciding what belongs behind a tool boundary, and logging every call across it. Runs entirely on synthetic data generated locally — no external API, no account, no proprietary source.

Why it's shaped this way

The whole dataset could be exposed through a single run_query(sql: str) tool. That's the wrong shape for an agent to call reliably: it pushes schema-learning onto the model at call time, and there's no way to scope or audit "what can be asked" per capability. Instead:

Tool	Contract
list_vehicles	Enumerate the fleet
get_vehicle_telemetry	Raw readings for one vehicle, time-bounded
fleet_health_summary	Latest snapshot + fleet averages
detect_anomalies	Rule-based flags: overheating, low battery, fault codes, harsh driving
get_maintenance_recommendations	Prioritized actions for one vehicle

A model composes these — summary → pick a flagged vehicle → pull its telemetry → get a recommendation — rather than writing free-form queries against raw rows. It also makes the audit story trivial: there are only five well-defined calls to log, so audit_log.jsonl (see connected_car_mcp/audit.py) is one line per call — timestamp, tool, arguments, duration, success/failure — with no custom logic per tool. A production deployment would emit the same record as structured logs via MCPServer's middleware hook (which sees every JSON-RPC call, tool or resource) rather than a local file; the decorator here keeps the demo runnable with zero extra infrastructure.

Anomaly thresholds are simple and explainable (engine_temp_c >= 110, not a trained model) on purpose — a fleet monitor's flags need to be auditable by a human, not just accurate.

Data

data/generate_telemetry.py generates a deterministic (fixed-seed), fully synthetic dataset: 12 vehicles, readings every 10 minutes over 3 days. A few vehicles are seeded with faults so the anomaly detector has real signal to find:

• CCV-004 — engine temperature ramps into critical range (cooling system failure)
• CCV-009 — battery voltage degrades over time (failing battery/alternator)
• CCV-002, CCV-011 — intermittent DTC fault codes
• CCV-006 — occasional harsh-driving speed spikes

data/telemetry.csv is committed so the repo runs immediately; regenerate it with:

python data/generate_telemetry.py

Running it

python -m venv .venv
.venv/Scripts/activate        # .venv/bin/activate on macOS/Linux
pip install -r requirements.txt

python -m connected_car_mcp.server   # starts the MCP server over stdio

To try it from Claude Desktop or another MCP client, point it at the module with cwd set to the repo root, e.g. in claude_desktop_config.json:

{
  "mcpServers": {
    "connected-car-fleet": {
      "command": "python",
      "args": ["-m", "connected_car_mcp.server"],
      "cwd": "/path/to/connected-car-mcp"
    }
  }
}

Then ask something like "Which vehicles in the fleet need attention right now, and why?" — the model will call fleet_health_summary, follow up with detect_anomalies on the flagged vehicles, and can call get_maintenance_recommendations to turn that into next actions.

Tests

pip install pytest
pytest tests/

Covers the data layer directly (fleet size, unknown-vehicle handling, and that the seeded faults actually get flagged) rather than round-tripping through the MCP protocol layer.

Project layout

connected_car_mcp/
  server.py       MCP tool + resource definitions
  data_store.py   Query layer over the telemetry CSV (pandas)
  audit.py        Per-call audit log decorator
data/
  generate_telemetry.py   Synthetic dataset generator
  telemetry.csv            Generated dataset (committed)
tests/
  test_data_store.py

License

MIT — see LICENSE.