From d6ebec4b81c979587c28fac9ea4f8806b2e2f5e0 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Tue, 4 Nov 2025 23:39:01 -0600 Subject: [PATCH 1/7] add management command tool for executing Django commands --- CHANGELOG.md | 4 + README.md | 1 + src/mcp_django/project/management.py | 206 +++++++++++++++++++++++++++ src/mcp_django/project/server.py | 77 ++++++++++ tests/test_management_command.py | 181 +++++++++++++++++++++++ tests/test_server.py | 1 + 6 files changed, 470 insertions(+) create mode 100644 src/mcp_django/project/management.py create mode 100644 tests/test_management_command.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ff608d..afeae8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,10 @@ For multi-package releases, use package names as subsections: ## [Unreleased] +### Added + +- Added `management_command` tool for executing Django management commands with arguments and options + ## [0.12.0] ### Added diff --git a/README.md b/README.md index 3c07b19..b5f7242 100644 --- a/README.md +++ b/README.md @@ -205,6 +205,7 @@ Read-only resources for project exploration without executing code (note that re | `list_apps` | List all installed Django applications with their models | | `list_models` | Get detailed information about Django models with optional filtering by app or scope | | `list_routes` | Introspect Django URL routes with filtering support for HTTP method, route name, or URL pattern | +| `management_command` | Execute Django management commands with arguments and options | #### Shell diff --git a/src/mcp_django/project/management.py b/src/mcp_django/project/management.py new file mode 100644 index 0000000..e3e6d2a --- /dev/null +++ b/src/mcp_django/project/management.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import logging +from contextlib import redirect_stderr +from contextlib import redirect_stdout +from dataclasses import dataclass +from dataclasses import field +from datetime import datetime +from io import StringIO + +from asgiref.sync import sync_to_async +from django.core.management import call_command +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import field_serializer + +logger = logging.getLogger(__name__) + + +@dataclass +class CommandResult: + """Result from successfully executing a management command.""" + + command: str + args: tuple[str, ...] + options: dict[str, str | int | bool] + stdout: str + stderr: str + timestamp: datetime = field(default_factory=datetime.now) + + def __post_init__(self): + logger.debug( + "%s created for command: %s", self.__class__.__name__, self.command + ) + if self.stdout: + logger.debug("%s.stdout: %s", self.__class__.__name__, self.stdout[:200]) + if self.stderr: + logger.debug("%s.stderr: %s", self.__class__.__name__, self.stderr[:200]) + + +@dataclass +class CommandErrorResult: + """Result from a management command that raised an exception.""" + + command: str + args: tuple[str, ...] + options: dict[str, str | int | bool] + exception: Exception + stdout: str + stderr: str + timestamp: datetime = field(default_factory=datetime.now) + + def __post_init__(self): + logger.debug( + "%s created for command: %s - exception type: %s", + self.__class__.__name__, + self.command, + type(self.exception).__name__, + ) + logger.debug("%s.message: %s", self.__class__.__name__, str(self.exception)) + if self.stdout: + logger.debug("%s.stdout: %s", self.__class__.__name__, self.stdout[:200]) + if self.stderr: + logger.debug("%s.stderr: %s", self.__class__.__name__, self.stderr[:200]) + + +Result = CommandResult | CommandErrorResult + + +class ManagementCommandOutput(BaseModel): + """Output from executing a Django management command.""" + + status: str # "success" or "error" + command: str + args: list[str] + options: dict[str, str | int | bool] + stdout: str + stderr: str + exception: ExceptionInfo | None = None + + @classmethod + def from_result(cls, result: Result) -> ManagementCommandOutput: + """Create output from a command result.""" + match result: + case CommandResult(): + return cls( + status="success", + command=result.command, + args=list(result.args), + options=result.options, + stdout=result.stdout, + stderr=result.stderr, + exception=None, + ) + case CommandErrorResult(): + return cls( + status="error", + command=result.command, + args=list(result.args), + options=result.options, + stdout=result.stdout, + stderr=result.stderr, + exception=ExceptionInfo( + type=type(result.exception).__name__, + message=str(result.exception), + ), + ) + + +class ExceptionInfo(BaseModel): + """Information about an exception that occurred during command execution.""" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + type: str + message: str + + +class ManagementCommandExecutor: + """Executor for Django management commands with output capture.""" + + async def execute( + self, + command: str, + args: list[str] | None = None, + options: dict[str, str | int | bool] | None = None, + ) -> Result: + """Execute a Django management command asynchronously. + + Args: + command: The management command name (e.g., 'migrate', 'check') + args: Positional arguments for the command + options: Keyword options for the command + + Returns: + CommandResult if successful, CommandErrorResult if an exception occurred + """ + return await sync_to_async(self._execute)(command, args, options) + + def _execute( + self, + command: str, + args: list[str] | None = None, + options: dict[str, str | int | bool] | None = None, + ) -> Result: + """Execute a Django management command synchronously. + + Captures stdout and stderr from the command execution. + + Args: + command: The management command name + args: Positional arguments for the command + options: Keyword options for the command + + Returns: + CommandResult if successful, CommandErrorResult if an exception occurred + """ + args = args or [] + options = options or {} + + args_tuple = tuple(args) + options_dict = dict(options) + + logger.info( + "Executing management command: %s with args=%s, options=%s", + command, + args_tuple, + options_dict, + ) + + stdout = StringIO() + stderr = StringIO() + + with redirect_stdout(stdout), redirect_stderr(stderr): + try: + call_command(command, *args_tuple, **options_dict) + + logger.debug("Management command executed successfully: %s", command) + + return CommandResult( + command=command, + args=args_tuple, + options=options_dict, + stdout=stdout.getvalue(), + stderr=stderr.getvalue(), + ) + + except Exception as e: + logger.error( + "Exception during management command execution: %s - Command: %s", + f"{type(e).__name__}: {e}", + command, + ) + logger.debug("Full traceback for error:", exc_info=True) + + return CommandErrorResult( + command=command, + args=args_tuple, + options=options_dict, + exception=e, + stdout=stdout.getvalue(), + stderr=stderr.getvalue(), + ) + + +management_command_executor = ManagementCommandExecutor() diff --git a/src/mcp_django/project/server.py b/src/mcp_django/project/server.py index d5072e8..8e0000c 100644 --- a/src/mcp_django/project/server.py +++ b/src/mcp_django/project/server.py @@ -10,6 +10,8 @@ from fastmcp import FastMCP from mcp.types import ToolAnnotations +from .management import ManagementCommandOutput +from .management import management_command_executor from .resources import AppResource from .resources import ModelResource from .resources import ProjectResource @@ -310,3 +312,78 @@ def get_setting( ), tags={PROJECT_TOOLSET}, )(get_setting) + + +@mcp.tool( + name="management_command", + annotations=ToolAnnotations( + title="Run Django Management Command", + destructiveHint=True, + openWorldHint=True, + ), + tags={PROJECT_TOOLSET}, +) +async def management_command( + ctx: Context, + command: Annotated[ + str, + "Management command name (e.g., 'migrate', 'check', 'collectstatic')", + ], + args: Annotated[ + list[str] | None, + "Positional arguments for the command", + ] = None, + options: Annotated[ + dict[str, str | int | bool] | None, + "Keyword options for the command (use underscores for dashes, e.g., 'run_syncdb' for '--run-syncdb')", + ] = None, +) -> ManagementCommandOutput: + """Execute a Django management command. + + Calls Django's call_command() to run management commands. Arguments and options + are passed directly to the command. Command output (stdout/stderr) is captured + and returned. + + Examples: + - Check for issues: command="check" + - Show migrations: command="showmigrations", args=["myapp"] + - Migrate with options: command="migrate", options={"verbosity": 2} + - Check with tag: command="check", options={"tag": "security"} + + Note: Management commands can modify your database and project state. Use with + caution, especially commands like migrate, flush, loaddata, etc. + """ + logger.info( + "management_command called - request_id: %s, client_id: %s, command: %s, args: %s, options: %s", + ctx.request_id, + ctx.client_id or "unknown", + command, + args, + options, + ) + + try: + result = await management_command_executor.execute(command, args, options) + output = ManagementCommandOutput.from_result(result) + + logger.debug( + "management_command completed - request_id: %s, status: %s", + ctx.request_id, + output.status, + ) + + if output.status == "error": + await ctx.debug( + f"Command failed: {output.exception.type}: {output.exception.message}" + ) + + return output + + except Exception as e: + logger.error( + "Unexpected error in management_command tool - request_id: %s: %s", + ctx.request_id, + e, + exc_info=True, + ) + raise diff --git a/tests/test_management_command.py b/tests/test_management_command.py new file mode 100644 index 0000000..bd43cf9 --- /dev/null +++ b/tests/test_management_command.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import pytest +import pytest_asyncio +from fastmcp import Client + +from mcp_django.server import mcp + +pytestmark = [pytest.mark.asyncio, pytest.mark.django_db] + + +@pytest_asyncio.fixture(autouse=True) +async def initialize_server(): + """Initialize the MCP server before tests.""" + await mcp.initialize() + + +async def test_management_command_check(): + """Test running the 'check' command (safe, read-only).""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "check", + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "check" + assert result.data.args == [] + assert result.data.exception is None + # The check command should produce some output to stderr + assert ( + "System check identified" in result.data.stderr or result.data.stderr == "" + ) + + +async def test_management_command_with_args(): + """Test running a command with positional arguments.""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "showmigrations", + "args": ["tests"], + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "showmigrations" + assert result.data.args == ["tests"] + assert result.data.exception is None + + +async def test_management_command_with_options(): + """Test running a command with keyword options.""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "check", + "options": {"verbosity": 2}, + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "check" + assert result.data.exception is None + + +async def test_management_command_with_args_and_options(): + """Test running a command with both args and options.""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "showmigrations", + "args": ["tests"], + "options": {"verbosity": 0}, + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "showmigrations" + assert result.data.args == ["tests"] + assert result.data.exception is None + + +async def test_management_command_invalid_command(): + """Test running an invalid/non-existent command.""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "this_command_does_not_exist", + }, + ) + + assert result.data is not None + assert result.data.status == "error" + assert result.data.command == "this_command_does_not_exist" + assert result.data.exception is not None + assert result.data.exception.type in [ + "CommandError", + "ManagementUtilityError", + ] + assert "Unknown command" in result.data.exception.message + + +async def test_management_command_makemigrations_dry_run(): + """Test running makemigrations with --dry-run (safe, read-only).""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "makemigrations", + "options": {"dry_run": True, "verbosity": 0}, + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "makemigrations" + assert result.data.args == [] + assert result.data.exception is None + + +async def test_management_command_diffsettings(): + """Test running diffsettings command (read-only introspection).""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "diffsettings", + "options": {"all": True}, + }, + ) + + assert result.data is not None + assert result.data.status == "success" + assert result.data.command == "diffsettings" + # Should output settings + assert len(result.data.stdout) > 0 + + +async def test_management_command_stdout_capture(): + """Test that stdout is properly captured from commands.""" + async with Client(mcp.server) as client: + result = await client.call_tool( + "project_management_command", + { + "command": "check", + "options": {"verbosity": 2}, + }, + ) + + assert result.data is not None + assert result.data.status == "success" + # With higher verbosity, check should produce output + assert isinstance(result.data.stdout, str) + assert isinstance(result.data.stderr, str) + + +async def test_management_command_list_in_main_server(): + """Test that management_command tool is listed in main server tools.""" + async with Client(mcp.server) as client: + tools = await client.list_tools() + tool_names = [tool.name for tool in tools] + + assert "project_management_command" in tool_names + + # Find the tool and check its metadata + mgmt_tool = next( + tool for tool in tools if tool.name == "project_management_command" + ) + assert mgmt_tool.description is not None + assert "management command" in mgmt_tool.description.lower() diff --git a/tests/test_server.py b/tests/test_server.py index 3136511..b56381f 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -50,6 +50,7 @@ async def test_tool_listing(): "project_list_models", "project_list_routes", "project_get_setting", + "project_management_command", "shell_execute", "shell_clear_history", "shell_export_history", From 74f311943381cc96d5b3ecf9bc2f773b806926ea Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Tue, 4 Nov 2025 23:47:00 -0600 Subject: [PATCH 2/7] move to dedicated toolset --- CHANGELOG.md | 2 +- README.md | 8 +- src/mcp_django/management_toolset/__init__.py | 9 ++ .../core.py} | 30 ++++ src/mcp_django/management_toolset/server.py | 133 ++++++++++++++++++ src/mcp_django/project/server.py | 77 ---------- src/mcp_django/server.py | 3 + tests/test_management_command.py | 67 +++++++-- tests/test_server.py | 3 +- 9 files changed, 242 insertions(+), 90 deletions(-) create mode 100644 src/mcp_django/management_toolset/__init__.py rename src/mcp_django/{project/management.py => management_toolset/core.py} (88%) create mode 100644 src/mcp_django/management_toolset/server.py diff --git a/CHANGELOG.md b/CHANGELOG.md index afeae8c..f64123d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,7 @@ For multi-package releases, use package names as subsections: ### Added -- Added `management_command` tool for executing Django management commands with arguments and options +- Added Management toolset with `command` and `list_commands` tools for executing and discovering Django management commands ## [0.12.0] diff --git a/README.md b/README.md index b5f7242..c6d32db 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,13 @@ Read-only resources for project exploration without executing code (note that re | `list_apps` | List all installed Django applications with their models | | `list_models` | Get detailed information about Django models with optional filtering by app or scope | | `list_routes` | Introspect Django URL routes with filtering support for HTTP method, route name, or URL pattern | -| `management_command` | Execute Django management commands with arguments and options | + +#### Management + +| Tool | Description | +|------|-------------| +| `command` | Execute Django management commands with arguments and options | +| `list_commands` | List all available Django management commands with their source apps | #### Shell diff --git a/src/mcp_django/management_toolset/__init__.py b/src/mcp_django/management_toolset/__init__.py new file mode 100644 index 0000000..d0a7eb0 --- /dev/null +++ b/src/mcp_django/management_toolset/__init__.py @@ -0,0 +1,9 @@ +from __future__ import annotations + +from .server import MANAGEMENT_TOOLSET +from .server import mcp + +__all__ = [ + "MANAGEMENT_TOOLSET", + "mcp", +] diff --git a/src/mcp_django/project/management.py b/src/mcp_django/management_toolset/core.py similarity index 88% rename from src/mcp_django/project/management.py rename to src/mcp_django/management_toolset/core.py index e3e6d2a..8ee296a 100644 --- a/src/mcp_django/project/management.py +++ b/src/mcp_django/management_toolset/core.py @@ -10,6 +10,7 @@ from asgiref.sync import sync_to_async from django.core.management import call_command +from django.core.management import get_commands from pydantic import BaseModel from pydantic import ConfigDict from pydantic import field_serializer @@ -204,3 +205,32 @@ def _execute( management_command_executor = ManagementCommandExecutor() + + +class CommandInfo(BaseModel): + """Information about a management command.""" + + name: str + app_name: str + + +def get_management_commands() -> list[CommandInfo]: + """Get list of all available Django management commands. + + Returns a list of management commands with their app origins, + sorted alphabetically by command name. + + Returns: + List of CommandInfo objects containing command name and source app. + """ + logger.info("Fetching available management commands") + + commands = get_commands() + command_list = [ + CommandInfo(name=name, app_name=app_name) + for name, app_name in sorted(commands.items()) + ] + + logger.debug("Found %d management commands", len(command_list)) + + return command_list diff --git a/src/mcp_django/management_toolset/server.py b/src/mcp_django/management_toolset/server.py new file mode 100644 index 0000000..a1ae503 --- /dev/null +++ b/src/mcp_django/management_toolset/server.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import logging +from typing import Annotated + +from fastmcp import Context +from fastmcp import FastMCP +from mcp.types import ToolAnnotations + +from .core import CommandInfo +from .core import ManagementCommandOutput +from .core import get_management_commands +from .core import management_command_executor + +logger = logging.getLogger(__name__) + +mcp = FastMCP( + name="Management", + instructions="Execute and discover Django management commands. Run commands with arguments and options, or list available commands to discover what's available in your project.", +) + +MANAGEMENT_TOOLSET = "management" + + +@mcp.tool( + name="command", + annotations=ToolAnnotations( + title="Run Django Management Command", + destructiveHint=True, + openWorldHint=True, + ), + tags={MANAGEMENT_TOOLSET}, +) +async def command( + ctx: Context, + command: Annotated[ + str, + "Management command name (e.g., 'migrate', 'check', 'collectstatic')", + ], + args: Annotated[ + list[str] | None, + "Positional arguments for the command", + ] = None, + options: Annotated[ + dict[str, str | int | bool] | None, + "Keyword options for the command (use underscores for dashes, e.g., 'run_syncdb' for '--run-syncdb')", + ] = None, +) -> ManagementCommandOutput: + """Execute a Django management command. + + Calls Django's call_command() to run management commands. Arguments and options + are passed directly to the command. Command output (stdout/stderr) is captured + and returned. + + Examples: + - Check for issues: command="check" + - Show migrations: command="showmigrations", args=["myapp"] + - Migrate with options: command="migrate", options={"verbosity": 2} + - Check with tag: command="check", options={"tag": "security"} + + Note: Management commands can modify your database and project state. Use with + caution, especially commands like migrate, flush, loaddata, etc. + """ + logger.info( + "management_command called - request_id: %s, client_id: %s, command: %s, args: %s, options: %s", + ctx.request_id, + ctx.client_id or "unknown", + command, + args, + options, + ) + + try: + result = await management_command_executor.execute(command, args, options) + output = ManagementCommandOutput.from_result(result) + + logger.debug( + "management_command completed - request_id: %s, status: %s", + ctx.request_id, + output.status, + ) + + if output.status == "error": + await ctx.debug( + f"Command failed: {output.exception.type}: {output.exception.message}" + ) + + return output + + except Exception as e: + logger.error( + "Unexpected error in management_command tool - request_id: %s: %s", + ctx.request_id, + e, + exc_info=True, + ) + raise + + +@mcp.tool( + name="list_commands", + annotations=ToolAnnotations( + title="List Django Management Commands", + readOnlyHint=True, + idempotentHint=True, + ), + tags={MANAGEMENT_TOOLSET}, +) +def list_commands(ctx: Context) -> list[CommandInfo]: + """List all available Django management commands. + + Returns a list of all management commands available in the current Django + project, including built-in Django commands and custom commands from + installed apps. Each command includes its name and the app that provides it. + + Useful for discovering what commands are available before executing them + with the command tool. + """ + logger.info( + "list_management_commands called - request_id: %s, client_id: %s", + ctx.request_id, + ctx.client_id or "unknown", + ) + + commands = get_management_commands() + + logger.debug( + "list_management_commands completed - request_id: %s, commands_count: %d", + ctx.request_id, + len(commands), + ) + + return commands diff --git a/src/mcp_django/project/server.py b/src/mcp_django/project/server.py index 8e0000c..d5072e8 100644 --- a/src/mcp_django/project/server.py +++ b/src/mcp_django/project/server.py @@ -10,8 +10,6 @@ from fastmcp import FastMCP from mcp.types import ToolAnnotations -from .management import ManagementCommandOutput -from .management import management_command_executor from .resources import AppResource from .resources import ModelResource from .resources import ProjectResource @@ -312,78 +310,3 @@ def get_setting( ), tags={PROJECT_TOOLSET}, )(get_setting) - - -@mcp.tool( - name="management_command", - annotations=ToolAnnotations( - title="Run Django Management Command", - destructiveHint=True, - openWorldHint=True, - ), - tags={PROJECT_TOOLSET}, -) -async def management_command( - ctx: Context, - command: Annotated[ - str, - "Management command name (e.g., 'migrate', 'check', 'collectstatic')", - ], - args: Annotated[ - list[str] | None, - "Positional arguments for the command", - ] = None, - options: Annotated[ - dict[str, str | int | bool] | None, - "Keyword options for the command (use underscores for dashes, e.g., 'run_syncdb' for '--run-syncdb')", - ] = None, -) -> ManagementCommandOutput: - """Execute a Django management command. - - Calls Django's call_command() to run management commands. Arguments and options - are passed directly to the command. Command output (stdout/stderr) is captured - and returned. - - Examples: - - Check for issues: command="check" - - Show migrations: command="showmigrations", args=["myapp"] - - Migrate with options: command="migrate", options={"verbosity": 2} - - Check with tag: command="check", options={"tag": "security"} - - Note: Management commands can modify your database and project state. Use with - caution, especially commands like migrate, flush, loaddata, etc. - """ - logger.info( - "management_command called - request_id: %s, client_id: %s, command: %s, args: %s, options: %s", - ctx.request_id, - ctx.client_id or "unknown", - command, - args, - options, - ) - - try: - result = await management_command_executor.execute(command, args, options) - output = ManagementCommandOutput.from_result(result) - - logger.debug( - "management_command completed - request_id: %s, status: %s", - ctx.request_id, - output.status, - ) - - if output.status == "error": - await ctx.debug( - f"Command failed: {output.exception.type}: {output.exception.message}" - ) - - return output - - except Exception as e: - logger.error( - "Unexpected error in management_command tool - request_id: %s: %s", - ctx.request_id, - e, - exc_info=True, - ) - raise diff --git a/src/mcp_django/server.py b/src/mcp_django/server.py index 8ad2a90..fcef104 100644 --- a/src/mcp_django/server.py +++ b/src/mcp_django/server.py @@ -6,6 +6,8 @@ from fastmcp import FastMCP +from mcp_django.management_toolset import MANAGEMENT_TOOLSET +from mcp_django.management_toolset import mcp as management_mcp from mcp_django.packages import DJANGOPACKAGES_TOOLSET from mcp_django.packages import mcp as packages_mcp from mcp_django.project import PROJECT_TOOLSET @@ -17,6 +19,7 @@ TOOLSETS = { DJANGOPACKAGES_TOOLSET: packages_mcp, + MANAGEMENT_TOOLSET: management_mcp, PROJECT_TOOLSET: project_mcp, SHELL_TOOLSET: shell_mcp, } diff --git a/tests/test_management_command.py b/tests/test_management_command.py index bd43cf9..c98aea7 100644 --- a/tests/test_management_command.py +++ b/tests/test_management_command.py @@ -19,7 +19,7 @@ async def test_management_command_check(): """Test running the 'check' command (safe, read-only).""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "check", }, @@ -40,7 +40,7 @@ async def test_management_command_with_args(): """Test running a command with positional arguments.""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "showmigrations", "args": ["tests"], @@ -58,7 +58,7 @@ async def test_management_command_with_options(): """Test running a command with keyword options.""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "check", "options": {"verbosity": 2}, @@ -75,7 +75,7 @@ async def test_management_command_with_args_and_options(): """Test running a command with both args and options.""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "showmigrations", "args": ["tests"], @@ -94,7 +94,7 @@ async def test_management_command_invalid_command(): """Test running an invalid/non-existent command.""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "this_command_does_not_exist", }, @@ -115,7 +115,7 @@ async def test_management_command_makemigrations_dry_run(): """Test running makemigrations with --dry-run (safe, read-only).""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "makemigrations", "options": {"dry_run": True, "verbosity": 0}, @@ -133,7 +133,7 @@ async def test_management_command_diffsettings(): """Test running diffsettings command (read-only introspection).""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "diffsettings", "options": {"all": True}, @@ -151,7 +151,7 @@ async def test_management_command_stdout_capture(): """Test that stdout is properly captured from commands.""" async with Client(mcp.server) as client: result = await client.call_tool( - "project_management_command", + "management_command", { "command": "check", "options": {"verbosity": 2}, @@ -171,11 +171,58 @@ async def test_management_command_list_in_main_server(): tools = await client.list_tools() tool_names = [tool.name for tool in tools] - assert "project_management_command" in tool_names + assert "management_command" in tool_names # Find the tool and check its metadata mgmt_tool = next( - tool for tool in tools if tool.name == "project_management_command" + tool for tool in tools if tool.name == "management_command" ) assert mgmt_tool.description is not None assert "management command" in mgmt_tool.description.lower() + + +async def test_list_management_commands(): + """Test listing all available management commands.""" + async with Client(mcp.server) as client: + result = await client.call_tool("management_list_commands", {}) + + assert result.data is not None + assert isinstance(result.data, list) + assert len(result.data) > 0 + + # Check that we have some standard Django commands + command_names = [cmd["name"] for cmd in result.data] + assert "check" in command_names + assert "migrate" in command_names + assert "showmigrations" in command_names + + # Verify structure of command info + first_cmd = result.data[0] + assert "name" in first_cmd + assert "app_name" in first_cmd + assert isinstance(first_cmd["name"], str) + assert isinstance(first_cmd["app_name"], str) + + +async def test_list_management_commands_includes_custom_commands(): + """Test that custom management commands are included in the list.""" + async with Client(mcp.server) as client: + result = await client.call_tool("management_list_commands", {}) + + assert result.data is not None + command_names = [cmd["name"] for cmd in result.data] + + # The mcp command from this project should be in the list + assert "mcp" in command_names + + +async def test_list_management_commands_sorted(): + """Test that management commands are sorted alphabetically.""" + async with Client(mcp.server) as client: + result = await client.call_tool("management_list_commands", {}) + + assert result.data is not None + command_names = [cmd["name"] for cmd in result.data] + + # Verify the list is sorted + assert command_names == sorted(command_names) diff --git a/tests/test_server.py b/tests/test_server.py index b56381f..5c5a70e 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -45,12 +45,13 @@ async def test_tool_listing(): "djangopackages_get_grid", "djangopackages_get_package", "djangopackages_search", + "management_command", + "management_list_commands", "project_get_project_info", "project_list_apps", "project_list_models", "project_list_routes", "project_get_setting", - "project_management_command", "shell_execute", "shell_clear_history", "shell_export_history", From c39e9e6a345fb54f94a0d4f1eca49dc8cae958c8 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Tue, 4 Nov 2025 23:50:36 -0600 Subject: [PATCH 3/7] clean up --- src/mcp_django/management_toolset/core.py | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/src/mcp_django/management_toolset/core.py b/src/mcp_django/management_toolset/core.py index 8ee296a..b2d36e1 100644 --- a/src/mcp_django/management_toolset/core.py +++ b/src/mcp_django/management_toolset/core.py @@ -13,15 +13,12 @@ from django.core.management import get_commands from pydantic import BaseModel from pydantic import ConfigDict -from pydantic import field_serializer logger = logging.getLogger(__name__) @dataclass class CommandResult: - """Result from successfully executing a management command.""" - command: str args: tuple[str, ...] options: dict[str, str | int | bool] @@ -41,8 +38,6 @@ def __post_init__(self): @dataclass class CommandErrorResult: - """Result from a management command that raised an exception.""" - command: str args: tuple[str, ...] options: dict[str, str | int | bool] @@ -69,8 +64,6 @@ def __post_init__(self): class ManagementCommandOutput(BaseModel): - """Output from executing a Django management command.""" - status: str # "success" or "error" command: str args: list[str] @@ -81,7 +74,6 @@ class ManagementCommandOutput(BaseModel): @classmethod def from_result(cls, result: Result) -> ManagementCommandOutput: - """Create output from a command result.""" match result: case CommandResult(): return cls( @@ -109,8 +101,6 @@ def from_result(cls, result: Result) -> ManagementCommandOutput: class ExceptionInfo(BaseModel): - """Information about an exception that occurred during command execution.""" - model_config = ConfigDict(arbitrary_types_allowed=True) type: str @@ -118,8 +108,6 @@ class ExceptionInfo(BaseModel): class ManagementCommandExecutor: - """Executor for Django management commands with output capture.""" - async def execute( self, command: str, @@ -208,8 +196,6 @@ def _execute( class CommandInfo(BaseModel): - """Information about a management command.""" - name: str app_name: str From 2325cc85d1b00690d3f2cbe89704ee44017f16b3 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Tue, 4 Nov 2025 23:51:32 -0600 Subject: [PATCH 4/7] rename --- src/mcp_django/{management_toolset => mgmt}/__init__.py | 0 src/mcp_django/{management_toolset => mgmt}/core.py | 0 src/mcp_django/{management_toolset => mgmt}/server.py | 0 src/mcp_django/server.py | 4 ++-- 4 files changed, 2 insertions(+), 2 deletions(-) rename src/mcp_django/{management_toolset => mgmt}/__init__.py (100%) rename src/mcp_django/{management_toolset => mgmt}/core.py (100%) rename src/mcp_django/{management_toolset => mgmt}/server.py (100%) diff --git a/src/mcp_django/management_toolset/__init__.py b/src/mcp_django/mgmt/__init__.py similarity index 100% rename from src/mcp_django/management_toolset/__init__.py rename to src/mcp_django/mgmt/__init__.py diff --git a/src/mcp_django/management_toolset/core.py b/src/mcp_django/mgmt/core.py similarity index 100% rename from src/mcp_django/management_toolset/core.py rename to src/mcp_django/mgmt/core.py diff --git a/src/mcp_django/management_toolset/server.py b/src/mcp_django/mgmt/server.py similarity index 100% rename from src/mcp_django/management_toolset/server.py rename to src/mcp_django/mgmt/server.py diff --git a/src/mcp_django/server.py b/src/mcp_django/server.py index fcef104..3fa2907 100644 --- a/src/mcp_django/server.py +++ b/src/mcp_django/server.py @@ -6,8 +6,8 @@ from fastmcp import FastMCP -from mcp_django.management_toolset import MANAGEMENT_TOOLSET -from mcp_django.management_toolset import mcp as management_mcp +from mcp_django.mgmt import MANAGEMENT_TOOLSET +from mcp_django.mgmt import mcp as management_mcp from mcp_django.packages import DJANGOPACKAGES_TOOLSET from mcp_django.packages import mcp as packages_mcp from mcp_django.project import PROJECT_TOOLSET From 7d8cf8f59266d893cdc1b47d5c54036824d260a0 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Tue, 4 Nov 2025 23:57:59 -0600 Subject: [PATCH 5/7] rename --- README.md | 2 +- src/mcp_django/mgmt/server.py | 8 ++++---- tests/test_management_command.py | 20 ++++++++++---------- tests/test_server.py | 2 +- 4 files changed, 16 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index c6d32db..e16b91e 100644 --- a/README.md +++ b/README.md @@ -210,7 +210,7 @@ Read-only resources for project exploration without executing code (note that re | Tool | Description | |------|-------------| -| `command` | Execute Django management commands with arguments and options | +| `execute_command` | Execute Django management commands with arguments and options | | `list_commands` | List all available Django management commands with their source apps | #### Shell diff --git a/src/mcp_django/mgmt/server.py b/src/mcp_django/mgmt/server.py index a1ae503..274e919 100644 --- a/src/mcp_django/mgmt/server.py +++ b/src/mcp_django/mgmt/server.py @@ -23,15 +23,15 @@ @mcp.tool( - name="command", + name="execute_command", annotations=ToolAnnotations( - title="Run Django Management Command", + title="Execute Django Management Command", destructiveHint=True, openWorldHint=True, ), tags={MANAGEMENT_TOOLSET}, ) -async def command( +async def execute_command( ctx: Context, command: Annotated[ str, @@ -114,7 +114,7 @@ def list_commands(ctx: Context) -> list[CommandInfo]: installed apps. Each command includes its name and the app that provides it. Useful for discovering what commands are available before executing them - with the command tool. + with the execute_command tool. """ logger.info( "list_management_commands called - request_id: %s, client_id: %s", diff --git a/tests/test_management_command.py b/tests/test_management_command.py index c98aea7..15592a9 100644 --- a/tests/test_management_command.py +++ b/tests/test_management_command.py @@ -19,7 +19,7 @@ async def test_management_command_check(): """Test running the 'check' command (safe, read-only).""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "check", }, @@ -40,7 +40,7 @@ async def test_management_command_with_args(): """Test running a command with positional arguments.""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "showmigrations", "args": ["tests"], @@ -58,7 +58,7 @@ async def test_management_command_with_options(): """Test running a command with keyword options.""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "check", "options": {"verbosity": 2}, @@ -75,7 +75,7 @@ async def test_management_command_with_args_and_options(): """Test running a command with both args and options.""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "showmigrations", "args": ["tests"], @@ -94,7 +94,7 @@ async def test_management_command_invalid_command(): """Test running an invalid/non-existent command.""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "this_command_does_not_exist", }, @@ -115,7 +115,7 @@ async def test_management_command_makemigrations_dry_run(): """Test running makemigrations with --dry-run (safe, read-only).""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "makemigrations", "options": {"dry_run": True, "verbosity": 0}, @@ -133,7 +133,7 @@ async def test_management_command_diffsettings(): """Test running diffsettings command (read-only introspection).""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "diffsettings", "options": {"all": True}, @@ -151,7 +151,7 @@ async def test_management_command_stdout_capture(): """Test that stdout is properly captured from commands.""" async with Client(mcp.server) as client: result = await client.call_tool( - "management_command", + "management_execute_command", { "command": "check", "options": {"verbosity": 2}, @@ -171,11 +171,11 @@ async def test_management_command_list_in_main_server(): tools = await client.list_tools() tool_names = [tool.name for tool in tools] - assert "management_command" in tool_names + assert "management_execute_command" in tool_names # Find the tool and check its metadata mgmt_tool = next( - tool for tool in tools if tool.name == "management_command" + tool for tool in tools if tool.name == "management_execute_command" ) assert mgmt_tool.description is not None assert "management command" in mgmt_tool.description.lower() diff --git a/tests/test_server.py b/tests/test_server.py index 5c5a70e..50ab95e 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -45,7 +45,7 @@ async def test_tool_listing(): "djangopackages_get_grid", "djangopackages_get_package", "djangopackages_search", - "management_command", + "management_execute_command", "management_list_commands", "project_get_project_info", "project_list_apps", From 3259468c13cb62f1800e00d16bd37491c5db3ae3 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Wed, 5 Nov 2025 00:12:02 -0600 Subject: [PATCH 6/7] coverage --- tests/test_management_command.py | 42 ++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_management_command.py b/tests/test_management_command.py index 15592a9..73767f8 100644 --- a/tests/test_management_command.py +++ b/tests/test_management_command.py @@ -4,6 +4,8 @@ import pytest_asyncio from fastmcp import Client +from mcp_django.mgmt import core +from mcp_django.mgmt.core import CommandErrorResult from mcp_django.server import mcp pytestmark = [pytest.mark.asyncio, pytest.mark.django_db] @@ -226,3 +228,43 @@ async def test_list_management_commands_sorted(): # Verify the list is sorted assert command_names == sorted(command_names) + + +async def test_management_command_unexpected_exception(monkeypatch): + """Test handling of unexpected exceptions in execute_command tool.""" + + # Mock the executor to raise an unexpected exception + async def mock_execute(*args, **kwargs): + raise RuntimeError("Unexpected error in executor") + + monkeypatch.setattr(core.management_command_executor, "execute", mock_execute) + + async with Client(mcp.server) as client: + # The tool should re-raise unexpected exceptions + with pytest.raises(Exception): # Will be wrapped by FastMCP + await client.call_tool( + "management_execute_command", + { + "command": "check", + }, + ) + + +async def test_command_error_result_with_stdout_stderr(): + """Test CommandErrorResult __post_init__ with stdout and stderr for coverage.""" + # Create a CommandErrorResult with both stdout and stderr + # This triggers lines 58 and 60 in mgmt/core.py + result = CommandErrorResult( + command="test_command", + args=("arg1",), + options={"opt": "value"}, + exception=Exception("Test exception"), + stdout="Some stdout output", + stderr="Some stderr output", + ) + + # Verify the result was created correctly + assert result.command == "test_command" + assert result.stdout == "Some stdout output" + assert result.stderr == "Some stderr output" + assert str(result.exception) == "Test exception" From f0fa38a3623a7494ad46494e4cc01d5d31252110 Mon Sep 17 00:00:00 2001 From: Josh Thomas Date: Wed, 5 Nov 2025 00:16:01 -0600 Subject: [PATCH 7/7] mypy --- src/mcp_django/mgmt/server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mcp_django/mgmt/server.py b/src/mcp_django/mgmt/server.py index 274e919..897b9ab 100644 --- a/src/mcp_django/mgmt/server.py +++ b/src/mcp_django/mgmt/server.py @@ -80,7 +80,7 @@ async def execute_command( output.status, ) - if output.status == "error": + if output.status == "error" and output.exception: await ctx.debug( f"Command failed: {output.exception.type}: {output.exception.message}" )