canmcpd

A proactive CAN bus daemon with an MCP interface

Status
Research
Version
No tagged release
Platform
Library / CLI
License
MIT
01 · Problem

A GUI sniffer only analyses when someone is looking. canmcpd keeps the analysis running so anomalies accrue whether or not an agent is attached, and questions like "anything anomalous in the last 5 minutes?" are answered from continuous monitoring rather than a cold query.

02 · Architecture

How it is put together.

canmcpd runs on a laptop, owns the CAN adapter, captures continuously and runs standing inference over the bus: a per-ID cadence model, novelty detection and optional DBC-decoded signal checks. It exposes live state and history to any MCP client over Streamable HTTP. Capture, analysis and the HTTP server share one asyncio event loop, so tools read an in-process state object with no IPC.

Capturecanmcpd.pyBusStatecanmcpd.pyAnalysis loopcanmcpd.pyAnomaly logcanmcpd.pyMCP servercanmcpd.pyframes (capture_loop)frames (capture_loop)check_timingcheck_timinganomaliesanomaliestool readstool readsCapturecanmcpd.pyBusStatecanmcpd.pyAnalysis loopcanmcpd.pyAnomaly logcanmcpd.pyMCP servercanmcpd.pyframes (capture_loop)frames (capture_loop)check_timingcheck_timinganomaliesanomaliestool readstool reads
FIG. 01Modules and data flows, from the repository layout and documentation.
  1. 1

    Capture

    python-can Bus and Notifier receive thread scheduling frames onto the asyncio loop via AsyncBufferedReader

    canmcpd.py
  2. 2

    BusState

    Ingests frames per (channel, id), keeps history and the score_frame() model hook

    canmcpd.py
  3. 3

    Analysis loop

    Periodic timing checks against each ID's own inter-arrival history

    canmcpd.py
  4. 4

    Anomaly log

    In-memory anomaly log with optional JSONL sink

    canmcpd.py
  5. 5

    MCP server

    uvicorn + FastMCP over Streamable HTTP; async tools read BusState

    canmcpd.py
03 · Capabilities

Specification.

Anomaly types
NEW_ID, SILENCE, REAPPEAR, STATIC_CHANGE, plus optional SCORE from a user-implemented hook
MCP tools
get_vehicle_state, get_recent_anomalies, get_bus_stats, list_ids, query_signal, describe_bus, send_frame; resource can://state
Transport
Streamable HTTP on 127.0.0.1:8127/mcp by default
Adapters
python-can interfaces: slcan, gs_usb, canalystii, socketcan, pcan
Decoding
Optional DBC via cantools; known-ID labels work without a DBC
Transmit
send_frame disabled unless safety.allow_transmit = true
Anomaly sink
Optional JSONL file for tailing by an agent or alert
Tested
Detector unit tests and an end-to-end test over SocketCAN vcan0 with a FastMCP client; not exercised on a moving vehicle
Runs on
Local CLI daemon (python canmcpd.py -c config.toml) that needs a CAN adapter or vcan and binds to 127.0.0.1 with no auth
Stack
  • Python 3.11+
  • FastMCP
  • python-can
  • cantools
  • asyncio
  • uvicorn
04 · Changelog

Log.

  1. canmcpd published
    A CAN bus daemon with an MCP interface.