ds_mcp_template/build/lib/mcp_template/examples/server_examples.py

"""
Server Examples showing how to use the modular MCP template
"""
import asyncio
from typing import List
from ..core.types import TransportType
from ..server.server_factory import MCPServerFactory
from ..tools.tool_registry import ToolRegistry
from ..resources.data_resources import DataResources


class ServerExamples:
    """Examples of creating different types of MCP servers"""

    @staticmethod
    async def create_math_server():
        """Create a server with only math tools"""
        print("🧮 Creating Math Server...")

        # Use tool registry to get math tools
        registry = ToolRegistry()
        math_tools = registry.get_tools_by_category('math')

        server = MCPServerFactory.create_server(
            name="Math Server",
            transport=TransportType.STDIO,
            tools=math_tools
        )

        print(f"✅ Math Server created with {len(math_tools)} tools")
        return server

    @staticmethod
    async def create_developer_server():
        """Create a comprehensive developer server"""
        print("👨‍💻 Creating Developer Server...")

        registry = ToolRegistry()
        tools = registry.get_tools_by_categories(['math', 'text', 'system'])

        # Add data resources
        resources = DataResources.get_resources()

        server = MCPServerFactory.create_server(
            name="Developer Server",
            transport=TransportType.SSE,
            host="localhost",
            port=8050,
            tools=tools,
            resources=resources
        )

        print(f"✅ Developer Server created with {len(tools)} tools and {len(resources)} resources")
        return server

    @staticmethod
    async def create_business_server():
        """Create a business-focused server"""
        print("💼 Creating Business Server...")

        registry = ToolRegistry()

        # Add custom business tools
        async def calculate_roi(initial_investment: float, final_value: float) -> str:
            """Calculate Return on Investment"""
            if initial_investment <= 0:
                raise ValueError("Initial investment must be positive")
            roi = ((final_value - initial_investment) / initial_investment) * 100
            return ".2f"

        async def format_currency(amount: float, currency: str = "USD") -> str:
            """Format amount as currency"""
            return ",.2f"

        from ..core.types import MCPTool
        business_tools = [
            MCPTool(
                name="calculate_roi",
                description="Calculate Return on Investment percentage",
                input_schema={
                    "type": "object",
                    "properties": {
                        "initial_investment": {"type": "number", "description": "Initial investment amount"},
                        "final_value": {"type": "number", "description": "Final value amount"},
                    },
                    "required": ["initial_investment", "final_value"],
                },
                handler=calculate_roi,
            ),
            MCPTool(
                name="format_currency",
                description="Format a number as currency",
                input_schema={
                    "type": "object",
                    "properties": {
                        "amount": {"type": "number", "description": "Amount to format"},
                        "currency": {"type": "string", "description": "Currency code", "default": "USD"},
                    },
                    "required": ["amount"],
                },
                handler=format_currency,
            ),
        ]

        # Add to registry
        registry.add_custom_tools(business_tools)

        # Get business-relevant tools
        all_tools = registry.get_tools_by_categories(['math', 'text'])
        all_tools.extend(business_tools)

        server = MCPServerFactory.create_server(
            name="Business Server",
            transport=TransportType.STDIO,
            tools=all_tools
        )

        print(f"✅ Business Server created with {len(all_tools)} tools")
        return server

    @staticmethod
    async def create_custom_server_with_config():
        """Create a server using configuration"""
        print("⚙️ Creating Custom Server with Configuration...")

        config = {
            "name": "Custom Config Server",
            "transport": "sse",
            "host": "localhost",
            "port": 8080,
            "tools": ["math", "text"],  # Tool categories to include
            "enable_resources": True,
        }

        registry = ToolRegistry()
        tools = registry.get_tools_by_categories(config["tools"])
        resources = DataResources.get_resources() if config.get("enable_resources") else []

        server = MCPServerFactory.create_server(
            name=config["name"],
            transport=config["transport"],
            host=config["host"],
            port=config["port"],
            tools=tools,
            resources=resources
        )

        print(f"✅ Custom Server created with configuration")
        print(f"   - Tools: {len(tools)}")
        print(f"   - Resources: {len(resources)}")
        print(f"   - Transport: {config['transport']}")
        return server

    @staticmethod
    async def demo_server_lifecycle():
        """Demonstrate complete server lifecycle"""
        print("🔄 Server Lifecycle Demo")

        # Create server
        server = await ServerExamples.create_math_server()

        # List available tools
        tools = await server.list_tools()
        print(f"📋 Available tools: {[t['name'] for t in tools]}")

        # Test a tool
        try:
            result = await server.call_tool("add", {"a": 10, "b": 5})
            print(f"🧮 Tool test - add(10, 5) = {result}")
        except Exception as e:
            print(f"❌ Tool test failed: {e}")

        print("✅ Server lifecycle demo completed")
        return server