From 0eeb028ccf2bed5d0ec90acab544a956b4445964 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Tue, 20 Jan 2026 17:28:15 +0000 Subject: [PATCH 01/10] Convert list_services to return structured output --- src/linux_mcp_server/commands.py | 4 +++- src/linux_mcp_server/tools/services.py | 26 ++++++++++++-------------- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index 67165b5d..81bbe345 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -91,7 +91,9 @@ class CommandGroup(BaseModel): # === Services === "list_services": CommandGroup( commands={ - "default": CommandSpec(args=("systemctl", "list-units", "--type=service", "--all", "--no-pager")), + "default": CommandSpec( + args=("systemctl", "list-units", "--type=service", "--all", "--no-pager", "--output=json") + ), } ), "running_services": CommandGroup( diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 50b4d11a..8e4c8d2d 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -1,7 +1,9 @@ """Service management tools.""" +import json import typing as t +from fastmcp.exceptions import ToolError from mcp.types import ToolAnnotations from pydantic import Field @@ -9,8 +11,6 @@ from linux_mcp_server.commands import get_command from linux_mcp_server.formatters import format_service_logs from linux_mcp_server.formatters import format_service_status -from linux_mcp_server.formatters import format_services_list -from linux_mcp_server.parsers import parse_service_count from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -27,27 +27,25 @@ @disallow_local_execution_in_containers async def list_services( host: Host = None, -) -> str: +) -> list[dict[str, str]]: """List all systemd services. Retrieves all systemd service units with their load state, active state, - sub-state, and description. Also includes a count of currently running services. + sub-state, and description. + + Returns: + list[dict[str, str]]: A list of dictionaries containing information about each service. + + Raises: + ToolError: If an error occurs while listing services. """ cmd = get_command("list_services") returncode, stdout, stderr = await cmd.run(host=host) if returncode != 0: - return f"Error listing services: {stderr}" - - # Get running services count - running_cmd = get_command("running_services") - returncode_summary, stdout_summary, _ = await running_cmd.run(host=host) - - running_count = None - if returncode_summary == 0: - running_count = parse_service_count(stdout_summary) + raise ToolError(f"Error listing services: {stderr}") - return format_services_list(stdout, running_count) + return t.cast(list[dict[str, str]], json.loads(stdout)) @mcp.tool( From 917957808ed999902f1f4b5836863e33d404cdee Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Tue, 20 Jan 2026 13:41:23 -0500 Subject: [PATCH 02/10] Add parse_systemctl_show Convert the Key=Value pairs into a dict[str, str] --- src/linux_mcp_server/parsers.py | 17 +++++++++++ tests/parsers/test_parse_systemctl_show.py | 33 ++++++++++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 tests/parsers/test_parse_systemctl_show.py diff --git a/src/linux_mcp_server/parsers.py b/src/linux_mcp_server/parsers.py index d30a7911..30716242 100644 --- a/src/linux_mcp_server/parsers.py +++ b/src/linux_mcp_server/parsers.py @@ -457,6 +457,23 @@ def parse_service_count(stdout: str) -> int: return count +def parse_systemctl_show(stdout: str) -> dict[str, str]: + """Parse systemctl show output into key-value pairs. + + Args: + stdout: Raw output from systemctl show command. + + Returns: + Dictionary of key-value pairs. + """ + result: dict[str, str] = {} + for line in stdout.strip().split("\n"): + if "=" in line: + key, value = line.split("=", 1) + result[key.strip()] = value.strip() + return result + + def parse_directory_listing( stdout: str, sort_by: str, diff --git a/tests/parsers/test_parse_systemctl_show.py b/tests/parsers/test_parse_systemctl_show.py new file mode 100644 index 00000000..e63134e9 --- /dev/null +++ b/tests/parsers/test_parse_systemctl_show.py @@ -0,0 +1,33 @@ +"""Tests for parse_systemctl_show""" + +import pytest + +from linux_mcp_server.parsers import parse_systemctl_show + + +@pytest.mark.parametrize( + "stdout, expected", + [ + ( + """ + ActiveState=active + SubState=running + LoadState=loaded + """, + { + "ActiveState": "active", + "SubState": "running", + "LoadState": "loaded", + }, + ), + ( + """ + Field=value + EmptyField= + """, + {"Field": "value", "EmptyField": ""}, + ), + ], +) +def test_parse_systemctl_show(stdout, expected): + assert parse_systemctl_show(stdout) == expected From fd419e674982f60f81df7e5536fc081712bcf57b Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Tue, 20 Jan 2026 18:57:18 +0000 Subject: [PATCH 03/10] Convert get_service_status to structured output --- src/linux_mcp_server/commands.py | 2 +- src/linux_mcp_server/tools/services.py | 29 ++++++++------ tests/tools/test_services.py | 55 ++++++++++++++------------ 3 files changed, 49 insertions(+), 37 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index 81bbe345..6ca94972 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -105,7 +105,7 @@ class CommandGroup(BaseModel): ), "service_status": CommandGroup( commands={ - "default": CommandSpec(args=("systemctl", "status", "{service_name}", "--no-pager", "--full")), + "default": CommandSpec(args=("systemctl", "show", "{service_name}", "--no-pager")), } ), "service_logs": CommandGroup( diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 8e4c8d2d..04efe9bf 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -10,7 +10,7 @@ from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command from linux_mcp_server.formatters import format_service_logs -from linux_mcp_server.formatters import format_service_status +from linux_mcp_server.parsers import parse_systemctl_show from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -65,27 +65,34 @@ async def get_service_status( ), ], host: Host = None, -) -> str: +) -> dict[str, str]: """Get status of a specific systemd service. Retrieves detailed service information including active/enabled state, - main PID, memory usage, CPU time, and recent log entries from the journal. + main PID, memory usage, etc. + + Returns: + A dictionary containing the service status information. + + Raises: + ToolError: If there was an error getting the service status. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: service_name = f"{service_name}.service" cmd = get_command("service_status") - _, stdout, stderr = await cmd.run(host=host, service_name=service_name) + returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name) + + if returncode != 0: + raise ToolError(f"Error getting service status: {stderr}") + + status = parse_systemctl_show(stdout) - # Note: systemctl status returns non-zero for inactive services, but that's expected - if not stdout and stderr: - # Service not found - if "not found" in stderr.lower() or "could not be found" in stderr.lower(): - return f"Service '{service_name}' not found on this system." - return f"Error getting service status: {stderr}" + if status.get("LoadState") == "not-found": + raise ToolError(f"Service '{service_name}' not found on this system.") - return format_service_status(stdout, service_name) + return status @mcp.tool( diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index f9d49710..44104502 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -4,6 +4,8 @@ import pytest +from fastmcp.exceptions import ToolError + @pytest.mark.skipif(sys.platform != "linux", reason="Only passes on Linux") class TestServices: @@ -17,30 +19,28 @@ async def test_list_services(self, mcp_client): assert all(any(n in result_text for n in case) for case in expected), "Did not find all expected values" - @pytest.mark.parametrize( - "service_name, expected", - ( - ("sshd.service", ("active", "inactive", "loaded", "not found")), - ("nonexistent-service-xyz123", ("not found", "could not", "error")), - ), - ) - async def test_get_service_status(self, mcp_client, service_name, expected): - result = await mcp_client.call_tool("get_service_status", arguments={"service_name": service_name}) - result_text = result.content[0].text.casefold() + async def test_get_service_status(self, mcp_client): + """Test getting service status returns structured data.""" + result = await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd.service"}) - assert any(n in result_text for n in expected), "Did not find any expected values" + # The tool returns structured data (dict), so check the structured content + assert result.structured_content is not None, "Expected structured data" + + # Verify we have service status fields + data = result.structured_content + assert "LoadState" in data or "ActiveState" in data, "Expected service status fields in structured data" + + # ActiveState should be one of the valid states + if "ActiveState" in data: + assert data["ActiveState"] in ("active", "inactive", "activating", "deactivating", "failed"), ( + f"Unexpected ActiveState: {data['ActiveState']}" + ) async def test_get_service_status_with_nonexistent_service(self, mcp_client): - result = await mcp_client.call_tool( - "get_service_status", arguments={"service_name": "nonexistent-service-xyz123"} - ) - result_text = result.content[0].text.casefold() - expected = ( - "not found", - "could not", - "error", - ) - assert any(n in result_text for n in expected), "Did not find any expected values" + """Test that nonexistent service raises ToolError.""" + + with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): + await mcp_client.call_tool("get_service_status", arguments={"service_name": "nonexistent-service-xyz123"}) async def test_get_service_logs(self, mcp_client): result = await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd.service", "lines": 5}) @@ -84,16 +84,21 @@ async def test_list_services_remote(self, mock_execute_with_fallback, mcp_client async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_client): """Test getting service status on a remote host.""" - mock_output = "● nginx.service - Nginx HTTP Server\n Loaded: loaded\n Active: active (running)" + # Mock systemctl show output (key=value format) + mock_output = "LoadState=loaded\nActiveState=active\nSubState=running\nDescription=Nginx HTTP Server" mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool( "get_service_status", arguments={"service_name": "nginx", "host": "remote.example.com"} ) - result_text = result.content[0].text.casefold() - assert "nginx.service" in result_text - assert "active" in result_text + # The tool now returns structured data + assert result.structured_content is not None + data = result.structured_content + + assert data.get("LoadState") == "loaded" + assert data.get("ActiveState") == "active" + assert data.get("SubState") == "running" mock_execute_with_fallback.assert_called() async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_client): From 93d60f5a8c67e964aa6e06be0f14f7cb98eaec74 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Wed, 21 Jan 2026 15:10:03 +0000 Subject: [PATCH 04/10] Convert get_service_logs to structured output --- src/linux_mcp_server/commands.py | 4 +- src/linux_mcp_server/tools/services.py | 17 +++--- tests/tools/test_services.py | 78 +++++++++++++++++--------- 3 files changed, 65 insertions(+), 34 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index 6ca94972..cc46775c 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -110,7 +110,9 @@ class CommandGroup(BaseModel): ), "service_logs": CommandGroup( commands={ - "default": CommandSpec(args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager")), + "default": CommandSpec( + args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager", "--output=json") + ), } ), # === Network === diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 04efe9bf..f7d297f1 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -9,7 +9,6 @@ from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command -from linux_mcp_server.formatters import format_service_logs from linux_mcp_server.parsers import parse_systemctl_show from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers @@ -113,11 +112,17 @@ async def get_service_logs( ], lines: t.Annotated[int, Field(description="Number of log lines to retrieve.", ge=1, le=10_000)] = 50, host: Host = None, -) -> str: +) -> list[dict[str, str]]: """Get recent logs for a specific systemd service. Retrieves journal entries for the specified service unit, including timestamps, priority levels, and log messages. + + Raises: + ToolError: If an error occurs while retrieving logs. + + Returns: + list[dict[str, str]]: A list of dictionaries containing log entries. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: @@ -127,11 +132,9 @@ async def get_service_logs( returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name, lines=lines) if returncode != 0: - if "not found" in stderr.lower() or "no entries" in stderr.lower(): - return f"No logs found for service '{service_name}'. The service may not exist or has no log entries." - return f"Error getting service logs: {stderr}" + raise ToolError(f"Error getting service logs: {stderr}") if is_empty_output(stdout): - return f"No log entries found for service '{service_name}'." + raise ToolError(f"No log entries found for service '{service_name}'.") - return format_service_logs(stdout, service_name, lines) + return t.cast(list[dict[str, str]], json.loads(stdout)) diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index 44104502..70e3ba71 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -42,29 +42,46 @@ async def test_get_service_status_with_nonexistent_service(self, mcp_client): with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): await mcp_client.call_tool("get_service_status", arguments={"service_name": "nonexistent-service-xyz123"}) - async def test_get_service_logs(self, mcp_client): + async def test_get_service_status_error(self, mock_execute_with_fallback, mcp_client): + """Test that get_service_status raises ToolError when systemctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to get unit file state: Connection refused") + + with pytest.raises(ToolError, match="Error getting service status: Failed to get unit file state"): + await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd"}) + + async def test_get_service_logs(self, mock_execute_with_fallback, mcp_client): + """Test getting service logs with mocked output.""" + mock_output = '[{"__REALTIME_TIMESTAMP": "1600000000000000", "MESSAGE": "sshd: session opened for user test", "PRIORITY": "6", "_PID": "1234"}, {"__REALTIME_TIMESTAMP": "1600000001000000", "MESSAGE": "sshd: session closed for user test", "PRIORITY": "6", "_PID": "1234"}]' + mock_execute_with_fallback.return_value = (0, mock_output, "") + result = await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd.service", "lines": 5}) - # Filter out empty lines, header lines (=), and journalctl boot markers (--) - result_lines = [ - line - for line in result.content[0].text.split("\n") - if line and not line.startswith("=") and not line.startswith("--") - ] - assert len(result_lines) <= 5, "Got more lines than expected" + assert result.structured_content is not None, "Expected structured data" + logs = result.structured_content["result"] + assert isinstance(logs, list) + assert len(logs) == 2 + assert logs[0]["MESSAGE"] == "sshd: session opened for user test" + mock_execute_with_fallback.assert_called() async def test_get_service_logs_with_nonexistent_service(self, mcp_client): - result = await mcp_client.call_tool( - "get_service_logs", arguments={"service_name": "nonexistent-service-xyz123", "lines": 10} - ) - result_text = result.content[0].text.casefold() - expected = ( - "not found", - "no entries", - "error", - ) + with pytest.raises(ToolError, match="No log entries found for service 'nonexistent-service-xyz123.service'."): + await mcp_client.call_tool( + "get_service_logs", arguments={"service_name": "nonexistent-service-xyz123", "lines": 10} + ) + + async def test_get_service_logs_error(self, mock_execute_with_fallback, mcp_client): + """Test that get_service_logs raises ToolError when journalctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to access journal: Permission denied") + + with pytest.raises(ToolError, match="Error getting service logs: Failed to access journal"): + await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd", "lines": 10}) + + async def test_list_services_error(self, mock_execute_with_fallback, mcp_client): + """Test that list_services raises ToolError when systemctl fails.""" + mock_execute_with_fallback.return_value = (1, "", "Failed to connect to bus: No such file or directory") - assert any(n in result_text for n in expected), "Did not find any expected values" + with pytest.raises(ToolError, match="Error listing services: Failed to connect to bus"): + await mcp_client.call_tool("list_services") class TestRemoteServices: @@ -72,14 +89,19 @@ class TestRemoteServices: async def test_list_services_remote(self, mock_execute_with_fallback, mcp_client): """Test listing services on a remote host.""" - mock_output = "UNIT LOAD ACTIVE SUB DESCRIPTION\nnginx.service loaded active running Nginx server\n" + mock_output = '[{"unit":"nginx.service","load":"loaded","active":"active","sub":"running","description":"Nginx HTTP Server"}]' mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool("list_services", arguments={"host": "remote.example.com"}) - result_text = result.content[0].text.casefold() - assert "nginx.service" in result_text - assert "system services" in result_text + assert result.structured_content is not None + services = result.structured_content["result"] + assert isinstance(services, list) + assert services[0]["unit"] == "nginx.service" + assert services[0]["load"] == "loaded" + assert services[0]["active"] == "active" + assert services[0]["sub"] == "running" + assert services[0]["description"] == "Nginx HTTP Server" mock_execute_with_fallback.assert_called() async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_client): @@ -103,14 +125,18 @@ async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_c async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_client): """Test getting service logs on a remote host.""" - mock_output = "Jan 01 12:00:00 host nginx[1234]: Starting Nginx\nJan 01 12:00:01 host nginx[1234]: Started" + mock_output = '[{"_REALTIME_TIMESTAMP": "Jan 01 12:00:00", "_PID": "1234", "MESSAGE": "Starting Nginx"}, {"_REALTIME_TIMESTAMP": "Jan 01 12:00:01", "_PID": "1234", "MESSAGE": "Started"}]' mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool( "get_service_logs", arguments={"service_name": "nginx", "host": "remote.example.com", "lines": 50} ) - result_text = result.content[0].text.casefold() - assert "nginx" in result_text - assert "starting" in result_text + assert result.structured_content is not None + data = result.structured_content["result"] + + assert isinstance(data, list) + assert len(data) == 2 + assert data[0]["MESSAGE"] == "Starting Nginx" + assert data[1]["MESSAGE"] == "Started" mock_execute_with_fallback.assert_called() From b8b4ab572d7fac9b29a960eb0a2f7577928f6f41 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Thu, 2 Apr 2026 11:23:02 -0400 Subject: [PATCH 05/10] Revert "Convert get_service_status to structured output" This reverts commit fd419e674982f60f81df7e5536fc081712bcf57b. --- src/linux_mcp_server/commands.py | 2 +- src/linux_mcp_server/tools/services.py | 29 ++++++-------- tests/tools/test_services.py | 55 ++++++++++++-------------- 3 files changed, 37 insertions(+), 49 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index cc46775c..af7eb079 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -105,7 +105,7 @@ class CommandGroup(BaseModel): ), "service_status": CommandGroup( commands={ - "default": CommandSpec(args=("systemctl", "show", "{service_name}", "--no-pager")), + "default": CommandSpec(args=("systemctl", "status", "{service_name}", "--no-pager", "--full")), } ), "service_logs": CommandGroup( diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index f7d297f1..6fda2b71 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -9,7 +9,7 @@ from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command -from linux_mcp_server.parsers import parse_systemctl_show +from linux_mcp_server.formatters import format_service_status from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -64,34 +64,27 @@ async def get_service_status( ), ], host: Host = None, -) -> dict[str, str]: +) -> str: """Get status of a specific systemd service. Retrieves detailed service information including active/enabled state, - main PID, memory usage, etc. - - Returns: - A dictionary containing the service status information. - - Raises: - ToolError: If there was an error getting the service status. + main PID, memory usage, CPU time, and recent log entries from the journal. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: service_name = f"{service_name}.service" cmd = get_command("service_status") - returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name) - - if returncode != 0: - raise ToolError(f"Error getting service status: {stderr}") - - status = parse_systemctl_show(stdout) + _, stdout, stderr = await cmd.run(host=host, service_name=service_name) - if status.get("LoadState") == "not-found": - raise ToolError(f"Service '{service_name}' not found on this system.") + # Note: systemctl status returns non-zero for inactive services, but that's expected + if not stdout and stderr: + # Service not found + if "not found" in stderr.lower() or "could not be found" in stderr.lower(): + return f"Service '{service_name}' not found on this system." + return f"Error getting service status: {stderr}" - return status + return format_service_status(stdout, service_name) @mcp.tool( diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index 70e3ba71..7be3d5dc 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -4,8 +4,6 @@ import pytest -from fastmcp.exceptions import ToolError - @pytest.mark.skipif(sys.platform != "linux", reason="Only passes on Linux") class TestServices: @@ -19,28 +17,30 @@ async def test_list_services(self, mcp_client): assert all(any(n in result_text for n in case) for case in expected), "Did not find all expected values" - async def test_get_service_status(self, mcp_client): - """Test getting service status returns structured data.""" - result = await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd.service"}) - - # The tool returns structured data (dict), so check the structured content - assert result.structured_content is not None, "Expected structured data" - - # Verify we have service status fields - data = result.structured_content - assert "LoadState" in data or "ActiveState" in data, "Expected service status fields in structured data" + @pytest.mark.parametrize( + "service_name, expected", + ( + ("sshd.service", ("active", "inactive", "loaded", "not found")), + ("nonexistent-service-xyz123", ("not found", "could not", "error")), + ), + ) + async def test_get_service_status(self, mcp_client, service_name, expected): + result = await mcp_client.call_tool("get_service_status", arguments={"service_name": service_name}) + result_text = result.content[0].text.casefold() - # ActiveState should be one of the valid states - if "ActiveState" in data: - assert data["ActiveState"] in ("active", "inactive", "activating", "deactivating", "failed"), ( - f"Unexpected ActiveState: {data['ActiveState']}" - ) + assert any(n in result_text for n in expected), "Did not find any expected values" async def test_get_service_status_with_nonexistent_service(self, mcp_client): - """Test that nonexistent service raises ToolError.""" - - with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): - await mcp_client.call_tool("get_service_status", arguments={"service_name": "nonexistent-service-xyz123"}) + result = await mcp_client.call_tool( + "get_service_status", arguments={"service_name": "nonexistent-service-xyz123"} + ) + result_text = result.content[0].text.casefold() + expected = ( + "not found", + "could not", + "error", + ) + assert any(n in result_text for n in expected), "Did not find any expected values" async def test_get_service_status_error(self, mock_execute_with_fallback, mcp_client): """Test that get_service_status raises ToolError when systemctl fails.""" @@ -106,21 +106,16 @@ async def test_list_services_remote(self, mock_execute_with_fallback, mcp_client async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_client): """Test getting service status on a remote host.""" - # Mock systemctl show output (key=value format) - mock_output = "LoadState=loaded\nActiveState=active\nSubState=running\nDescription=Nginx HTTP Server" + mock_output = "● nginx.service - Nginx HTTP Server\n Loaded: loaded\n Active: active (running)" mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool( "get_service_status", arguments={"service_name": "nginx", "host": "remote.example.com"} ) + result_text = result.content[0].text.casefold() - # The tool now returns structured data - assert result.structured_content is not None - data = result.structured_content - - assert data.get("LoadState") == "loaded" - assert data.get("ActiveState") == "active" - assert data.get("SubState") == "running" + assert "nginx.service" in result_text + assert "active" in result_text mock_execute_with_fallback.assert_called() async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_client): From abcba535ef6ec4367fd2bd8b85ee29e47e7bf58c Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Thu, 2 Apr 2026 11:23:02 -0400 Subject: [PATCH 06/10] Revert "Add parse_systemctl_show" This reverts commit 917957808ed999902f1f4b5836863e33d404cdee. --- src/linux_mcp_server/parsers.py | 17 ----------- tests/parsers/test_parse_systemctl_show.py | 33 ---------------------- 2 files changed, 50 deletions(-) delete mode 100644 tests/parsers/test_parse_systemctl_show.py diff --git a/src/linux_mcp_server/parsers.py b/src/linux_mcp_server/parsers.py index 30716242..d30a7911 100644 --- a/src/linux_mcp_server/parsers.py +++ b/src/linux_mcp_server/parsers.py @@ -457,23 +457,6 @@ def parse_service_count(stdout: str) -> int: return count -def parse_systemctl_show(stdout: str) -> dict[str, str]: - """Parse systemctl show output into key-value pairs. - - Args: - stdout: Raw output from systemctl show command. - - Returns: - Dictionary of key-value pairs. - """ - result: dict[str, str] = {} - for line in stdout.strip().split("\n"): - if "=" in line: - key, value = line.split("=", 1) - result[key.strip()] = value.strip() - return result - - def parse_directory_listing( stdout: str, sort_by: str, diff --git a/tests/parsers/test_parse_systemctl_show.py b/tests/parsers/test_parse_systemctl_show.py deleted file mode 100644 index e63134e9..00000000 --- a/tests/parsers/test_parse_systemctl_show.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Tests for parse_systemctl_show""" - -import pytest - -from linux_mcp_server.parsers import parse_systemctl_show - - -@pytest.mark.parametrize( - "stdout, expected", - [ - ( - """ - ActiveState=active - SubState=running - LoadState=loaded - """, - { - "ActiveState": "active", - "SubState": "running", - "LoadState": "loaded", - }, - ), - ( - """ - Field=value - EmptyField= - """, - {"Field": "value", "EmptyField": ""}, - ), - ], -) -def test_parse_systemctl_show(stdout, expected): - assert parse_systemctl_show(stdout) == expected From 81ebd6620ec6e892731b3c5114a6362cc62d78ae Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Thu, 2 Apr 2026 14:30:59 +0000 Subject: [PATCH 07/10] Update get_service_status to raise ToolError instead of returning error strings --- src/linux_mcp_server/formatters.py | 15 -------------- src/linux_mcp_server/tools/services.py | 22 +++++++++++++++------ tests/test_formatters.py | 12 ------------ tests/tools/test_services.py | 27 +++++++------------------- 4 files changed, 23 insertions(+), 53 deletions(-) diff --git a/src/linux_mcp_server/formatters.py b/src/linux_mcp_server/formatters.py index 2cf49090..f330b253 100644 --- a/src/linux_mcp_server/formatters.py +++ b/src/linux_mcp_server/formatters.py @@ -189,21 +189,6 @@ def format_services_list(stdout: str, running_count: int | None = None) -> str: return "\n".join(lines) -def format_service_status(stdout: str, service_name: str) -> str: - """Format service status output. - - Args: - stdout: Raw output from systemctl status. - service_name: Name of the service. - - Returns: - Formatted string representation. - """ - lines = [f"=== Status of {service_name} ===\n"] - lines.append(stdout) - return "\n".join(lines) - - def format_service_logs(stdout: str, service_name: str, lines_count: int) -> str: """Format service logs output. diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 6fda2b71..7b4560c8 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -9,7 +9,6 @@ from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command -from linux_mcp_server.formatters import format_service_status from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -68,23 +67,34 @@ async def get_service_status( """Get status of a specific systemd service. Retrieves detailed service information including active/enabled state, - main PID, memory usage, CPU time, and recent log entries from the journal. + main PID, memory usage, etc. + + Returns: + A string containing the output of the systemctl status command. + + Raises: + ToolError: If there was an error getting the service status. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: service_name = f"{service_name}.service" cmd = get_command("service_status") - _, stdout, stderr = await cmd.run(host=host, service_name=service_name) + returncode, stdout, stderr = await cmd.run(host=host, service_name=service_name) + + # systemctl status uses exit status 4 when the unit does not exist (see systemctl(1)). + # This is stable across locales; stderr text alone may be translated. + if returncode == 4: + raise ToolError(f"Service '{service_name}' not found on this system.") # Note: systemctl status returns non-zero for inactive services, but that's expected if not stdout and stderr: # Service not found if "not found" in stderr.lower() or "could not be found" in stderr.lower(): - return f"Service '{service_name}' not found on this system." - return f"Error getting service status: {stderr}" + raise ToolError(f"Service '{service_name}' not found on this system.") + raise ToolError(f"Error getting service status: {stderr}") - return format_service_status(stdout, service_name) + return stdout @mcp.tool( diff --git a/tests/test_formatters.py b/tests/test_formatters.py index 37de51b0..f6f3a996 100644 --- a/tests/test_formatters.py +++ b/tests/test_formatters.py @@ -8,7 +8,6 @@ from linux_mcp_server.formatters import format_process_detail from linux_mcp_server.formatters import format_process_list from linux_mcp_server.formatters import format_service_logs -from linux_mcp_server.formatters import format_service_status from linux_mcp_server.formatters import format_services_list from linux_mcp_server.models import ListeningPort from linux_mcp_server.models import NetworkConnection @@ -221,17 +220,6 @@ def test_format_with_count(self): assert "Summary: 5 services currently running" in result -class TestFormatServiceStatus: - """Tests for format_service_status function.""" - - def test_format_service_status(self): - """Test formatting service status.""" - stdout = "● ssh.service - OpenBSD Secure Shell server\n Active: active (running)" - result = format_service_status(stdout, "ssh.service") - assert "=== Status of ssh.service ===" in result - assert "Active: active (running)" in result - - class TestFormatServiceLogs: """Tests for format_service_logs function.""" diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index 7be3d5dc..2676bdc2 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -4,6 +4,8 @@ import pytest +from fastmcp.exceptions import ToolError + @pytest.mark.skipif(sys.platform != "linux", reason="Only passes on Linux") class TestServices: @@ -17,30 +19,15 @@ async def test_list_services(self, mcp_client): assert all(any(n in result_text for n in case) for case in expected), "Did not find all expected values" - @pytest.mark.parametrize( - "service_name, expected", - ( - ("sshd.service", ("active", "inactive", "loaded", "not found")), - ("nonexistent-service-xyz123", ("not found", "could not", "error")), - ), - ) - async def test_get_service_status(self, mcp_client, service_name, expected): - result = await mcp_client.call_tool("get_service_status", arguments={"service_name": service_name}) + async def test_get_service_status(self, mcp_client): + result = await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd.service"}) result_text = result.content[0].text.casefold() - + expected = ("active", "inactive", "loaded", "not found") assert any(n in result_text for n in expected), "Did not find any expected values" async def test_get_service_status_with_nonexistent_service(self, mcp_client): - result = await mcp_client.call_tool( - "get_service_status", arguments={"service_name": "nonexistent-service-xyz123"} - ) - result_text = result.content[0].text.casefold() - expected = ( - "not found", - "could not", - "error", - ) - assert any(n in result_text for n in expected), "Did not find any expected values" + with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): + await mcp_client.call_tool("get_service_status", arguments={"service_name": "nonexistent-service-xyz123"}) async def test_get_service_status_error(self, mock_execute_with_fallback, mcp_client): """Test that get_service_status raises ToolError when systemctl fails.""" From df692572efbac02a36d2a6d211b985e6ec0b2edb Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Thu, 2 Apr 2026 15:13:08 +0000 Subject: [PATCH 08/10] Mock the sshd.service service So that tests can pass even when sshd.service is not installed. --- tests/tools/test_services.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index 2676bdc2..72c681d2 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -19,11 +19,18 @@ async def test_list_services(self, mcp_client): assert all(any(n in result_text for n in case) for case in expected), "Did not find all expected values" - async def test_get_service_status(self, mcp_client): + async def test_get_service_status(self, mock_execute_with_fallback, mcp_client): + """Test getting service status with mocked systemctl output (no real sshd unit required).""" + mock_output = "● sshd.service - OpenSSH server\n Loaded: loaded\n Active: active (running)" + mock_execute_with_fallback.return_value = (0, mock_output, "") + result = await mcp_client.call_tool("get_service_status", arguments={"service_name": "sshd.service"}) result_text = result.content[0].text.casefold() - expected = ("active", "inactive", "loaded", "not found") - assert any(n in result_text for n in expected), "Did not find any expected values" + + assert "sshd.service" in result_text + assert "loaded" in result_text + assert "active" in result_text + mock_execute_with_fallback.assert_called() async def test_get_service_status_with_nonexistent_service(self, mcp_client): with pytest.raises(ToolError, match="Service 'nonexistent-service-xyz123.service' not found on this system."): From ad53afdd5f4dcacd6856172d9d9e89ec549f62f7 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Wed, 15 Apr 2026 06:52:12 -0400 Subject: [PATCH 09/10] Refactor get_service_logs to return structured LogEntries --- src/linux_mcp_server/commands.py | 2 +- src/linux_mcp_server/tools/services.py | 13 ++++++++----- tests/tools/test_services.py | 27 +++++++++++++++----------- 3 files changed, 25 insertions(+), 17 deletions(-) diff --git a/src/linux_mcp_server/commands.py b/src/linux_mcp_server/commands.py index af7eb079..c98198da 100644 --- a/src/linux_mcp_server/commands.py +++ b/src/linux_mcp_server/commands.py @@ -111,7 +111,7 @@ class CommandGroup(BaseModel): "service_logs": CommandGroup( commands={ "default": CommandSpec( - args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager", "--output=json") + args=("journalctl", "-u", "{service_name}", "-n", "{lines}", "--no-pager"), ), } ), diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 7b4560c8..5bf958b1 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -9,6 +9,7 @@ from linux_mcp_server.audit import log_tool_call from linux_mcp_server.commands import get_command +from linux_mcp_server.models import LogEntries from linux_mcp_server.server import mcp from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers from linux_mcp_server.utils.types import Host @@ -115,17 +116,17 @@ async def get_service_logs( ], lines: t.Annotated[int, Field(description="Number of log lines to retrieve.", ge=1, le=10_000)] = 50, host: Host = None, -) -> list[dict[str, str]]: +) -> LogEntries: """Get recent logs for a specific systemd service. - Retrieves journal entries for the specified service unit, including - timestamps, priority levels, and log messages. + Retrieves journal lines for the specified service unit (default journalctl + short format, same style as get_journal_logs). Raises: ToolError: If an error occurs while retrieving logs. Returns: - list[dict[str, str]]: A list of dictionaries containing log entries. + LogEntries: Structured log lines, unit name, and line count. """ # Ensure service name has .service suffix if not present if not service_name.endswith(".service") and "." not in service_name: @@ -140,4 +141,6 @@ async def get_service_logs( if is_empty_output(stdout): raise ToolError(f"No log entries found for service '{service_name}'.") - return t.cast(list[dict[str, str]], json.loads(stdout)) + entries = [line for line in stdout.strip().splitlines() if line] + + return LogEntries(entries=entries, unit=service_name) diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index 72c681d2..e809279f 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -45,16 +45,20 @@ async def test_get_service_status_error(self, mock_execute_with_fallback, mcp_cl async def test_get_service_logs(self, mock_execute_with_fallback, mcp_client): """Test getting service logs with mocked output.""" - mock_output = '[{"__REALTIME_TIMESTAMP": "1600000000000000", "MESSAGE": "sshd: session opened for user test", "PRIORITY": "6", "_PID": "1234"}, {"__REALTIME_TIMESTAMP": "1600000001000000", "MESSAGE": "sshd: session closed for user test", "PRIORITY": "6", "_PID": "1234"}]' + mock_output = ( + "Jan 01 12:00:00 host sshd[1234]: session opened for user test\n" + "Jan 01 12:00:01 host sshd[1234]: session closed for user test" + ) mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool("get_service_logs", arguments={"service_name": "sshd.service", "lines": 5}) assert result.structured_content is not None, "Expected structured data" - logs = result.structured_content["result"] - assert isinstance(logs, list) - assert len(logs) == 2 - assert logs[0]["MESSAGE"] == "sshd: session opened for user test" + content = result.structured_content + assert content["lines_count"] == 2 + assert len(content["entries"]) == 2 + assert "session opened for user test" in content["entries"][0] + assert content["unit"] == "sshd.service" mock_execute_with_fallback.assert_called() async def test_get_service_logs_with_nonexistent_service(self, mcp_client): @@ -114,7 +118,7 @@ async def test_get_service_status_remote(self, mock_execute_with_fallback, mcp_c async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_client): """Test getting service logs on a remote host.""" - mock_output = '[{"_REALTIME_TIMESTAMP": "Jan 01 12:00:00", "_PID": "1234", "MESSAGE": "Starting Nginx"}, {"_REALTIME_TIMESTAMP": "Jan 01 12:00:01", "_PID": "1234", "MESSAGE": "Started"}]' + mock_output = "Jan 01 12:00:00 remote nginx[1]: Starting Nginx\nJan 01 12:00:01 remote nginx[1]: Started" mock_execute_with_fallback.return_value = (0, mock_output, "") result = await mcp_client.call_tool( @@ -122,10 +126,11 @@ async def test_get_service_logs_remote(self, mock_execute_with_fallback, mcp_cli ) assert result.structured_content is not None - data = result.structured_content["result"] + content = result.structured_content - assert isinstance(data, list) - assert len(data) == 2 - assert data[0]["MESSAGE"] == "Starting Nginx" - assert data[1]["MESSAGE"] == "Started" + assert content["lines_count"] == 2 + assert len(content["entries"]) == 2 + assert "Starting Nginx" in content["entries"][0] + assert "Started" in content["entries"][1] + assert content["unit"] == "nginx.service" mock_execute_with_fallback.assert_called() From d4b1d31799691935e883367df404c9dc80c08da1 Mon Sep 17 00:00:00 2001 From: Link Dupont Date: Wed, 15 Apr 2026 07:43:46 -0400 Subject: [PATCH 10/10] Enhance get_service_logs documentation and error handling Modified the test for nonexistent services to ensure it correctly raises ToolError when no log entries are found. --- src/linux_mcp_server/tools/services.py | 2 +- tests/tools/test_services.py | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/src/linux_mcp_server/tools/services.py b/src/linux_mcp_server/tools/services.py index 5bf958b1..ae9560d9 100644 --- a/src/linux_mcp_server/tools/services.py +++ b/src/linux_mcp_server/tools/services.py @@ -123,7 +123,7 @@ async def get_service_logs( short format, same style as get_journal_logs). Raises: - ToolError: If an error occurs while retrieving logs. + ToolError: If journal output is empty, or journalctl fails. Returns: LogEntries: Structured log lines, unit name, and line count. diff --git a/tests/tools/test_services.py b/tests/tools/test_services.py index e809279f..c93a9a0e 100644 --- a/tests/tools/test_services.py +++ b/tests/tools/test_services.py @@ -61,12 +61,17 @@ async def test_get_service_logs(self, mock_execute_with_fallback, mcp_client): assert content["unit"] == "sshd.service" mock_execute_with_fallback.assert_called() - async def test_get_service_logs_with_nonexistent_service(self, mcp_client): + async def test_get_service_logs_with_nonexistent_service(self, mock_execute_with_fallback, mcp_client): + """Empty journal (e.g. unknown unit with no lines) raises; real journalctl output is not deterministic.""" + mock_execute_with_fallback.return_value = (0, "", "") + with pytest.raises(ToolError, match="No log entries found for service 'nonexistent-service-xyz123.service'."): await mcp_client.call_tool( "get_service_logs", arguments={"service_name": "nonexistent-service-xyz123", "lines": 10} ) + mock_execute_with_fallback.assert_called() + async def test_get_service_logs_error(self, mock_execute_with_fallback, mcp_client): """Test that get_service_logs raises ToolError when journalctl fails.""" mock_execute_with_fallback.return_value = (1, "", "Failed to access journal: Permission denied")