Nanonets MCP Server

#!/usr/bin/env python3 """ Test script for Nanonets MCP Server OCR functionality """ import asyncio import json import time import requests import base64 from pathlib import Path async def test_mcp_tools(): """Test the MCP tools functionality locally""" try: import sys sys.path.append('/home/arne/src/nanonets_mcp') from nanonets_mcp.server import mcp print("🔍 Testing MCP tools locally...") # Test 1: List available tools tools = await mcp.list_tools() print(f"✅ Available tools: {[tool.name for tool in tools]}") # Test 2: Get supported formats formats = await mcp.call_tool('get_supported_formats', {}) print(f"✅ Supported formats: {len(formats)} items") # Test 3: Load test image test_image_path = Path('/home/arne/src/nanonets_mcp/tests/research_document_b64.txt') if test_image_path.exists(): with open(test_image_path, 'r') as f: image_b64 = f.read() print(f"📷 Loaded test image: {len(image_b64)} characters") # Test 4: OCR processing print("🤖 Starting OCR processing...") start_time = time.time() result = await mcp.call_tool('ocr_image_to_markdown', { 'image_data': image_b64, 'image_format': 'png' }) end_time = time.time() processing_time = end_time - start_time print(f"✅ OCR completed in {processing_time:.2f} seconds") print("📄 OCR Result:") print("=" * 80) # Extract result if hasattr(result, '__iter__') and len(result) >= 1: content = result[0] if hasattr(content, 'text'): ocr_text = content.text else: ocr_text = str(content) else: ocr_text = str(result) print(ocr_text) print("=" * 80) # Save result with open('/home/arne/src/nanonets_mcp/tests/ocr_result.txt', 'w') as f: f.write(f"OCR Processing Time: {processing_time:.2f} seconds\\n") f.write(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}\\n") f.write("=" * 80 + "\\n") f.write(ocr_text) print(f"💾 Result saved to: /home/arne/src/nanonets_mcp/tests/ocr_result.txt") return True else: print(f"❌ Test image not found: {test_image_path}") return False except Exception as e: print(f"❌ Error: {e}") import traceback traceback.print_exc() return False def test_http_server(): """Test the HTTP server endpoints""" print("🌐 Testing HTTP server...") try: # Test basic connectivity response = requests.get('http://localhost:8005', timeout=5) print(f"✅ Server responding: {response.status_code}") # Test SSE endpoint (just check if it responds) try: response = requests.get('http://localhost:8005/sse', timeout=2, stream=True) print(f"✅ SSE endpoint responding: {response.status_code}") except requests.exceptions.ReadTimeout: print("✅ SSE endpoint streaming (timeout expected)") return True except Exception as e: print(f"❌ HTTP server error: {e}") return False async def main(): """Main test function""" print("🧪 Nanonets MCP Server Test Suite") print("=" * 50) # Test 1: HTTP Server http_ok = test_http_server() # Test 2: MCP Tools mcp_ok = await test_mcp_tools() print("\\n📊 Test Results:") print("=" * 50) print(f"HTTP Server: {'✅ PASS' if http_ok else '❌ FAIL'}") print(f"MCP Tools: {'✅ PASS' if mcp_ok else '❌ FAIL'}") if http_ok and mcp_ok: print("\\n🎉 ALL TESTS PASSED! Nanonets MCP Server is fully functional!") else: print("\\n⚠️ Some tests failed. Check the logs above.") if __name__ == "__main__": asyncio.run(main())