Upload web_search_oracle_addon.py

https://huggingface.co/spaces/Agents-MCP-Hackathon/web-search-mcp

src/addons/web_search_oracle_addon.py

@@ -0,0 +1,455 @@
+"""
+Web Search Oracle Add-on - MCP-powered web search information system.
+"""
+
+import time
+import json
+import random
+import string
+import asyncio
+from typing import Dict, List, Optional
+import gradio as gr
+from abc import ABC, abstractmethod
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+from contextlib import AsyncExitStack
+
+from ..interfaces.npc_addon import NPCAddon
+
+
+class IWebSearchService(ABC):
+    """Interface for Web Search service operations."""
+    
+    @abstractmethod
+    def search_web(self, query: str, engine: str = "brave") -> str:
+        """Search the web for information."""
+        pass
+    
+    @abstractmethod
+    def connect_to_mcp(self) -> str:
+        """Connect to MCP web search server."""
+        pass
+
+
+class WebSearchOracleService(IWebSearchService, NPCAddon):
+    """Service for managing Web Search Oracle MCP integration."""
+    
+    def __init__(self):
+        super().__init__()
+        self.connected = False
+        self.last_connection_attempt = 0
+        self.connection_cooldown = 30  # 30 seconds between connection attempts
+        self.server_url = "https://agents-mcp-hackathon-web-search-mcp.hf.space/gradio_api/mcp/sse"
+        self.session = None
+        self.tools = []
+        self.exit_stack = None
+        self.search_engines = [
+            "brave", "duckduckgo", "searxng", "serpapi", "serper", "tavily"
+        ]
+        # Set up event loop for async operations
+        try:
+            self.loop = asyncio.get_event_loop()
+        except RuntimeError:
+            self.loop = asyncio.new_event_loop()
+    
+    @property
+    def addon_id(self) -> str:
+        """Unique identifier for this add-on"""
+        return "web_search_oracle"
+    
+    @property
+    def addon_name(self) -> str:
+        """Display name for this add-on"""
+        return "Web Search Oracle (MCP)"
+    
+    @property
+    def npc_config(self) -> Dict:
+        """NPC configuration for auto-placement in world"""
+        return {
+            'id': 'web_search_oracle',
+            'name': 'Web Search Oracle (MCP)',
+            'x': 450, 'y': 300,
+            'char': '🔍',
+            'type': 'mcp',
+            'description': 'MCP-powered web search information service'
+        }
+    
+    @property
+    def ui_tab_name(self) -> str:
+        """UI tab name for this addon"""
+        return "Web Search Oracle"
+    
+    def get_interface(self) -> gr.Component:
+        """Return Gradio interface for this add-on"""
+        with gr.Column() as interface:
+            gr.Markdown("""
+            ## 🔍 Web Search Oracle (MCP)
+            
+            *I commune with the digital spirits through the mystical MCP protocol to bring you knowledge from across the web!*
+            
+            **Ask me to search for anything:**
+            - Current news and information
+            - Technical documentation
+            - Research topics
+            - General knowledge queries
+            - Powered by Model Context Protocol (MCP)
+            
+            *Available search engines: Brave, DuckDuckGo, SearxNG, SerpAPI, Serper, Tavily*
+            """)
+            
+            # Connection status
+            connection_status = gr.HTML(
+                value=f"<div style='color: {'green' if self.connected else 'red'};'>{'🟢 Connected to search spirits' if self.connected else '🔴 Disconnected from search realm'}</div>"
+            )
+            
+            with gr.Row():
+                query_input = gr.Textbox(
+                    label="Search Query",
+                    placeholder="e.g., latest news about AI, Python programming tutorials",
+                    scale=3
+                )
+                search_btn = gr.Button("🔍 Consult Search Spirits", variant="primary", scale=1)
+            
+            with gr.Row():
+                engine_dropdown = gr.Dropdown(
+                    choices=self.search_engines,
+                    value="brave",
+                    label="Search Engine",
+                    scale=2
+                )
+                max_results = gr.Slider(
+                    minimum=1,
+                    maximum=10,
+                    value=5,
+                    step=1,
+                    label="Max Results",
+                    scale=1
+                )
+            
+            search_output = gr.Textbox(
+                label="🔍 Search Wisdom",
+                lines=12,
+                interactive=False,
+                placeholder="Enter a search query and I will consult the search spirits..."
+            )
+            
+            # Example queries
+            with gr.Row():
+                gr.Examples(
+                    examples=[
+                        ["latest AI developments"],
+                        ["Python programming best practices"],
+                        ["climate change news 2024"],
+                        ["machine learning tutorials"],
+                        ["cryptocurrency market trends"],
+                        ["space exploration news"],
+                        ["renewable energy technologies"]
+                    ],
+                    inputs=[query_input],
+                    label="🌐 Try These Searches"
+                )
+            
+            # Connection controls
+            with gr.Row():
+                connect_btn = gr.Button("🔗 Connect to MCP", variant="secondary")
+                status_btn = gr.Button("📊 Check Status", variant="secondary")
+                tools_btn = gr.Button("🛠️ List Tools", variant="secondary")
+            
+            def handle_search_request(query: str, engine: str, max_results: int):
+                if not query.strip():
+                    return "❓ Please enter a search query to find information."
+                return self.search_web(query, engine, max_results)
+            
+            def handle_connect():
+                result = self.connect_to_mcp()
+                # Update connection status
+                new_status = f"<div style='color: {'green' if self.connected else 'red'};'>{'🟢 Connected to search spirits' if self.connected else '🔴 Disconnected from search realm'}</div>"
+                return result, new_status
+            
+            def handle_status():
+                status = "🟢 Connected" if self.connected else "🔴 Disconnected"
+                return f"🔍 **Web Search Oracle Status**\n\nConnection: {status}\nLast update: {time.strftime('%H:%M')}\nAvailable engines: {', '.join(self.search_engines)}"
+            
+            def handle_list_tools():
+                if not self.connected:
+                    return "❌ Not connected to MCP server. Please connect first."
+                if not self.tools:
+                    return "❌ No tools available."
+                
+                tool_info = "🛠️ **Available Search Tools:**\n\n"
+                for tool in self.tools:
+                    tool_info += f"• **{tool.name}**: {tool.description}\n"
+                return tool_info
+            
+            # Wire up events
+            search_btn.click(
+                handle_search_request,
+                inputs=[query_input, engine_dropdown, max_results],
+                outputs=[search_output]
+            )
+            
+            query_input.submit(
+                handle_search_request,
+                inputs=[query_input, engine_dropdown, max_results], 
+                outputs=[search_output]
+            )
+            
+            connect_btn.click(
+                handle_connect,
+                outputs=[search_output, connection_status]
+            )
+            
+            status_btn.click(
+                handle_status,
+                outputs=[search_output]
+            )
+            
+            tools_btn.click(
+                handle_list_tools,
+                outputs=[search_output]
+            )
+        
+        return interface
+    
+    def connect_to_mcp(self) -> str:
+        """Connect to MCP web search server."""
+        current_time = time.time()
+        if current_time - self.last_connection_attempt < self.connection_cooldown:
+            return "⏳ Please wait before retrying connection..."
+        
+        self.last_connection_attempt = current_time
+        
+        try:
+            return self.loop.run_until_complete(self._connect())
+        except Exception as e:
+            self.connected = False
+            return f"❌ Connection failed: {str(e)}"
+    
+    async def _connect(self) -> str:
+        """Async connect to MCP server using SSE."""
+        try:
+            # Clean up previous connection
+            if self.exit_stack:
+                await self.exit_stack.aclose()
+
+            self.exit_stack = AsyncExitStack()
+
+            # Connect to SSE MCP server
+            sse_transport = await self.exit_stack.enter_async_context(
+                sse_client(self.server_url)
+            )
+            read_stream, write_callable = sse_transport
+
+            self.session = await self.exit_stack.enter_async_context(
+                ClientSession(read_stream, write_callable)
+            )
+            await self.session.initialize()
+
+            # Get available tools
+            response = await self.session.list_tools()
+            self.tools = response.tools
+
+            self.connected = True
+            tool_names = [tool.name for tool in self.tools]
+            return f"✅ Connected to web search MCP server!\nAvailable tools: {', '.join(tool_names)}"
+
+        except Exception as e:
+            self.connected = False
+            return f"❌ Connection failed: {str(e)}"
+    
+    def search_web(self, query: str, engine: str = "brave", max_results: int = 5) -> str:
+        """Search the web using actual MCP server"""
+        if not self.connected:
+            # Try to auto-connect
+            connect_result = self.connect_to_mcp()
+            if not self.connected:
+                return f"❌ Failed to connect to search server. {connect_result}"
+        
+        if not query.strip():
+            return "❌ Please enter a search query"
+        
+        try:
+            return self.loop.run_until_complete(self._search_web(query, engine, max_results))
+        except Exception as e:
+            return f"❌ Error performing search: {str(e)}"
+    
+    async def _search_web(self, query: str, engine: str, max_results: int) -> str:
+        """Async search web using MCP."""
+        try:
+            # Find the appropriate search tool based on engine
+            tool_name = f"web_search_mcp_{engine}_search"
+            search_tool = next((tool for tool in self.tools if tool.name == tool_name), None)
+            
+            if not search_tool:
+                # Try to find any search tool if specific engine not found
+                search_tool = next((tool for tool in self.tools if 'search' in tool.name.lower()), None)
+                if not search_tool:
+                    return "❌ No search tools found on server"
+
+            # Call the tool
+            params = {"query": query}
+            result = await self.session.call_tool(search_tool.name, params)
+
+            # Extract content properly
+            content_text = ""
+            if hasattr(result, 'content') and result.content:
+                if isinstance(result.content, list):
+                    for content_item in result.content:
+                        if hasattr(content_item, 'text'):
+                            content_text += content_item.text
+                        elif hasattr(content_item, 'content'):
+                            content_text += str(content_item.content)
+                        else:
+                            content_text += str(content_item)
+                elif hasattr(result.content, 'text'):
+                    content_text = result.content.text
+                else:
+                    content_text = str(result.content)
+
+            if not content_text:
+                return "❌ No content received from server"
+
+            try:
+                # Try to parse as JSON
+                parsed = json.loads(content_text)
+                if isinstance(parsed, dict):
+                    if 'error' in parsed:
+                        return f"❌ Error: {parsed['error']}"
+
+                    # Format search results nicely
+                    formatted = f"🔍 **Search Results for: '{query}'**\n"
+                    formatted += f"🌐 **Engine: {engine.title()}**\n\n"
+                    
+                    # Handle different response formats
+                    if 'data' in parsed and isinstance(parsed['data'], list):
+                        results = parsed['data'][:max_results]
+                        for i, result_item in enumerate(results, 1):
+                            if isinstance(result_item, list) and len(result_item) >= 3:
+                                title, link, body = result_item[0], result_item[1], result_item[2]
+                                formatted += f"**{i}. {title}**\n"
+                                formatted += f"🔗 {link}\n"
+                                formatted += f"📝 {body[:200]}{'...' if len(body) > 200 else ''}\n\n"
+                            elif isinstance(result_item, dict):
+                                title = result_item.get('title', 'No title')
+                                link = result_item.get('link', 'No link')
+                                body = result_item.get('body', 'No description')
+                                formatted += f"**{i}. {title}**\n"
+                                formatted += f"🔗 {link}\n"
+                                formatted += f"📝 {body[:200]}{'...' if len(body) > 200 else ''}\n\n"
+                    elif 'results' in parsed:
+                        results = parsed['results'][:max_results]
+                        for i, result_item in enumerate(results, 1):
+                            title = result_item.get('title', 'No title')
+                            link = result_item.get('url', result_item.get('link', 'No link'))
+                            body = result_item.get('snippet', result_item.get('body', 'No description'))
+                            formatted += f"**{i}. {title}**\n"
+                            formatted += f"🔗 {link}\n"
+                            formatted += f"📝 {body[:200]}{'...' if len(body) > 200 else ''}\n\n"
+                    else:
+                        # Fallback to raw JSON display
+                        formatted += f"✅ Raw search data:\n```json\n{json.dumps(parsed, indent=2)}\n```"
+                    
+                    formatted += f"\n⏰ Search completed: {time.strftime('%H:%M')}\n⚡ **Powered by MCP**"
+                    return formatted
+
+            except json.JSONDecodeError:
+                # If not JSON, try to format as text
+                formatted = f"🔍 **Search Results for: '{query}'**\n"
+                formatted += f"🌐 **Engine: {engine.title()}**\n\n"
+                formatted += f"✅ Search data:\n```\n{content_text[:1000]}{'...' if len(content_text) > 1000 else ''}\n```\n\n"
+                formatted += f"⏰ Search completed: {time.strftime('%H:%M')}\n⚡ **Powered by MCP**"
+                return formatted
+
+            return f"🔍 **Search Results for: '{query}'**\n\n✅ Raw result:\n{content_text[:1000]}{'...' if len(content_text) > 1000 else ''}\n\n⚡ **Powered by MCP**"
+
+        except Exception as e:
+            return f"❌ Failed to search: {str(e)}"
+    
+    def handle_command(self, player_id: str, command: str) -> str:
+        """Handle Web Search Oracle commands."""
+        parts = command.strip().split(' ', 2)
+        cmd = parts[0].lower()
+        
+        if cmd == "search" and len(parts) > 1:
+            query = parts[1]
+            engine = "brave"  # default
+            if len(parts) > 2 and parts[2] in self.search_engines:
+                engine = parts[2]
+            return self.search_web(query, engine)
+        elif cmd == "engines":
+            return f"🔍 **Available Search Engines:**\n\n{', '.join(self.search_engines)}\n\n**Usage:** `search <query> [engine]`"
+        elif cmd == "connect":
+            return self.connect_to_mcp()
+        elif cmd == "status":
+            status = "🟢 Connected" if self.connected else "🔴 Disconnected"
+            return f"🔍 **Web Search Oracle Status**\n\nConnection: {status}\nLast update: {time.strftime('%H:%M')}\nAvailable engines: {', '.join(self.search_engines)}"
+        elif cmd == "tools":
+            if not self.connected:
+                return "❌ Not connected to MCP server. Use `connect` first."
+            if not self.tools:
+                return "❌ No tools available."
+            
+            tool_info = "🛠️ **Available Search Tools:**\n\n"
+            for tool in self.tools:
+                tool_info += f"• **{tool.name}**: {tool.description}\n"
+            return tool_info
+        elif cmd == "help":
+            return """🔍 **Web Search Oracle Commands:**
+
+**search** `<query>` `[engine]` - Search the web (e.g., 'search AI news brave')
+**engines** - List available search engines
+**connect** - Connect to MCP search server
+**status** - Check connection status
+**tools** - List available MCP tools
+**help** - Show this help
+
+🌐 **Available Engines:**
+{engines}
+
+🔍 **Example Commands:**
+• search latest AI developments
+• search Python tutorials duckduckgo
+• search climate change news searxng
+
+⚡ **Powered by MCP (Model Context Protocol)**""".format(engines=', '.join(self.search_engines))
+        else:
+            return "❓ Invalid command. Try: `search <query>`, `engines`, `connect`, `status`, `tools`, or `help`"
+
+
+# Global Web Search Oracle service instance
+web_search_oracle_service = WebSearchOracleService()
+
+
+def auto_register(game_engine):
+    """Auto-register the Web Search Oracle addon with the game engine.
+    
+    This function makes the addon self-contained by handling its own registration.
+    """
+    try:
+        # Create the web search oracle NPC definition
+        web_search_oracle_npc = {
+            'id': 'web_search_oracle_auto',
+            'name': '🔍 Web Search Oracle (Auto)',
+            'x': 450, 'y': 150,
+            'char': '🔍',
+            'type': 'mcp',
+            'personality': 'web_search_oracle',
+            'description': 'Self-contained MCP-powered web search information service'
+        }
+        
+        # Register the NPC with the NPC service
+        npc_service = game_engine.get_npc_service()
+        npc_service.register_npc('web_search_oracle_auto', web_search_oracle_npc)
+        
+        # Register the addon for handling private message commands
+        world = game_engine.get_world()
+        if not hasattr(world, 'addon_npcs'):
+            world.addon_npcs = {}
+        world.addon_npcs['web_search_oracle_auto'] = web_search_oracle_service
+        
+        print("[WebSearchOracleAddon] Auto-registered successfully as self-contained addon")
+        return True
+        
+    except Exception as e:
+        print(f"[WebSearchOracleAddon] Error during auto-registration: {e}")
+        return False