Merge pull request #2 from ntk148v/sse

feat(transport): support SSE transport

Author: ntk148v

Makefile

@@ -0,0 +1,34 @@
+.PHONY: setup run dev install deploy test clean format type-check
+
+# Set the Python version from cookiecutter or default to 3.12
+PYTHON_VERSION := 3.12
+
+# Setup with uv
+setup:
+	# Check if uv is installed, install if not
+	@which uv >/dev/null || (curl -LsSf https://astral.sh/uv/install.sh | sh)
+	# Create a virtual environment
+	uv venv
+	# Install dependencies with development extras
+	uv pip install -e ".[dev]"
+	@echo "✅ Environment setup complete. Activate it with 'source .venv/bin/activate' (Unix/macOS) or '.venv\\Scripts\activate' (Windows)"
+
+# Run the server directly
+run:
+	uv run src/alertmanager-mcp-server/server.py
+
+# Run in development mode with MCP inspector
+dev:
+	mcp dev src/alertmanager-mcp-server/server.py
+
+# Install in Claude desktop
+install:
+	mcp install src/alertmanager-mcp-server/server.py
+
+# Docker build
+docker-build:
+	docker build -t kiennt26/alertmanager-mcp-server:latest .
+
+# Run with Docker
+docker-run:
+	docker run -p 8000:8000 kiennt26/alertmanager-mcp-server:latest

pyproject.toml

@@ -4,7 +4,10 @@ version = "1.0.0"
 description = "MCP Server for Prometheus Alertmanager integration"
 readme = "README.md"
 requires-python = ">=3.12"
-dependencies = ["mcp[cli]>=1.8.1", "requests>=2.32.3"]
+dependencies = [
+    "mcp[cli]>=1.8.1",
+    "requests>=2.32.3",
+]
 
 [project.scripts]
 alertmanager-mcp-server = "alertmanager_mcp_server.server:run_server"

src/alertmanager_mcp_server/server.py

@@ -2,12 +2,17 @@
 import os
 from dataclasses import dataclass
 from typing import Any, Dict, Optional, List
-import sys
 
-import dotenv
-import requests
+from mcp.server import Server
 from mcp.server.fastmcp import FastMCP
+from mcp.server.sse import SseServerTransport
 from requests.compat import urljoin
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.routing import Mount, Route
+import dotenv
+import requests
+import uvicorn
 
 dotenv.load_dotenv()
 mcp = FastMCP("Alertmanager MCP")
@@ -279,17 +284,90 @@ def setup_environment():
     return True
 
 
+def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
+    """Create a Starlette application that can serve the provided MCP server with SSE.
+
+    Sets up a Starlette web application with routes for SSE (Server-Sent Events)
+    communication with the MCP server.
+
+    Args:
+        mcp_server: The MCP server instance to connect
+        debug: Whether to enable debug mode for the Starlette app
+
+    Returns:
+        A configured Starlette application
+    """
+    # Create an SSE transport with a base path for messages
+    sse = SseServerTransport("/messages/")
+
+    async def handle_sse(request: Request) -> None:
+        """Handler for SSE connections.
+
+        Establishes an SSE connection and connects it to the MCP server.
+
+        Args:
+            request: The incoming HTTP request
+        """
+        # Connect the SSE transport to the request
+        async with sse.connect_sse(
+                request.scope,
+                request.receive,
+                request._send,  # noqa: SLF001
+        ) as (read_stream, write_stream):
+            # Run the MCP server with the SSE streams
+            await mcp_server.run(
+                read_stream,
+                write_stream,
+                mcp_server.create_initialization_options(),
+            )
+
+    # Create and return the Starlette application with routes
+    return Starlette(
+        debug=debug,
+        routes=[
+            Route("/sse", endpoint=handle_sse),  # Endpoint for SSE connections
+            # Endpoint for posting messages
+            Mount("/messages/", app=sse.handle_post_message),
+        ],
+    )
+
+
 def run_server():
     """Main entry point for the Prometheus Alertmanager MCP Server"""
-    # Setup environment
-    if not setup_environment():
-        sys.exit(1)
-
+    setup_environment()
+    # Get the underlying MCP server from the FastMCP instance
+    mcp_server = mcp._mcp_server  # noqa: WPS437
+
+    import argparse
+
+    # Set up command-line argument parsing
+    parser = argparse.ArgumentParser(
+        description='Run MCP server with configurable transport')
+    # Allow choosing between stdio and SSE transport modes
+    parser.add_argument('--transport', choices=['stdio', 'sse'], default='stdio',
+                        help='Transport mode (stdio or sse)')
+    # Host configuration for SSE mode
+    parser.add_argument('--host', default='0.0.0.0',
+                        help='Host to bind to (for SSE mode)')
+    # Port configuration for SSE mode
+    parser.add_argument('--port', type=int, default=8000,
+                        help='Port to listen on (for SSE mode)')
+    args = parser.parse_args()
     print("\nStarting Prometheus Alertmanager MCP Server...")
-    print("Running server in standard mode...")
 
-    # Run the server with the stdio transport
-    mcp.run(transport="stdio")
+    # Launch the server with the selected transport mode
+    if args.transport == 'stdio':
+        print("Running server with stdio transport (default)")
+        # Run with stdio transport (default)
+        # This mode communicates through standard input/output
+        mcp.run(transport='stdio')
+    else:
+        print("Running server with SSE transport (web-based)")
+        # Run with SSE transport (web-based)
+        # Create a Starlette app to serve the MCP server
+        starlette_app = create_starlette_app(mcp_server, debug=True)
+        # Start the web server with the configured host and port
+        uvicorn.run(starlette_app, host=args.host, port=args.port)
 
 
 if __name__ == "__main__":

tests/test_server.py

@@ -266,12 +266,3 @@ def test_run_server_success(mock_mcp, mock_setup_env):
         mock_mcp.run.assert_called_once_with(transport="stdio")
         assert any("Starting Prometheus Alertmanager MCP Server" in str(call)
                    for call in mock_print.call_args_list)
-
-
-@patch("alertmanager_mcp_server.server.setup_environment", return_value=False)
-def test_run_server_exit(mock_setup_env):
-    with patch("sys.exit", side_effect=SystemExit) as mock_exit:
-        with pytest.raises(SystemExit):
-            server.run_server()
-        mock_setup_env.assert_called_once()
-        mock_exit.assert_called_once_with(1)