enhanced-mcp-tools/examples/fastmcp_server_patterns.py

#!/usr/bin/env python3
"""
FastMCP Server Startup Patterns - Complete Reference Implementation

This example demonstrates the correct patterns for starting FastMCP servers
in both stdio and HTTP modes, with proper error handling and CLI integration.

🛡️ FastMCP Expert Architecture Pattern:
- Intelligent Transport Selection: Auto-detect transport based on arguments
- Progressive Disclosure: Security-first server initialization
- Mixin Composition: Modular tool registration with proper prefixes
- Production Ready: Comprehensive error handling and logging
"""

import argparse
import sys
from typing import Optional

from fastmcp import FastMCP


class ServerManager:
    """FastMCP Server Manager with intelligent transport selection"""

    def __init__(self, name: str = "FastMCP Server"):
        self.name = name
        self.app: Optional[FastMCP] = None

    def create_server(self) -> FastMCP:
        """Create and configure FastMCP server"""
        if self.app is None:
            self.app = FastMCP(self.name)
            self._register_tools()
        return self.app

    def _register_tools(self):
        """Register example tools - replace with your actual tools"""
        @self.app.tool
        def hello_world(name: str = "World") -> str:
            """Say hello to someone"""
            return f"Hello, {name}!"

        @self.app.tool
        def get_status() -> dict:
            """Get server status"""
            return {
                "name": self.name,
                "status": "running",
                "transport": "active"
            }

    def run_stdio(self) -> None:
        """Run server in stdio mode (for MCP clients like Claude Desktop)"""
        try:
            print(f"🚀 {self.name} - stdio mode", file=sys.stderr)
            print("📋 Ready for MCP client communication", file=sys.stderr)

            app = self.create_server()
            app.run(transport="stdio")

        except Exception as e:
            print(f"❌ Stdio mode error: {e}", file=sys.stderr)
            sys.exit(1)

    def run_http(self,
                 host: str = "localhost",
                 port: int = 8000,
                 transport: str = "sse") -> None:
        """Run server in HTTP mode with specified transport

        Args:
            host: Host to bind to
            port: Port to bind to
            transport: HTTP transport type ("sse" or "streamable-http")
        """
        try:
            transport_name = {
                "sse": "SSE (Server-Sent Events)",
                "streamable-http": "Streamable HTTP"
            }.get(transport, transport)

            print(f"🚀 {self.name} - HTTP server", file=sys.stderr)
            print(f"🌐 Server: http://{host}:{port}", file=sys.stderr)
            print(f"📡 Transport: {transport_name}", file=sys.stderr)

            app = self.create_server()
            app.run(transport=transport, host=host, port=port)

        except Exception as e:
            print(f"❌ HTTP server error: {e}", file=sys.stderr)
            sys.exit(1)

    def get_package_version(self) -> str:
        """Get package version for startup banner"""
        try:
            from importlib.metadata import version
            return version("enhanced-mcp-tools")
        except:
            return "1.0.0"


def create_cli_parser() -> argparse.ArgumentParser:
    """Create comprehensive CLI argument parser for FastMCP server"""
    parser = argparse.ArgumentParser(
        prog="fastmcp-server",
        description="FastMCP Server with intelligent transport selection",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
FastMCP Transport Modes:

STDIO Mode (for MCP clients):
  --stdio                         # Direct MCP protocol communication

HTTP Modes (for web/API access):
  --transport sse                 # Server-Sent Events (recommended)
  --transport streamable-http     # Streamable HTTP responses

Examples:
  %(prog)s --stdio                        # MCP client mode
  %(prog)s --transport sse                # HTTP with SSE (default)
  %(prog)s --transport streamable-http    # HTTP with streaming
  %(prog)s --host 0.0.0.0 --port 8080     # Custom host/port

For uvx usage:
  uvx my-fastmcp-server --stdio           # Direct execution
        """
    )

    # Transport selection
    transport_group = parser.add_mutually_exclusive_group()
    transport_group.add_argument(
        "--stdio",
        action="store_true",
        help="Run in stdio mode (for MCP clients like Claude Desktop)"
    )
    transport_group.add_argument(
        "--transport",
        choices=["sse", "streamable-http"],
        default="sse",
        help="HTTP transport type (default: sse)"
    )

    # Server configuration
    parser.add_argument(
        "--host",
        default="localhost",
        help="Host to bind to (default: localhost)"
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8000,
        help="Port to bind to (default: 8000)"
    )
    parser.add_argument(
        "--name",
        default="FastMCP Server",
        help="Server name (default: FastMCP Server)"
    )

    # Utility options
    parser.add_argument(
        "--version",
        action="version",
        version="FastMCP Server Patterns 1.0.0"
    )
    parser.add_argument(
        "--list-tools",
        action="store_true",
        help="List available tools and exit"
    )

    return parser


async def list_tools_async(server_manager: ServerManager):
    """List all available tools asynchronously"""
    try:
        app = server_manager.create_server()
        tools = await app.get_tools()

        print(f"📋 {server_manager.name} - Available Tools:")
        print("=" * 60)

        for i, tool_name in enumerate(sorted(tools), 1):
            print(f"  {i:2d}. {tool_name}")

        print(f"\n🎯 Total: {len(tools)} tools available")
        print("\n💡 Usage:")
        print("  --stdio           # For MCP clients")
        print("  --transport sse   # HTTP with SSE")

    except Exception as e:
        print(f"❌ Failed to list tools: {e}", file=sys.stderr)
        sys.exit(1)


def main():
    """Main entry point demonstrating FastMCP server patterns"""
    parser = create_cli_parser()
    args = parser.parse_args()

    # Create server manager
    server_manager = ServerManager(args.name)

    # Handle list-tools option
    if args.list_tools:
        import asyncio
        asyncio.run(list_tools_async(server_manager))
        return

    # Intelligent transport selection
    try:
        if args.stdio:
            # stdio mode for MCP clients
            server_manager.run_stdio()
        else:
            # HTTP mode with selected transport
            server_manager.run_http(
                host=args.host,
                port=args.port,
                transport=args.transport
            )

    except KeyboardInterrupt:
        print(f"\n👋 Shutting down {server_manager.name}", file=sys.stderr)
        sys.exit(0)
    except Exception as e:
        print(f"❌ Server startup failed: {e}", file=sys.stderr)
        sys.exit(1)


# FastMCP Server Startup Pattern Reference
# =====================================

def demo_all_startup_patterns():
    """Demonstrate all FastMCP server startup patterns"""

    # Pattern 1: Basic stdio mode
    def pattern_stdio():
        app = FastMCP("Demo Server")
        app.run(transport="stdio")

    # Pattern 2: HTTP with SSE transport (recommended)
    def pattern_http_sse():
        app = FastMCP("Demo Server")
        app.run(transport="sse", host="localhost", port=8000)

    # Pattern 3: HTTP with streamable transport
    def pattern_http_streamable():
        app = FastMCP("Demo Server")
        app.run(transport="streamable-http", host="localhost", port=8000)

    # Pattern 4: Async startup (for advanced use cases)
    async def pattern_async():
        app = FastMCP("Demo Server")
        await app.run_async(transport="sse", host="localhost", port=8000)

    # Pattern 5: Conditional transport selection
    def pattern_intelligent_selection(use_stdio: bool = False):
        app = FastMCP("Demo Server")

        if use_stdio:
            # For MCP clients
            app.run(transport="stdio")
        else:
            # For HTTP clients
            app.run(transport="sse", host="localhost", port=8000)

    print("🛡️ FastMCP Expert Patterns:")
    print("1. stdio: app.run(transport='stdio')")
    print("2. HTTP SSE: app.run(transport='sse', host='host', port=port)")
    print("3. HTTP Streamable: app.run(transport='streamable-http', host='host', port=port)")
    print("4. Async: await app.run_async(transport='sse', ...)")
    print("5. Intelligent Selection: Conditional based on args")


if __name__ == "__main__":
    main()