ds_mcp_template/run_mcp_server.py

#!/usr/bin/env python3
"""
MCP Modular Server Runner
Run the modular MCP server with dynamic transport selection
"""
import argparse
import sys
import os
import logging

# Default logging configuration (can be overridden by --verbose)
logging.getLogger("mcp").setLevel(logging.WARNING)
logging.getLogger("uvicorn").setLevel(logging.WARNING)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)

# Add the src directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

from mcp_template.server.modular_server import create_default_server

# Global server object for MCP dev command compatibility
mcp_server = None
mcp = None  # Global MCP object for `mcp dev` command compatibility


def get_dev_mcp():
    """Get the MCP object for dev mode"""
    global mcp
    if mcp is None:
        try:
            # Create a basic server instance for dev mode
            dev_server = create_default_server("stdio", 8050)
            mcp = dev_server.mcp  # Extract the FastMCP instance
        except Exception as e:
            print(f"Warning: Could not create MCP server for dev mode: {e}")
            mcp = None
    return mcp


# Initialize MCP object for dev command
mcp = get_dev_mcp()


def get_mcp_server(transport: str = "stdio", port: int = 8050):
    """Get or create the MCP server instance"""
    global mcp_server

    if mcp_server is None:
        mcp_server = create_default_server(transport, port)

    return mcp_server


def run_server(transport: str = "stdio", port: int = 8050) -> None:
    """Run the modular MCP server with the specified transport method and port"""
    supported_transports = ["stdio", "sse", "streamable-http"]

    if transport not in supported_transports:
        raise ValueError(f"Unknown transport: {transport}. Supported: {supported_transports}")

    print(f"Starting MCP Modular Server with {transport.upper()} transport on port {port}")

    # Create and run the server
    server = get_mcp_server(transport, port)

    if transport == "stdio":
        print("Server ready for stdio communication")
    elif transport == "sse":
        print(f"Server ready for SSE communication on http://0.0.0.0:{port}/sse")
    elif transport == "streamable-http":
        print(f"Server ready for Streamable HTTP communication on port {port}")

    # Run the server
    server.run(transport)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Run MCP Modular Server",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python run_mcp_server.py                           # Run with stdio on port 8050 (default)
  python run_mcp_server.py --transport stdio         # Run with stdio explicitly
  python run_mcp_server.py --transport sse           # Run with SSE transport
  python run_mcp_server.py --transport sse --port 8080  # Run SSE on custom port
  python run_mcp_server.py --transport streamable-http --port 9000  # Run Streamable HTTP on port 9000
  python run_mcp_server.py --dev                     # Run in development mode (SSE on port 3000)
  python run_mcp_server.py --dev --verbose           # Dev mode with detailed logging

MCP Dev Command:
  mcp dev dev_run.py                                 # Use MCP dev command with dedicated dev file

The server will automatically discover and register:
- Tools from the tools directory
- Prompts from the prompts directory
- Resources from the resources directory
        """
    )
    parser.add_argument(
        "--transport",
        choices=["stdio", "sse", "streamable-http"],
        default="stdio",
        help="Transport method to use (default: stdio)"
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8050,
        help="Port to run the server on (default: 8050, only used for HTTP transports)"
    )
    parser.add_argument(
        "--dev",
        action="store_true",
        help="Run in development mode with SSE transport on port 3000"
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Enable verbose logging (shows all MCP and server messages)"
    )

    args = parser.parse_args()

    # Configure logging based on verbose flag
    if args.verbose:
        logging.getLogger("mcp").setLevel(logging.INFO)
        logging.getLogger("uvicorn").setLevel(logging.INFO)
        logging.getLogger("uvicorn.access").setLevel(logging.INFO)
        print("Verbose logging enabled")
    else:
        # Keep the reduced logging for cleaner output
        logging.getLogger("mcp").setLevel(logging.WARNING)
        logging.getLogger("uvicorn").setLevel(logging.WARNING)
        logging.getLogger("uvicorn.access").setLevel(logging.WARNING)

    if args.dev:
        # In dev mode, run with SSE transport on a standard dev port
        # This is what MCP dev command expects
        print("Starting MCP server in development mode...")
        server = get_mcp_server("sse", 3000)  # Standard dev port
        print(f"🌐 Server ready for SSE communication on http://0.0.0.0:3000/sse")
        server.run("sse")

    run_server(args.transport, args.port)