From ad83ea56a0185fe655a91da8cab1997840cbe974 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Wed, 15 Apr 2026 12:08:37 -0400 Subject: [PATCH 1/2] Add 'get_network_routes' tool --- src/linux_mcp_server/commands.py | 5 ++ src/linux_mcp_server/formatters.py | 32 +++++++++++ src/linux_mcp_server/models.py | 12 ++++ src/linux_mcp_server/parsers.py | 76 ++++++++++++++++++++++++++ src/linux_mcp_server/tools/__init__.py | 2 + src/linux_mcp_server/tools/network.py | 28 ++++++++++ tests/tools/test_network.py | 57 +++++++++++++++++++ 7 files changed, 212 insertions(+) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index 14823e81..aaaafd42 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -135,6 +135,11 @@ class CommandGroup(BaseModel): "stats": CommandSpec(args=("cat", "/proc/net/dev")), } ), + "network_routes": CommandGroup( + commands={ + "default": CommandSpec(args=("ip", "route")), + } + ), # === Logs === "journal_logs": CommandGroup( commands={ diff --git a/src/linux_mcp_server/formatters.py b/src/linux_mcp_server/formatters.py index 2cf49090..f6414980 100644 --- a/src/linux_mcp_server/formatters.py +++ b/src/linux_mcp_server/formatters.py @@ -8,6 +8,7 @@ from linux_mcp_server.models import NetworkConnection from linux_mcp_server.models import NetworkInterface from linux_mcp_server.models import ProcessInfo +from linux_mcp_server.models import Route from linux_mcp_server.utils import format_bytes @@ -140,6 +141,37 @@ def format_network_interfaces( return "\n".join(lines) +def format_routes( + routes: list[Route], + header: str = "=== Routing Table ===\n", +) -> str: + """Format routing table entries into a readable string. + + Args: + routes: List of Route objects. + header: Header text for the output. + + Returns: + Formatted string representation. + """ + lines = [header] + lines.append( + f"{'Destination':<20} {'Gateway':<18} {'Device':<10} {'Protocol':<10} {'Scope':<10} {'Source':<18} {'Metric'}" + ) + lines.append("-" * 110) + + for route in routes: + gateway = route.gateway or "-" + metric = str(route.metric) if route.metric is not None else "-" + lines.append( + f"{route.destination:<20} {gateway:<18} {route.device:<10} {route.protocol:<10} " + f"{route.scope:<10} {route.source:<18} {metric}" + ) + + lines.append(f"\n\nTotal routes: {len(routes)}") + return "\n".join(lines) + + def format_process_detail( ps_output: str, proc_status: dict[str, str] | None = None, diff --git a/src/linux_mcp_server/models.py b/src/linux_mcp_server/models.py index ff1f3fa1..e0f3fda8 100644 --- a/src/linux_mcp_server/models.py +++ b/src/linux_mcp_server/models.py @@ -43,6 +43,18 @@ class ListeningPort(BaseModel): process: str = "" +class Route(BaseModel): + """Parsed route entry from ip route output.""" + + destination: str + gateway: str = "" + device: str = "" + protocol: str = "" + scope: str = "" + source: str = "" + metric: int | None = None + + class NetworkInterface(BaseModel): """Parsed network interface information.""" diff --git a/src/linux_mcp_server/parsers.py b/src/linux_mcp_server/parsers.py index d30a7911..63376916 100644 --- a/src/linux_mcp_server/parsers.py +++ b/src/linux_mcp_server/parsers.py @@ -13,6 +13,7 @@ from linux_mcp_server.models import NetworkInterface from linux_mcp_server.models import NodeEntry from linux_mcp_server.models import ProcessInfo +from linux_mcp_server.models import Route from linux_mcp_server.models import SwapInfo from linux_mcp_server.models import SystemInfo from linux_mcp_server.models import SystemMemory @@ -280,6 +281,81 @@ def parse_ip_brief(stdout: str) -> dict[str, NetworkInterface]: return interfaces +def parse_ip_route(stdout: str) -> list[Route]: + """Parse ip route output into Route objects. + + Handles standard ``ip route`` output where each line describes a route. + Key-value tokens (e.g. ``via``, ``dev``, ``proto``, ``scope``, ``src``, + ``metric``) are extracted into the corresponding model fields. + + Args: + stdout: Raw output from ip route command. + + Returns: + List of Route objects. + """ + routes: list[Route] = [] + lines = stdout.strip().split("\n") + + for line in lines: + line = line.strip() + if not line: + continue + + parts = line.split() + if not parts: + continue + + destination = parts[0] + gateway = "" + device = "" + protocol = "" + scope = "" + source = "" + metric: int | None = None + + i = 1 + while i < len(parts): + token = parts[i] + if token == "via" and i + 1 < len(parts): + gateway = parts[i + 1] + i += 2 + elif token == "dev" and i + 1 < len(parts): + device = parts[i + 1] + i += 2 + elif token == "proto" and i + 1 < len(parts): + protocol = parts[i + 1] + i += 2 + elif token == "scope" and i + 1 < len(parts): + scope = parts[i + 1] + i += 2 + elif token == "src" and i + 1 < len(parts): + source = parts[i + 1] + i += 2 + elif token == "metric" and i + 1 < len(parts): + try: + metric = int(parts[i + 1]) + except ValueError: + pass + i += 2 + else: + i += 1 + + routes.append( + Route( + destination=destination, + gateway=gateway, + device=device, + protocol=protocol, + scope=scope, + source=source, + metric=metric, + ) + ) + + return routes + + def parse_system_info(results: dict[str, str]) -> SystemInfo: """Parse system info command results into SystemInfo object. diff --git a/src/linux_mcp_server/tools/__init__.py b/src/linux_mcp_server/tools/__init__.py index 954b167e..f388d66a 100644 --- a/src/linux_mcp_server/tools/__init__.py +++ b/src/linux_mcp_server/tools/__init__.py @@ -7,6 +7,7 @@ from linux_mcp_server.tools.network import get_listening_ports from linux_mcp_server.tools.network import get_network_connections from linux_mcp_server.tools.network import get_network_interfaces +from linux_mcp_server.tools.network import get_network_routes # processes from linux_mcp_server.tools.processes import get_process_info @@ -49,6 +50,7 @@ "get_memory_information", "get_network_connections", "get_network_interfaces", + "get_network_routes", "get_process_info", "get_service_logs", "get_service_status", diff --git a/src/linux_mcp_server/tools/network.py b/src/linux_mcp_server/tools/network.py index 4b21f19b..4d9e29d2 100644 --- a/src/linux_mcp_server/tools/network.py +++ b/src/linux_mcp_server/tools/network.py @@ -7,7 +7,9 @@ from linux_mcp_server.formatters import format_listening_ports from linux_mcp_server.formatters import format_network_connections from linux_mcp_server.formatters import format_network_interfaces +from linux_mcp_server.formatters import format_routes from linux_mcp_server.parsers import parse_ip_brief +from linux_mcp_server.parsers import parse_ip_route from linux_mcp_server.parsers import parse_proc_net_dev from linux_mcp_server.parsers import parse_ss_connections from linux_mcp_server.parsers import parse_ss_listening @@ -103,3 +105,29 @@ async def get_listening_ports( ports = parse_ss_listening(stdout) return format_listening_ports(ports) return f"Error getting listening ports: return code {returncode}, stderr: {stderr}" + + +@mcp.tool( + title="Get network routes", + description="Get the system routing table showing network routes, gateways, and interfaces.", + tags={"connectivity", "network", "routing"}, + annotations=ToolAnnotations(readOnlyHint=True), +) +@log_tool_call +@disallow_local_execution_in_containers +async def get_network_routes( + host: Host = None, +) -> str: + """Get network routing table. + + Retrieves the system routing table including destination networks, + gateways, devices, protocols, scopes, source addresses, and metrics. + """ + cmd = get_command("network_routes") + + returncode, stdout, stderr = await cmd.run(host=host) + + if is_successful_output(returncode, stdout): + routes = parse_ip_route(stdout) + return format_routes(routes) + return f"Error getting network routes: return code {returncode}, stderr: {stderr}" diff --git a/tests/tools/test_network.py b/tests/tools/test_network.py index 504fe8a2..27a8f0e3 100644 --- a/tests/tools/test_network.py +++ b/tests/tools/test_network.py @@ -211,3 +211,60 @@ async def test_get_listening_ports_error(self, mcp_client, mock_execute): match = re.compile(r"error calling tool.*raised intentionally", flags=re.I) with pytest.raises(ToolError, match=match): await mcp_client.call_tool("get_listening_ports") + + +class TestGetNetworkRoutes: + """Test get_network_routes function.""" + + @pytest.mark.parametrize( + ("host", "mock_output", "expected_content"), + [ + pytest.param( + None, + "default via 192.168.1.1 dev eth0 proto dhcp src 192.168.1.100 metric 100\n" + "192.168.1.0/24 dev eth0 proto kernel scope link src 192.168.1.100\n" + "172.17.0.0/16 dev docker0 proto kernel scope link src 172.17.0.1", + ["192.168.1.1", "eth0", "Total routes: 3"], + id="local", + ), + pytest.param( + "remote.host", + "default via 10.0.0.1 dev ens5 proto dhcp metric 100\n" + "10.0.0.0/24 dev ens5 proto kernel scope link src 10.0.0.5", + ["10.0.0.1", "ens5"], + id="remote", + ), + ], + ) + async def test_get_network_routes_success(self, mcp_client, mock_execute, host, mock_output, expected_content): + """Test getting network routes with success.""" + mock_execute.return_value = (0, mock_output, "") + result = await mcp_client.call_tool("get_network_routes", arguments={"host": host}) + result_text = result.content[0].text.casefold() + + assert "routing table" in result_text + assert all(content.casefold() in result_text for content in expected_content), ( + "Did not find all expected values" + ) + + @pytest.mark.parametrize( + ("return_value",), + [ + pytest.param((1, "", "Command not found"), id="command_fails"), + pytest.param((0, "", ""), id="empty_output"), + ], + ) + async def test_get_network_routes_failure(self, mcp_client, mock_execute, return_value): + """Test getting network routes when command fails or returns empty.""" + mock_execute.return_value = return_value + result = await mcp_client.call_tool("get_network_routes") + result_text = result.content[0].text.casefold() + + assert "error" in result_text + + async def test_get_network_routes_error(self, mcp_client, mock_execute): + """Test getting network routes with general error.""" + mock_execute.side_effect = ValueError("Raised intentionally") + match = re.compile(r"error calling tool.*raised intentionally", flags=re.I) + with pytest.raises(ToolError, match=match): + await mcp_client.call_tool("get_network_routes") From e68b34749a13fa329058a113e5c7ef3180951c13 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Tue, 23 Jun 2026 10:42:16 -0400 Subject: [PATCH 2/2] refactor(network): use JSON output for routing table parsing - Switch `ip route` to `ip -json route` for structured output - Update Route model with validation aliases matching JSON fields - Remove text parsing and formatting functions - Change get_network_routes to return list[Route] instead of formatted string - Raise ToolError on failure instead of returning error text - Add "fixed" tag to indicate stable tool interface --- src/linux_mcp_server/commands.py | 2 +- src/linux_mcp_server/formatters.py | 32 ----------- src/linux_mcp_server/models.py | 11 ++-- src/linux_mcp_server/parsers.py | 76 --------------------------- src/linux_mcp_server/tools/network.py | 15 +++--- tests/test_server.py | 1 + tests/tools/test_network.py | 53 +++++++++++++++---- 7 files changed, 59 insertions(+), 131 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index aaaafd42..2f4c6f95 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -137,7 +137,7 @@ class CommandGroup(BaseModel): ), "network_routes": CommandGroup( commands={ - "default": CommandSpec(args=("ip", "route")), + "default": CommandSpec(args=("ip", "-json", "route")), } ), # === Logs === diff --git a/src/linux_mcp_server/formatters.py b/src/linux_mcp_server/formatters.py index f6414980..2cf49090 100644 --- a/src/linux_mcp_server/formatters.py +++ b/src/linux_mcp_server/formatters.py @@ -8,7 +8,6 @@ from linux_mcp_server.models import NetworkConnection from linux_mcp_server.models import NetworkInterface from linux_mcp_server.models import ProcessInfo -from linux_mcp_server.models import Route from linux_mcp_server.utils import format_bytes @@ -141,37 +140,6 @@ def format_network_interfaces( return "\n".join(lines) -def format_routes( - routes: list[Route], - header: str = "=== Routing Table ===\n", -) -> str: - """Format routing table entries into a readable string. - - Args: - routes: List of Route objects. - header: Header text for the output. - - Returns: - Formatted string representation. - """ - lines = [header] - lines.append( - f"{'Destination':<20} {'Gateway':<18} {'Device':<10} {'Protocol':<10} {'Scope':<10} {'Source':<18} {'Metric'}" - ) - lines.append("-" * 110) - - for route in routes: - gateway = route.gateway or "-" - metric = str(route.metric) if route.metric is not None else "-" - lines.append( - f"{route.destination:<20} {gateway:<18} {route.device:<10} {route.protocol:<10} " - f"{route.scope:<10} {route.source:<18} {metric}" - ) - - lines.append(f"\n\nTotal routes: {len(routes)}") - return "\n".join(lines) - - def format_process_detail( ps_output: str, proc_status: dict[str, str] | None = None, diff --git a/src/linux_mcp_server/models.py b/src/linux_mcp_server/models.py index e0f3fda8..45049d5e 100644 --- a/src/linux_mcp_server/models.py +++ b/src/linux_mcp_server/models.py @@ -4,6 +4,7 @@ from pathlib import Path from pydantic import BaseModel +from pydantic import ConfigDict from pydantic import Field from pydantic import field_serializer from pydantic import model_validator @@ -44,14 +45,16 @@ class ListeningPort(BaseModel): class Route(BaseModel): - """Parsed route entry from ip route output.""" + """Parsed route entry from ip -json route output.""" - destination: str + model_config = ConfigDict(populate_by_name=True) + + destination: str = Field(validation_alias="dst") gateway: str = "" - device: str = "" + device: str = Field(default="", validation_alias="dev") protocol: str = "" scope: str = "" - source: str = "" + source: str = Field(default="", validation_alias="prefsrc") metric: int | None = None diff --git a/src/linux_mcp_server/parsers.py b/src/linux_mcp_server/parsers.py index 63376916..d30a7911 100644 --- a/src/linux_mcp_server/parsers.py +++ b/src/linux_mcp_server/parsers.py @@ -13,7 +13,6 @@ from linux_mcp_server.models import NetworkInterface from linux_mcp_server.models import NodeEntry from linux_mcp_server.models import ProcessInfo -from linux_mcp_server.models import Route from linux_mcp_server.models import SwapInfo from linux_mcp_server.models import SystemInfo from linux_mcp_server.models import SystemMemory @@ -281,81 +280,6 @@ def parse_ip_brief(stdout: str) -> dict[str, NetworkInterface]: return interfaces -def parse_ip_route(stdout: str) -> list[Route]: - """Parse ip route output into Route objects. - - Handles standard ``ip route`` output where each line describes a route. - Key-value tokens (e.g. ``via``, ``dev``, ``proto``, ``scope``, ``src``, - ``metric``) are extracted into the corresponding model fields. - - Args: - stdout: Raw output from ip route command. - - Returns: - List of Route objects. - """ - routes: list[Route] = [] - lines = stdout.strip().split("\n") - - for line in lines: - line = line.strip() - if not line: - continue - - parts = line.split() - if not parts: - continue - - destination = parts[0] - gateway = "" - device = "" - protocol = "" - scope = "" - source = "" - metric: int | None = None - - i = 1 - while i < len(parts): - token = parts[i] - if token == "via" and i + 1 < len(parts): - gateway = parts[i + 1] - i += 2 - elif token == "dev" and i + 1 < len(parts): - device = parts[i + 1] - i += 2 - elif token == "proto" and i + 1 < len(parts): - protocol = parts[i + 1] - i += 2 - elif token == "scope" and i + 1 < len(parts): - scope = parts[i + 1] - i += 2 - elif token == "src" and i + 1 < len(parts): - source = parts[i + 1] - i += 2 - elif token == "metric" and i + 1 < len(parts): - try: - metric = int(parts[i + 1]) - except ValueError: - pass - i += 2 - else: - i += 1 - - routes.append( - Route( - destination=destination, - gateway=gateway, - device=device, - protocol=protocol, - scope=scope, - source=source, - metric=metric, - ) - ) - - return routes - - def parse_system_info(results: dict[str, str]) -> SystemInfo: """Parse system info command results into SystemInfo object. diff --git a/src/linux_mcp_server/tools/network.py b/src/linux_mcp_server/tools/network.py index 4d9e29d2..e294e62f 100644 --- a/src/linux_mcp_server/tools/network.py +++ b/src/linux_mcp_server/tools/network.py @@ -1,5 +1,8 @@ """Network diagnostic tools.""" +import json + +from fastmcp.exceptions import ToolError from mcp.types import ToolAnnotations from linux_mcp_server.audit import log_tool_call @@ -7,9 +10,8 @@ from linux_mcp_server.formatters import format_listening_ports from linux_mcp_server.formatters import format_network_connections from linux_mcp_server.formatters import format_network_interfaces -from linux_mcp_server.formatters import format_routes +from linux_mcp_server.models import Route from linux_mcp_server.parsers import parse_ip_brief -from linux_mcp_server.parsers import parse_ip_route from linux_mcp_server.parsers import parse_proc_net_dev from linux_mcp_server.parsers import parse_ss_connections from linux_mcp_server.parsers import parse_ss_listening @@ -110,14 +112,14 @@ async def get_listening_ports( @mcp.tool( title="Get network routes", description="Get the system routing table showing network routes, gateways, and interfaces.", - tags={"connectivity", "network", "routing"}, + tags={"fixed", "connectivity", "network", "routing"}, annotations=ToolAnnotations(readOnlyHint=True), ) @log_tool_call @disallow_local_execution_in_containers async def get_network_routes( host: Host = None, -) -> str: +) -> list[Route]: """Get network routing table. Retrieves the system routing table including destination networks, @@ -128,6 +130,5 @@ async def get_network_routes( returncode, stdout, stderr = await cmd.run(host=host) if is_successful_output(returncode, stdout): - routes = parse_ip_route(stdout) - return format_routes(routes) - return f"Error getting network routes: return code {returncode}, stderr: {stderr}" + return [Route.model_validate(entry) for entry in json.loads(stdout)] + raise ToolError(f"Error getting network routes: return code {returncode}, stderr: {stderr}") diff --git a/tests/test_server.py b/tests/test_server.py index ec26a011..b7fb1213 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -31,6 +31,7 @@ "get_memory_information", "get_network_connections", "get_network_interfaces", + "get_network_routes", "get_process_info", "get_service_logs", "get_service_status", diff --git a/tests/tools/test_network.py b/tests/tools/test_network.py index 27a8f0e3..27ed2f6f 100644 --- a/tests/tools/test_network.py +++ b/tests/tools/test_network.py @@ -1,5 +1,6 @@ """Tests for network diagnostic tools.""" +import json import re import pytest @@ -221,16 +222,49 @@ class TestGetNetworkRoutes: [ pytest.param( None, - "default via 192.168.1.1 dev eth0 proto dhcp src 192.168.1.100 metric 100\n" - "192.168.1.0/24 dev eth0 proto kernel scope link src 192.168.1.100\n" - "172.17.0.0/16 dev docker0 proto kernel scope link src 172.17.0.1", - ["192.168.1.1", "eth0", "Total routes: 3"], + json.dumps( + [ + { + "dst": "default", + "gateway": "192.168.1.1", + "dev": "eth0", + "protocol": "dhcp", + "prefsrc": "192.168.1.100", + "metric": 100, + }, + { + "dst": "192.168.1.0/24", + "dev": "eth0", + "protocol": "kernel", + "scope": "link", + "prefsrc": "192.168.1.100", + }, + { + "dst": "172.17.0.0/16", + "dev": "docker0", + "protocol": "kernel", + "scope": "link", + "prefsrc": "172.17.0.1", + }, + ] + ), + ["192.168.1.1", "eth0", "192.168.1.100"], id="local", ), pytest.param( "remote.host", - "default via 10.0.0.1 dev ens5 proto dhcp metric 100\n" - "10.0.0.0/24 dev ens5 proto kernel scope link src 10.0.0.5", + json.dumps( + [ + {"dst": "default", "gateway": "10.0.0.1", "dev": "ens5", "protocol": "dhcp", "metric": 100}, + { + "dst": "10.0.0.0/24", + "dev": "ens5", + "protocol": "kernel", + "scope": "link", + "prefsrc": "10.0.0.5", + }, + ] + ), ["10.0.0.1", "ens5"], id="remote", ), @@ -242,7 +276,6 @@ async def test_get_network_routes_success(self, mcp_client, mock_execute, host, result = await mcp_client.call_tool("get_network_routes", arguments={"host": host}) result_text = result.content[0].text.casefold() - assert "routing table" in result_text assert all(content.casefold() in result_text for content in expected_content), ( "Did not find all expected values" ) @@ -257,10 +290,8 @@ async def test_get_network_routes_success(self, mcp_client, mock_execute, host, async def test_get_network_routes_failure(self, mcp_client, mock_execute, return_value): """Test getting network routes when command fails or returns empty.""" mock_execute.return_value = return_value - result = await mcp_client.call_tool("get_network_routes") - result_text = result.content[0].text.casefold() - - assert "error" in result_text + with pytest.raises(ToolError, match="Error getting network routes"): + await mcp_client.call_tool("get_network_routes") async def test_get_network_routes_error(self, mcp_client, mock_execute): """Test getting network routes with general error."""