using async processing

requirements.txt

@@ -1,3 +1,4 @@
+aiohttp
 fastapi
 uvicorn[standard]
 httpx

routers/slack.py

@@ -1,6 +1,8 @@
-from fastapi import APIRouter, Request, HTTPException
+from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
 import json
 import logging
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
 from crew.crew import SlackCrew
 from slack.utils import verify_slack_signature
 from slack.workflows import send_to_workflow
@@ -13,10 +15,46 @@ router = APIRouter(
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+# Thread pool for CPU-bound tasks
+executor = ThreadPoolExecutor(max_workers=4)
+
+async def process_with_crew_async(text: str, user: str, channel: str):
+    """Process message with CrewAI in background"""
+    try:
+        crew = SlackCrew().crew()
+        
+        # Run the crew kickoff in a thread pool since it's likely CPU-bound
+        loop = asyncio.get_event_loop()
+        crew_response = await loop.run_in_executor(
+            executor, 
+            lambda: crew.kickoff(inputs={"question": text})
+        )
+        
+        logger.info(f"AI processed: {text[:50]}... -> Response length: {len(str(crew_response))}")
+        
+        # Send the AI response to workflow
+        workflow_data = {
+            "user": user,
+            "query": text,
+            "message": str(crew_response),
+        }
+        
+        # Send to workflow asynchronously
+        await send_to_workflow(workflow_data)
+        
+    except Exception as e:
+        logger.error(f"Background crew processing error: {e}")
+        # Optionally send error to workflow or handle gracefully
+        error_data = {
+            "user": user,
+            "query": text,
+            "message": f"Sorry, I encountered an error processing your request: {str(e)}",
+        }
+        await send_to_workflow(error_data)
+
 @router.post("/slack")
-async def handle_slack(request: Request):
-    """Single endpoint for all Slack requests"""
-    crew = SlackCrew().crew()
+async def handle_slack(request: Request, background_tasks: BackgroundTasks):
+    """Single endpoint for all Slack requests - now with async processing"""
     try:
         # Get request data
         timestamp = request.headers.get("X-Slack-Request-Timestamp", "")
@@ -64,29 +102,21 @@ async def handle_slack(request: Request):
         if payload.get("event", {}).get("subtype") == "bot_message":
             return {"status": "ignored"}
         
-        # Process with AI if needed
-        crew_response = ""
+        # Process with AI in background if needed
         if text:
-            crew_response = crew.kickoff(inputs = {
-                "question": text,
-            })
-            logger.info(f"AI processed: {text[:50]}...")
-        
-        # Prepare workflow data
-        workflow_data = {
-            "user": user,
-            "query": text,
-            "message": text
-        }
-        
-        # Add crew response to message if available
-        if crew_response:
-            workflow_data["message"] = f"{crew_response}"
-        
-        # Send to workflow
-        send_to_workflow(workflow_data)
+            # Add the crew processing to background tasks
+            background_tasks.add_task(process_with_crew_async, text, user, channel)
+            
+            # Optionally send immediate acknowledgment
+            ack_data = {
+                "user": user,
+                "query": text,
+                "message": "Processing your request... I'll get back to you shortly!",
+            }
+            await send_to_workflow(ack_data)
         
-        return {"status": "ok"}
+        # Return immediately
+        return {"status": "ok", "message": "Processing in background"}
         
     except HTTPException:
         raise

slack/workflows.py

@@ -2,14 +2,14 @@ import os
 import logging
 import requests
 import json
+import aiohttp
 
 SLACK_WORKFLOW_URL = os.getenv("SLACK_WORKFLOW_URL", "")
 
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
 
-def send_to_workflow(data: dict) -> bool:
-    """Send data to Slack workflow"""
+async def send_to_workflow(data: dict) -> bool:
     if not SLACK_WORKFLOW_URL:
         logger.info("No workflow URL configured")
         return False
@@ -21,14 +21,23 @@ def send_to_workflow(data: dict) -> bool:
         }
         
         logger.info(f"Sending to workflow: {payload}")
-        response = requests.post(SLACK_WORKFLOW_URL, headers=headers, data=payload, timeout=10)
-        logger.info(f"Workflow response status: {response.status_code}")
-        logger.info(f"Workflow response body: {response.text}")
-        response.raise_for_status()
-        logger.info("Sent to workflow successfully")
-        return True
-    except requests.exceptions.HTTPError as e:
-        logger.error(f"Workflow HTTP error {response.status_code}: {response.text}")
+        
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                SLACK_WORKFLOW_URL, 
+                headers=headers, 
+                data=payload,
+                timeout=aiohttp.ClientTimeout(total=10)
+            ) as response:
+                response_text = await response.text()
+                logger.info(f"Workflow response status: {response.status}")
+                logger.info(f"Workflow response body: {response_text}")
+                response.raise_for_status()
+                logger.info("Sent to workflow successfully")
+                return True
+                
+    except aiohttp.ClientError as e:
+        logger.error(f"Workflow HTTP error: {e}")
         return False
     except Exception as e:
         logger.error(f"Workflow error: {e}")