ds_mcp_template/examples/modular_demo.py

#!/usr/bin/env python3
"""
Modular MCP Template Demo
Demonstrates the new modular architecture with separate tool, prompt, and resource folders
"""

import asyncio
import sys
import os

# Add src to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))

from server.server_factory import MCPServerFactory
from tools.tool_registry import ToolRegistry
from resources.data_resources import DataResources
from clients.client_factory import AIClientFactory


async def demo_modular_server_creation():
    """Demo: Creating servers using the modular architecture"""
    print("🏗️  Modular MCP Server Creation Demo")
    print("=" * 50)

    # 1. Using Tool Registry for category-based tool selection
    print("\n1️⃣ Tool Registry - Category-based Tool Selection")
    registry = ToolRegistry()

    print("Available tool categories:")
    categories = registry.get_available_categories()
    for category in categories:
        tools = registry.get_tools_by_category(category)
        print(f"  • {category}: {len(tools)} tools")

    # Create a math-focused server
    math_tools = registry.get_tools_by_category('math')
    math_server = MCPServerFactory.create_server(
        name="Math Server",
        tools=math_tools,
        transport="stdio"
    )
    print(f"\n✅ Created Math Server with {len(math_tools)} tools")

    # 2. Using Server Factory with categories
    print("\n2️⃣ Server Factory - Direct Category Selection")
    dev_server = MCPServerFactory.create_server_from_categories(
        name="Developer Server",
        categories=["math", "text", "system"],
        transport="stdio"
    )
    print("✅ Created Developer Server with math, text, and system tools")

    # 3. Adding custom tools to registry
    print("\n3️⃣ Custom Tools - Extending the Registry")
    async def generate_password(length: int = 12) -> str:
        """Generate a secure random password"""
        import secrets
        import string
        chars = string.ascii_letters + string.digits + "!@#$%^&*"
        return ''.join(secrets.choice(chars) for _ in range(length))

    from core.types import MCPTool
    custom_tool = MCPTool(
        name="generate_password",
        description="Generate a secure random password",
        input_schema={
            "type": "object",
            "properties": {
                "length": {"type": "integer", "description": "Password length", "default": 12}
            }
        },
        handler=generate_password,
    )

    registry.add_custom_tool(custom_tool)
    print("✅ Added custom password generation tool")

    # 4. Resources integration
    print("\n4️⃣ Resources - Static Data Integration")
    resources = DataResources.get_resources()
    print(f"Available data resources: {len(resources)}")
    for resource in resources:
        print(f"  • {resource.name}: {resource.uri}")

    # Create server with resources
    resource_server = MCPServerFactory.create_server(
        name="Resource Server",
        tools=math_tools,
        resources=resources,
        transport="stdio"
    )
    print("✅ Created server with both tools and resources")

    # 5. Test the servers
    print("\n5️⃣ Server Testing")
    print("Testing Math Server:")

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

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

    # Test resources
    print("\nTesting Resource Server:")
    resources_list = await resource_server.list_resources()
    print(f"  📄 Available resources: {len(resources_list)}")

    if resources_list:
        # Read a resource
        resource_content = await resource_server.read_resource(resources_list[0]["uri"])
        print(f"  📖 Read resource: {resources_list[0]['name']} ({len(resource_content)} chars)")

    print("\n✅ Modular server creation demo completed!")


async def demo_ai_client_integration():
    """Demo: AI client integration with modular servers"""
    print("\n🤖 AI Client Integration Demo")
    print("=" * 40)

    try:
        # Note: This demo requires API keys to be set
        print("Note: This demo requires OPENAI_API_KEY environment variable")

        # Create AI client (would work with proper API key)
        ai_client = AIClientFactory.create_openai_client(
            model_name="gpt-4o",
            api_key=os.getenv("OPENAI_API_KEY", "demo-key")
        )
        print("✅ Created OpenAI client (requires valid API key for actual use)")

        # Create a server for AI integration
        registry = ToolRegistry()
        tools = registry.get_tools_by_categories(["math", "text"])

        server = MCPServerFactory.create_server(
            name="AI Integration Server",
            tools=tools,
            transport="stdio"
        )
        print(f"✅ Created server with {len(tools)} tools for AI integration")

        print("\nExample AI queries that would work:")
        print("  • 'What is 25 * 18?'")
        print("  • 'Count the words in: The quick brown fox jumps over the lazy dog'")
        print("  • 'Convert this to uppercase: hello world'")
        print("  • 'Calculate BMI for 70kg and 1.75m height'")

    except Exception as e:
        print(f"❌ AI Client demo setup failed: {e}")
        print("This is expected without proper API key configuration")


async def main():
    """Main demo function"""
    print("🎭 Modular MCP Template Demo")
    print("Demonstrating the new modular architecture with separate folders")
    print("=" * 70)

    try:
        await demo_modular_server_creation()
        await demo_ai_client_integration()

        print("\n🎉 All demos completed successfully!")
        print("\n📚 Key Features Demonstrated:")
        print("  • 🔧 Modular tool organization by category")
        print("  • 📄 Separate resource management")
        print("  • 🏗️  Flexible server creation with tool registry")
        print("  • 🤖 AI client abstraction and integration")
        print("  • ⚙️  Easy configuration and customization")

        print("\n🚀 Next Steps:")
        print("  • Add your API keys to .env file")
        print("  • Create custom tools in src/tools/")
        print("  • Add custom resources in src/resources/")
        print("  • Create custom prompts in src/prompts/")
        print("  • Run the examples with: python examples/modular_demo.py")

    except KeyboardInterrupt:
        print("\n👋 Demo interrupted by user")
    except Exception as e:
        print(f"❌ Error running demo: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())