Mikrotik MCP Server

MikroTik MCP服务器

MikroTik RouterOS设备的全面模型上下文协议（MCP）服务器实现，通过组织良好、生产就绪的Python代码库提供对设备管理、监控和配置的编程访问。

特性

🔍 日志管理

• 全面的日志检索：使用高级筛选和格式化选项访问系统日志
• 客户端过滤：针对主题、消息和日志级别的强大过滤功能
• 专用日志类型：调试、错误、警告和信息日志的专用方法
• 缓冲支架：从单独的内存日志缓冲区访问日志
• 额外信息：使用其他元数据和上下文检索日志

🖥️ 系统管理

• 资源监控：实时系统资源使用和性能指标
• 健康评估：具有状态指标的综合系统健康评估
• 性能指标：CPU负载、内存使用情况、磁盘空间和正常运行时间监控
• 主动警报：资源阈值的警告和关键状态检测

🌐 IP和网络管理

• IP 地址管理：查看和过滤跨接口的IP地址配置
• 路由表：访问和分析IP路由配置和网关设置
• IP池：管理DHCP地址池和分配范围
• 网络摘要：网络拓扑和配置的全面概述

🔌 接口管理

• 界面发现：检索具有详细属性的所有已配置接口
• 接口类型：支持以太网、无线、网桥和其他接口类型
• 状态监测：实时接口状态、MTU、MAC地址和操作状态
• 筛选选项：按名称、类型、状态和注释进行高级筛选

🛡️ 防火墙管理

• 规则管理：全面的防火墙规则检索和过滤
• NAT配置：网络地址转换规则管理
• 芒果规则：数据包修改和QoS规则处理
• 地址列表：安全策略的IP地址列表管理

🌐 DHCP管理

• 服务器配置：DHCP服务器设置和配置管理
• 租赁管理：主动DHCP租约监控和跟踪
• 网络配置：DHCP网络和子网管理
• 客户管理：DHCP客户端配置和状态

🛠️ MCP集成

• 资源发现：所有可用资源和工具的丰富元数据
• 工具集成：具有参数验证的综合工具模式
• 提示系统：常见任务和故障排除的智能提示
• 错误处理：具有详细错误消息的稳健错误处理

建筑

该项目遵循模块化、领域驱动的架构：

src/mcp_mikrotik/
├── __init__.py          # Main package exports
├── base.py              # Base client with common functionality
├── models.py            # Shared type definitions
├── client.py            # Main client combining all specialized clients
├── logs/                # Log management module
│   ├── __init__.py
│   ├── models.py        # Log-specific type definitions
│   └── client.py        # Specialized logs client
├── system/              # System management module
│   ├── __init__.py
│   ├── models.py        # System-specific type definitions
│   └── client.py        # Specialized system client
├── ip/                  # IP and network management module
│   ├── __init__.py
│   ├── models.py        # IP-specific type definitions
│   └── client.py        # Specialized IP client
├── interface/           # Interface management module
│   ├── __init__.py
│   ├── models.py        # Interface-specific type definitions
│   └── client.py        # Specialized interface client
├── firewall/            # Firewall management module
│   ├── __init__.py
│   ├── models.py        # Firewall-specific type definitions
│   └── client.py        # Specialized firewall client
└── dhcp/                # DHCP management module
    ├── __init__.py
    ├── models.py        # DHCP-specific type definitions
    └── client.py        # Specialized DHCP client

安装

先决条件

• Python 3.11或更高版本
• 访问MikroTik RouterOS设备
• 设备的网络连接

设置

1. 克隆存储库:

   git clone 
   cd mikrotik-mcp

1. 创建虚拟环境:

   python -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate

1. 安装依赖项:

   pip install -r requirements.txt

1. 配置环境:

   cp config/example.env .env
   # Edit .env with your MikroTik device details

配置

创建一个 .env 使用您的MikroTik设备配置文件：

# MikroTik device connection settings
MIKROTIK_HOST=192.168.88.1
MIKROTIK_USERNAME=admin
MIKROTIK_PASSWORD=your_password_here
MIKROTIK_PORT=443
MIKROTIK_USE_SSL=true

# Optional: Debug mode
DEBUG=true

# Optional: Log level
LOG_LEVEL=INFO

用法

基本用法

from src.mcp_mikrotik import MikroTikClient

# Initialize client
config = {
    "host": "192.168.88.1",
    "username": "admin",
    "password": "your_password",
    "useSSL": True
}

client = MikroTikClient(config)

# Test connection
is_connected = await client.test_connection()
print(f"Connected: {is_connected}")

# Get system information
system_info = await client.get_system_info()
print(f"System uptime: {system_info.get('uptime')}")

# Get recent logs
logs = await client.get_logs({"brief": True, "max_logs": 10})
for log in logs:
    print(f"{log['time']}: {log['message']}")

日志管理

# Get specific log types
debug_logs = await client.get_debug_logs()
error_logs = await client.get_error_logs()
warning_logs = await client.get_warning_logs()

# Filter logs by condition
system_logs = await client.find_logs('topics~"system"')
dhcp_logs = await client.find_logs('topics~"dhcp" and message~"assigned"')

# Get logs with extra information
detailed_logs = await client.get_logs_with_extra_info({"brief": False})

系统监控

# Get comprehensive system health
health = await client.get_system_health()
print(f"System status: {health['status']}")
print(f"Memory usage: {health['memory_usage_percent']}%")
print(f"Disk usage: {health['disk_usage_percent']}%")

# Get detailed resource information
resources = await client.get_system_resources()
for resource in resources:
    print(f"CPU: {resource['cpu_count']} cores, {resource['cpu_frequency']} MHz")

网络管理

# Get IP addresses
addresses = await client.get_ip_addresses()
for addr in addresses:
    print(f"{addr['address']} on {addr['interface']}")

# Get routing table
routes = await client.get_ip_routes()
for route in routes:
    print(f"{route['dst_address']} -> {route['gateway']}")

# Get network summary
summary = await client.get_network_summary()
print(f"Total IP addresses: {summary['ip_addresses_count']}")
print(f"Total routes: {summary['ip_routes_count']}")

接口管理

# Get all interfaces
interfaces = await client.get_interfaces()
for interface in interfaces:
    print(f"{interface['name']}: {interface['type']} - {interface['status']}")

# Get specific interface types
ethernet_interfaces = await client.get_ethernet_interfaces()
wireless_interfaces = await client.get_wireless_interfaces()
bridge_interfaces = await client.get_bridge_interfaces()

# Filter interfaces by properties
lan_interfaces = await client.get_interfaces({
    "type": "ether",
    "running": True,
    "comment": "LAN"
})

防火墙管理

# Get firewall rules
firewall_rules = await client.get_firewall_rules()
for rule in firewall_rules:
    print(f"{rule['chain']}: {rule['action']} {rule['src_address']} -> {rule['dst_address']}")

# Get NAT rules
nat_rules = await client.get_nat_rules()
for rule in nat_rules:
    print(f"NAT: {rule['chain']} {rule['action']}")

# Get address lists
address_lists = await client.get_address_lists()
for addr_list in address_lists:
    print(f"List: {addr_list['name']} - {addr_list['address']}")

DHCP管理

# Get DHCP servers
dhcp_servers = await client.get_dhcp_servers()
for server in dhcp_servers:
    print(f"DHCP Server: {server['name']} on {server['interface']}")

# Get active leases
dhcp_leases = await client.get_dhcp_leases()
for lease in dhcp_leases:
    print(f"Lease: {lease['address']} -> {lease['mac_address']}")

# Get DHCP networks
dhcp_networks = await client.get_dhcp_networks()
for network in dhcp_networks:
    print(f"Network: {network['name']} - {network['address']}")

MCP服务器

运行服务器

# Run the MCP server
python -m server.server

# Or use the installed script
mikrotik-mcp

可用资源

MCP服务器提供以下资源：

• mikrotik://logs/recent -最近的系统日志
• mikrotik://logs/debug -调试日志
• mikrotik://logs/error -错误日志
• mikrotik://logs/warning -警告日志
• mikrotik://logs/info -信息日志
• mikrotik://logs/detailed -包含额外信息的详细日志
• mikrotik://system/info -系统信息
• mikrotik://system/health -系统健康指标
• mikrotik://ip/addresses -IP地址配置
• mikrotik://ip/routes -IP路由表
• mikrotik://ip/pools -IP地址池
• mikrotik://ip/network_summary -网络配置摘要
• mikrotik://interface/all -所有接口配置
• mikrotik://interface/ethernet -以太网接口配置
• mikrotik://interface/wireless -无线接口配置
• mikrotik://interface/bridge -网桥接口配置
• mikrotik://firewall/rules -防火墙规则配置
• mikrotik://firewall/nat -NAT规则配置
• mikrotik://firewall/address-lists -防火墙地址列表
• mikrotik://dhcp/servers -DHCP服务器配置
• mikrotik://dhcp/leases -活动DHCP租约
• mikrotik://dhcp/networks -DHCP网络配置

可用工具

• get_logs -通过筛选检索系统日志
• get_debug_logs -获取调试日志
• get_error_logs -获取错误日志
• get_warning_logs -获取警告日志
• get_info_logs -获取信息日志
• get_logs_from_buffer -从特定缓冲区获取日志
• get_logs_with_extra_info -获取包含额外元数据的日志
• get_system_info -获取系统信息
• get_system_health -获取系统健康指标
• get_ip_addresses -获取IP地址配置
• get_ip_routes -获取IP路由表
• get_ip_pools -获取IP地址池
• get_network_summary -获取网络配置摘要
• get_interfaces -获取所有接口配置
• get_ethernet_interfaces -获取以太网接口配置
• get_wireless_interfaces -获取无线接口配置
• get_bridge_interfaces -获取网桥接口配置
• get_firewall_rules -获取防火墙规则配置
• get_nat_rules -获取NAT规则配置
• get_mangle_rules -获取混乱的规则配置
• get_address_lists -获取防火墙地址列表
• get_dhcp_servers -获取DHCP服务器配置
• get_dhcp_leases -获取活动DHCP租约
• get_dhcp_networks -获取DHCP网络配置
• get_dhcp_clients -获取DHCP客户端配置
• test_connection -测试设备连接性

发展

运行测试

# Run all tests
pytest

# Run specific test modules
pytest tests/test_logs_client.py
pytest tests/test_system_client.py
pytest tests/test_ip_client.py

# Run with coverage
pytest --cov=src/mcp_mikrotik --cov-report=html

代码质量

# Format code with Black
black src/ tests/

# Sort imports with isort
isort src/ tests/

# Lint with flake8
flake8 src/ tests/

# Type checking with mypy
mypy src/

项目结构

mikrotik-mcp/
├── src/mcp_mikrotik/           # Main package source
│   ├── __init__.py             # Package exports
│   ├── base.py                 # Base client functionality
│   ├── models.py               # Shared type definitions
│   ├── client.py               # Main client interface
│   ├── logs/                   # Log management module
│   ├── system/                 # System management module
│   └── ip/                     # IP management module
├── server/                     # MCP server implementation
├── tests/                      # Comprehensive test suite
├── config/                     # Configuration examples
├── requirements.txt            # Python dependencies
├── pyproject.toml             # Project configuration
└── README.md                  # This file

api参考

MicroTik客户端

提供对所有功能访问的主客户端类：

class MikroTikClient:
    def __init__(self, config: MikroTikConfig)
    
    # Log management
    async def get_logs(options: Optional[GetLogsArgs] = None, max_logs: int = 1000)
    async def get_debug_logs(options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def get_error_logs(options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def get_warning_logs(options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def get_info_logs(options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def get_logs_from_buffer(buffer_name: str, options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def get_logs_with_extra_info(options: Optional[Dict[str, Any]] = None, max_logs: int = 1000)
    async def find_logs(where: str, options: Optional[Dict[str, Any]] = None)
    
    # System management
    async def get_system_info()
    async def get_system_resources()
    async def get_system_health()
    
    # IP management
    async def get_ip_addresses(options: Optional[GetIPAddressesArgs] = None)
    async def get_ip_routes(options: Optional[GetIPRoutesArgs] = None)
    async def get_ip_pools()
    async def get_network_summary()
    
    # Interface management
    async def get_interfaces(options: Optional[GetInterfacesArgs] = None)
    async def get_ethernet_interfaces()
    async def get_wireless_interfaces()
    async def get_bridge_interfaces()
    
    # Firewall management
    async def get_firewall_rules(options: Optional[GetFirewallRulesArgs] = None)
    async def get_nat_rules()
    async def get_mangle_rules()
    async def get_address_lists()
    
    # DHCP management
    async def get_dhcp_servers(options: Optional[GetDHCPServersArgs] = None)
    async def get_dhcp_leases()
    async def get_dhcp_networks()
    async def get_dhcp_clients()
    
    # Connection testing
    async def test_connection()

专业客户

通过专用客户端访问专用功能：

# Log management
logs_client = client.logs
system_client = client.system
ip_client = client.ip
interface_client = client.interface
firewall_client = client.firewall
dhcp_client = client.dhcp

# Use specialized methods
debug_logs = await logs_client.get_debug_logs()
system_health = await system_client.get_system_health()
network_summary = await ip_client.get_network_summary()
ethernet_interfaces = await interface_client.get_ethernet_interfaces()
firewall_rules = await firewall_client.get_firewall_rules()
dhcp_leases = await dhcp_client.get_dhcp_leases()

贡献

1. 分叉存储库
2. 创建要素分支
3. 进行更改
4. 添加新功能的测试
5. 确保所有测试通过
6. 提交拉取请求

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

支持

有关支持和问题：

• 在GitHub上创建问题
• 检查文档
• 查看测试示例

路线图

• \[x\] 接口管理模块
• \[x\] 防火墙配置模块
• \[x\] DHCP服务器管理
• \[\]无线管理模块
• \[\]VPN配置
• \[\]用户管理
• \[\]备份和恢复功能
• \[\]实时监控和警报
• \[\]配置模板
• \[\]批量操作支持