integrate RabbitMQ with CloudAMQP

app/auth/routes.py

@@ -1,5 +1,4 @@
 from datetime import datetime, timezone
-from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
 from motor.motor_asyncio import AsyncIOMotorDatabase
@@ -17,11 +16,18 @@ from app.auth.jwt_handler import (
     verify_refresh_token,
 )
 from app.config import settings
+from app.rabbitmq.publishers import (
+    publish_user_registered,
+    publish_user_logged_in,
+    publish_user_logged_out,
+    publish_token_refreshed,
+)
 
 router = APIRouter(prefix="/auth", tags=["Authentication"])
 pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
 
+
 async def get_current_user(
     token: str = Depends(oauth2_scheme),
     db: AsyncIOMotorDatabase = Depends(get_db),
@@ -37,6 +43,7 @@ async def get_current_user(
         raise HTTPException(status_code=401, detail="User not found")
     return UserPublic(**user)
 
+
 @router.post("/register", response_model=UserPublic, status_code=status.HTTP_201_CREATED)
 async def register(user: UserCreate, db: AsyncIOMotorDatabase = Depends(get_db)):
     username = user.username.strip().lower()
@@ -52,8 +59,13 @@ async def register(user: UserCreate, db: AsyncIOMotorDatabase = Depends(get_db))
         "created_at": datetime.utcnow(),
     }
     await db.users.insert_one(doc)
+
+    # Publish registration event
+    await publish_user_registered(username, email, user.company)
+
     return UserPublic(**doc)
 
+
 @router.post("/login", response_model=Token)
 async def login(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
     username = form_data.username.strip().lower()
@@ -63,24 +75,28 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get
     access_token = create_access_token(username)
     refresh_token = create_refresh_token(username)
     payload = decode_token(refresh_token)
-    await db.sessions.insert_one(
-        {
-            "user_id": username,
-            "refresh_token_hash": hash_refresh_token(refresh_token),
-            "created_at": datetime.now(timezone.utc),
-            "expires_at": datetime.fromtimestamp(payload["exp"], tz=timezone.utc),
-            "revoked_at": None,
-        }
-    )
+    await db.sessions.insert_one({
+        "user_id": username,
+        "refresh_token_hash": hash_refresh_token(refresh_token),
+        "created_at": datetime.now(timezone.utc),
+        "expires_at": datetime.fromtimestamp(payload["exp"], tz=timezone.utc),
+        "revoked_at": None,
+    })
+
+    # Publish login event
+    await publish_user_logged_in(username)
+
     return Token(
         access_token=access_token,
         refresh_token=refresh_token,
         expires_in=settings.access_token_expire_minutes * 60,
     )
 
+
 class RefreshIn(BaseModel):
     refresh_token: str
 
+
 @router.post("/refresh", response_model=Token)
 async def refresh_token(payload: RefreshIn, db=Depends(get_db)):
     try:
@@ -91,11 +107,7 @@ async def refresh_token(payload: RefreshIn, db=Depends(get_db)):
         raise HTTPException(status_code=401, detail="Invalid token type")
     username = decoded.get("sub")
     session_doc = await db.sessions.find_one(
-        {
-            "user_id": username,
-            "revoked_at": None,
-            "expires_at": {"$gt": datetime.now(timezone.utc)},
-        },
+        {"user_id": username, "revoked_at": None, "expires_at": {"$gt": datetime.now(timezone.utc)}},
         sort=[("created_at", -1)],
     )
     if not session_doc or not verify_refresh_token(payload.refresh_token, session_doc["refresh_token_hash"]):
@@ -103,19 +115,27 @@ async def refresh_token(payload: RefreshIn, db=Depends(get_db)):
     new_access = create_access_token(username)
     new_refresh = create_refresh_token(username)
     await db.sessions.update_one(
-        {"_id": session_doc["_id"]}, {"$set": {"revoked_at": datetime.now(timezone.utc)}}
+        {"_id": session_doc["_id"]},
+        {"$set": {"revoked_at": datetime.now(timezone.utc)}}
     )
     payload_new = decode_token(new_refresh)
-    await db.sessions.insert_one(
-        {
-            "user_id": username,
-            "refresh_token_hash": hash_refresh_token(new_refresh),
-            "created_at": datetime.now(timezone.utc),
-            "expires_at": datetime.fromtimestamp(payload_new["exp"], tz=timezone.utc),
-            "revoked_at": None,
-        }
+    await db.sessions.insert_one({
+        "user_id": username,
+        "refresh_token_hash": hash_refresh_token(new_refresh),
+        "created_at": datetime.now(timezone.utc),
+        "expires_at": datetime.fromtimestamp(payload_new["exp"], tz=timezone.utc),
+        "revoked_at": None,
+    })
+
+    # Publish token refresh event
+    await publish_token_refreshed(username)
+
+    return Token(
+        access_token=new_access,
+        refresh_token=new_refresh,
+        expires_in=settings.access_token_expire_minutes * 60,
     )
-    return Token(access_token=new_access, refresh_token=new_refresh, expires_in=settings.access_token_expire_minutes * 60)
+
 
 @router.post("/logout")
 async def logout(payload: RefreshIn, db=Depends(get_db)):
@@ -125,21 +145,21 @@ async def logout(payload: RefreshIn, db=Depends(get_db)):
         return {"ok": True}
     username = decoded.get("sub")
     session_doc = await db.sessions.find_one(
-        {
-            "user_id": username,
-            "revoked_at": None,
-            "expires_at": {"$gt": datetime.now(timezone.utc)},
-        },
+        {"user_id": username, "revoked_at": None, "expires_at": {"$gt": datetime.now(timezone.utc)}},
         sort=[("created_at", -1)],
     )
     if not session_doc:
         return {"ok": True}
     if verify_refresh_token(payload.refresh_token, session_doc["refresh_token_hash"]):
         await db.sessions.update_one(
-            {"_id": session_doc["_id"]}, {"$set": {"revoked_at": datetime.now(timezone.utc)}}
+            {"_id": session_doc["_id"]},
+            {"$set": {"revoked_at": datetime.now(timezone.utc)}}
         )
+        await publish_user_logged_out(username)
+
     return {"ok": True}
 
+
 @router.get("/profile", response_model=UserPublic)
 async def read_users_me(current_user: UserPublic = Depends(get_current_user)):
     return current_user

app/config.py

@@ -2,14 +2,17 @@ import os
 from pydantic_settings import BaseSettings
 
 class Settings(BaseSettings):
-    mongo_uri: str = os.getenv("MONGO_URI")
-    database_name: str = os.getenv("DATABASE_NAME")
-    groq_api_key: str = os.getenv("GROQ_API_KEY")
-    secret_key: str = os.getenv("SECRET_KEY")
-    algorithm: str = os.getenv("ALGORITHM")
+    mongo_uri: str = os.getenv("MONGO_URI", "")
+    database_name: str = os.getenv("DATABASE_NAME", "echoloft")
+    groq_api_key: str = os.getenv("GROQ_API_KEY", "")
+    secret_key: str = os.getenv("SECRET_KEY", "changeme")
+    algorithm: str = os.getenv("ALGORITHM", "HS256")
     access_token_expire_minutes: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 30))
     refresh_token_expire_days: int = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", 7))
 
+    # RabbitMQ — defaults to CloudAMQP free tier URL or local
+    rabbitmq_url: str = os.getenv("RABBITMQ_URL", "amqp://guest:guest@localhost/")
+
     class Config:
         env_file = ".env"
         env_file_encoding = "utf-8"

app/main.py

@@ -1,16 +1,42 @@
+import logging
+from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from app.auth.routes import router as auth_router
 from app.rag.routes import router as rag_router
 from app.logging_config import setup_logging
+from app.rabbitmq.connection import rabbitmq
+from app.rabbitmq.consumers import start_consumers
 
 setup_logging()
+logger = logging.getLogger(__name__)
 
-app = FastAPI(title="GrokRAG API", description="SaaS RAG Chat with Groq", version="1.1.0")
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # ── Startup ──
+    logger.info("Connecting to RabbitMQ...")
+    await rabbitmq.connect()
+    await start_consumers()
+    yield
+    # ── Shutdown ──
+    logger.info("Disconnecting from RabbitMQ...")
+    await rabbitmq.disconnect()
+
+
+app = FastAPI(
+    title="EchoLoft AI API",
+    description="SaaS RAG Chat with Groq + RabbitMQ event streaming",
+    version="1.2.0",
+    lifespan=lifespan,
+)
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["https://echoloftai.vercel.app/"],
+    allow_origins=[
+        "http://localhost:3000",
+        "https://echoloftai.vercel.app",
+    ],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
@@ -19,6 +45,19 @@ app.add_middleware(
 app.include_router(auth_router)
 app.include_router(rag_router, prefix="/rag")
 
+
 @app.get("/")
 async def root():
-    return {"message": "Welcome to GrokRAG API"}
+    return {
+        "message": "Welcome to EchoLoft AI API",
+        "version": "1.2.0",
+        "rabbitmq": "connected" if rabbitmq.is_connected else "disconnected",
+    }
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "rabbitmq": "connected" if rabbitmq.is_connected else "disconnected",
+    }

app/rabbitmq/__init__.py

@@ -0,0 +1,19 @@
+from .connection import rabbitmq
+from .publishers import (
+    publish_user_registered,
+    publish_user_logged_in,
+    publish_user_logged_out,
+    publish_token_refreshed,
+    publish_conversation_created,
+    publish_message_sent,
+)
+
+__all__ = [
+    "rabbitmq",
+    "publish_user_registered",
+    "publish_user_logged_in",
+    "publish_user_logged_out",
+    "publish_token_refreshed",
+    "publish_conversation_created",
+    "publish_message_sent",
+]

app/rabbitmq/connection.py

@@ -0,0 +1,93 @@
+import asyncio
+import logging
+import json
+from typing import Optional, Callable
+import aio_pika
+from aio_pika import Message, DeliveryMode
+from aio_pika.abc import AbstractRobustConnection, AbstractChannel, AbstractQueue
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+class RabbitMQManager:
+    def __init__(self):
+        self.connection: Optional[AbstractRobustConnection] = None
+        self.channel: Optional[AbstractChannel] = None
+
+    async def connect(self):
+        try:
+            self.connection = await aio_pika.connect_robust(
+                settings.rabbitmq_url,
+                reconnect_interval=5,
+            )
+            self.channel = await self.connection.channel()
+            await self.channel.set_qos(prefetch_count=10)
+            logger.info("RabbitMQ connected successfully")
+        except Exception as e:
+            logger.error(f"RabbitMQ connection failed: {e}")
+            self.connection = None
+            self.channel = None
+
+    async def disconnect(self):
+        try:
+            if self.connection and not self.connection.is_closed:
+                await self.connection.close()
+                logger.info("RabbitMQ disconnected")
+        except Exception as e:
+            logger.error(f"RabbitMQ disconnect error: {e}")
+
+    async def declare_queue(self, queue_name: str, durable: bool = True) -> Optional[AbstractQueue]:
+        if not self.channel:
+            logger.warning("RabbitMQ channel not available")
+            return None
+        try:
+            queue = await self.channel.declare_queue(
+                queue_name,
+                durable=durable,
+                arguments={"x-message-ttl": 86400000}  # 24hr TTL
+            )
+            return queue
+        except Exception as e:
+            logger.error(f"Queue declare error: {e}")
+            return None
+
+    async def publish(self, queue_name: str, payload: dict) -> bool:
+        if not self.channel:
+            logger.warning("RabbitMQ not connected — skipping publish")
+            return False
+        try:
+            await self.declare_queue(queue_name)
+            message = Message(
+                body=json.dumps(payload).encode(),
+                delivery_mode=DeliveryMode.PERSISTENT,
+                content_type="application/json",
+            )
+            await self.channel.default_exchange.publish(
+                message,
+                routing_key=queue_name,
+            )
+            logger.info(f"Published to {queue_name}: {payload.get('event', 'unknown')}")
+            return True
+        except Exception as e:
+            logger.error(f"Publish error: {e}")
+            return False
+
+    async def consume(self, queue_name: str, callback: Callable):
+        if not self.channel:
+            logger.warning("RabbitMQ not connected — skipping consume")
+            return
+        try:
+            queue = await self.declare_queue(queue_name)
+            if queue:
+                await queue.consume(callback)
+                logger.info(f"Consuming from queue: {queue_name}")
+        except Exception as e:
+            logger.error(f"Consume error: {e}")
+
+    @property
+    def is_connected(self) -> bool:
+        return self.connection is not None and not self.connection.is_closed
+
+
+# Singleton instance
+rabbitmq = RabbitMQManager()

app/rabbitmq/consumers.py

@@ -0,0 +1,59 @@
+import json
+import logging
+from aio_pika.abc import AbstractIncomingMessage
+from app.rabbitmq.connection import rabbitmq
+from app.rabbitmq.queues import (
+    QUEUE_AUTH_EVENTS, QUEUE_CHAT_EVENTS,
+    QUEUE_AUDIT_LOG, QUEUE_NOTIFICATIONS
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def handle_auth_event(message: AbstractIncomingMessage):
+    async with message.process():
+        try:
+            data = json.loads(message.body.decode())
+            event = data.get("event")
+            logger.info(f"[AUTH EVENT] {event} — user: {data.get('username')}")
+            # Add your business logic here:
+            # e.g. send welcome email on user.registered
+            # e.g. alert on suspicious login patterns
+        except Exception as e:
+            logger.error(f"Auth event handler error: {e}")
+
+
+async def handle_chat_event(message: AbstractIncomingMessage):
+    async with message.process():
+        try:
+            data = json.loads(message.body.decode())
+            event = data.get("event")
+            logger.info(f"[CHAT EVENT] {event} — conv: {data.get('conversation_id')}")
+            # Add your business logic here:
+            # e.g. track usage metrics
+            # e.g. send notifications
+        except Exception as e:
+            logger.error(f"Chat event handler error: {e}")
+
+
+async def handle_audit_log(message: AbstractIncomingMessage):
+    async with message.process():
+        try:
+            data = json.loads(message.body.decode())
+            logger.info(f"[AUDIT] {data.get('event')} — user: {data.get('user')} at {data.get('timestamp')}")
+            # Add your business logic here:
+            # e.g. write to audit collection in MongoDB
+            # e.g. forward to external SIEM
+        except Exception as e:
+            logger.error(f"Audit log handler error: {e}")
+
+
+async def start_consumers():
+    """Register all consumers. Called on app startup."""
+    if not rabbitmq.is_connected:
+        logger.warning("RabbitMQ not connected — consumers not started")
+        return
+    await rabbitmq.consume(QUEUE_AUTH_EVENTS, handle_auth_event)
+    await rabbitmq.consume(QUEUE_CHAT_EVENTS, handle_chat_event)
+    await rabbitmq.consume(QUEUE_AUDIT_LOG, handle_audit_log)
+    logger.info("All RabbitMQ consumers started")

app/rabbitmq/publishers.py

@@ -0,0 +1,70 @@
+from datetime import datetime, timezone
+from app.rabbitmq.connection import rabbitmq
+from app.rabbitmq.queues import (
+    QUEUE_AUTH_EVENTS, QUEUE_CHAT_EVENTS,
+    QUEUE_AUDIT_LOG, AuthEvent, ChatEvent
+)
+
+
+async def publish_user_registered(username: str, email: str, company: str):
+    await rabbitmq.publish(QUEUE_AUTH_EVENTS, {
+        "event": AuthEvent.USER_REGISTERED,
+        "username": username,
+        "email": email,
+        "company": company,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+    await rabbitmq.publish(QUEUE_AUDIT_LOG, {
+        "event": AuthEvent.USER_REGISTERED,
+        "user": username,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def publish_user_logged_in(username: str):
+    await rabbitmq.publish(QUEUE_AUTH_EVENTS, {
+        "event": AuthEvent.USER_LOGGED_IN,
+        "username": username,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+    await rabbitmq.publish(QUEUE_AUDIT_LOG, {
+        "event": AuthEvent.USER_LOGGED_IN,
+        "user": username,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def publish_user_logged_out(username: str):
+    await rabbitmq.publish(QUEUE_AUTH_EVENTS, {
+        "event": AuthEvent.USER_LOGGED_OUT,
+        "username": username,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def publish_token_refreshed(username: str):
+    await rabbitmq.publish(QUEUE_AUTH_EVENTS, {
+        "event": AuthEvent.TOKEN_REFRESHED,
+        "username": username,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def publish_conversation_created(user_id: str, conversation_id: str):
+    await rabbitmq.publish(QUEUE_CHAT_EVENTS, {
+        "event": ChatEvent.CONVERSATION_CREATED,
+        "user_id": user_id,
+        "conversation_id": conversation_id,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def publish_message_sent(user_id: str, conversation_id: str, model: str, message_preview: str):
+    await rabbitmq.publish(QUEUE_CHAT_EVENTS, {
+        "event": ChatEvent.MESSAGE_SENT,
+        "user_id": user_id,
+        "conversation_id": conversation_id,
+        "model": model,
+        "message_preview": message_preview[:100],
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })

app/rabbitmq/queues.py

@@ -0,0 +1,24 @@
+# Queue Names
+QUEUE_CHAT_EVENTS     = "chat.events"
+QUEUE_AUTH_EVENTS     = "auth.events"
+QUEUE_RAG_JOBS        = "rag.jobs"
+QUEUE_NOTIFICATIONS   = "notifications"
+QUEUE_AUDIT_LOG       = "audit.log"
+
+# Event Types
+class AuthEvent:
+    USER_REGISTERED  = "user.registered"
+    USER_LOGGED_IN   = "user.logged_in"
+    USER_LOGGED_OUT  = "user.logged_out"
+    TOKEN_REFRESHED  = "token.refreshed"
+
+class ChatEvent:
+    MESSAGE_SENT      = "message.sent"
+    MESSAGE_RECEIVED  = "message.received"
+    CONVERSATION_CREATED = "conversation.created"
+
+class RagEvent:
+    JOB_QUEUED    = "rag.job.queued"
+    JOB_STARTED   = "rag.job.started"
+    JOB_COMPLETED = "rag.job.completed"
+    JOB_FAILED    = "rag.job.failed"

app/rag/routes.py

@@ -15,6 +15,10 @@ from app.auth.models import UserPublic
 from app.rag.models import ALLOWED_MODELS, Message
 from app.rag.rag_processor import build_context_from_files, web_search
 from app.config import settings
+from app.rabbitmq.publishers import (
+    publish_conversation_created,
+    publish_message_sent,
+)
 
 router = APIRouter(tags=["RAG Chat"])
 logger = logging.getLogger(__name__)
@@ -34,6 +38,7 @@ WEB_SEARCH_TOOL = {
     },
 }
 
+
 @router.post("/conversations", status_code=status.HTTP_201_CREATED)
 async def create_conversation(
     current_user: UserPublic = Depends(get_current_user),
@@ -41,8 +46,13 @@ async def create_conversation(
 ):
     conv = ConversationDB(user_id=current_user.username)
     result = await db.conversations.insert_one(conv.dict(exclude={"id"}))
-    conv.id = str(result.inserted_id)
-    return {"conversation_id": conv.id}
+    conv_id = str(result.inserted_id)
+
+    # Publish conversation created event
+    await publish_conversation_created(current_user.username, conv_id)
+
+    return {"conversation_id": conv_id}
+
 
 @router.get("/conversations/{conv_id}")
 async def get_conversation(
@@ -52,7 +62,7 @@ async def get_conversation(
 ):
     try:
         oid = ObjectId(conv_id)
-    except:
+    except Exception:
         raise HTTPException(status_code=400, detail="Invalid conversation ID")
     conv = await db.conversations.find_one({"_id": oid, "user_id": current_user.username})
     if not conv:
@@ -61,6 +71,7 @@ async def get_conversation(
     del conv["_id"]
     return conv
 
+
 @router.post("/conversations/{conv_id}/messages")
 async def send_message(
     conv_id: str,
@@ -75,39 +86,41 @@ async def send_message(
         raise HTTPException(status_code=400, detail="Invalid model")
     try:
         oid = ObjectId(conv_id)
-    except:
+    except Exception:
         raise HTTPException(status_code=400, detail="Invalid conversation ID")
     conv = await db.conversations.find_one({"_id": oid, "user_id": current_user.username})
     if not conv:
         raise HTTPException(status_code=404, detail="Conversation not found")
 
-    # Load messages
     messages = [Message(**m) for m in conv.get("messages", [])]
 
-    # Build RAG context if files
     rag_context = ""
     if files:
         rag_context = build_context_from_files(files, message)
-    
-    # System prompt with context
-    system_msg = {"role": "system", "content": SYSTEM_PROMPT + (f"\n\nContext: {rag_context}" if rag_context else "")}
 
-    # Append user message
+    system_msg = {
+        "role": "system",
+        "content": SYSTEM_PROMPT + (f"\n\nContext:\n{rag_context}" if rag_context else "")
+    }
+
     user_msg = Message(role="user", content=message)
     messages.append(user_msg)
 
-    # Groq client
-    client = Groq(api_key=settings.groq_api_key)
+    # Publish message sent event
+    await publish_message_sent(
+        user_id=current_user.username,
+        conversation_id=conv_id,
+        model=model,
+        message_preview=message,
+    )
 
-    # Tools if enabled
+    client = Groq(api_key=settings.groq_api_key)
     tools = [WEB_SEARCH_TOOL] if enable_web_search else None
 
-    # Tool loop for reasoning and multiple calls (up to 3 iterations)
-    chat_history = [
-        system_msg if isinstance(system_msg, dict) else system_msg.dict()
-    ] + [
-            m if isinstance(m, dict) else m.dict() for m in messages
-            ]
+    chat_history = [system_msg] + [
+        m if isinstance(m, dict) else m.dict() for m in messages
+    ]
+
     max_tool_loops = 3
     for _ in range(max_tool_loops):
         completion = client.chat.completions.create(
@@ -122,25 +135,21 @@ async def send_message(
         )
         choice = completion.choices[0].message
         if not choice.tool_calls:
-            # No more tools, prepare to stream
             break
         for tool_call in choice.tool_calls:
             if tool_call.function.name == "web_search":
                 args = json.loads(tool_call.function.arguments)
-                query = args["query"]
-                result = web_search(query)
-                tool_response = {
+                result = web_search(args["query"])
+                chat_history.append({
                     "role": "tool",
                     "tool_call_id": tool_call.id,
                     "name": "web_search",
                     "content": result,
-                }
-                chat_history.append(tool_response)
+                })
     else:
         logger.warning("Max tool loops reached")
         raise HTTPException(status_code=500, detail="Too many tool calls")
 
-    # Final streaming call
     completion = client.chat.completions.create(
         model=model,
         messages=chat_history,
@@ -151,18 +160,19 @@ async def send_message(
         stop=None,
     )
 
-    # Stream response
     async def generate():
         response_content = ""
         for chunk in completion:
             content = chunk.choices[0].delta.content or ""
             response_content += content
             yield content
-        # Save to DB
         messages.append(Message(role="assistant", content=response_content))
         await db.conversations.update_one(
             {"_id": oid},
-            {"$set": {"messages": [m.dict() for m in messages], "updated_at": datetime.utcnow()}}
+            {"$set": {
+                "messages": [m.dict() for m in messages],
+                "updated_at": datetime.utcnow()
+            }}
         )
 
     return StreamingResponse(generate(), media_type="text/event-stream")

requirements.txt

@@ -19,5 +19,5 @@ openpyxl==3.1.2
 pandas==2.2.2
 duckduckgo-search==6.1.9
 python-multipart==0.0.9
-
+pika==9.4.1
 numpy==1.26.4