example_mcp_client.py

#!/usr/bin/env python3
"""
Example MCP client for TTS service.
This demonstrates how to connect to and use the TTS service via MCP protocol.

Note: This requires the 'mcp' package to be installed:
pip install mcp>=1.0.0
"""

import asyncio
import json
import sys
import subprocess
import os

try:
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
    MCP_CLIENT_AVAILABLE = True
except ImportError:
    MCP_CLIENT_AVAILABLE = False
    print("⚠️ MCP client not available. Install with: pip install mcp>=1.0.0")

async def test_tts_mcp_connection():
    """Test connection and basic functionality with TTS MCP service"""
    
    if not MCP_CLIENT_AVAILABLE:
        print("❌ Cannot run test - MCP client library not available")
        return False
    
    print("🔌 Testing TTS MCP Connection...")
    
    try:
        # Set up server parameters to run our TTS service in MCP-only mode
        server_params = StdioServerParameters(
            command="python3",
            args=["app.py", "--mcp-only"],
            cwd=os.path.dirname(os.path.abspath(__file__))
        )
        
        print("🚀 Starting TTS service in MCP-only mode...")
        
        # Connect to the TTS service
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                
                # Initialize the session
                print("🤝 Initializing MCP session...")
                await session.initialize()
                print("✅ MCP session initialized")
                
                # List available tools
                print("\n📋 Listing available tools...")
                tools_result = await session.list_tools()
                
                if tools_result.tools:
                    print(f"✅ Found {len(tools_result.tools)} tools:")
                    for tool in tools_result.tools:
                        print(f"   - {tool.name}: {tool.description}")
                else:
                    print("❌ No tools found")
                    return False
                
                # Test 1: Get system info
                print("\n🔍 Testing tts_get_info...")
                info_result = await session.call_tool("tts_get_info", {})
                
                if info_result.content:
                    info_data = json.loads(info_result.content[0].text)
                    print("✅ System info retrieved:")
                    print(f"   - MCP Status: {info_data.get('mcp_status')}")
                    print(f"   - Available Tools: {len(info_data.get('available_tools', []))}")
                    
                    # Show voice presets
                    voice_presets = info_data.get('voice_presets', [])
                    if voice_presets:
                        print(f"   - Voice Presets: {len(voice_presets)} available")
                        print(f"     Example: {voice_presets[0]['code']} - {voice_presets[0]['description']}")
                else:
                    print("❌ Failed to get system info")
                    return False
                
                # Test 2: Simple text synthesis (this will fail without GPU, but tests the protocol)
                print("\n🎵 Testing tts_synthesize...")
                try:
                    synthesis_result = await session.call_tool("tts_synthesize", {
                        "text": "Hello from MCP! This is a test of the text-to-speech service.",
                        "voice_preset": "v2/en_speaker_6"
                    })
                    
                    if synthesis_result.content:
                        result_data = json.loads(synthesis_result.content[0].text)
                        print(f"✅ Synthesis request processed:")
                        print(f"   - Status: {result_data.get('status', 'Unknown')}")
                        print(f"   - Success: {result_data.get('success', False)}")
                        
                        if result_data.get('success'):
                            print(f"   - Audio file: {result_data.get('audio_file')}")
                        else:
                            print("   ⚠️ Synthesis failed (expected without GPU)")
                    else:
                        print("❌ No synthesis result received")
                
                except Exception as e:
                    print(f"⚠️ Synthesis test failed (expected without GPU): {e}")
                
                # Test 3: Batch synthesis
                print("\n📦 Testing tts_batch_synthesize...")
                try:
                    batch_result = await session.call_tool("tts_batch_synthesize", {
                        "text_list": [
                            "First test sentence.",
                            "Second test sentence.",
                            "Third test sentence."
                        ],
                        "voice_preset": "v2/en_speaker_1"
                    })
                    
                    if batch_result.content:
                        batch_data = json.loads(batch_result.content[0].text)
                        print(f"✅ Batch synthesis request processed:")
                        print(f"   - Total items: {batch_data.get('total_items', 0)}")
                        print(f"   - Results: {len(batch_data.get('results', []))}")
                        
                except Exception as e:
                    print(f"⚠️ Batch synthesis test failed (expected without GPU): {e}")
                
                # Test 4: Error handling
                print("\n🚨 Testing error handling...")
                try:
                    error_result = await session.call_tool("nonexistent_tool", {})
                    
                    if error_result.content:
                        error_data = json.loads(error_result.content[0].text)
                        if "error" in error_data:
                            print("✅ Error handling works correctly")
                            print(f"   - Error message: {error_data['error']}")
                        else:
                            print("❌ Unexpected error response format")
                            return False
                
                except Exception as e:
                    print(f"✅ Error correctly caught: {e}")
                
                print("\n🎉 All MCP protocol tests completed successfully!")
                return True
                
    except Exception as e:
        print(f"❌ MCP connection test failed: {e}")
        return False

def show_usage_examples():
    """Show practical usage examples"""
    print("\n" + "=" * 60)
    print("📖 TTS MCP Integration Usage Examples")
    print("=" * 60)
    
    print("\n1️⃣ Basic MCP Client Usage:")
    print("""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_tts():
    server_params = StdioServerParameters(
        command="python3",
        args=["app.py", "--mcp-only"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Generate speech
            result = await session.call_tool("tts_synthesize", {
                "text": "Hello, world!",
                "voice_preset": "v2/en_speaker_6"
            })
            
            print(result.content[0].text)

asyncio.run(use_tts())
""")
    
    print("\n2️⃣ Integration with Claude Code or AI Assistants:")
    print("""
# The TTS service can now be used as an MCP tool by:
# - Claude Code (AI coding assistant)
# - Other MCP-compatible AI systems
# - Custom applications using MCP protocol

# Available tools:
# - tts_synthesize: Convert text to speech
# - tts_batch_synthesize: Convert multiple texts
# - tts_get_info: Get service capabilities
""")
    
    print("\n3️⃣ Running Modes:")
    print("""
# Dual mode (both Gradio UI + MCP):
python3 app.py

# MCP-only mode (for integration):
python3 app.py --mcp-only

# Access Gradio UI at: http://localhost:7860
# Access MCP via: stdio protocol
""")

async def main():
    """Main test function"""
    print("=" * 60)
    print("🎤 TTS MCP Integration Test & Demo")
    print("=" * 60)
    
    if not MCP_CLIENT_AVAILABLE:
        print("\n📦 To test MCP functionality, install the MCP client:")
        print("   pip install mcp>=1.0.0")
        show_usage_examples()
        return 1
    
    # Run the MCP connection test
    success = await test_tts_mcp_connection()
    
    # Show usage examples regardless of test result
    show_usage_examples()
    
    print("\n" + "=" * 60)
    if success:
        print("✅ TTS MCP Integration Test PASSED!")
        print("\n🚀 The TTS service is ready for:")
        print("   • Gradio web interface usage")
        print("   • MCP protocol integration")
        print("   • AI assistant tool usage")
        print("   • Programmatic API access")
        return 0
    else:
        print("⚠️ TTS MCP Integration Test completed with warnings")
        print("   (Some failures expected without GPU/dependencies)")
        return 0

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