CME Prediction Markets MCP Server

#!/usr/bin/env python3 """ MCP Server Interaction Demo Shows different ways to interact with the CME MCP server """ import asyncio import json from typing import Dict, Any import httpx from datetime import datetime, timedelta # Base URL for the API server BASE_URL = "http://localhost:8080" class MCPInteractionDemo: """Demonstrates different ways to interact with the MCP server.""" def __init__(self): self.client = httpx.AsyncClient(timeout=30.0) async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.client.aclose() async def check_health(self): """Check if the server is running.""" print("🏥 Health Check") print("=" * 50) try: response = await self.client.get(f"{BASE_URL}/health") result = response.json() print(f"✅ Server Status: {result}") return True except Exception as e: print(f"❌ Server not accessible: {e}") return False async def list_mcp_tools(self): """List available MCP tools via API.""" print("\n🛠️ Available MCP Tools") print("=" * 50) try: response = await self.client.get(f"{BASE_URL}/mcp/tools") tools = response.json() for tool in tools.get("tools", []): print(f"📦 {tool['name']}") print(f" Description: {tool.get('description', 'N/A')}") print(f" Schema: {json.dumps(tool.get('inputSchema', {}), indent=2)}") print() except Exception as e: print(f"❌ Error listing tools: {e}") async def test_contract_info_tool(self): """Test the get_contract_info MCP tool.""" print("\n📊 Testing Contract Info Tool") print("=" * 50) # Test with a sample contract payload = { "name": "get_contract_info", "arguments": { "symbol": "BTC_95000_YES" } } try: response = await self.client.post( f"{BASE_URL}/mcp/call_tool", json=payload ) result = response.json() print(f"✅ Contract Info Result:") print(json.dumps(result, indent=2)) except Exception as e: print(f"❌ Error testing contract info: {e}") async def test_trading_data_tool(self): """Test the query_trading_data MCP tool.""" print("\n📈 Testing Trading Data Tool") print("=" * 50) # Create date range for last 7 days end_time = datetime.now() start_time = end_time - timedelta(days=7) payload = { "name": "query_trading_data", "arguments": { "contract_symbol": "BTC_95000_YES", "start_time": start_time.isoformat(), "end_time": end_time.isoformat(), "aggregation": "hour" } } try: response = await self.client.post( f"{BASE_URL}/mcp/call_tool", json=payload ) result = response.json() print(f"✅ Trading Data Result:") print(json.dumps(result, indent=2)) except Exception as e: print(f"❌ Error testing trading data: {e}") async def test_claim_verification_tool(self): """Test the verify_claim MCP tool.""" print("\n🔍 Testing Claim Verification Tool") print("=" * 50) payload = { "name": "verify_claim", "arguments": { "claim_text": "Bitcoin reached $95,000 on December 15th, 2025", "source": "market_analysis", "confidence_threshold": 0.8 } } try: response = await self.client.post( f"{BASE_URL}/mcp/call_tool", json=payload ) result = response.json() print(f"✅ Claim Verification Result:") print(json.dumps(result, indent=2)) except Exception as e: print(f"❌ Error testing claim verification: {e}") async def show_swagger_ui_access(self): """Show how to access interactive API documentation.""" print("\n📚 Interactive API Documentation") print("=" * 50) print(f"🌐 Swagger UI: {BASE_URL}/docs") print(f"📄 OpenAPI Schema: {BASE_URL}/openapi.json") print("\n💡 Use the Swagger UI to:") print(" • Explore all available endpoints") print(" • Test API calls interactively") print(" • View request/response schemas") print(" • Generate client code") async def demonstrate_direct_mcp_usage(self): """Show how to use MCP tools directly (not via HTTP API).""" print("\n🔧 Direct MCP Tool Usage Example") print("=" * 50) # This would be the code for direct MCP integration mcp_example = ''' # Example: Direct MCP tool usage (in Python client) import asyncio from src.mcp.tools.query_trading_data import query_trading_data async def use_mcp_tool_directly(): """Direct tool usage without HTTP layer.""" arguments = { "contract_symbol": "BTC_95000_YES", "start_time": "2025-12-19T00:00:00Z", "end_time": "2025-12-19T23:59:59Z", "aggregation": "hour" } result = await query_trading_data(arguments) return result # Run the tool result = await use_mcp_tool_directly() print(result) ''' print("💻 Python Code Example:") print(mcp_example) async def run_full_demo(self): """Run the complete interaction demo.""" print("🚀 CME MCP Server Interaction Demo") print("=" * 60) # Check if server is running if not await self.check_health(): print("\n❌ Please start the server first:") print(" poetry run uvicorn src.main:app --reload --host 0.0.0.0 --port 8080") return # Run all demo functions await self.list_mcp_tools() await self.test_contract_info_tool() await self.test_trading_data_tool() await self.test_claim_verification_tool() await self.show_swagger_ui_access() await self.demonstrate_direct_mcp_usage() print("\n🎉 Demo Complete!") print("\n📖 Best Practices for MCP Interaction:") print(" 1. Use Swagger UI for exploration and testing") print(" 2. Use HTTP API for external integrations") print(" 3. Use direct MCP tools for internal Python code") print(" 4. Always check tool schemas before calling") print(" 5. Handle errors gracefully in production") async def main(): """Main demo function.""" async with MCPInteractionDemo() as demo: await demo.run_full_demo() if __name__ == "__main__": asyncio.run(main())