Opinion.trade MCP Server

#!/usr/bin/env python3 """ Integration test for Opinion.trade MCP Server Tests: Markets, Orderbooks, and Trading """ import os import sys from pathlib import Path # Add src to path sys.path.insert(0, str(Path(__file__).parent / "src")) from mcp_opinion.config import OpinionConfig from mcp_opinion.public_client import PublicClient from mcp_opinion.trading_client import TradingClient def test_config(): """Test configuration loading""" print("\n" + "="*60) print("TEST 1: Configuration Loading") print("="*60) try: config = OpinionConfig.from_env() print(f"✓ API Key loaded: {config.api_key[:10]}...") print(f"✓ Chain ID: {config.chain_id}") print(f"✓ API Host: {config.api_host}") print(f"✓ API Timeout: {config.api_timeout}s") print(f"✓ Enable Trading: {config.enable_trading}") if config.enable_trading: print(f"✓ Private Key loaded: {config.private_key[:10]}...") print(f"✓ Mode: {config.get_mode_description()}") return True, config except Exception as e: print(f"✗ Config Error: {e}") import traceback traceback.print_exc() return False, None async def test_markets(config): """Test reading markets""" print("\n" + "="*60) print("TEST 2: Reading Markets") print("="*60) try: client = PublicClient( api_key=config.api_key, base_url=config.api_host, timeout=config.api_timeout ) # Get markets print("Fetching markets...") markets = await client.get_markets() print(f"✓ Found {len(markets)} markets") if markets: market = markets[0] print(f"\nFirst Market Details:") print(f" ID: {market.get('id', 'N/A')}") q = str(market.get('question', 'N/A')) print(f" Question: {q[:70]}...") print(f" YES Price: {market.get('yes_price', 'N/A')}") print(f" NO Price: {market.get('no_price', 'N/A')}") print(f" Volume: {market.get('volume', 'N/A')}") print(f" Status: {market.get('status', 'N/A')}") return True except Exception as e: print(f"✗ Markets Error: {e}") import traceback traceback.print_exc() return False async def test_orderbook(config): """Test reading orderbook""" print("\n" + "="*60) print("TEST 3: Reading Orderbook") print("="*60) try: client = PublicClient( api_key=config.api_key, base_url=config.api_host, timeout=config.api_timeout ) # Get a market first print("Fetching markets...") markets = await client.get_markets() if not markets: print("✗ No markets available") return False market_id = markets[0].get('id') print(f"Reading orderbook for market ID: {market_id}") # Get orderbook orderbook = await client.get_orderbook(market_id) print(f"✓ Got orderbook data") if orderbook: print(f"\nOrderbook Details:") yes_bids = orderbook.get('yes_bids', []) yes_asks = orderbook.get('yes_asks', []) no_bids = orderbook.get('no_bids', []) no_asks = orderbook.get('no_asks', []) print(f" YES Bids: {len(yes_bids)} levels") print(f" YES Asks: {len(yes_asks)} levels") print(f" NO Bids: {len(no_bids)} levels") print(f" NO Asks: {len(no_asks)} levels") # Show first level if available if yes_bids: bid = yes_bids[0] print(f"\n First YES Bid: {bid}") if yes_asks: ask = yes_asks[0] print(f" First YES Ask: {ask}") return True except Exception as e: print(f"✗ Orderbook Error: {e}") import traceback traceback.print_exc() return False async def test_trading(config): """Test trading functionality""" print("\n" + "="*60) print("TEST 4: Trading Capability Check") print("="*60) try: if not config.enable_trading: print("✗ Trading mode disabled (no private key found)") print(f" Mode: {config.get_mode_description()}") print(" To enable trading, set OPINION_PRIVATE_KEY in .env") return False # Check if chain_id is mainnet if config.chain_id != 56: print(f"⚠ IMPORTANT: Opinion.trade SDK only supports mainnet (chain_id=56)") print(f" Your current setting: chain_id={config.chain_id}") print(f" Change OPINION_CHAIN_ID to 56 in .env to use trading features") print(f"\n✗ Trading test skipped (testnet not supported by SDK)") return False print(f"✓ Trading mode enabled") print(f"✓ Mode: {config.get_mode_description()}") client = TradingClient(config) # Get markets for trading print("\nFetching markets for trading...") markets = await client.get_markets() if not markets: print("✗ No markets available") return False market = markets[0] market_id = market.get('id') print(f"✓ Trading client initialized") print(f"\nAvailable Market for Testing:") print(f" ID: {market_id}") q = str(market.get('question', 'N/A')) print(f" Question: {q[:70]}...") print(f" YES Price: {market.get('yes_price', 'N/A')}") print(f" NO Price: {market.get('no_price', 'N/A')}") # Test get_balance print(f"\nFetching account balance...") try: balance = await client.get_balance() print(f"✓ Account Balance: ${balance}") except Exception as e: print(f"⚠ Could not fetch balance: {e}") # Show trading info print(f"\n✓ Ready to place trades!") print(f" Example trade parameters:") print(f" - token_id: {market_id}") print(f" - side: 'BUY' or 'SELL'") print(f" - amount: 10.0 (amount in USD)") print(f" - price: 0.5 (price per share, 0-1)") return True except Exception as e: print(f"✗ Trading Error: {e}") import traceback traceback.print_exc() return False async def run_tests(): """Run all tests""" print("\n" + "="*60) print("OPINION.TRADE MCP SERVER - INTEGRATION TESTS") print("="*60) results = [] config = None # Config test (sync) success, config = test_config() results.append(("Configuration", success)) if not config: print("\n✗ Cannot run other tests without valid configuration") return 1 # Async tests results.append(("Markets", await test_markets(config))) results.append(("Orderbook", await test_orderbook(config))) results.append(("Trading", await test_trading(config))) # Summary print("\n" + "="*60) print("TEST SUMMARY") print("="*60) passed = sum(1 for _, r in results if r) total = len(results) for test_name, result in results: status = "✓ PASS" if result else "✗ FAIL" print(f"{status}: {test_name}") print(f"\nTotal: {passed}/{total} tests passed") # Special note about trading if config.chain_id != 56 and config.enable_trading: print("\n⚠ NOTE: Trading mode requires mainnet (chain_id=56)") print(f" Your .env is set to chain_id={config.chain_id}") print(" To use trading features, update: OPINION_CHAIN_ID=56") if passed >= 3: print("\n✓ Core functionality working! MCP server is ready.") return 0 else: print(f"\n✗ {total - passed} test(s) failed. See details above.") return 1 if __name__ == "__main__": import asyncio exit_code = asyncio.run(run_tests()) sys.exit(exit_code)