fix (double wrapping of server results)

src/pipeline.py

@@ -95,6 +95,7 @@
 """
 Farmer.chat Pipeline - Main Orchestrator
 Coordinates Router → Executor → Compiler stages for alert generation
+FIXED: Proper async handling for FastAPI integration
 """
 
 from typing import Dict, Any, Optional
@@ -108,8 +109,8 @@ class FarmerChatPipeline:
     Main pipeline orchestrating the complete alert generation workflow.
     
     Architecture:
-    Stage 1: QueryRouter - Determines which MCP servers to query with priorities
-    Stage 2: MCPExecutor - Executes parallel calls to selected MCP servers
+    Stage 1: QueryRouter - Determines which MCP servers to query
+    Stage 2: MCPExecutor - Executes parallel calls to MCP servers  
     Stage 3: ResponseCompiler - Compiles results into actionable alert summary
     """
     
@@ -131,29 +132,15 @@ class FarmerChatPipeline:
         
         print(f"✓ Pipeline initialized for location: {location['latitude']:.4f}°N, {location['longitude']:.4f}°E")
     
-    def generate_alert(
+    async def generate_alert(
         self, 
         location: Optional[Dict[str, float]] = None,
         location_name: str = ""
     ) -> Dict[str, Any]:
         """
         Generate comprehensive alert summary for a location.
-        
-        Always queries ALL MCP servers. The compiler extracts only
-        the alerting/concerning information from the comprehensive data.
-        
-        Args:
-            location: Optional location override. Uses default if not provided.
-            location_name: Human-readable location name for context
-            
-        Returns:
-            Dict containing:
-                - alert_summary: Compiled alert text (only concerning info)
-                - location: Location coordinates used
-                - mcp_results: Raw results from each MCP server
+        ASYNC version for FastAPI endpoints.
         """
-        
-        # Use provided location or default
         query_location = location or self.location
         
         print(f"\n{'='*60}")
@@ -166,21 +153,21 @@ class FarmerChatPipeline:
         routing = self.router.route_alert_query(query_location)
         print(f"✓ Routing complete: All 5 servers will be queried")
         
-        # Stage 2: Execute - Query all MCP servers in parallel
+        # Stage 2: Execute - Query all MCP servers in parallel (ASYNC)
         print("\nStage 2: Executing parallel MCP server calls...")
-        mcp_results = self.executor.execute_parallel(routing, query_location)
+        mcp_results = await self.executor.execute_parallel_async(routing, query_location)
         
         success_count = sum(1 for r in mcp_results.values() if r.get("status") == "success")
         print(f"✓ Execution complete: {success_count}/{len(mcp_results)} servers responded successfully")
         
         # Stage 3: Compile - Extract ONLY alerting information
-        print("\nStage 3: Compiling alert summary (extracting concerning info only)...")
+        print("\nStage 3: Compiling alert summary...")
         alert_summary = self.compiler.compile_alert_summary(
             mcp_results, 
             query_location,
             location_name
         )
-        print("✓ Alert summary generated (focusing on alerts/concerns only)")
+        print("✓ Alert summary generated")
         
         print(f"\n{'='*60}\n")
         
@@ -191,29 +178,15 @@ class FarmerChatPipeline:
             "mcp_results": mcp_results
         }
     
-    def process_query(
+    async def process_query(
         self, 
         query: str, 
         location: Optional[Dict[str, float]] = None
     ) -> Dict[str, Any]:
         """
         Process a specific farmer query through the pipeline.
-        
-        For now, also queries all servers to ensure comprehensive data.
-        The compiler extracts query-relevant information.
-        
-        Args:
-            query: Farmer's question or request
-            location: Optional location override
-            
-        Returns:
-            Dict containing:
-                - response: Compiled response text
-                - location: Location coordinates used
-                - mcp_results: Raw results from MCP servers
+        ASYNC version for FastAPI endpoints.
         """
-        
-        # Use provided location or default
         query_location = location or self.location
         
         print(f"\n{'='*60}")
@@ -221,14 +194,14 @@ class FarmerChatPipeline:
         print(f"Location: {query_location['latitude']:.4f}°N, {query_location['longitude']:.4f}°E")
         print(f"{'='*60}\n")
         
-        # Stage 1: Route - For now, query all servers
+        # Stage 1: Route
         print("Stage 1: Routing to all MCP servers...")
         routing = self.router.route_query(query, query_location)
-        print(f"✓ Routing complete: All servers will be queried")
+        print(f"✓ Routing complete")
         
-        # Stage 2: Execute MCP calls
+        # Stage 2: Execute MCP calls (ASYNC)
         print("\nStage 2: Executing MCP server calls...")
-        mcp_results = self.executor.execute_parallel(routing, query_location)
+        mcp_results = await self.executor.execute_parallel_async(routing, query_location)
         
         success_count = sum(1 for r in mcp_results.values() if r.get("status") == "success")
         print(f"✓ Execution complete: {success_count}/{len(mcp_results)} servers responded")
@@ -247,31 +220,19 @@ class FarmerChatPipeline:
         }
     
     def update_location(self, location: Dict[str, float]):
-        """
-        Update default location for pipeline.
-        
-        Args:
-            location: New default location dict with 'latitude' and 'longitude' keys
-        """
+        """Update default location for pipeline."""
         self.location = location
         print(f"✓ Default location updated to: {location['latitude']:.4f}°N, {location['longitude']:.4f}°E")
     
     def get_server_status(self) -> Dict[str, str]:
-        """
-        Get status of all MCP servers.
-        
-        Returns:
-            Dict mapping server names to "available" or "unavailable"
-        """
+        """Get status of all MCP servers."""
         status = {}
         for server_name, server in self.servers.items():
             try:
-                # Try to check if server is responsive
                 if hasattr(server, 'health_check'):
                     status[server_name] = "available" if server.health_check() else "unavailable"
                 else:
-                    status[server_name] = "available"  # Assume available if no health check
+                    status[server_name] = "available"
             except:
                 status[server_name] = "unavailable"
-        
         return status