fix(sse_app): Expose SSE App correctly for production setup

Author: timothedelion

packages/developer_mcp_server/pyproject.toml

@@ -38,7 +38,7 @@ sentry = [
 
 
 [project.scripts]
-developer-mcp-server = "developer_mcp_server.server:run_mcp_server"
+developer-mcp-server = "developer_mcp_server.run:run_mcp_server"
 
 [build-system]
 requires = ["hatchling"]

packages/developer_mcp_server/src/developer_mcp_server/run.py

@@ -0,0 +1,63 @@
+"""Runtime entry points for the Developer MCP server.
+
+This module provides different ways to run the MCP server:
+- stdio: Standard input/output transport (default for CLI tools)
+- http: HTTP/SSE transport using uvicorn (for local development)
+"""
+
+import logging
+import os
+
+from gg_api_core.sentry_integration import init_sentry
+
+from developer_mcp_server.server import mcp
+
+logger = logging.getLogger(__name__)
+
+
+def run_stdio():
+    """Run the MCP server over stdio transport.
+
+    This is the default mode for MCP servers, used when the server
+    is invoked as a subprocess by MCP clients like Claude Desktop.
+    """
+    init_sentry()
+    logger.info("Developer MCP server running on stdio")
+    mcp.run()
+
+
+def run_http_with_uvicorn():
+    """Run the MCP server over HTTP using uvicorn ASGI server.
+
+    This is meant for local development. For production ready setup,
+    better use gunicorn with uvicorn ASGI workers via ggmcp_http.sse_app:app
+    """
+    init_sentry()
+
+    # Get host and port from environment variables
+    mcp_port = int(os.environ.get("MCP_PORT", "8000"))
+    mcp_host = os.environ.get("MCP_HOST", "127.0.0.1")
+
+    # Use HTTP/SSE transport with uvicorn
+    import uvicorn
+
+    logger.info(f"Starting Developer MCP server on {mcp_host}:{mcp_port}")
+    uvicorn.run(mcp.sse_app(), host=mcp_host, port=mcp_port)
+
+
+def run_mcp_server():
+    """Run the MCP server with transport auto-detection.
+
+    If MCP_PORT is set, uses HTTP/SSE transport.
+    Otherwise, uses stdio transport (default).
+    """
+    mcp_port = os.environ.get("MCP_PORT")
+
+    if mcp_port:
+        run_http_with_uvicorn()
+    else:
+        run_stdio()
+
+
+if __name__ == "__main__":
+    run_mcp_server()

packages/developer_mcp_server/src/developer_mcp_server/server.py

@@ -1,11 +1,9 @@
 """GitGuardian MCP server for developers with remediation tools."""
 
 import logging
-import os
 
 from gg_api_core.mcp_server import get_mcp_server
 from gg_api_core.scopes import set_developer_scopes
-from gg_api_core.sentry_integration import init_sentry
 
 from developer_mcp_server.add_health_check import add_health_check
 from developer_mcp_server.register_tools import DEVELOPER_INSTRUCTIONS, register_developer_tools
@@ -21,40 +19,10 @@
     log_level="DEBUG",
     instructions=DEVELOPER_INSTRUCTIONS,
 )
-logger.info("Created Developer GitGuardianFastMCP instance")
 
 register_developer_tools(mcp)
 add_health_check(mcp)
 
-
 set_developer_scopes()
 
-
-def run_mcp_server():
-    logger.info("Starting Developer MCP server...")
-
-    # Initialize Sentry if configured (optional)
-    sentry_enabled = init_sentry()
-    if sentry_enabled:
-        logger.info("Sentry monitoring is enabled")
-    else:
-        logger.debug("Sentry monitoring is not configured")
-
-    # Check if HTTP/SSE transport is requested via environment variables
-    mcp_port = os.environ.get("MCP_PORT")
-    mcp_host = os.environ.get("MCP_HOST", "127.0.0.1")
-
-    if mcp_port:
-        # Use HTTP/SSE transport
-        import uvicorn
-
-        logger.info(f"Starting MCP server with HTTP/SSE transport on {mcp_host}:{mcp_port}")
-        uvicorn.run(mcp.sse_app(), host=mcp_host, port=int(mcp_port))
-    else:
-        # Use default stdio transport
-        logger.info("Starting MCP server with stdio transport (default)")
-        mcp.run()
-
-
-if __name__ == "__main__":
-    run_mcp_server()
+logger.info("Developer MCP server instance created and configured")

packages/developer_mcp_server/src/developer_mcp_server/sse_app.py

@@ -0,0 +1,3 @@
+from developer_mcp_server.server import mcp
+
+sse_app = mcp.sse_app()

packages/secops_mcp_server/pyproject.toml

@@ -29,7 +29,8 @@ dependencies = [
     "httpx>=0.24.0",
     "pydantic>=2.0.0",
     "gg-api-core",
-    "uvicorn>=0.27.0"
+    "uvicorn>=0.27.0",
+    "gunicorn>=23.0.0"
 ]
 
 [project.optional-dependencies]
@@ -38,8 +39,11 @@ sentry = [
 ]
 
 [project.scripts]
-secops-mcp-server = "secops_mcp_server.server:run_mcp_server"
+secops-mcp-server = "secops_mcp_server.run:run_mcp_server"
 
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/secops_mcp_server", "src/config"]

packages/secops_mcp_server/src/config/__init__.py

@@ -0,0 +1 @@
+"""Configuration modules for the MCP server."""

packages/secops_mcp_server/src/config/gunicorn.py

@@ -0,0 +1,86 @@
+"""Gunicorn configuration and custom logger for MCP server.
+
+This module provides a custom logger class for gunicorn that integrates
+with Python's logging system for consistent log formatting.
+"""
+
+import logging
+
+
+class GunicornLogger:
+    """Custom logger class for gunicorn.
+
+    This logger integrates gunicorn's logging with Python's standard logging
+    to ensure consistent log formatting across the application.
+    """
+
+    def __init__(self, cfg):
+        """Initialize the GunicornLogger.
+
+        Args:
+            cfg: Gunicorn configuration object
+        """
+        self.cfg = cfg
+        self._error_log = None
+        self._access_log = None
+
+    @property
+    def error_log(self):
+        """Get the error logger."""
+        if not self._error_log:
+            self._error_log = logging.getLogger("gunicorn.error")
+        return self._error_log
+
+    @property
+    def access_log(self):
+        """Get the access logger."""
+        if not self._access_log:
+            self._access_log = logging.getLogger("gunicorn.access")
+        return self._access_log
+
+    def critical(self, msg, *args, **kwargs):
+        """Log critical message."""
+        self.error_log.critical(msg, *args, **kwargs)
+
+    def error(self, msg, *args, **kwargs):
+        """Log error message."""
+        self.error_log.error(msg, *args, **kwargs)
+
+    def warning(self, msg, *args, **kwargs):
+        """Log warning message."""
+        self.error_log.warning(msg, *args, **kwargs)
+
+    def info(self, msg, *args, **kwargs):
+        """Log info message."""
+        self.error_log.info(msg, *args, **kwargs)
+
+    def debug(self, msg, *args, **kwargs):
+        """Log debug message."""
+        self.error_log.debug(msg, *args, **kwargs)
+
+    def exception(self, msg, *args, **kwargs):
+        """Log exception message."""
+        self.error_log.exception(msg, *args, **kwargs)
+
+    def log(self, lvl, msg, *args, **kwargs):
+        """Log message at specific level."""
+        self.error_log.log(lvl, msg, *args, **kwargs)
+
+    def access(self, resp, req, environ, request_time):
+        """Log access information."""
+        self.access_log.info(
+            '%s - "%s %s %s" %d',
+            environ.get("REMOTE_ADDR", "-"),
+            environ.get("REQUEST_METHOD", "-"),
+            environ.get("PATH_INFO", "-"),
+            environ.get("SERVER_PROTOCOL", "-"),
+            resp.status_code,
+        )
+
+    def reopen_files(self):
+        """Reopen log files (for log rotation)."""
+        pass
+
+    def close_on_exec(self):
+        """Close log files on exec."""
+        pass

packages/secops_mcp_server/src/secops_mcp_server/run.py

@@ -0,0 +1,64 @@
+"""Runtime entry points for the SecOps MCP server.
+
+This module provides different ways to run the MCP server:
+- stdio: Standard input/output transport (default for CLI tools)
+- http: HTTP/SSE transport using uvicorn (for local development)
+"""
+
+import logging
+import os
+
+from gg_api_core.sentry_integration import init_sentry
+
+from secops_mcp_server.server import mcp
+
+logger = logging.getLogger(__name__)
+
+
+def run_stdio():
+    """Run the MCP server over stdio transport.
+
+    This is the default mode for MCP servers, used when the server
+    is invoked as a subprocess by MCP clients like Claude Desktop.
+    """
+    init_sentry()
+    # Run the server in stdio mode
+    logger.info("SecOps MCP server running on stdio")
+    mcp.run()
+
+
+def run_http_with_uvicorn():
+    """Run the MCP server over HTTP using uvicorn ASGI server.
+
+    This is meant for local development. For production ready setup,
+    better use gunicorn with uvicorn ASGI workers via ggmcp_http.sse_app:app
+    """
+
+    init_sentry()
+    # Get host and port from environment variables
+    mcp_port = int(os.environ.get("MCP_PORT", "8000"))
+    mcp_host = os.environ.get("MCP_HOST", "127.0.0.1")
+
+    # Use HTTP/SSE transport with uvicorn
+    import uvicorn
+
+    logger.info(f"Starting SecOps MCP server on {mcp_host}:{mcp_port}")
+    uvicorn.run(mcp.sse_app(), host=mcp_host, port=mcp_port)
+
+
+def run_mcp_server():
+    """Run the MCP server with transport auto-detection.
+
+    If MCP_PORT is set, uses HTTP/SSE transport.
+    Otherwise, uses stdio transport (default).
+    """
+    mcp_port = os.environ.get("MCP_PORT")
+
+    if mcp_port:
+        run_http_with_uvicorn()
+    else:
+        run_stdio()
+
+
+if __name__ == "__main__":
+    run_mcp_server()

packages/secops_mcp_server/src/secops_mcp_server/server.py

@@ -1,15 +1,13 @@
 """GitGuardian MCP server for SecOps teams with incident management tools."""
 
 import logging
-import os
 from typing import Any, Literal
 
 from developer_mcp_server.add_health_check import add_health_check
 from developer_mcp_server.register_tools import register_developer_tools
 from fastmcp.exceptions import ToolError
 from gg_api_core.mcp_server import get_mcp_server, register_common_tools
 from gg_api_core.scopes import set_secops_scopes
-from gg_api_core.sentry_integration import init_sentry
 from gg_api_core.tools.assign_incident import assign_incident
 from gg_api_core.tools.create_code_fix_request import create_code_fix_request
 from gg_api_core.tools.list_users import list_users
@@ -118,7 +116,6 @@ class ListHoneytokensParams(BaseModel):
     log_level="DEBUG",
     instructions=SECOPS_INSTRUCTIONS,
 )
-logger.debug("Created SecOps GitGuardianFastMCP instance")
 
 register_developer_tools(mcp)
 add_health_check(mcp)
@@ -208,37 +205,4 @@ async def get_current_token_info() -> dict[str, Any]:
 
 register_common_tools(mcp)
 
-
-def run_mcp_server():
-    logger.info("Starting SecOps MCP server...")
-
-    # Initialize Sentry if configured (optional)
-    sentry_enabled = init_sentry()
-    if sentry_enabled:
-        logger.info("Sentry monitoring is enabled")
-    else:
-        logger.debug("Sentry monitoring is not configured")
-
-    # Check if HTTP/SSE transport is requested via environment variables
-    mcp_port = os.environ.get("MCP_PORT")
-    mcp_host = os.environ.get("MCP_HOST", "127.0.0.1")
-
-    if mcp_port:
-        # Use HTTP/SSE transport
-        import uvicorn
-
-        logger.info(f"Starting MCP server with HTTP/SSE transport on {mcp_host}:{mcp_port}")
-        uvicorn.run(mcp.sse_app(), host=mcp_host, port=int(mcp_port))
-    else:
-        # Use default stdio transport
-        logger.info("Starting MCP server with stdio transport (default)")
-        mcp.run()
-
-
-# Entrypoint for the Docker container and Kubernetes deployment
-def run_mcp_server_over_http():
-    run_mcp_server()
-
-
-if __name__ == "__main__":
-    run_mcp_server()
+logger.info("SecOps MCP server instance created and configured")

packages/secops_mcp_server/src/secops_mcp_server/sse_app.py

@@ -0,0 +1,25 @@
+"""ASGI application for MCP server over HTTP/SSE.
+
+This module exports the ASGI application for use with ASGI servers like gunicorn + uvicorn.
+It imports the configured MCP server and exposes its SSE application.
+
+This module is specifically for production deployment with gunicorn.
+For local development, use the run_http_with_uvicorn() function instead.
+"""
+
+import logging
+
+from gg_api_core.sentry_integration import init_sentry
+
+from secops_mcp_server.server import mcp
+
+logger = logging.getLogger(__name__)
+
+# Initialize Sentry for production deployment
+init_sentry()
+
+# Export the ASGI application
+# This will be used by gunicorn as: secops_mcp_server.sse_app:app
+app = mcp.sse_app()
+
+logger.info("MCP SSE application initialized for HTTP/SSE transport")