fix (asyncio inside FastAPI bug)

src/executor.py

@@ -268,15 +268,15 @@
 #         return results
         
 #         return results
-
 """
 MCP Executor - Stage 2
 Executes parallel calls to MCP servers based on routing decisions
-FIXED: Simpler async handling to prevent deadlocks
+FIXED: 
+1. Proper async handling for FastAPI (no asyncio.run inside existing loop)
+2. Fixed double-wrapping of server results
 """
 
 from typing import Dict, Any
-from concurrent.futures import ThreadPoolExecutor, as_completed
 import asyncio
 import inspect
 
@@ -284,8 +284,7 @@ import inspect
 class MCPExecutor:
     """
     Executes MCP server calls based on routing decisions.
-    Integrates with existing server implementations in src/servers/
-    Handles both sync and async server methods safely.
+    Properly handles async servers within FastAPI's event loop.
     """
     
     def __init__(self, servers: Dict[str, Any]):
@@ -294,110 +293,38 @@ class MCPExecutor:
         
         Args:
             servers: Dict mapping server names to initialized server objects
-                    e.g., {"weather": WeatherServer(), "soil": SoilPropertiesServer(), ...}
         """
         self.servers = servers
     
-    def execute_parallel(self, routing: Dict[str, bool], location: Dict[str, float]) -> Dict[str, Any]:
+    async def execute_parallel_async(self, routing: Dict[str, bool], location: Dict[str, float]) -> Dict[str, Any]:
         """
-        Execute MCP server calls in parallel based on routing.
+        Execute MCP server calls in parallel (async version for FastAPI).
         
         Args:
-            routing: Simple dict with server names as keys and True/False as values
+            routing: Dict with server names as keys and True/False as values
             location: Dict with 'latitude' and 'longitude' keys
             
         Returns:
-            Dict mapping server names to their results with metadata
+            Dict mapping server names to their results
         """
         results = {}
-        
-        # For async servers, we need to run them differently
-        # Separate sync and async servers
-        sync_tasks = []
-        async_tasks = []
+        tasks = []
+        server_names = []
         
         for server_name, should_query in routing.items():
             if should_query and server_name in self.servers:
                 server = self.servers[server_name]
-                task = {
-                    "server_name": server_name,
-                    "server": server,
-                    "location": location
-                }
-                
-                # Check if server method is async
-                if hasattr(server, 'get_data'):
-                    method = getattr(server, 'get_data')
-                    if inspect.iscoroutinefunction(method):
-                        async_tasks.append(task)
-                    else:
-                        sync_tasks.append(task)
-                else:
-                    sync_tasks.append(task)
-        
-        # Execute sync servers in parallel with ThreadPoolExecutor
-        if sync_tasks:
-            with ThreadPoolExecutor(max_workers=5) as executor:
-                futures = {
-                    executor.submit(self._call_sync_server, task): task 
-                    for task in sync_tasks
-                }
-                
-                for future in as_completed(futures):
-                    task = futures[future]
-                    server_name = task["server_name"]
-                    
-                    try:
-                        result = future.result(timeout=30)
-                        results[server_name] = {
-                            "data": result,
-                            "status": "success"
-                        }
-                        print(f"✓ {server_name.upper()}: Retrieved successfully")
-                    except Exception as e:
-                        results[server_name] = {
-                            "data": None,
-                            "status": "error",
-                            "error": str(e)
-                        }
-                        print(f"✗ {server_name.upper()}: Error - {str(e)}")
+                tasks.append(self._call_server(server, server_name, location))
+                server_names.append(server_name)
         
-        # Execute async servers together in single event loop
-        if async_tasks:
-            try:
-                async_results = asyncio.run(self._execute_async_batch(async_tasks))
-                results.update(async_results)
-            except Exception as e:
-                # If batch fails, mark all as failed
-                for task in async_tasks:
-                    results[task["server_name"]] = {
-                        "data": None,
-                        "status": "error",
-                        "error": f"Async batch execution failed: {str(e)}"
-                    }
-                    print(f"✗ {task['server_name'].upper()}: Async batch error")
+        if not tasks:
+            return results
         
-        return results
-    
-    async def _execute_async_batch(self, tasks: list) -> Dict[str, Any]:
-        """
-        Execute multiple async server calls concurrently in a single event loop.
-        This is safer than creating multiple event loops.
-        """
-        results = {}
-        
-        # Create async tasks for all servers
-        async_calls = []
-        for task in tasks:
-            async_calls.append(self._call_async_server(task))
-        
-        # Execute all async calls concurrently
-        task_results = await asyncio.gather(*async_calls, return_exceptions=True)
+        # Execute all tasks concurrently
+        task_results = await asyncio.gather(*tasks, return_exceptions=True)
         
         # Process results
-        for task, result in zip(tasks, task_results):
-            server_name = task["server_name"]
-            
+        for server_name, result in zip(server_names, task_results):
             if isinstance(result, Exception):
                 results[server_name] = {
                     "data": None,
@@ -406,76 +333,66 @@ class MCPExecutor:
                 }
                 print(f"✗ {server_name.upper()}: Error - {str(result)}")
             else:
-                results[server_name] = {
-                    "data": result,
-                    "status": "success"
-                }
-                print(f"✓ {server_name.upper()}: Retrieved successfully")
-        
-        return results
-    
-    async def _call_async_server(self, task: Dict[str, Any]) -> Any:
-        """Call individual async MCP server"""
-        server = task["server"]
-        location = task["location"]
-        
-        if hasattr(server, 'get_data'):
-            return await server.get_data(location['latitude'], location['longitude'])
-        else:
-            raise AttributeError(f"Server {task['server_name']} has no get_data method")
-    
-    def _call_sync_server(self, task: Dict[str, Any]) -> Any:
-        """Call individual sync MCP server"""
-        server = task["server"]
-        location = task["location"]
-        
-        if hasattr(server, 'get_data'):
-            return server.get_data(location['latitude'], location['longitude'])
-        elif hasattr(server, 'query'):
-            return server.query(location)
-        elif hasattr(server, 'fetch_data'):
-            return server.fetch_data(location['latitude'], location['longitude'])
-        else:
-            raise AttributeError(f"Server {task['server_name']} has no compatible query method")
-    
-    def execute_sequential(self, routing: Dict[str, bool], location: Dict[str, float]) -> Dict[str, Any]:
-        """
-        Execute MCP server calls sequentially (fallback if parallel fails).
-        """
-        results = {}
-        
-        for server_name, should_query in routing.items():
-            if should_query and server_name in self.servers:
-                try:
-                    server = self.servers[server_name]
-                    
-                    # Check if async
-                    if hasattr(server, 'get_data') and inspect.iscoroutinefunction(server.get_data):
-                        # Run async method
-                        result = asyncio.run(server.get_data(location['latitude'], location['longitude']))
+                # FIX: Handle servers that return {"status": ..., "data": ...}
+                # Don't double-wrap!
+                if isinstance(result, dict) and "status" in result:
+                    # Server already returned proper format
+                    if result.get("status") == "success":
+                        results[server_name] = {
+                            "data": result.get("data"),  # Extract actual data
+                            "status": "success"
+                        }
                     else:
-                        # Run sync method
-                        task = {
-                            "server_name": server_name,
-                            "server": server,
-                            "location": location
+                        results[server_name] = {
+                            "data": None,
+                            "status": "error",
+                            "error": result.get("error", "Unknown error")
                         }
-                        result = self._call_sync_server(task)
-                    
+                else:
+                    # Server returned raw data
                     results[server_name] = {
                         "data": result,
                         "status": "success"
                     }
-                    print(f"✓ {server_name.upper()}: Retrieved successfully")
-                    
-                except Exception as e:
-                    results[server_name] = {
-                        "data": None,
-                        "status": "error",
-                        "error": str(e)
-                    }
-                    print(f"✗ {server_name.upper()}: Error - {str(e)}")
+                print(f"✓ {server_name.upper()}: Retrieved successfully")
         
         return results
+    
+    def execute_parallel(self, routing: Dict[str, bool], location: Dict[str, float]) -> Dict[str, Any]:
+        """
+        Execute MCP server calls in parallel (sync wrapper).
         
-        return results
+        Detects if we're already in an async context and handles appropriately.
+        """
+        try:
+            # Check if there's already a running event loop
+            loop = asyncio.get_running_loop()
+            # We're in an async context - need to use nest_asyncio or return a coroutine
+            # For FastAPI, the endpoint should be async and call execute_parallel_async directly
+            raise RuntimeError(
+                "execute_parallel called from async context. "
+                "Use 'await executor.execute_parallel_async()' instead."
+            )
+        except RuntimeError:
+            # No running loop - safe to use asyncio.run
+            return asyncio.run(self.execute_parallel_async(routing, location))
+    
+    async def _call_server(self, server: Any, server_name: str, location: Dict[str, float]) -> Any:
+        """
+        Call individual MCP server, handling both sync and async methods.
+        """
+        lat = location['latitude']
+        lon = location['longitude']
+        
+        if hasattr(server, 'get_data'):
+            method = getattr(server, 'get_data')
+            
+            if inspect.iscoroutinefunction(method):
+                # Async method - await it
+                return await method(lat, lon)
+            else:
+                # Sync method - run in executor to not block
+                loop = asyncio.get_event_loop()
+                return await loop.run_in_executor(None, method, lat, lon)
+        else:
+            raise AttributeError(f"Server {server_name} has no get_data method")