mcp-lnav-canbus

mcp-lnav-canbus

A Model Context Protocol (MCP) server that exposes lnav log file analysis capabilities to AI assistants, specifically optimized for Kvaser Plain Text Log Frame CAN bus log processing.

Category
访问服务器

README

MCP lnav CAN bus Server

A Model Context Protocol (MCP) server that exposes lnav log file analysis capabilities to AI assistants, specifically optimized for Kvaser Plain Text Log Frame CAN bus log processing.

Overview

This Python-based MCP server provides session-based access to CAN bus log files with structured JSON output, enabling AI models to analyze, query, and extract insights from Kvaser CAN bus logs through the Model Context Protocol.

Architecture

┌─────────────────────┐     ┌──────────────────────┐     ┌─────────────────┐
│  MCP Client         │────▶│  MCP Server          │────▶│  lnav CLI       │
│  (Claude, Cursor)   │◀────│  (Python, stdio)     │◀────│  (v0.12.0+)     │
└─────────────────────┘     └──────────────────────┘     └─────────────────┘
                                   │
                                   ▼
                            ┌──────────────────────┐
                            │  Session Manager     │
                            │  (stateful context)  │
                            └──────────────────────┘
                                   │
                                   ▼
                            ┌──────────────────────┐
                            │  Kvaser CAN Logs     │
                            │  (.txt, .log)        │
                            └──────────────────────┘

Requirements

Runtime Dependencies

  • Python: 3.10 or higher
  • lnav: 0.12.0 or higher (must be installed and available in PATH)
  • MCP Python SDK: 2.0.0+ (v2 beta) or 1.27+ (v1 stable)

Supported Log Format

Primary Format: Kvaser Plain Text Log Frame

# Column format: Timestamp Channel Type Dir ID DLC Data
0.000000 0        Rx   std 0x123 8 00 01 02 03 04 05 06 07
0.001000 0        Tx   std 0x456 8 AA BB CC DD EE FF 00 11

Also Supported (via lnav auto-detection):

  • Vector ASC
  • CANalyzer TRC
  • Generic CSV with timestamp/channel/id/data columns

Installation

1. Install lnav

# macOS
brew install lnav

# Ubuntu/Debian
sudo apt-get install lnav

# Or download from https://github.com/tstack/lnav/releases

Verify installation:

lnav --version

2. Install the MCP Server

# Using pip
pip install mcp-lnav-canbus

# Using uv
uv add mcp-lnav-canbus

Configuration

Claude Desktop

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json Linux: ~/.config/claude/claude_desktop_config.json Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "lnav-canbus": {
      "command": "python",
      "args": ["-m", "mcp_lnav_canbus"],
      "env": {
        "LNAV_PATH": "/usr/local/bin/lnav"
      }
    }
  }
}

Cursor

.cursor/mcp.json in project root:

{
  "mcpServers": {
    "lnav-canbus": {
      "command": "python",
      "args": ["-m", "mcp_lnav_canbus"],
      "cwd": "/path/to/your/project",
      "env": {
        "LNAV_PATH": "/usr/local/bin/lnav"
      }
    }
  }
}

VS Code (with MCP extension)

.vscode/mcp.json:

{
  "servers": {
    "lnav-canbus": {
      "type": "stdio",
      "command": "python",
      "args": ["-m", "mcp_lnav_canbus"]
    }
  }
}

Tools

open_can_log

Open a Kvaser CAN bus log file in the current session.

Parameters:

  • file_path (string, required): Path to the Kvaser CAN log file
  • session_id (string, optional): Session identifier (auto-generated if not provided)

Example:

{
  "name": "open_can_log",
  "arguments": {
    "file_path": "/data/can_trace.txt"
  }
}

Response:

{
  "success": true,
  "session_id": "sess_abc123",
  "file_info": {
    "path": "/data/can_trace.txt",
    "format": "kvaser",
    "message_count": 15420,
    "time_range": {
      "start": "2024-01-15T10:00:00.000000",
      "end": "2024-01-15T10:05:30.123456"
    },
    "channel_count": 2,
    "unique_ids": 47
  }
}

query_can_messages

Execute SQL queries against CAN bus log data using lnav's SQL engine.

Parameters:

  • query (string, required): SQLite query to execute
  • limit (integer, optional, default: 100): Maximum rows to return

Example:

{
  "name": "query_can_messages",
  "arguments": {
    "query": "SELECT can_id, COUNT(*) as count FROM log GROUP BY can_id ORDER BY count DESC LIMIT 10"
  }
}

Response:

{
  "success": true,
  "columns": ["can_id", "count"],
  "rows": [
    ["0x123", 1542],
    ["0x456", 987],
    ["0x789", 654]
  ],
  "row_count": 3,
  "execution_time_ms": 45
}

extract_can_frames

Extract specific CAN frames by ID, channel, or time range.

Parameters:

  • can_id (string, optional): CAN ID to filter (hex format, e.g., '0x123')
  • channel (integer, optional): CAN channel number (0-based)
  • start_time (string, optional): Start timestamp
  • end_time (string, optional): End timestamp
  • limit (integer, optional, default: 1000): Maximum frames to return

Example:

{
  "name": "extract_can_frames",
  "arguments": {
    "can_id": "0x123",
    "start_time": "0.000000",
    "end_time": "1.000000",
    "limit": 100
  }
}

Response:

{
  "success": true,
  "frames": [
    {
      "timestamp": "0.000000",
      "channel": 0,
      "type": "Rx",
      "can_id": "0x123",
      "dlc": 8,
      "data": [0, 1, 2, 3, 4, 5, 6, 7]
    }
  ],
  "frame_count": 1,
  "total_matching": 154
}

get_statistics

Generate statistical analysis of CAN bus traffic.

Parameters:

  • stat_type (string, required): "frequency", "timing", "load", "errors", or "summary"
  • interval (string, optional): Time interval for aggregation (e.g., '1s', '100ms')
  • can_id (string, optional): CAN ID to filter statistics

Example:

{
  "name": "get_statistics",
  "arguments": {
    "stat_type": "frequency",
    "interval": "1s"
  }
}

Response:

{
  "success": true,
  "stat_type": "frequency",
  "statistics": {
    "by_id": [
      {"can_id": "0x123", "count": 1542, "rate_hz": 100.0},
      {"can_id": "0x456", "count": 987, "rate_hz": 64.2}
    ],
    "total_messages": 15420,
    "duration_seconds": 330.123456,
    "average_rate_hz": 46.7
  }
}

find_errors

Find error frames and anomalies in CAN bus logs.

Parameters:

  • error_type (string, optional, default: "all"): "all", "crc", "bit", "stuff", "form", "ack", "timeout", or "gap"
  • channel (integer, optional): Channel filter

Example:

{
  "name": "find_errors",
  "arguments": {
    "error_type": "crc"
  }
}

Response:

{
  "success": true,
  "errors": [
    {
      "timestamp": "1.234567",
      "channel": 0,
      "error_type": "crc",
      "can_id": "0x123",
      "message": "CRC error detected"
    }
  ],
  "error_count": 1,
  "error_rate_per_minute": 0.18
}

analyze_payload_noise

Analyze CAN bus payload repetition to detect noise in logs using MD5 hashes. This tool identifies repeated payloads that may represent noise or meaningless data in the log.

Parameters:

  • min_repetitions (integer, optional, default: 10): Minimum number of repetitions to consider as noise
  • session_id (string, optional): Session identifier

Example:

{
  "name": "analyze_payload_noise",
  "arguments": {
    "min_repetitions": 5
  }
}

Response:

{
  "success": true,
  "analysis": {
    "total_frames": 1000,
    "unique_payloads": 25,
    "noise_patterns_found": 3,
    "total_noisy_frames": 850,
    "noise_percentage": 85.0,
    "min_repetitions_threshold": 5
  },
  "noise_patterns": [
    {
      "payload_md5": "0ee0646c1c77d8131cc8f4ee65c7673b",
      "repetition_count": 500,
      "payload_hex": "01 02 03 04 05 06 07 08",
      "first_timestamp": "0.000000",
      "last_timestamp": "10.500000",
      "time_span_seconds": 10.5,
      "can_ids": ["0x123"],
      "channels": [0],
      "noise_ratio": 0.5
    }
  ]
}

scan_payload_for_value

Scan CAN bus payload bytes for a specific decimal value. Searches for frames where consecutive bytes in the payload match the binary representation of the given decimal value. Supports both big-endian and little-endian byte orders.

Parameters:

  • byte_value (integer, required): Decimal value to search for (0 to unlimited, auto-calculates byte width)
  • endianness (string, optional, default: "big"): Byte order ("big" or "little")
  • start_offset (integer, optional, default: 0): Start searching from this byte offset
  • end_offset (integer, optional, default: None): Stop searching at this byte offset (None searches to end)
  • limit (integer, optional, default: 1000): Maximum frames to return
  • session_id (string, optional): Session identifier

Example:

{
  "name": "scan_payload_for_value",
  "arguments": {
    "byte_value": 65535,
    "endianness": "big",
    "limit": 10
  }
}

Response:

{
  "success": true,
  "search_criteria": {
    "byte_value": 65535,
    "endianness": "big",
    "start_offset": 0,
    "end_offset": null,
    "byte_width": 2,
    "target_bytes_hex": "FF FF"
  },
  "matches": [
    {
      "timestamp": "0.001000",
      "channel": 0,
      "type": "Tx",
      "direction": "std",
      "can_id": "0x456",
      "dlc": 8,
      "data": [170, 187, 255, 255, 238, 239, 0, 17],
      "payload_hex": "AA BB FF FF EE EF 00 11",
      "match_position": 2,
      "match_bytes": [255, 255],
      "byte_width": 2
    }
  ],
  "match_count": 1,
  "frames_scanned": 10,
  "frames_skipped_too_short": 0
}

Use Cases:

  • Find frames containing specific sensor values (e.g., temperature = 255)
  • Search for error codes encoded in payload bytes
  • Locate frames with specific multi-byte values (e.g., 16-bit counters, 32-bit timestamps)
  • Debug communication issues by finding frames with unexpected byte patterns

get_session_info

Retrieve information about the current session state.

Parameters: None

Response:

{
  "success": true,
  "session_id": "sess_abc123",
  "files_loaded": [
    {
      "path": "/data/can_trace.txt",
      "format": "kvaser",
      "message_count": 15420,
      "loaded_at": "2024-01-15T14:30:00Z"
    }
  ],
  "active_filters": [],
  "lnav_version": "0.12.0",
  "session_duration_seconds": 120
}

close_session

Close the current session and release resources.

Parameters: None

Response:

{
  "success": true,
  "session_id": "sess_abc123",
  "message": "Session closed successfully"
}

Resources

The server provides the following MCP resources:

canbus://session/{session_id}/log

Access the current session's loaded log file content.

Example URI: canbus://session/sess_abc123/log

canbus://session/{session_id}/messages/{can_id}

Access specific CAN ID messages from the session.

Example URI: canbus://session/sess_abc123/messages/0x123

canbus://session/{session_id}/errors

Access error frames from the session's log file.

Example URI: canbus://session/sess_abc123/errors

Prompts

canbus-analyze

Generate a comprehensive analysis of a Kvaser CAN bus log file.

Parameters:

  • file_path (string, required): Path to the Kvaser CAN log file
  • analysis_type (string, optional): "quick", "detailed", "errors", "traffic", or "custom" (default: "detailed")
  • focus_ids (array, optional): List of CAN IDs to focus analysis on

Example:

/canbus-analyze file_path="/data/can_trace.txt" analysis_type="detailed"

canbus-decode-frame

Help decode and interpret a specific CAN frame.

Parameters:

  • can_id (string, required): CAN ID to decode
  • data_bytes (array, required): Data bytes to decode
  • dbc_file (string, optional): Path to DBC file for signal decoding

Example:

/canbus-decode-frame can_id="0x123" data_bytes=[0,1,2,3,4,5,6,7]

Session Lifecycle

Session Creation

  1. Client initiates MCP connection via stdio
  2. Server creates session context with unique session ID
  3. Server spawns lnav process for the session
  4. Session state initialized (empty file list, no filters)

Session Usage

  1. Client calls open_can_log to load a file
  2. Server loads file into lnav, updates session state
  3. Client makes queries using session context
  4. Server maintains lnav process and state between calls

Session Termination

  1. Client calls close_session or disconnects
  2. Server terminates lnav process
  3. Server releases session resources
  4. Session state cleared

Server Configuration

Transport

Transport Protocol: stdio only

The server communicates via standard input/output streams, suitable for local CLI integration and subprocess spawning.

Session Management

The server maintains session-based state between calls:

  • Each client connection establishes a session context
  • Session persists for the lifetime of the MCP connection
  • Loaded files remain in session memory for subsequent queries
  • Session state includes: loaded files, active filters, query history, lnav process handle

Path Validation

Pass-through to lnav: The server does not validate file paths. Path validation and error handling is delegated to lnav, which returns appropriate error messages for invalid paths.

Environment Variables

Variable Description Default
LNAV_PATH Path to lnav binary lnav (from PATH)
LNAV_TIMEOUT Default timeout for lnav operations (seconds) 30
MAX_SESSION_COUNT Maximum concurrent sessions 10
LOG_LEVEL Logging verbosity INFO

Server Startup

# Basic usage
python -m mcp_lnav_canbus

# With custom lnav path
LNAV_PATH=/usr/local/bin/lnav python -m mcp_lnav_canbus

# With debug logging
LOG_LEVEL=DEBUG python -m mcp_lnav_canbus

Error Handling

Error Response Format

{
  "success": false,
  "error": {
    "code": "FILE_NOT_FOUND",
    "message": "File not found: /path/to/log.txt",
    "details": {
      "path": "/path/to/log.txt",
      "lnav_error": "Unable to open file"
    }
  }
}

Error Codes

Code Description
FILE_NOT_FOUND Specified file path does not exist
INVALID_FORMAT Log file format not recognized
QUERY_ERROR SQL query syntax error
SESSION_ERROR Session not found or expired
LNAV_ERROR lnav process error
TIMEOUT Operation timed out
INTERNAL_ERROR Internal server error

Usage Examples

Example 1: Basic Log Analysis

// Request
{
  "tool": "open_can_log",
  "arguments": {
    "file_path": "/data/can_trace.txt"
  }
}

// Response
{
  "success": true,
  "session_id": "sess_abc123",
  "file_info": {
    "path": "/data/can_trace.txt",
    "format": "kvaser",
    "message_count": 15420,
    "time_range": {
      "start": "2024-01-15T10:00:00.000000",
      "end": "2024-01-15T10:05:30.123456"
    },
    "channel_count": 2,
    "unique_ids": 47
  }
}

Example 2: Query by CAN ID

// Request
{
  "tool": "query_can_messages",
  "arguments": {
    "query": "SELECT can_id, COUNT(*) as count FROM log GROUP BY can_id ORDER BY count DESC LIMIT 10"
  }
}

// Response
{
  "success": true,
  "columns": ["can_id", "count"],
  "rows": [
    ["0x123", 1542],
    ["0x456", 987],
    ["0x789", 654]
  ],
  "row_count": 3
}

Example 3: Error Detection

// Request
{
  "tool": "find_errors",
  "arguments": {
    "error_type": "all"
  }
}

// Response
{
  "success": true,
  "errors": [
    {
      "timestamp": "1.234567",
      "channel": 0,
      "error_type": "crc",
      "can_id": "0x123",
      "message": "CRC error detected"
    }
  ],
  "error_count": 1,
  "error_rate_per_minute": 0.18
}

Security Considerations

  • Server runs with same permissions as host process
  • File access limited to paths provided in tool calls
  • No network access performed (stdio transport only)
  • SQL queries executed in sandboxed lnav session
  • Session isolation prevents cross-session data access
  • Command injection prevented via parameterized queries

Troubleshooting

Common Issues

lnav not found:

Error: LNAV_PATH not set or lnav not in PATH
Solution: Set LNAV_PATH environment variable or install lnav

Invalid log format:

Error: INVALID_FORMAT
Solution: Ensure log file is Kvaser Plain Text Log Frame format

Session timeout:

Error: SESSION_ERROR
Solution: Check session_id is valid and session hasn't expired

Query syntax error:

Error: QUERY_ERROR
Solution: Verify SQLite syntax in query parameter

Performance Considerations

  • Large log files indexed on open (progress reported to client)
  • SQL queries executed via lnav's SQLite integration
  • Result limiting prevents memory exhaustion
  • Timeout enforcement prevents hung operations

Limitations

  • Binary formats (BLF, MF4) require external conversion tools
  • Real-time log tailing is not supported (batch processing only)
  • DBC signal decoding requires external DBC parser integration
  • Maximum file size recommendations apply for performance

Development

Running in Development Mode

Use the MCP Inspector to test the server:

uv run mcp dev server.py

Testing

# Run tests
pytest tests/

# Run with coverage
pytest --cov=mcp_lnav_canbus tests/

Version Compatibility

Component Minimum Version Recommended Version
Python 3.10 3.11+
lnav 0.12.0 0.14.0+
MCP SDK 1.27.0 or 2.0.0b1 2.0.0b1+

Contributing

Contributions are welcome! Please see our Contributing Guide for details.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Acknowledgments

Resources

推荐服务器

Baidu Map

Baidu Map

百度地图核心API现已全面兼容MCP协议,是国内首家兼容MCP协议的地图服务商。

官方
精选
JavaScript
Playwright MCP Server

Playwright MCP Server

一个模型上下文协议服务器,它使大型语言模型能够通过结构化的可访问性快照与网页进行交互,而无需视觉模型或屏幕截图。

官方
精选
TypeScript
Magic Component Platform (MCP)

Magic Component Platform (MCP)

一个由人工智能驱动的工具,可以从自然语言描述生成现代化的用户界面组件,并与流行的集成开发环境(IDE)集成,从而简化用户界面开发流程。

官方
精选
本地
TypeScript
Audiense Insights MCP Server

Audiense Insights MCP Server

通过模型上下文协议启用与 Audiense Insights 账户的交互,从而促进营销洞察和受众数据的提取和分析,包括人口统计信息、行为和影响者互动。

官方
精选
本地
TypeScript
VeyraX

VeyraX

一个单一的 MCP 工具,连接你所有喜爱的工具:Gmail、日历以及其他 40 多个工具。

官方
精选
本地
graphlit-mcp-server

graphlit-mcp-server

模型上下文协议 (MCP) 服务器实现了 MCP 客户端与 Graphlit 服务之间的集成。 除了网络爬取之外,还可以将任何内容(从 Slack 到 Gmail 再到播客订阅源)导入到 Graphlit 项目中,然后从 MCP 客户端检索相关内容。

官方
精选
TypeScript
Kagi MCP Server

Kagi MCP Server

一个 MCP 服务器,集成了 Kagi 搜索功能和 Claude AI,使 Claude 能够在回答需要最新信息的问题时执行实时网络搜索。

官方
精选
Python
e2b-mcp-server

e2b-mcp-server

使用 MCP 通过 e2b 运行代码。

官方
精选
Neon MCP Server

Neon MCP Server

用于与 Neon 管理 API 和数据库交互的 MCP 服务器

官方
精选
Exa MCP Server

Exa MCP Server

模型上下文协议(MCP)服务器允许像 Claude 这样的 AI 助手使用 Exa AI 搜索 API 进行网络搜索。这种设置允许 AI 模型以安全和受控的方式获取实时的网络信息。

官方
精选