Add comprehensive logging (#5)

joshuadavidthomas · web-flow · commit 0a75d656024d · 2025-08-19T12:49:05.000-05:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,6 +18,12 @@ and this project attempts to adhere to [Semantic Versioning](https://semver.org/
 
 ## [Unreleased]
 
+### Added
+
+- Comprehensive logging for debugging MCP/LLM interactions
+- `--debug` flag for the `mcp_shell` management command to enable detailed logging
+- Request and client ID tracking in server logs
+
 ### Changed
 
 - **Internal**: Refactored results to use separate dataclasses and a tagged union
diff --git a/src/mcp_django_shell/management/commands/mcp_shell.py b/src/mcp_django_shell/management/commands/mcp_shell.py
@@ -1,12 +1,59 @@
 from __future__ import annotations
 
+import logging
+import os
+import signal
+import sys
 from typing import Any
 
 from django.core.management.base import BaseCommand
 
 from mcp_django_shell.server import mcp
 
+logger = logging.getLogger(__name__)
+
 
 class Command(BaseCommand):
+    help = "Run the MCP Django Shell server"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--debug",
+            action="store_true",
+            help="Enable debug logging",
+        )
+
     def handle(self, *args: Any, **options: Any):
-        mcp.run()
+        if options.get("debug"):
+            logging.basicConfig(
+                level=logging.DEBUG,
+                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+            )
+            logger.debug("Debug logging enabled")
+
+        logger.info("Starting MCP Django Shell server")
+        logger.debug(
+            "Django settings module: %s",
+            os.environ.get("DJANGO_SETTINGS_MODULE", "Not set"),
+        )
+
+        def signal_handler(signum, frame):
+            logger.info("Received signal %s, shutting down MCP server", signum)
+
+            sys.exit(0)
+
+        signal.signal(signal.SIGINT, signal_handler)
+        signal.signal(signal.SIGTERM, signal_handler)
+
+        try:
+            logger.info("MCP server ready and listening")
+
+            mcp.run()
+
+        except Exception as e:
+            logger.error("MCP server crashed: %s", e, exc_info=True)
+
+            raise
+
+        finally:
+            logger.info("MCP Django Shell server stopped")
diff --git a/src/mcp_django_shell/server.py b/src/mcp_django_shell/server.py
@@ -1,18 +1,23 @@
 from __future__ import annotations
 
+import logging
+
+from fastmcp import Context
 from fastmcp import FastMCP
 
 from .shell import DjangoShell
+from .shell import ErrorResult
 
 mcp = FastMCP(
     name="Django Shell",
     instructions="Provides a stateful Django shell environment for executing Python code, managing sessions, and exporting command history.",
 )
 shell = DjangoShell()
+logger = logging.getLogger(__name__)
 
 
 @mcp.tool
-async def django_shell(code: str, timeout: int | None = None) -> str:
+async def django_shell(code: str, ctx: Context, timeout: int | None = None) -> str:
     """Execute Python code in a stateful Django shell session.
 
     Django is pre-configured and ready to use with your project. You can import and use any Django
@@ -26,16 +31,63 @@ async def django_shell(code: str, timeout: int | None = None) -> str:
     **NOTE**: that only synchronous Django ORM operations are supported - use standard methods like
     `.filter()` and `.get()` rather than their async counterparts (`.afilter()`, `.aget()`).
     """
-    result = await shell.execute(code, timeout=timeout)
-    return result.output
+    logger.info(
+        "django_shell tool called - request_id: %s, client_id: %s, code: %s",
+        ctx.request_id,
+        ctx.client_id or "unknown",
+        (code[:100] + "..." if len(code) > 100 else code).replace("\n", "\\n"),
+    )
+    logger.debug(
+        "Full code for django_shell - request_id: %s: %s", ctx.request_id, code
+    )
+
+    try:
+        result = await shell.execute(code, timeout=timeout)
+
+        logger.debug(
+            "django_shell execution completed - request_id: %s, result type: %s",
+            ctx.request_id,
+            type(result).__name__,
+        )
+        if isinstance(result, ErrorResult):
+            await ctx.debug(
+                f"Execution failed: {type(result.exception).__name__}",
+                extra={
+                    "error_type": type(result.exception).__name__,
+                    "code_length": len(code),
+                },
+            )
+
+        return result.output
+
+    except Exception as e:
+        logger.error(
+            "Unexpected error in django_shell tool - request_id: %s: %s",
+            ctx.request_id,
+            e,
+            exc_info=True,
+        )
+        raise
 
 
 @mcp.tool
-def django_reset() -> str:
+async def django_reset(ctx: Context) -> str:
     """
     Reset the Django shell session, clearing all variables and history.
 
     Use this when you want to start fresh or if the session state becomes corrupted.
     """
+    logger.info(
+        "django_reset tool called - request_id: %s, client_id: %s",
+        ctx.request_id,
+        ctx.client_id or "unknown",
+    )
+    await ctx.debug("Django shell session reset")
+
     shell.reset()
+
+    logger.debug(
+        "Django shell session reset completed - request_id: %s", ctx.request_id
+    )
+
     return "Django shell session has been reset. All previously set variables and history cleared."
diff --git a/src/mcp_django_shell/shell.py b/src/mcp_django_shell/shell.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import traceback
 from contextlib import redirect_stderr
 from contextlib import redirect_stdout
@@ -12,19 +13,33 @@
 
 from asgiref.sync import sync_to_async
 
+logger = logging.getLogger(__name__)
+
 
 class DjangoShell:
     def __init__(self):
+        logger.debug("Initializing %s", self.__class__.__name__)
+
         from django import setup
         from django.apps import apps
 
         if not apps.ready:  # pragma: no cover
+            logger.info("Django not initialized, running django.setup()")
+
             setup()
 
+            logger.debug("Django setup completed")
+        else:
+            logger.debug("Django already initialized, skipping setup")
+
         self.globals: dict[str, Any] = {}
         self.history: list[Result] = []
 
+        logger.info("Shell initialized successfully")
+
     def reset(self):
+        logger.info("Shell reset - clearing globals and history")
+
         self.globals = {}
         self.history = []
 
@@ -51,20 +66,44 @@ def _execute(self, code: str, timeout: int | None = None) -> Result:
         Note: This synchronous method contains the actual execution logic.
         Use `execute()` for async contexts or `_execute()` for sync/testing.
         """
+        code_preview = (code[:100] + "..." if len(code) > 100 else code).replace(
+            "\n", "\\n"
+        )
+        logger.info("Executing code: %s", code_preview)
+
         stdout = StringIO()
         stderr = StringIO()
 
         with redirect_stdout(stdout), redirect_stderr(stderr):
             try:
                 code, setup, code_type = parse_code(code)
 
+                logger.debug(
+                    "Execution type: %s, has setup: %s", code_type, bool(setup)
+                )
+                logger.debug(
+                    "Code to execute: %s",
+                    code[:200] + "..." if len(code) > 200 else code,
+                )
+
                 # Execute setup, if any (only applicable to expressions)
                 if setup:
+                    logger.debug(
+                        "Setup code: %s",
+                        setup[:200] + "..." if len(setup) > 200 else setup,
+                    )
+
                     exec(setup, self.globals)
 
                 match code_type:
                     case "expression":
                         value = eval(code, self.globals)
+
+                        logger.debug(
+                            "Expression executed successfully, result type: %s",
+                            type(value).__name__,
+                        )
+
                         return self.save_result(
                             ExpressionResult(
                                 code=code,
@@ -75,6 +114,9 @@ def _execute(self, code: str, timeout: int | None = None) -> Result:
                         )
                     case "statement":
                         exec(code, self.globals)
+
+                        logger.debug("Statement executed successfully")
+
                         return self.save_result(
                             StatementResult(
                                 code=code,
@@ -84,6 +126,13 @@ def _execute(self, code: str, timeout: int | None = None) -> Result:
                         )
 
             except Exception as e:
+                logger.error(
+                    "Exception during code execution: %s - Code: %s",
+                    f"{type(e).__name__}: {e}",
+                    code_preview,
+                )
+                logger.debug("Full traceback for error:", exc_info=True)
+
                 return self.save_result(
                     ErrorResult(
                         code=code,
@@ -136,6 +185,21 @@ class ExpressionResult:
     stderr: str
     timestamp: datetime = field(default_factory=datetime.now)
 
+    def __post_init__(self):
+        logger.debug(
+            "%s created - value type: %s",
+            self.__class__.__name__,
+            type(self.value).__name__,
+        )
+        logger.debug("%s.value: %s", self.__class__.__name__, repr(self.value)[:200])
+        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])
+        logger.debug(
+            "%s output (for LLM): %s", self.__class__.__name__, self.output[:500]
+        )
+
     @property
     def output(self) -> str:
         value = repr(self.value)
@@ -170,6 +234,16 @@ class StatementResult:
     stderr: str
     timestamp: datetime = field(default_factory=datetime.now)
 
+    def __post_init__(self):
+        logger.debug("%s created", self.__class__.__name__)
+        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])
+        logger.debug(
+            "%s output (for LLM): %s", self.__class__.__name__, self.output[:500]
+        )
+
     @property
     def output(self) -> str:
         return self.stdout or "OK"
@@ -183,11 +257,27 @@ class ErrorResult:
     stderr: str
     timestamp: datetime = field(default_factory=datetime.now)
 
+    def __post_init__(self):
+        logger.debug(
+            "%s created - exception type: %s",
+            self.__class__.__name__,
+            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])
+        logger.debug(
+            "%s output (filtered for LLM): %s",
+            self.__class__.__name__,
+            self.output[:500],
+        )
+
     @property
     def output(self) -> str:
         error_type = self.exception.__class__.__name__
 
-        # Format the stored exception's traceback
         tb_str = "".join(
             traceback.format_exception(
                 type(self.exception), self.exception, self.exception.__traceback__
diff --git a/tests/test_server.py b/tests/test_server.py
@@ -1,10 +1,14 @@
 from __future__ import annotations
 
+from unittest.mock import AsyncMock
+
 import pytest
 import pytest_asyncio
 from fastmcp import Client
+from fastmcp.exceptions import ToolError
 
 from mcp_django_shell.server import mcp
+from mcp_django_shell.server import shell
 
 pytestmark = pytest.mark.asyncio
 
@@ -45,6 +49,16 @@ async def test_django_shell_tool_orm():
         assert result.data is not None
 
 
+async def test_django_shell_tool_unexpected_error(monkeypatch):
+    monkeypatch.setattr(
+        shell, "execute", AsyncMock(side_effect=RuntimeError("Unexpected error"))
+    )
+
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="Unexpected error"):
+            await client.call_tool("django_shell", {"code": "2 + 2"})
+
+
 async def test_django_reset_session():
     async with Client(mcp) as client:
         await client.call_tool("django_shell", {"code": "x = 42"})
diff --git a/tests/test_shell.py b/tests/test_shell.py
