Add management command toolset with command execution & discovery (#60)

Author: joshuadavidthomas

CHANGELOG.md

@@ -26,6 +26,10 @@ For multi-package releases, use package names as subsections:
 
 ## [Unreleased]
 
+### Added
+
+- Added Management toolset with `command` and `list_commands` tools for executing and discovering Django management commands
+
 ## [0.12.0]
 
 ### Added

README.md

@@ -206,6 +206,13 @@ Read-only resources for project exploration without executing code (note that re
 | `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
+
+| Tool | Description |
+|------|-------------|
+| `execute_command` | Execute Django management commands with arguments and options |
+| `list_commands` | List all available Django management commands with their source apps |
+
 #### Shell
 
 | Tool | Description |

src/mcp_django/mgmt/__init__.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from .server import MANAGEMENT_TOOLSET
+from .server import mcp
+
+__all__ = [
+    "MANAGEMENT_TOOLSET",
+    "mcp",
+]

src/mcp_django/mgmt/core.py

@@ -0,0 +1,222 @@
+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 django.core.management import get_commands
+from pydantic import BaseModel
+from pydantic import ConfigDict
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CommandResult:
+    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:
+    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):
+    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:
+        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):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    type: str
+    message: str
+
+
+class ManagementCommandExecutor:
+    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()
+
+
+class CommandInfo(BaseModel):
+    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

src/mcp_django/mgmt/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="execute_command",
+    annotations=ToolAnnotations(
+        title="Execute Django Management Command",
+        destructiveHint=True,
+        openWorldHint=True,
+    ),
+    tags={MANAGEMENT_TOOLSET},
+)
+async def execute_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" and output.exception:
+            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 execute_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