twilio notifications (#9)

* twilio

* phone number validation

* enforce user preference

---------

Co-authored-by: Kartik Sarangmath <kartiksarangmath@Kartiks-MacBook-Air.local>

.env.example

@@ -33,4 +33,11 @@ SENTRY_DSN=123.us.sentry.io/456
 # STRIPE_SECRET_KEY=sk_test_...
 # STRIPE_WEBHOOK_SECRET=whsec_...
 # STRIPE_PRO_PRICE_ID=price_...  # $9 unlimited agents
-# STRIPE_ENTERPRISE_PRICE_ID=price_...  # $500 unlimited + enterprise features
+# STRIPE_ENTERPRISE_PRICE_ID=price_...  # $500 unlimited + enterprise features
+
+# Twilio Configuration (optional - for SMS and Email notifications)
+# TWILIO_ACCOUNT_SID=ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+# TWILIO_AUTH_TOKEN=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+# TWILIO_FROM_PHONE_NUMBER=+1234567890  # Your Twilio phone number in E.164 format
+# TWILIO_SENDGRID_API_KEY=SG.xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+# TWILIO_FROM_EMAIL=notifications@example.com  # Verified sender email address

backend/api/push_notifications.py

@@ -11,7 +11,6 @@ import logging
 from backend.auth.dependencies import get_current_user_id
 from shared.database.session import get_db
 from shared.database import PushToken
-from servers.shared.notifications import push_service
 
 router = APIRouter(prefix="/push", tags=["push_notifications"])
 
@@ -109,36 +108,3 @@ def get_my_push_tokens(
         )
         for token in tokens
     ]
-
-
-@router.post("/send-test-push", response_model=dict)
-async def send_test_push_notification(
-    user_id: UUID = Depends(get_current_user_id),
-    db: Session = Depends(get_db),
-):
-    """Send a real test push notification using Expo Push API (tests complete flow including when app is closed)"""
-    try:
-        # Send test notification using the same system that question notifications use
-        success = await push_service.send_notification(
-            db=db,
-            user_id=user_id,
-            title="Test Notification",
-            body="Push notifications are working correctly. You'll receive alerts when your agents need input.",
-            data={"type": "test_notification", "source": "backend_api"},
-        )
-
-        if success:
-            return {
-                "success": True,
-                "message": "Real test notification sent via Expo Push API! Check your device.",
-            }
-        else:
-            return {
-                "success": False,
-                "message": "Failed to send notification. Check if you have active push tokens registered.",
-            }
-
-    except Exception as e:
-        raise HTTPException(
-            status_code=500, detail=f"Error sending test notification: {str(e)}"
-        )

backend/api/user_settings.py

@@ -0,0 +1,165 @@
+"""
+User Settings API endpoints for managing notification preferences and other user settings.
+"""
+
+import re
+from fastapi import APIRouter, Depends, HTTPException
+from shared.database.models import User
+from shared.database.session import get_db
+from sqlalchemy.orm import Session
+
+from ..auth.dependencies import get_current_user
+from ..models import (
+    UserNotificationSettingsRequest,
+    UserNotificationSettingsResponse,
+)
+
+router = APIRouter(tags=["user-settings"])
+
+
+def validate_e164_phone_number(phone: str) -> bool:
+    """
+    Validate if a phone number is in E.164 format.
+    E.164 format: +[country code][subscriber number]
+    - Must start with +
+    - Country code: 1-3 digits
+    - Total length: 8-15 digits (excluding the +)
+    """
+    if not phone:
+        return True  # Empty is valid (user clearing their number)
+
+    # E.164 regex pattern
+    pattern = r"^\+[1-9]\d{1,14}$"
+    return bool(re.match(pattern, phone))
+
+
+@router.get(
+    "/user/notification-settings", response_model=UserNotificationSettingsResponse
+)
+async def get_notification_settings(current_user: User = Depends(get_current_user)):
+    """Get current user's notification settings"""
+    return UserNotificationSettingsResponse(
+        push_notifications_enabled=current_user.push_notifications_enabled,
+        email_notifications_enabled=current_user.email_notifications_enabled,
+        sms_notifications_enabled=current_user.sms_notifications_enabled,
+        phone_number=current_user.phone_number,
+        notification_email=current_user.notification_email or current_user.email,
+    )
+
+
+@router.put(
+    "/user/notification-settings", response_model=UserNotificationSettingsResponse
+)
+async def update_notification_settings(
+    request: UserNotificationSettingsRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Update current user's notification settings"""
+    if request.push_notifications_enabled is not None:
+        current_user.push_notifications_enabled = request.push_notifications_enabled
+
+    if request.email_notifications_enabled is not None:
+        current_user.email_notifications_enabled = request.email_notifications_enabled
+
+    if request.sms_notifications_enabled is not None:
+        current_user.sms_notifications_enabled = request.sms_notifications_enabled
+
+    if request.phone_number is not None:
+        # Validate E.164 format
+        if not validate_e164_phone_number(request.phone_number):
+            raise HTTPException(
+                status_code=400,
+                detail="Phone number must be in E.164 format (e.g., +12125551234). Must start with + followed by country code and number (8-15 digits total).",
+            )
+        current_user.phone_number = request.phone_number
+
+    if request.notification_email is not None:
+        # If empty string, set to None to use default email
+        current_user.notification_email = request.notification_email or None
+
+    # Additional validation: If SMS is enabled, phone number must be provided
+    if current_user.sms_notifications_enabled and not current_user.phone_number:
+        raise HTTPException(
+            status_code=400,
+            detail="Phone number is required when SMS notifications are enabled.",
+        )
+
+    db.commit()
+    db.refresh(current_user)
+
+    return UserNotificationSettingsResponse(
+        push_notifications_enabled=current_user.push_notifications_enabled,
+        email_notifications_enabled=current_user.email_notifications_enabled,
+        sms_notifications_enabled=current_user.sms_notifications_enabled,
+        phone_number=current_user.phone_number,
+        notification_email=current_user.notification_email or current_user.email,
+    )
+
+
+@router.post("/user/test-notification")
+async def test_notification(
+    notification_type: str = "all",  # "push", "sms", "email", or "all"
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Send a test notification to verify settings"""
+    results = {"push": False, "sms": False, "email": False}
+
+    # Test push notification
+    if notification_type in ["push", "all"] and current_user.push_notifications_enabled:
+        try:
+            from servers.shared.notifications import push_service
+
+            results["push"] = await push_service.send_notification(
+                db=db,
+                user_id=current_user.id,
+                title="Test Notification",
+                body="This is a test notification from Omnara.",
+            )
+        except Exception:
+            results["push"] = False
+
+    # Test email notification
+    if (
+        notification_type in ["email", "all"]
+        and current_user.email_notifications_enabled
+    ):
+        try:
+            from servers.shared.twilio_service import twilio_service
+
+            email_results = twilio_service.send_notification(
+                db=db,
+                user_id=current_user.id,
+                title="Test Notification",
+                body="This is a test notification from Omnara. If you received this, your email notification settings are working correctly.",
+                send_email=True,
+                send_sms=False,
+            )
+            results["email"] = email_results.get("email", False)
+        except Exception:
+            pass
+
+    # Test SMS notification
+    if notification_type in ["sms", "all"] and current_user.sms_notifications_enabled:
+        try:
+            from servers.shared.twilio_service import twilio_service
+
+            sms_results = twilio_service.send_notification(
+                db=db,
+                user_id=current_user.id,
+                title="Test Notification",
+                body="Test notification body",
+                sms_body="Omnara test notification. Your settings are working!",
+                send_email=False,
+                send_sms=True,
+            )
+            results["sms"] = sms_results.get("sms", False)
+        except Exception:
+            pass
+
+    return {
+        "status": "success",
+        "results": results,
+        "message": "Test notifications sent. Check your devices.",
+    }

backend/main.py

@@ -7,7 +7,14 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 import sentry_sdk
 from shared.config import settings
-from .api import agents, questions, user_agents, push_notifications, billing
+from .api import (
+    agents,
+    questions,
+    user_agents,
+    push_notifications,
+    billing,
+    user_settings,
+)
 from .auth import routes as auth_routes
 
 # Configure logging
@@ -68,6 +75,7 @@ app.include_router(agents.router, prefix=settings.api_v1_prefix)
 app.include_router(questions.router, prefix=settings.api_v1_prefix)
 app.include_router(user_agents.router, prefix=settings.api_v1_prefix)
 app.include_router(push_notifications.router, prefix=settings.api_v1_prefix)
+app.include_router(user_settings.router, prefix=settings.api_v1_prefix)
 
 # Conditionally include billing router if Stripe is configured
 if settings.stripe_secret_key:

backend/models.py

@@ -43,6 +43,33 @@ class UserFeedbackResponse(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
 
+# ============================================================================
+# User Settings Models
+# ============================================================================
+
+
+class UserNotificationSettingsRequest(BaseModel):
+    push_notifications_enabled: Optional[bool] = None
+    email_notifications_enabled: Optional[bool] = None
+    sms_notifications_enabled: Optional[bool] = None
+    phone_number: Optional[str] = Field(
+        None, description="Phone number in E.164 format (e.g., +1234567890)"
+    )
+    notification_email: Optional[str] = Field(
+        None, description="Email for notifications (defaults to account email)"
+    )
+
+
+class UserNotificationSettingsResponse(BaseModel):
+    push_notifications_enabled: bool
+    email_notifications_enabled: bool
+    sms_notifications_enabled: bool
+    phone_number: Optional[str]
+    notification_email: str  # Always returns an email (account email as fallback)
+
+    model_config = ConfigDict(from_attributes=True)
+
+
 # ============================================================================
 # Agent Models
 # ============================================================================

servers/fastapi_server/routers.py

@@ -48,6 +48,9 @@ def log_step(
             step_description=request.step_description,
             user_id=user_id,
             agent_instance_id=request.agent_instance_id,
+            send_email=request.send_email,
+            send_sms=request.send_sms,
+            send_push=request.send_push,
         )
 
         return LogStepResponse(
@@ -90,6 +93,9 @@ async def ask_question(
             agent_instance_id=request.agent_instance_id,
             question_text=request.question_text,
             user_id=user_id,
+            send_email=request.send_email,
+            send_sms=request.send_sms,
+            send_push=request.send_push,
         )
         db.commit()
 

servers/requirements.txt

@@ -11,5 +11,9 @@ python-jose[cryptography]>=3.3.0
 # Push notifications
 exponent-server-sdk>=2.1.0
 
+# Twilio notifications
+twilio>=8.0.0
+sendgrid>=6.10.0
+
 # Shared dependencies
 -r ../shared/requirements.txt

servers/shared/core/agents.py

@@ -50,6 +50,9 @@ def process_log_step(
     step_description: str,
     user_id: str,
     agent_instance_id: str | None = None,
+    send_email: bool | None = None,
+    send_sms: bool | None = None,
+    send_push: bool | None = None,
 ) -> tuple[str, int, list[str]]:
     """Process a log step operation with all common logic.
 
@@ -72,8 +75,8 @@ def process_log_step(
     else:
         instance = create_agent_instance(db, agent_type_obj.id, user_id)
 
-    # Create step
-    step = log_step(db, instance.id, step_description)
+    # Create step with notification preferences
+    step = log_step(db, instance.id, step_description, send_email, send_sms, send_push)
 
     # Get unretrieved feedback
     feedback = get_and_mark_unretrieved_feedback(db, instance.id)
@@ -86,6 +89,9 @@ async def create_agent_question(
     agent_instance_id: str,
     question_text: str,
     user_id: str,
+    send_email: bool | None = None,
+    send_sms: bool | None = None,
+    send_push: bool | None = None,
 ):
     """Create a question with validation and send push notification.
 
@@ -102,8 +108,10 @@ async def create_agent_question(
     instance = validate_agent_access(db, agent_instance_id, user_id)
 
     # Create question
-    # Note: Push notification sent by create_question() function
-    question = await create_question(db, instance.id, question_text)
+    # Note: Notifications sent by create_question() function based on parameters
+    question = await create_question(
+        db, instance.id, question_text, send_email, send_sms, send_push
+    )
 
     return question
 

servers/shared/db/queries.py

@@ -11,11 +11,14 @@ from shared.database import (
     AgentStep,
     AgentUserFeedback,
     UserAgent,
+    User,
 )
 from shared.database.billing_operations import check_agent_limit
 from sqlalchemy import func
 from sqlalchemy.orm import Session
 from fastmcp import Context
+from servers.shared.notifications import push_service
+from servers.shared.twilio_service import twilio_service
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +66,14 @@ def get_agent_instance(db: Session, instance_id: str) -> AgentInstance | None:
     return db.query(AgentInstance).filter(AgentInstance.id == instance_id).first()
 
 
-def log_step(db: Session, instance_id: UUID, description: str) -> AgentStep:
+def log_step(
+    db: Session,
+    instance_id: UUID,
+    description: str,
+    send_email: bool | None = None,
+    send_sms: bool | None = None,
+    send_push: bool | None = None,
+) -> AgentStep:
     """Log a new step for an agent instance"""
     # Get the next step number
     max_step = (
@@ -82,11 +92,74 @@ def log_step(db: Session, instance_id: UUID, description: str) -> AgentStep:
     db.add(step)
     db.commit()
     db.refresh(step)
+
+    # Send notifications if requested (all default to False for log steps)
+    if send_email or send_sms or send_push:
+        # Get instance details for notifications
+        instance = (
+            db.query(AgentInstance).filter(AgentInstance.id == instance_id).first()
+        )
+        if instance:
+            user = db.query(User).filter(User.id == instance.user_id).first()
+
+            if user:
+                agent_name = (
+                    instance.user_agent.name if instance.user_agent else "Agent"
+                )
+
+                # Override defaults - for log steps, all notifications default to False
+                should_send_push = send_push if send_push is not None else False
+                should_send_email = send_email if send_email is not None else False
+                should_send_sms = send_sms if send_sms is not None else False
+
+                # Send push notification if explicitly enabled
+                if should_send_push:
+                    try:
+                        asyncio.create_task(
+                            push_service.send_step_notification(
+                                db=db,
+                                user_id=instance.user_id,
+                                instance_id=str(instance.id),
+                                step_number=step.step_number,
+                                agent_name=agent_name,
+                                step_description=description,
+                            )
+                        )
+                    except Exception as e:
+                        logger.error(
+                            f"Failed to send push notification for step {step.id}: {e}"
+                        )
+
+                # Send Twilio notifications if explicitly enabled
+                if should_send_email or should_send_sms:
+                    try:
+                        asyncio.create_task(
+                            twilio_service.send_step_notification(
+                                db=db,
+                                user_id=instance.user_id,
+                                instance_id=str(instance.id),
+                                step_number=step.step_number,
+                                agent_name=agent_name,
+                                step_description=description,
+                                send_email=should_send_email,
+                                send_sms=should_send_sms,
+                            )
+                        )
+                    except Exception as e:
+                        logger.error(
+                            f"Failed to send Twilio notification for step {step.id}: {e}"
+                        )
+
     return step
 
 
 async def create_question(
-    db: Session, instance_id: UUID, question_text: str
+    db: Session,
+    instance_id: UUID,
+    question_text: str,
+    send_email: bool | None = None,
+    send_sms: bool | None = None,
+    send_push: bool | None = None,
 ) -> AgentQuestion:
     """Create a new question for an agent instance"""
     # Mark any existing active questions as inactive
@@ -107,27 +180,61 @@ async def create_question(
     db.commit()
     db.refresh(question)
 
-    # Send push notification
-    try:
-        from servers.shared.notifications import push_service
+    # Send notifications based on user preferences
+    if instance:
+        # Get user for checking preferences
+        user = db.query(User).filter(User.id == instance.user_id).first()
 
-        # Get agent name from instance
-        if instance:
+        if user:
             agent_name = instance.user_agent.name if instance.user_agent else "Agent"
 
-            await push_service.send_question_notification(
-                db=db,
-                user_id=instance.user_id,
-                instance_id=str(instance.id),
-                question_id=str(question.id),
-                agent_name=agent_name,
-                question_text=question_text,
+            # Determine notification preferences
+            # For questions: push defaults to True (or user preference), email/SMS default to False
+            should_send_push = (
+                send_push if send_push is not None else user.push_notifications_enabled
             )
-    except Exception as e:
-        # Don't fail the question creation if push notification fails
-        logger.error(
-            f"Failed to send push notification for question {question.id}: {e}"
-        )
+            should_send_email = (
+                send_email
+                if send_email is not None
+                else user.email_notifications_enabled
+            )
+            should_send_sms = (
+                send_sms if send_sms is not None else user.sms_notifications_enabled
+            )
+
+            # Send push notification if enabled
+            if should_send_push:
+                try:
+                    await push_service.send_question_notification(
+                        db=db,
+                        user_id=instance.user_id,
+                        instance_id=str(instance.id),
+                        question_id=str(question.id),
+                        agent_name=agent_name,
+                        question_text=question_text,
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to send push notification for question {question.id}: {e}"
+                    )
+
+            # Send Twilio notification if enabled (email and/or SMS)
+            if should_send_email or should_send_sms:
+                try:
+                    twilio_service.send_question_notification(
+                        db=db,
+                        user_id=instance.user_id,
+                        instance_id=str(instance.id),
+                        question_id=str(question.id),
+                        agent_name=agent_name,
+                        question_text=question_text,
+                        send_email=should_send_email,
+                        send_sms=should_send_sms,
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to send Twilio notification for question {question.id}: {e}"
+                    )
 
     return question
 

servers/shared/models/base.py

@@ -25,6 +25,17 @@ class BaseLogStepRequest(BaseModel):
     step_description: str = Field(
         ..., description="Clear description of what the agent is doing"
     )
+    send_email: bool | None = Field(
+        None,
+        description="Whether to send email notification (overrides user preference)",
+    )
+    send_sms: bool | None = Field(
+        None, description="Whether to send SMS notification (overrides user preference)"
+    )
+    send_push: bool | None = Field(
+        None,
+        description="Whether to send push notification (overrides user preference)",
+    )
 
 
 class BaseAskQuestionRequest(BaseModel):
@@ -32,6 +43,17 @@ class BaseAskQuestionRequest(BaseModel):
 
     agent_instance_id: str = Field(..., description="Agent instance ID")
     question_text: str = Field(..., description="Question to ask the user")
+    send_email: bool | None = Field(
+        None,
+        description="Whether to send email notification (overrides user preference)",
+    )
+    send_sms: bool | None = Field(
+        None, description="Whether to send SMS notification (overrides user preference)"
+    )
+    send_push: bool | None = Field(
+        None,
+        description="Whether to send push notification (overrides user preference)",
+    )
 
 
 class BaseEndSessionRequest(BaseModel):

servers/shared/notification_base.py

@@ -0,0 +1,50 @@
+"""Abstract base class for notification services"""
+
+from abc import ABC, abstractmethod
+from typing import Dict, Union
+from uuid import UUID
+from sqlalchemy.orm import Session
+
+
+class NotificationServiceBase(ABC):
+    """Abstract base class defining the interface for notification services"""
+
+    @abstractmethod
+    def send_notification(
+        self, db: Session, user_id: UUID, title: str, body: str, **kwargs
+    ) -> Union[bool, Dict[str, bool]]:
+        """Send a general notification
+
+        Returns:
+            bool for single-channel services (push)
+            Dict[str, bool] for multi-channel services (email/SMS)
+        """
+        pass
+
+    @abstractmethod
+    def send_question_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        instance_id: str,
+        question_id: str,
+        agent_name: str,
+        question_text: str,
+        **kwargs,
+    ) -> Union[bool, Dict[str, bool]]:
+        """Send notification for a new agent question"""
+        pass
+
+    @abstractmethod
+    async def send_step_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        instance_id: str,
+        step_number: int,
+        agent_name: str,
+        step_description: str,
+        **kwargs,
+    ) -> Union[bool, Dict[str, bool]]:
+        """Send notification for a new agent step"""
+        pass

servers/shared/notifications.py

@@ -15,12 +15,13 @@ from exponent_server_sdk import (
 )
 import requests.exceptions
 
-from shared.database import PushToken
+from shared.database import PushToken, User
+from .notification_base import NotificationServiceBase
 
 logger = logging.getLogger(__name__)
 
 
-class PushNotificationService:
+class PushNotificationService(NotificationServiceBase):
     """Service for sending push notifications via Expo"""
 
     def __init__(self):
@@ -36,6 +37,12 @@ class PushNotificationService:
     ) -> bool:
         """Send push notification to all user's devices"""
         try:
+            # First check if user has push notifications enabled
+            user = db.query(User).filter(User.id == user_id).first()
+            if not user or not user.push_notifications_enabled:
+                logger.info(f"Push notifications disabled for user {user_id}")
+                return False
+
             # Get user's active push tokens
             tokens = (
                 db.query(PushToken)
@@ -210,6 +217,39 @@ class PushNotificationService:
             data=data,
         )
 
+    async def send_step_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        instance_id: str,
+        step_number: int,
+        agent_name: str,
+        step_description: str,
+    ) -> bool:
+        """Send notification for new agent step"""
+        # Format agent name for display
+        display_name = agent_name.replace("_", " ").title()
+        title = f"{display_name} - Step {step_number}"
+
+        # Truncate step description for notification
+        body = step_description
+        if len(body) > 100:
+            body = body[:97] + "..."
+
+        data = {
+            "type": "new_step",
+            "instanceId": instance_id,
+            "stepNumber": step_number,
+        }
+
+        return await self.send_notification(
+            db=db,
+            user_id=user_id,
+            title=title,
+            body=body,
+            data=data,
+        )
+
     @staticmethod
     def _chunks(lst, n):
         """Yield successive n-sized chunks from lst."""

servers/shared/twilio_service.py

@@ -0,0 +1,261 @@
+"""Twilio notification service for SMS and email notifications"""
+
+import logging
+from typing import Optional
+from uuid import UUID
+from sqlalchemy.orm import Session
+
+from twilio.rest import Client
+from twilio.base.exceptions import TwilioException
+from sendgrid import SendGridAPIClient
+from sendgrid.helpers.mail import Mail
+
+from shared.config import settings
+from shared.database import User
+from .notification_base import NotificationServiceBase
+
+logger = logging.getLogger(__name__)
+
+
+class TwilioNotificationService(NotificationServiceBase):
+    """Service for sending notifications via Twilio (SMS) and SendGrid (Email)"""
+
+    def __init__(self):
+        self.twilio_client = None
+        self.sendgrid_client = None
+
+        # Initialize Twilio client if credentials are provided
+        if settings.twilio_account_sid and settings.twilio_auth_token:
+            try:
+                self.twilio_client = Client(
+                    settings.twilio_account_sid, settings.twilio_auth_token
+                )
+                logger.info("Twilio client initialized successfully")
+            except Exception as e:
+                logger.error(f"Failed to initialize Twilio client: {e}")
+
+        # Initialize SendGrid client if API key is provided
+        if settings.twilio_sendgrid_api_key:
+            try:
+                self.sendgrid_client = SendGridAPIClient(
+                    settings.twilio_sendgrid_api_key
+                )
+                logger.info("SendGrid client initialized successfully")
+            except Exception as e:
+                logger.error(f"Failed to initialize SendGrid client: {e}")
+
+    def send_sms(
+        self, to_number: str, body: str, from_number: Optional[str] = None
+    ) -> bool:
+        """Send SMS notification via Twilio"""
+        if not self.twilio_client:
+            logger.warning("Twilio client not configured, skipping SMS")
+            return False
+
+        if not to_number:
+            logger.warning("No phone number provided for SMS")
+            return False
+
+        from_number = from_number or settings.twilio_from_phone_number
+        if not from_number:
+            logger.error("No from phone number configured")
+            return False
+
+        try:
+            message = self.twilio_client.messages.create(
+                body=body, from_=from_number, to=to_number
+            )
+            logger.info(f"SMS sent successfully: {message.sid}")
+            return True
+        except TwilioException as e:
+            logger.error(f"Twilio error sending SMS: {e}")
+            return False
+        except Exception as e:
+            logger.error(f"Unexpected error sending SMS: {e}")
+            return False
+
+    def send_email(
+        self,
+        to_email: str,
+        subject: str,
+        body: str,
+        html_body: Optional[str] = None,
+        from_email: Optional[str] = None,
+    ) -> bool:
+        """Send email notification via SendGrid"""
+        if not self.sendgrid_client:
+            logger.warning("SendGrid client not configured, skipping email")
+            return False
+
+        if not to_email:
+            logger.warning("No email address provided")
+            return False
+
+        from_email = from_email or settings.twilio_from_email
+        if not from_email:
+            logger.error("No from email configured")
+            return False
+
+        try:
+            message = Mail(
+                from_email=from_email,
+                to_emails=to_email,
+                subject=subject,
+                plain_text_content=body,
+                html_content=html_body,
+            )
+
+            response = self.sendgrid_client.send(message)
+            logger.info(f"Email sent successfully: {response.status_code}")
+            return response.status_code in [200, 201, 202]
+        except Exception as e:
+            logger.error(f"Error sending email: {e}")
+            return False
+
+    def send_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        title: str,
+        body: str,
+        sms_body: Optional[str] = None,
+        send_email: bool | None = None,
+        send_sms: bool | None = None,
+    ) -> dict[str, bool]:
+        """Send notification via user's preferred channels"""
+        results = {"sms": False, "email": False}
+
+        try:
+            # Get user preferences
+            user = db.query(User).filter(User.id == user_id).first()
+            if not user:
+                logger.error(f"User {user_id} not found")
+                return results
+
+            # User preferences are the ultimate authority
+            # Only send if BOTH the user has it enabled AND the API call requests it (or doesn't specify)
+            should_send_email = user.email_notifications_enabled and (
+                send_email if send_email is not None else True
+            )
+            should_send_sms = user.sms_notifications_enabled and (
+                send_sms if send_sms is not None else True
+            )
+
+            # Log when notifications are blocked by user preferences
+            if send_email and not user.email_notifications_enabled:
+                logger.info(
+                    f"Email notification blocked by user preferences for user {user_id}"
+                )
+            if send_sms and not user.sms_notifications_enabled:
+                logger.info(
+                    f"SMS notification blocked by user preferences for user {user_id}"
+                )
+
+            # Send SMS if enabled and phone number is available
+            if should_send_sms and user.phone_number and self.twilio_client:
+                sms_message = sms_body or body
+                # Truncate SMS to 160 characters
+                if len(sms_message) > 160:
+                    sms_message = sms_message[:157] + "..."
+                results["sms"] = self.send_sms(user.phone_number, sms_message)
+
+            # Send email if enabled
+            notification_email = user.notification_email or user.email
+            if should_send_email and notification_email and self.sendgrid_client:
+                results["email"] = self.send_email(
+                    notification_email, subject=title, body=body
+                )
+
+            return results
+
+        except Exception as e:
+            logger.error(f"Error sending Twilio notifications: {e}")
+            return results
+
+    def send_question_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        instance_id: str,
+        question_id: str,
+        agent_name: str,
+        question_text: str,
+        send_email: bool | None = None,
+        send_sms: bool | None = None,
+    ) -> dict[str, bool]:
+        """Send notification for new agent question"""
+        # Format agent name for display
+        display_name = agent_name.replace("_", " ").title()
+        title = f"{display_name} needs your input"
+
+        # Email body with more detail
+        email_body = f"""
+Your agent {display_name} has a question:
+
+{question_text}
+
+You can respond at: {settings.frontend_urls[0]}/agents/{instance_id}/questions/{question_id}
+
+Best regards,
+The Omnara Team
+"""
+
+        # SMS body (shorter)
+        sms_body = f"{display_name}: {question_text}"
+
+        return self.send_notification(
+            db=db,
+            user_id=user_id,
+            title=title,
+            body=email_body,
+            sms_body=sms_body,
+            send_email=send_email,
+            send_sms=send_sms,
+        )
+
+    async def send_step_notification(
+        self,
+        db: Session,
+        user_id: UUID,
+        instance_id: str,
+        step_number: int,
+        agent_name: str,
+        step_description: str,
+        send_email: bool | None = None,
+        send_sms: bool | None = None,
+    ) -> dict[str, bool]:
+        """Send notification for new agent step"""
+        # Format agent name for display
+        display_name = agent_name.replace("_", " ").title()
+        title = f"{display_name} - Step {step_number}"
+
+        # Email body with more detail
+        email_body = f"""
+Your agent {display_name} has logged a new step:
+
+Step {step_number}: {step_description}
+
+You can view the full session at: {settings.frontend_urls[0]}/agents/{instance_id}
+
+Best regards,
+The Omnara Team
+"""
+
+        # SMS body (shorter)
+        sms_body = f"{display_name} Step {step_number}: {step_description}"
+        if len(sms_body) > 160:
+            sms_body = f"{display_name} Step {step_number}: {step_description[:140]}..."
+
+        return self.send_notification(
+            db=db,
+            user_id=user_id,
+            title=title,
+            body=email_body,
+            sms_body=sms_body,
+            send_email=send_email,
+            send_sms=send_sms,
+        )
+
+
+# Singleton instance
+twilio_service = TwilioNotificationService()

shared/alembic/versions/84cd4a8c9a18_add_notification_preferences_to_users_.py

@@ -0,0 +1,68 @@
+"""Add notification preferences to users table
+
+Revision ID: 84cd4a8c9a18
+Revises: ae5dce0d9dd3
+Create Date: 2025-07-12 18:36:07.840763
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "84cd4a8c9a18"
+down_revision: Union[str, None] = "ae5dce0d9dd3"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    # Add columns with server defaults for existing rows
+    op.add_column(
+        "users",
+        sa.Column(
+            "push_notifications_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default="true",
+        ),
+    )
+    op.add_column(
+        "users",
+        sa.Column(
+            "email_notifications_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default="false",
+        ),
+    )
+    op.add_column(
+        "users",
+        sa.Column(
+            "sms_notifications_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default="false",
+        ),
+    )
+    op.add_column(
+        "users", sa.Column("phone_number", sa.String(length=20), nullable=True)
+    )
+    op.add_column(
+        "users", sa.Column("notification_email", sa.String(length=255), nullable=True)
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("users", "notification_email")
+    op.drop_column("users", "phone_number")
+    op.drop_column("users", "sms_notifications_enabled")
+    op.drop_column("users", "email_notifications_enabled")
+    op.drop_column("users", "push_notifications_enabled")
+    # ### end Alembic commands ###

shared/config/settings.py

@@ -98,6 +98,13 @@ class Settings(BaseSettings):
     enterprise_plan_agent_limit: int = -1  # Unlimited
     enterprise_plan_price: float = 500
 
+    # Twilio Configuration
+    twilio_account_sid: str = ""
+    twilio_auth_token: str = ""
+    twilio_from_phone_number: str = ""  # Format: +1234567890
+    twilio_sendgrid_api_key: str = ""  # For email notifications via SendGrid
+    twilio_from_email: str = ""  # Sender email address
+
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
 
 

shared/database/models.py

@@ -37,6 +37,17 @@ class User(Base):
         default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)
     )
 
+    # Notification preferences
+    push_notifications_enabled: Mapped[bool] = mapped_column(default=True)
+    email_notifications_enabled: Mapped[bool] = mapped_column(default=True)
+    sms_notifications_enabled: Mapped[bool] = mapped_column(default=False)
+    phone_number: Mapped[str | None] = mapped_column(
+        String(20), default=None
+    )  # E.164 format
+    notification_email: Mapped[str | None] = mapped_column(
+        String(255), default=None
+    )  # Defaults to email if not set
+
     # Relationships
     agent_instances: Mapped[list["AgentInstance"]] = relationship(
         "AgentInstance", back_populates="user"