ds_mcp_template/examples/demo_client.py

#!/usr/bin/env python3
"""
MCP Template Demo Client
Demonstrates how to connect to and interact with MCP servers
"""

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 clients.client_factory import AIClientFactory
from transport.transport_manager import TransportManager
from core.types import TransportType


class MCPDemoClient:
    """Demo client for interacting with MCP servers"""

    def __init__(self, ai_provider="openai", model="gpt-4o"):
        self.ai_provider = ai_provider
        self.model = model
        self.ai_client = None
        self.transport_manager = TransportManager()

    async def initialize(self):
        """Initialize the demo client"""
        print(f"🤖 Initializing {self.ai_provider} client with {self.model}...")

        # Initialize AI client
        self.ai_client = AIClientFactory.create_client(
            provider=self.ai_provider,
            model_name=self.model
        )

        await self.ai_client.initialize()
        print("✅ AI client initialized")

    async def demo_stdio_interaction(self):
        """Demo direct STDIO interaction with MCP server"""
        print("\n🔧 Demo: Direct STDIO Interaction")
        print("This demonstrates direct tool calls without AI involvement")

        # For this demo, we'll simulate MCP server responses
        # In a real scenario, you'd connect to an actual MCP server

        print("📋 Available tools (simulated):")
        print("  • add: Add two numbers together")
        print("  • multiply: Multiply two numbers")
        print("  • greet_user: Generate personalized greeting")
        print("  • calculate_bmi: Calculate BMI and health category")

        # Simulate some tool calls
        print("\n🧮 Simulating tool calls:")

        # Simulate add tool
        print("  add(5, 3) = 8")

        # Simulate greet tool
        print("  greet_user('Alice', 'casual') = 'Hey Alice! Welcome aboard! 🎉'")

        # Simulate BMI tool
        print("  calculate_bmi(70, 1.75) = 'Your BMI is 22.9 (Normal weight)'")

    async def demo_ai_with_tools(self):
        """Demo AI client with MCP tool integration"""
        print("\n🧠 Demo: AI Client with MCP Tools")
        print("This demonstrates how AI can use MCP tools to perform tasks")

        # Sample queries that would benefit from MCP tools
        queries = [
            "What is 15 + 27?",
            "Can you greet Sarah in a professional manner?",
            "What's the BMI for someone who weighs 80kg and is 1.8m tall?",
            "Calculate 12 * 8 for me"
        ]

        print("💭 Sample queries for AI + MCP integration:")
        for i, query in enumerate(queries, 1):
            print(f"  {i}. {query}")

        print("\n📝 Note: To run actual AI+MCP integration, you need:")
        print("  1. A running MCP server (see demo_server.py)")
        print("  2. Proper API keys in environment variables")
        print("  3. Network connection for AI provider")

    async def demo_transport_switching(self):
        """Demo transport layer switching capabilities"""
        print("\n🔄 Demo: Transport Layer Switching")
        print("The MCP template supports easy switching between transports:")

        print("📡 SSE (Server-Sent Events):")
        print("  • Best for: Web applications, remote connections")
        print("  • Protocol: HTTP-based, real-time communication")
        print("  • Use case: Browser-based MCP clients")

        print("\n💻 STDIO (Standard Input/Output):")
        print("  • Best for: Local applications, direct process communication")
        print("  • Protocol: Direct pipes between processes")
        print("  • Use case: CLI tools, local development")

        print("\n🔧 Easy switching example:")
        print("  # Switch to SSE transport")
        print("  transport_manager.switch_transport('sse', host='localhost', port=8050)")
        print("  ")
        print("  # Switch to STDIO transport")
        print("  transport_manager.switch_transport('stdio')")

    async def demo_configuration(self):
        """Demo configuration management"""
        print("\n⚙️  Demo: Configuration Management")
        print("The MCP template supports flexible configuration:")

        print("📄 Configuration sources (in order of priority):")
        print("  1. Environment variables")
        print("  2. JSON configuration files")
        print("  3. Default values")

        print("\n🌍 Environment variables example:")
        print("  MCP_SERVER_NAME=MyServer")
        print("  MCP_TRANSPORT=sse")
        print("  MCP_AI_PROVIDER=openai")
        print("  OPENAI_API_KEY=your_key_here")

        print("\n📋 JSON config example:")
        print("  {")
        print('    "server": {"name": "MyServer", "port": 8080},')
        print('    "client": {"provider": "openai", "model": "gpt-4o"}')
        print("  }")

    async def run_demo(self):
        """Run the complete demo"""
        print("🎭 MCP Template Demo Client")
        print("=" * 50)

        await self.initialize()
        await self.demo_stdio_interaction()
        await self.demo_ai_with_tools()
        await self.demo_transport_switching()
        await self.demo_configuration()

        print("\n🎉 Demo completed!")
        print("\n📚 Next steps:")
        print("  • Run 'python examples/demo_server.py calculator' for a calculator server")
        print("  • Run 'python examples/demo_server.py knowledge' for a knowledge base server")
        print("  • Check the tests/ directory for comprehensive test examples")
        print("  • Read docs in README.md for detailed usage instructions")


async def main():
    """Main demo function"""
    if len(sys.argv) > 1:
        ai_provider = sys.argv[1]
        model = sys.argv[2] if len(sys.argv) > 2 else None
    else:
        ai_provider = "openai"
        model = "gpt-4o"

    print(f"🤖 Using AI provider: {ai_provider}")
    if model:
        print(f"🧠 Using model: {model}")

    client = MCPDemoClient(ai_provider, model)

    try:
        await client.run_demo()
    except KeyboardInterrupt:
        print("\n👋 Demo interrupted by user")
    except Exception as e:
        print(f"❌ Error running demo: {e}")
        print("💡 Make sure you have the required dependencies installed:")
        print("   pip install openai aiohttp python-dotenv")
        sys.exit(1)


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