From ab37f15b2746f38f41195d03598115102e5c54f4 Mon Sep 17 00:00:00 2001 From: Amit Bhakt Date: Wed, 21 Oct 2026 23:51:56 +0530 Subject: [PATCH 01/10] feat: Add native WhatsApp Cloud API serving with SQLite session persistence - Add Agent.serve_whatsapp() and Server.serve_whatsapp() - Implement SQLiteSessionStore with WAL mode, session timeout, and message idempotency - Implement WhatsAppServer with HMAC-SHA256 signature verification and background task processing - Implement MetaWhatsAppClient with message chunking and read receipts - Implement MetaDiscoveryClient for automated Phone Number ID discovery and 3-tier webhook sync (adapted from WAVoice) - Add 17 unit and integration tests covering security, idempotency, session isolation, and auto-discovery --- .gitignore | 5 + src/sdialog/agents.py | 59 +++ src/sdialog/server.py | 63 ++++ src/sdialog/whatsapp/__init__.py | 25 ++ src/sdialog/whatsapp/client.py | 121 ++++++ src/sdialog/whatsapp/config.py | 64 ++++ src/sdialog/whatsapp/discovery.py | 181 +++++++++ src/sdialog/whatsapp/server.py | 331 +++++++++++++++++ src/sdialog/whatsapp/session_store.py | 214 +++++++++++ tests/test_whatsapp.py | 514 ++++++++++++++++++++++++++ 10 files changed, 1577 insertions(+) create mode 100644 src/sdialog/whatsapp/__init__.py create mode 100644 src/sdialog/whatsapp/client.py create mode 100644 src/sdialog/whatsapp/config.py create mode 100644 src/sdialog/whatsapp/discovery.py create mode 100644 src/sdialog/whatsapp/server.py create mode 100644 src/sdialog/whatsapp/session_store.py create mode 100644 tests/test_whatsapp.py diff --git a/.gitignore b/.gitignore index 79120d5..61f9e2a 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,8 @@ tutorials/01_audio/dscaper_data_impulse_response/* *audio_dialog.json ./src/sdialog/sdialog tutorials/01_audio/*.png + +# SQLite +*.db +*.db-wal +*.db-shm diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index 92abbbe..954e6e4 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -670,6 +670,65 @@ def serve(self, return Server.serve(agents=self, host=host, port=port, stateless=stateless, debug=debug, log_level=log_level) + def serve_whatsapp(self, + port: int = 1234, + host: str = "0.0.0.0", + phone_number_id: Optional[str] = None, + access_token: Optional[str] = None, + verify_token: Optional[str] = None, + app_secret: Optional[str] = None, + db_path: str = "whatsapp_sessions.db", + webhook_path: str = "/webhook", + session_timeout: int = 1800, + tunnel: bool = False, + auto_setup: bool = True, + log_level: str = "info"): + """ + Starts a Meta WhatsApp Cloud API webhook server to interact with the agent. + + :param port: Port number to bind the server to. + :type port: int + :param host: Host address to bind the server to. + :type host: str + :param phone_number_id: Meta WhatsApp Phone Number ID (or WHATSAPP_PHONE_NUMBER_ID env var). + :type phone_number_id: Optional[str] + :param access_token: Meta Graph API Access Token (or WHATSAPP_ACCESS_TOKEN env var). + :type access_token: Optional[str] + :param verify_token: Webhook verification token (or WHATSAPP_VERIFY_TOKEN env var). + :type verify_token: Optional[str] + :param app_secret: Meta App Secret for HMAC validation (or WHATSAPP_APP_SECRET env var). + :type app_secret: Optional[str] + :param db_path: Path to the SQLite session database. + :type db_path: str + :param webhook_path: Webhook URL path. + :type webhook_path: str + :param session_timeout: Seconds of inactivity before a user session resets. + :type session_timeout: int + :param tunnel: If True, attempts to spawn a public ngrok tunnel. + :type tunnel: bool + :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :type auto_setup: bool + :param log_level: Logging level for the server. + :type log_level: str + """ + from .whatsapp import WhatsAppServer, WhatsAppConfig + + config = WhatsAppConfig( + port=port, + host=host, + phone_number_id=phone_number_id, + access_token=access_token, + verify_token=verify_token, + app_secret=app_secret, + db_path=db_path, + webhook_path=webhook_path, + session_timeout=session_timeout, + tunnel=tunnel, + auto_setup=auto_setup, + log_level=log_level + ) + return WhatsAppServer.serve(agent=self, config=config) + def response_lookahead(self, message: str = None): """ Generates a response without updating the agent's memory. diff --git a/src/sdialog/server.py b/src/sdialog/server.py index 5119000..bf0420e 100644 --- a/src/sdialog/server.py +++ b/src/sdialog/server.py @@ -318,6 +318,69 @@ async def serve_async(cls, server = uvicorn.Server(config) await server.serve() + @classmethod + def serve_whatsapp(cls, + agent: Agent, + port: int = 1234, + host: str = "0.0.0.0", + phone_number_id: Optional[str] = None, + access_token: Optional[str] = None, + verify_token: Optional[str] = None, + app_secret: Optional[str] = None, + db_path: str = "whatsapp_sessions.db", + webhook_path: str = "/webhook", + session_timeout: int = 1800, + tunnel: bool = False, + auto_setup: bool = True, + log_level: str = "info") -> None: + """ + Serve an SDialog agent via Meta WhatsApp Cloud API webhook. + + :param agent: The SDialog agent to serve. + :type agent: Agent + :param port: Port number to listen on. + :type port: int + :param host: Host address to bind to. + :type host: str + :param phone_number_id: Meta WhatsApp Phone Number ID. + :type phone_number_id: Optional[str] + :param access_token: Meta Graph API Access Token. + :type access_token: Optional[str] + :param verify_token: Webhook verification token. + :type verify_token: Optional[str] + :param app_secret: Meta App Secret for HMAC validation. + :type app_secret: Optional[str] + :param db_path: Path to the SQLite session database. + :type db_path: str + :param webhook_path: Webhook URL path. + :type webhook_path: str + :param session_timeout: Inactivity seconds before resetting session. + :type session_timeout: int + :param tunnel: If True, attempts to spawn a public ngrok tunnel. + :type tunnel: bool + :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :type auto_setup: bool + :param log_level: Logging level for the server. + :type log_level: str + """ + from .whatsapp import WhatsAppServer, WhatsAppConfig + + config = WhatsAppConfig( + port=port, + host=host, + phone_number_id=phone_number_id, + access_token=access_token, + verify_token=verify_token, + app_secret=app_secret, + db_path=db_path, + webhook_path=webhook_path, + session_timeout=session_timeout, + tunnel=tunnel, + auto_setup=auto_setup, + log_level=log_level + ) + return WhatsAppServer.serve(agent=agent, config=config) + @classmethod def serve_in_thread(cls, agents: Union[Agent, List[Agent]], diff --git a/src/sdialog/whatsapp/__init__.py b/src/sdialog/whatsapp/__init__.py new file mode 100644 index 0000000..6cc40ca --- /dev/null +++ b/src/sdialog/whatsapp/__init__.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +WhatsApp integration for SDialog agents. + +This package provides native WhatsApp serving capabilities for SDialog agents, +enabling two-way text conversations over the Meta WhatsApp Cloud API with SQLite +persistence and per-user session isolation. +""" + +from .config import WhatsAppConfig +from .server import WhatsAppServer +from .session_store import SQLiteSessionStore +from .client import MetaWhatsAppClient +from .discovery import MetaDiscoveryClient + +__all__ = [ + "WhatsAppConfig", + "WhatsAppServer", + "SQLiteSessionStore", + "MetaWhatsAppClient", + "MetaDiscoveryClient", +] diff --git a/src/sdialog/whatsapp/client.py b/src/sdialog/whatsapp/client.py new file mode 100644 index 0000000..b4393af --- /dev/null +++ b/src/sdialog/whatsapp/client.py @@ -0,0 +1,121 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Meta WhatsApp Cloud API client. +""" + +import logging +from typing import List +import httpx + +from .config import WhatsAppConfig + +logger = logging.getLogger(__name__) + +# WhatsApp limits single message text body to 4096 characters +MAX_WHATSAPP_TEXT_LENGTH = 4096 + + +class MetaWhatsAppClient: + """Async client for interacting with the Meta WhatsApp Cloud API.""" + + def __init__(self, config: WhatsAppConfig): + self.config = config + self.base_url = f"https://graph.facebook.com/{config.api_version}/{config.phone_number_id}" + self.headers = { + "Authorization": f"Bearer {config.access_token}", + "Content-Type": "application/json", + } + + async def send_text_message(self, to: str, text: str) -> bool: + """ + Send a text message to a WhatsApp user. + + If the text exceeds WhatsApp's maximum length (4096 characters), + it is split and sent in chunks. + """ + if not text: + return False + + chunks = self._chunk_text(text, MAX_WHATSAPP_TEXT_LENGTH) + success = True + + async with httpx.AsyncClient(timeout=10.0) as client: + for chunk in chunks: + payload = { + "messaging_product": "whatsapp", + "recipient_type": "individual", + "to": to, + "type": "text", + "text": {"preview_url": False, "body": chunk}, + } + + try: + response = await client.post( + f"{self.base_url}/messages", + json=payload, + headers=self.headers, + ) + + if response.status_code in (200, 201): + logger.debug(f"[WhatsApp] Sent message to {to}") + else: + logger.error( + f"[WhatsApp] Failed to send message to {to}. " + f"HTTP {response.status_code}: {response.text}" + ) + success = False + except Exception as e: + logger.error(f"[WhatsApp] Exception while sending message to {to}: {e}") + success = False + + return success + + async def mark_as_read(self, message_id: str) -> bool: + """Mark an incoming WhatsApp message as read (blue double check).""" + if not message_id: + return False + + payload = { + "messaging_product": "whatsapp", + "status": "read", + "message_id": message_id, + } + + async with httpx.AsyncClient(timeout=5.0) as client: + try: + response = await client.post( + f"{self.base_url}/messages", + json=payload, + headers=self.headers, + ) + return response.status_code == 200 + except Exception as e: + logger.debug(f"[WhatsApp] Could not mark message {message_id} as read: {e}") + return False + + @staticmethod + def _chunk_text(text: str, max_length: int) -> List[str]: + """Split long messages into WhatsApp-safe chunks without splitting words where possible.""" + if len(text) <= max_length: + return [text] + + chunks = [] + while text: + if len(text) <= max_length: + chunks.append(text) + break + + # Find last whitespace within max_length + split_idx = text.rfind("\n", 0, max_length) + if split_idx == -1: + split_idx = text.rfind(" ", 0, max_length) + if split_idx == -1: + split_idx = max_length + + chunks.append(text[:split_idx].strip()) + text = text[split_idx:].lstrip() + + return chunks diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py new file mode 100644 index 0000000..9843fc0 --- /dev/null +++ b/src/sdialog/whatsapp/config.py @@ -0,0 +1,64 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Configuration models for SDialog WhatsApp server. +""" + +import os +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass +class WhatsAppConfig: + """ + Configuration parameters for serving an SDialog agent on WhatsApp. + + Parameters will fall back to standard environment variables if not provided: + - ``WHATSAPP_PHONE_NUMBER_ID`` + - ``WHATSAPP_ACCESS_TOKEN`` + - ``WHATSAPP_VERIFY_TOKEN`` + - ``WHATSAPP_APP_SECRET`` + """ + + port: int = 1234 + host: str = "0.0.0.0" + phone_number_id: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_PHONE_NUMBER_ID") + ) + access_token: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_ACCESS_TOKEN") or os.environ.get("WHATSAPP_TOKEN") + ) + verify_token: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_VERIFY_TOKEN") + or os.environ.get("WHATSAPP_WEBHOOK_VERIFY_TOKEN", "sdialog_whatsapp_secret") + ) + app_secret: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_APP_SECRET") + ) + api_version: str = "v21.0" + db_path: str = "whatsapp_sessions.db" + webhook_path: str = "/webhook" + session_timeout: int = 1800 # 30 minutes idle timeout + tunnel: bool = False + auto_setup: bool = True # Automatically discover IDs and configure Meta webhooks + max_history_turns: int = 40 # Context window sliding budget + enforce_signature: bool = True + mark_as_read: bool = True + log_level: str = "info" + + def validate(self, dry_run: bool = False) -> None: + """Validate that required credentials are set when not in dry_run mode.""" + if not dry_run: + if not self.phone_number_id: + raise ValueError( + "WhatsApp phone_number_id is required. Pass it to serve_whatsapp() " + "or set the WHATSAPP_PHONE_NUMBER_ID environment variable." + ) + if not self.access_token: + raise ValueError( + "WhatsApp access_token is required. Pass it to serve_whatsapp() " + "or set the WHATSAPP_ACCESS_TOKEN environment variable." + ) diff --git a/src/sdialog/whatsapp/discovery.py b/src/sdialog/whatsapp/discovery.py new file mode 100644 index 0000000..2a82dff --- /dev/null +++ b/src/sdialog/whatsapp/discovery.py @@ -0,0 +1,181 @@ +# SPDX-FileCopyrightText: Copyright © 2026 Amit Bhakt +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-License-Identifier: MIT +# +# Portions of this module are adapted from WAVoice: +# "WAVoice: WebRTC Gateway and DevOps Automation Engine for WhatsApp Voice AI" +# Repository: https://github.com/amitbhakt/wavoice +# DOI: 10.5281/zenodo.23537489 +# Author: Amit Bhakt +# Licensed under the MIT License. +""" +Meta WhatsApp Cloud API auto-discovery and multi-tier webhook configuration. + +This module automates the onboarding process for WhatsApp Cloud API agents: +1. Token inspection (extracts App ID, WABA ID, permissions) +2. Automatic discovery of Phone Number ID and WABA ID +3. 3-Tier Webhook synchronization (App-level, WABA-level, and Phone-number-level) +""" + +import logging +from typing import Optional, Dict, Any, List +import httpx + +logger = logging.getLogger(__name__) + +GRAPH_API_VERSION = "v21.0" +GRAPH_BASE_URL = f"https://graph.facebook.com/{GRAPH_API_VERSION}" + + +class MetaDiscoveryClient: + """ + Client for automating Meta WhatsApp Cloud API credentials discovery and webhook setup. + + Adapted from WAVoice (MIT License, https://github.com/amitbhakt/wavoice). + """ + + def __init__(self, access_token: str, app_secret: Optional[str] = None): + self.access_token = access_token + self.app_secret = app_secret + + @property + def _headers(self) -> Dict[str, str]: + return { + "Authorization": f"Bearer {self.access_token}", + "Content-Type": "application/json", + } + + async def inspect_token(self) -> Dict[str, Any]: + """ + Inspect the provided User/System Access Token via debug_token endpoint. + + Returns metadata including app_id, scopes, user_id, and granular target IDs (WABAs). + """ + url = f"{GRAPH_BASE_URL}/debug_token" + params = {"input_token": self.access_token, "access_token": self.access_token} + + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(url, params=params) + data = resp.json() + if resp.status_code != 200: + error_msg = data.get("error", {}).get("message", "Token inspection failed") + raise ValueError(f"Invalid Meta Access Token: {error_msg}") + return data.get("data", {}) + + async def discover_waba_id(self) -> Optional[str]: + """Auto-discover WhatsApp Business Account (WABA) ID from token permissions.""" + try: + token_data = await self.inspect_token() + granular_scopes = token_data.get("granular_scopes", []) + for scope_info in granular_scopes: + if scope_info.get("scope") in ("whatsapp_business_management", "whatsapp_business_messaging"): + target_ids = scope_info.get("target_ids", []) + if target_ids: + return str(target_ids[0]) + except Exception as e: + logger.debug(f"[WhatsApp] Could not auto-discover WABA ID: {e}") + return None + + async def list_phone_numbers(self, waba_id: str) -> List[Dict[str, Any]]: + """List all phone numbers under a WhatsApp Business Account.""" + url = f"{GRAPH_BASE_URL}/{waba_id}/phone_numbers" + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(url, headers=self._headers) + if resp.status_code == 200: + return resp.json().get("data", []) + return [] + + async def discover_phone_number_id(self, waba_id: Optional[str] = None) -> Optional[str]: + """Auto-discover the primary WhatsApp Phone Number ID.""" + target_waba = waba_id or await self.discover_waba_id() + if not target_waba: + return None + + numbers = await self.list_phone_numbers(target_waba) + if numbers: + phone_id = str(numbers[0].get("id")) + display_num = numbers[0].get("display_phone_number", "unknown") + logger.info(f"[WhatsApp] Auto-discovered Phone Number ID: {phone_id} ({display_num})") + return phone_id + return None + + async def configure_webhooks( + self, + callback_url: str, + verify_token: str, + phone_number_id: Optional[str] = None, + waba_id: Optional[str] = None, + ) -> Dict[str, bool]: + """ + Programmatically synchronize all 3 tiers of Meta WhatsApp webhooks: + 1. App-level subscription (POST /{app_id}/subscriptions) + 2. WABA-level subscription (POST /{waba_id}/subscribed_apps) + 3. Phone-number-level override (POST /{phone_number_id}) + """ + results = {"app_level": False, "waba_level": False, "phone_level": False} + + token_info = await self.inspect_token() + app_id = token_info.get("app_id") + target_waba = waba_id or await self.discover_waba_id() + + async with httpx.AsyncClient(timeout=15.0) as client: + # Tier 1: App-Level Webhook Subscription (requires App Access Token) + if app_id and self.app_secret: + try: + app_access_token = f"{app_id}|{self.app_secret}" + app_sub_url = f"{GRAPH_BASE_URL}/{app_id}/subscriptions" + payload = { + "object": "whatsapp_business_account", + "callback_url": callback_url, + "verify_token": verify_token, + "fields": "messages", + "include_values": "true", + "access_token": app_access_token, + } + resp = await client.post(app_sub_url, data=payload) + if resp.status_code == 200 and resp.json().get("success"): + results["app_level"] = True + logger.info("[WhatsApp] Configured App-level webhook subscription") + else: + logger.warning(f"[WhatsApp] App-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set App-level webhook: {e}") + + # Tier 2: WABA-Level Subscribed Apps + if target_waba: + try: + waba_sub_url = f"{GRAPH_BASE_URL}/{target_waba}/subscribed_apps" + waba_payload = { + "override_callback_uri": callback_url, + "verify_token": verify_token, + } + resp = await client.post(waba_sub_url, headers=self._headers, json=waba_payload) + if resp.status_code == 200 and resp.json().get("success"): + results["waba_level"] = True + logger.info(f"[WhatsApp] Configured WABA-level webhook override on {target_waba}") + else: + logger.warning(f"[WhatsApp] WABA-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set WABA-level webhook: {e}") + + # Tier 3: Phone-Number-Level Webhook Override + target_phone = phone_number_id or await self.discover_phone_number_id(target_waba) + if target_phone: + try: + phone_url = f"{GRAPH_BASE_URL}/{target_phone}" + phone_payload = { + "webhook_configuration": { + "override_callback_uri": callback_url, + "verify_token": verify_token, + } + } + resp = await client.post(phone_url, headers=self._headers, json=phone_payload) + if resp.status_code == 200 and resp.json().get("success"): + results["phone_level"] = True + logger.info(f"[WhatsApp] Configured Phone-level webhook override on {target_phone}") + else: + logger.warning(f"[WhatsApp] Phone-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set Phone-level webhook: {e}") + + return results diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py new file mode 100644 index 0000000..a6fa958 --- /dev/null +++ b/src/sdialog/whatsapp/server.py @@ -0,0 +1,331 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +FastAPI webhook server for SDialog WhatsApp agent serving. +""" + +import hmac +import hashlib +import asyncio +import logging +import threading +from typing import Optional, Dict, Any + +from fastapi import FastAPI, Request, Response, BackgroundTasks, HTTPException, Query +from fastapi.responses import JSONResponse +import uvicorn + +from .config import WhatsAppConfig +from .client import MetaWhatsAppClient +from .session_store import ( + SQLiteSessionStore, + restore_agent_state, + dump_agent_state, +) + +logger = logging.getLogger(__name__) + + +class WhatsAppServer: + """ + FastAPI server exposing Meta WhatsApp Cloud API webhook endpoints for SDialog agents. + """ + + _app: Optional[FastAPI] = None + _config: Optional[WhatsAppConfig] = None + _agent: Any = None + _store: Optional[SQLiteSessionStore] = None + _client: Optional[MetaWhatsAppClient] = None + _user_locks: Dict[str, asyncio.Lock] = {} + + @classmethod + def get_user_lock(cls, wa_id: str) -> asyncio.Lock: + """Get or create an async lock for a specific WhatsApp sender ID.""" + if wa_id not in cls._user_locks: + cls._user_locks[wa_id] = asyncio.Lock() + return cls._user_locks[wa_id] + + @classmethod + def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: + """Create and configure the FastAPI WhatsApp webhook application.""" + cls._agent = agent + cls._config = config + cls._store = SQLiteSessionStore(db_path=config.db_path) + cls._client = MetaWhatsAppClient(config) + cls._user_locks.clear() + + app = FastAPI( + title="SDialog WhatsApp Agent Server", + description="Meta WhatsApp Cloud API Webhook Server for SDialog Agents", + version="1.0.0", + ) + + @app.get(config.webhook_path) + async def verify_webhook( + hub_mode: str = Query(None, alias="hub.mode"), + hub_verify_token: str = Query(None, alias="hub.verify_token"), + hub_challenge: str = Query(None, alias="hub.challenge"), + ): + """Handle Meta WhatsApp webhook verification challenge.""" + if hub_mode == "subscribe" and hub_verify_token == config.verify_token: + logger.info("[WhatsApp] Webhook verification challenge accepted") + return Response(content=hub_challenge, media_type="text/plain") + logger.warning("[WhatsApp] Webhook verification failed: token mismatch") + return Response(content="Verification failed", status_code=403) + + @app.post(config.webhook_path) + async def receive_webhook(request: Request, background_tasks: BackgroundTasks): + """ + Receive incoming WhatsApp webhook events. + + Validates HMAC signature, checks idempotency, and queues background processing. + Returns 200 OK immediately to satisfy Meta's 3-second SLA. + """ + raw_body = await request.body() + + # 1. HMAC Signature Verification + if config.app_secret and config.enforce_signature: + sig_header = request.headers.get("X-Hub-Signature-256", "") + if not cls._verify_signature(raw_body, sig_header, config.app_secret): + logger.warning("[WhatsApp] Rejected request with invalid HMAC signature") + raise HTTPException(status_code=401, detail="Invalid signature") + + # 2. Parse payload + try: + payload = await request.json() + except Exception: + return JSONResponse({"status": "invalid_json"}, status_code=400) + + # 3. Guard: Extract messages from Meta payload structure + messages, sender_name = cls._extract_messages(payload) + if not messages: + # Normal delivery status update (sent, delivered, read) - acknowledge and ignore + return {"status": "ok"} + + for msg in messages: + wamid = msg.get("id") + sender_id = msg.get("from") + msg_type = msg.get("type") + + if not sender_id or not wamid: + continue + + # 4. Idempotency check: drop duplicate webhooks from Meta retries + if cls._store.is_message_processed(wamid): + logger.info(f"[WhatsApp] Duplicate message {wamid} ignored") + continue + + cls._store.mark_message_processed(wamid) + + # 5. Route by message type + if msg_type == "text": + text_body = msg.get("text", {}).get("body", "").strip() + if text_body: + background_tasks.add_task( + cls._handle_user_turn, + sender_id=sender_id, + text=text_body, + wamid=wamid, + ) + else: + # Non-text media (audio, image, document, location, etc.) + background_tasks.add_task( + cls._handle_unsupported_media, + sender_id=sender_id, + msg_type=msg_type, + ) + + return {"status": "ok"} + + @app.get("/health") + async def health(): + """Health check endpoint.""" + return {"status": "healthy", "service": "sdialog-whatsapp"} + + cls._app = app + return app + + @classmethod + async def _handle_user_turn(cls, sender_id: str, text: str, wamid: str) -> None: + """Process incoming user utterance under a per-user lock and reply via WhatsApp.""" + user_lock = cls.get_user_lock(sender_id) + + async with user_lock: + try: + # Mark message as read (blue checkmark) if configured + if cls._config.mark_as_read and cls._client: + await cls._client.mark_as_read(wamid) + + # Check for explicit session reset commands + if text.lower() in ("/reset", "/clear", "/restart"): + cls._store.reset_session(sender_id) + await cls._client.send_text_message( + to=sender_id, + text="Your conversation has been reset. How can I help you?", + ) + return + + # Load user state from SQLite + state = cls._store.load_session( + sender_id, timeout_seconds=cls._config.session_timeout + ) + + # Reset agent memory and restore user session state + cls._agent.reset() + restore_agent_state(cls._agent, state) + + # Run inference in a threadpool so synchronous LLMs don't block the event loop + logger.debug(f"[WhatsApp] Processing turn for user {sender_id}: '{text[:30]}...'") + response = await asyncio.to_thread(cls._agent, text) + + if not response: + logger.warning(f"[WhatsApp] Agent returned empty response for {sender_id}") + return + + # Save updated user state to SQLite + new_state = dump_agent_state( + cls._agent, max_history_turns=cls._config.max_history_turns + ) + cls._store.save_session(sender_id, new_state) + + # Dispatch reply via Meta WhatsApp Cloud API + await cls._client.send_text_message(to=sender_id, text=response) + + except Exception as e: + logger.error(f"[WhatsApp] Error processing message for {sender_id}: {e}", exc_info=True) + try: + await cls._client.send_text_message( + to=sender_id, + text="I encountered a problem processing your request. Please try again in a moment.", + ) + except Exception: + pass + + @classmethod + async def _handle_unsupported_media(cls, sender_id: str, msg_type: str) -> None: + """Notify user that non-text media messages are currently unsupported.""" + logger.info(f"[WhatsApp] Received unsupported message type '{msg_type}' from {sender_id}") + if cls._client: + await cls._client.send_text_message( + to=sender_id, + text="I currently only support text messages. Please send your query as text.", + ) + + @staticmethod + def _verify_signature(raw_body: bytes, sig_header: str, app_secret: str) -> bool: + """Verify the X-Hub-Signature-256 header using HMAC-SHA256 in constant time.""" + if not sig_header or not sig_header.startswith("sha256="): + return False + expected_hash = sig_header[7:] + computed_hash = hmac.new( + app_secret.encode("utf-8"), + msg=raw_body, + digestmod=hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(computed_hash, expected_hash) + + @staticmethod + def _extract_messages(payload: Dict[str, Any]): + """Safely extract message objects and contact details from Meta webhook payload.""" + messages = [] + sender_name = None + try: + entry = payload.get("entry", [])[0] + change = entry.get("changes", [])[0].get("value", {}) + messages = change.get("messages", []) + contacts = change.get("contacts", []) + if contacts: + sender_name = contacts[0].get("profile", {}).get("name") + except (IndexError, KeyError, AttributeError): + pass + return messages, sender_name + + @classmethod + def serve(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> None: + """Start the WhatsApp webhook server with optional auto-discovery and webhook sync.""" + config = config or WhatsAppConfig() + + # 1. Auto-discover Phone Number ID if missing but access token is present + if not config.phone_number_id and config.access_token: + from .discovery import MetaDiscoveryClient + discovery = MetaDiscoveryClient(config.access_token, config.app_secret) + try: + config.phone_number_id = asyncio.run(discovery.discover_phone_number_id()) + except Exception as e: + logger.debug(f"[WhatsApp] Auto-discovery skipped or failed: {e}") + + config.validate(dry_run=False) + + # 2. Optional tunnel setup for local testing + public_url = None + if config.tunnel: + public_url = cls._start_tunnel(config) + + # 3. Auto-configure Meta webhooks if requested and tunnel/public URL is active + if config.auto_setup and public_url and config.app_secret: + from .discovery import MetaDiscoveryClient + discovery = MetaDiscoveryClient(config.access_token, config.app_secret) + callback_url = f"{public_url}{config.webhook_path}" + logger.info(f"[WhatsApp] Auto-configuring Meta webhooks to {callback_url}...") + try: + res = asyncio.run(discovery.configure_webhooks( + callback_url=callback_url, + verify_token=config.verify_token, + phone_number_id=config.phone_number_id + )) + logger.info(f"[WhatsApp] Auto-configuration results: {res}") + except Exception as e: + logger.warning(f"[WhatsApp] Auto-configuration encountered an error: {e}") + + app = cls.create_app(agent, config) + + logger.info(f"[WhatsApp] Starting server on {config.host}:{config.port}") + logger.info(f"[WhatsApp] Webhook endpoint: {config.webhook_path}") + logger.info(f"[WhatsApp] Verify Token: {config.verify_token}") + + uvicorn.run(app, host=config.host, port=config.port, log_level=config.log_level) + + @classmethod + def serve_in_thread(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> threading.Thread: + """Start the WhatsApp webhook server in a background daemon thread (useful for Jupyter / tests).""" + config = config or WhatsAppConfig() + app = cls.create_app(agent, config) + + def run_server(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + uv_config = uvicorn.Config( + app, + host=config.host, + port=config.port, + log_level=config.log_level, + loop="asyncio", + ) + server = uvicorn.Server(uv_config) + loop.run_until_complete(server.serve()) + + thread = threading.Thread(target=run_server, daemon=True) + thread.start() + logger.info(f"[WhatsApp] Threaded server started on {config.host}:{config.port}") + return thread + + @staticmethod + def _start_tunnel(config: WhatsAppConfig) -> Optional[str]: + """Attempt to spawn an ngrok tunnel if pyngrok is installed. Returns public URL.""" + try: + from pyngrok import ngrok + public_url = ngrok.connect(config.port, bind_tls=True).public_url + logger.info("=" * 60) + logger.info(f"[WhatsApp] Public Tunnel URL: {public_url}{config.webhook_path}") + logger.info(f"[WhatsApp] Meta Callback URL: {public_url}{config.webhook_path}") + logger.info(f"[WhatsApp] Meta Verify Token: {config.verify_token}") + logger.info("=" * 60) + return public_url + except ImportError: + logger.warning( + "[WhatsApp] tunnel=True requested, but 'pyngrok' is not installed. " + "Install it with `pip install pyngrok` or use a tool like ngrok/cloudflared manually." + ) + return None diff --git a/src/sdialog/whatsapp/session_store.py b/src/sdialog/whatsapp/session_store.py new file mode 100644 index 0000000..948592a --- /dev/null +++ b/src/sdialog/whatsapp/session_store.py @@ -0,0 +1,214 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +SQLite-backed multi-user session and idempotency store for WhatsApp. +""" + +import json +import logging +import sqlite3 +from datetime import datetime +from typing import Optional, Any, Dict, List + +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage + +logger = logging.getLogger(__name__) + +_TYPE_TO_MSG_CLASS = { + "HumanMessage": HumanMessage, + "AIMessage": AIMessage, + "SystemMessage": SystemMessage, +} + + +class SQLiteSessionStore: + """ + Thread-safe SQLite store for WhatsApp user sessions and webhook idempotency. + + Features: + - WAL mode for concurrent non-blocking reads/writes. + - Idempotency tracking to reject duplicate Meta webhook deliveries. + - Automatic session expiration after inactivity timeout. + """ + + def __init__(self, db_path: str = "whatsapp_sessions.db"): + self.db_path = db_path + self._init_db() + + def _get_connection(self) -> sqlite3.Connection: + """Create a new SQLite connection with WAL mode and busy timeout configured.""" + conn = sqlite3.connect(self.db_path, timeout=10.0) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode = WAL;") + conn.execute("PRAGMA busy_timeout = 5000;") + conn.execute("PRAGMA synchronous = NORMAL;") + return conn + + def _init_db(self) -> None: + """Initialize database tables and indexes.""" + with self._get_connection() as conn: + conn.executescript(""" + CREATE TABLE IF NOT EXISTS whatsapp_sessions ( + wa_id TEXT PRIMARY KEY, + state TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_active TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS processed_messages ( + wamid TEXT PRIMARY KEY, + received_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_processed_received + ON processed_messages(received_at); + """) + + def load_session(self, wa_id: str, timeout_seconds: int = 1800) -> Optional[Dict[str, Any]]: + """ + Load session state for a user. + + Returns None if session does not exist or has exceeded timeout_seconds of inactivity. + """ + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT state, last_active FROM whatsapp_sessions WHERE wa_id = ?", + (wa_id,) + ) + row = cursor.fetchone() + if not row: + return None + + state_json, last_active_str = row["state"], row["last_active"] + + # Parse last_active timestamp + try: + # Handle SQLite timestamp formats + if "T" in last_active_str: + last_active = datetime.fromisoformat(last_active_str) + else: + last_active = datetime.strptime(last_active_str, "%Y-%m-%d %H:%M:%S") + except Exception: + last_active = datetime.now() + + # Check timeout + elapsed = (datetime.now() - last_active).total_seconds() + if elapsed > timeout_seconds: + logger.info(f"[WhatsApp] Session for user {wa_id} expired ({elapsed:.0f}s > {timeout_seconds}s)") + self.reset_session(wa_id) + return None + + try: + return json.loads(state_json) + except json.JSONDecodeError: + logger.warning(f"[WhatsApp] Corrupted session JSON for {wa_id}, resetting") + self.reset_session(wa_id) + return None + + def save_session(self, wa_id: str, state: Dict[str, Any]) -> None: + """Save or update the session state for a user with the current timestamp.""" + state_json = json.dumps(state) + now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO whatsapp_sessions (wa_id, state, last_active) + VALUES (?, ?, ?) + ON CONFLICT(wa_id) DO UPDATE SET + state = excluded.state, + last_active = excluded.last_active; + """, + (wa_id, state_json, now_str) + ) + + def reset_session(self, wa_id: str) -> None: + """Reset and remove session state for a user.""" + with self._get_connection() as conn: + conn.execute("DELETE FROM whatsapp_sessions WHERE wa_id = ?", (wa_id,)) + logger.info(f"[WhatsApp] Session reset for {wa_id}") + + def is_message_processed(self, wamid: str) -> bool: + """Check whether a message ID has already been received and processed.""" + if not wamid: + return False + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT 1 FROM processed_messages WHERE wamid = ?", + (wamid,) + ) + return cursor.fetchone() is not None + + def mark_message_processed(self, wamid: str) -> None: + """Record a message ID in the idempotency table.""" + if not wamid: + return + now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with self._get_connection() as conn: + conn.execute( + "INSERT OR IGNORE INTO processed_messages (wamid, received_at) VALUES (?, ?)", + (wamid, now_str) + ) + + def cleanup_old_records(self, retention_days: int = 7) -> None: + """Purge idempotency records older than retention_days.""" + with self._get_connection() as conn: + conn.execute( + f"DELETE FROM processed_messages WHERE received_at < datetime('now', '-{retention_days} days')" + ) + + +# --------------------------------------------------------------------------- +# State Bridge: Supports both future Agent.get_state() and interim fallback +# --------------------------------------------------------------------------- + +def restore_agent_state(agent: Any, state: Optional[Dict[str, Any]]) -> None: + """Restore agent state. Prefers native agent.set_state() if present.""" + if not state: + return + + if hasattr(agent, "set_state") and callable(agent.set_state): + agent.set_state(state) + return + + # Fallback bridge before core SDialog implements Agent.set_state(): + memory_data = state.get("memory", []) + restored_msgs: List[BaseMessage] = [] + for item in memory_data: + msg_cls = _TYPE_TO_MSG_CLASS.get(item.get("type"), HumanMessage) + restored_msgs.append(msg_cls(content=item.get("content", ""))) + + if restored_msgs: + # Keep initial system prompt / persona prompt if not present in saved turns + if agent.memory and isinstance(agent.memory[0], SystemMessage): + agent.memory[:] = [agent.memory[0]] + restored_msgs + else: + agent.memory[:] = restored_msgs + + +def dump_agent_state(agent: Any, max_history_turns: int = 40) -> Dict[str, Any]: + """Serialize agent state. Prefers native agent.get_state() if present.""" + if hasattr(agent, "get_state") and callable(agent.get_state): + return agent.get_state() + + # Fallback bridge before core SDialog implements Agent.get_state(): + memory_items = [] + # Skip the primary system message (turn 0) in the dynamic history + history = agent.memory[1:] if (agent.memory and isinstance(agent.memory[0], SystemMessage)) else agent.memory + + # Apply sliding window budget + if len(history) > max_history_turns: + history = history[-max_history_turns:] + + for msg in history: + memory_items.append({ + "type": msg.__class__.__name__, + "content": msg.content + }) + + return { + "version": 1, + "memory": memory_items, + "finished": getattr(agent, "_finished", False) + } diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py new file mode 100644 index 0000000..7a31ee8 --- /dev/null +++ b/tests/test_whatsapp.py @@ -0,0 +1,514 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Unit and integration tests for SDialog WhatsApp server. +""" + +import hmac +import hashlib +import json +import os +import tempfile +import time +import pytest +from unittest.mock import AsyncMock, patch, MagicMock + +from fastapi.testclient import TestClient +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage + +from sdialog.whatsapp import ( + WhatsAppConfig, + WhatsAppServer, + SQLiteSessionStore, + MetaWhatsAppClient, +) +from sdialog.whatsapp.session_store import restore_agent_state, dump_agent_state + + +class MockWhatsAppAgent: + """Mock agent for WhatsApp testing.""" + + def __init__(self, name="TestBot"): + self.name = name + self.memory = [SystemMessage(content="You are a helpful test bot.")] + self.call_count = 0 + self.last_input = None + + def __call__(self, message: str, **kwargs) -> str: + self.call_count += 1 + self.last_input = message + self.memory.append(HumanMessage(content=message)) + reply = f"Bot reply to: {message}" + self.memory.append(AIMessage(content=reply)) + return reply + + def reset(self, seed=None): + self.memory = [SystemMessage(content="You are a helpful test bot.")] + self.call_count = 0 + + +@pytest.fixture +def temp_db(): + """Create a temporary SQLite database file for testing.""" + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + yield path + if os.path.exists(path): + os.remove(path) + + +@pytest.fixture +def test_config(temp_db): + """WhatsAppConfig configured for testing.""" + return WhatsAppConfig( + port=1234, + phone_number_id="1234567890", + access_token="test_access_token", + verify_token="test_verify_token", + app_secret="test_app_secret", + db_path=temp_db, + session_timeout=3600, + enforce_signature=True, + ) + + +# --------------------------------------------------------------------------- +# Test 1: Config +# --------------------------------------------------------------------------- + +def test_whatsapp_config_defaults(): + config = WhatsAppConfig(verify_token="custom_token") + assert config.port == 1234 + assert config.verify_token == "custom_token" + assert config.session_timeout == 1800 + assert config.max_history_turns == 40 + + +def test_whatsapp_config_validation(): + config = WhatsAppConfig(phone_number_id=None, access_token=None) + with pytest.raises(ValueError, match="phone_number_id is required"): + config.validate(dry_run=False) + + config = WhatsAppConfig(phone_number_id="123", access_token=None) + with pytest.raises(ValueError, match="access_token is required"): + config.validate(dry_run=False) + + # Valid config + config = WhatsAppConfig(phone_number_id="123", access_token="token") + config.validate(dry_run=False) # Should not raise + + +# --------------------------------------------------------------------------- +# Test 2: SQLiteSessionStore +# --------------------------------------------------------------------------- + +def test_session_store_save_load(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wa_id = "15551234567" + + # Initially empty + assert store.load_session(wa_id) is None + + # Save and reload + state = {"version": 1, "memory": [{"type": "HumanMessage", "content": "Hello"}]} + store.save_session(wa_id, state) + + loaded = store.load_session(wa_id) + assert loaded is not None + assert loaded["memory"][0]["content"] == "Hello" + + # Reset + store.reset_session(wa_id) + assert store.load_session(wa_id) is None + + +def test_session_store_timeout_expiration(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wa_id = "15551234567" + + state = {"version": 1, "memory": []} + store.save_session(wa_id, state) + + # With timeout_seconds=0, it should be expired immediately + time.sleep(0.01) + expired = store.load_session(wa_id, timeout_seconds=0) + assert expired is None + + +def test_session_store_idempotency(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wamid = "wamid.HBgLMTIzNDU2" + + assert not store.is_message_processed(wamid) + store.mark_message_processed(wamid) + assert store.is_message_processed(wamid) + + # Idempotent re-marking + store.mark_message_processed(wamid) + assert store.is_message_processed(wamid) + + +def test_state_bridge_restore_and_dump(): + agent = MockWhatsAppAgent() + agent.memory.append(HumanMessage(content="User 1")) + agent.memory.append(AIMessage(content="Bot 1")) + + # Dump state + state = dump_agent_state(agent, max_history_turns=10) + assert len(state["memory"]) == 2 + assert state["memory"][0]["content"] == "User 1" + assert state["memory"][1]["content"] == "Bot 1" + + # Restore on a fresh agent + new_agent = MockWhatsAppAgent() + restore_agent_state(new_agent, state) + assert len(new_agent.memory) == 3 # SystemMessage + 2 turns + assert isinstance(new_agent.memory[1], HumanMessage) + assert new_agent.memory[1].content == "User 1" + + +# --------------------------------------------------------------------------- +# Test 3: MetaWhatsAppClient +# --------------------------------------------------------------------------- + +def test_client_text_chunking(): + # Verify chunking does not break words prematurely + short_text = "Hello world" + assert MetaWhatsAppClient._chunk_text(short_text, 100) == [short_text] + + long_text = "First paragraph.\n\nSecond paragraph." + chunks = MetaWhatsAppClient._chunk_text(long_text, 20) + assert len(chunks) == 2 + assert chunks[0] == "First paragraph." + assert chunks[1] == "Second paragraph." + + +def test_client_send_text_mocked(test_config): + import asyncio + client = MetaWhatsAppClient(test_config) + + async def _test(): + with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_post.return_value = mock_response + + success = await client.send_text_message("15551234567", "Hello from test!") + assert success is True + assert mock_post.called + + called_url = mock_post.call_args[0][0] + assert "1234567890/messages" in called_url + + asyncio.run(_test()) + + +# --------------------------------------------------------------------------- +# Test 4: Webhook Verification (GET /webhook) +# --------------------------------------------------------------------------- + +def test_webhook_verification_success(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + response = client.get( + "/webhook", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "test_verify_token", + "hub.challenge": "test_challenge_code_123", + }, + ) + assert response.status_code == 200 + assert response.text == "test_challenge_code_123" + + +def test_webhook_verification_failure(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + response = client.get( + "/webhook", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "wrong_token", + "hub.challenge": "challenge", + }, + ) + assert response.status_code == 403 + + +# --------------------------------------------------------------------------- +# Test 5: Inbound Messages & HMAC (POST /webhook) +# --------------------------------------------------------------------------- + +def _create_meta_payload(sender="15551234567", text="Hello agent", wamid="wamid.123"): + """Helper to create a standard Meta WhatsApp webhook payload.""" + return { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "123456", + "changes": [ + { + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "metadata": { + "display_phone_number": "15550001", + "phone_number_id": "1234567890", + }, + "contacts": [ + { + "profile": {"name": "Alice"}, + "wa_id": sender, + } + ], + "messages": [ + { + "from": sender, + "id": wamid, + "timestamp": "1700000000", + "type": "text", + "text": {"body": text}, + } + ], + }, + } + ], + } + ], + } + + +def _compute_hmac(body: bytes, secret: str) -> str: + """Helper to compute Meta X-Hub-Signature-256 header value.""" + digest = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + return f"sha256={digest}" + + +def test_webhook_hmac_enforcement(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + payload = json.dumps(_create_meta_payload()).encode() + + # 1. Missing signature + resp = client.post("/webhook", content=payload, headers={"Content-Type": "application/json"}) + assert resp.status_code == 401 + + # 2. Invalid signature + resp = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": "sha256=invalid"}, + ) + assert resp.status_code == 401 + + # 3. Valid signature + valid_sig = _compute_hmac(payload, test_config.app_secret) + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + resp = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": valid_sig}, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "ok"} + + +def test_webhook_status_receipt_ignored(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + # Status receipt payload (e.g. message delivered) + status_payload = { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "123456", + "changes": [ + { + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "statuses": [{"id": "wamid.123", "status": "delivered"}], + }, + } + ], + } + ], + } + raw = json.dumps(status_payload).encode() + sig = _compute_hmac(raw, test_config.app_secret) + + resp = client.post( + "/webhook", + content=raw, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "ok"} + assert agent.call_count == 0 + + +def test_webhook_idempotency_duplicate_dropped(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + payload = json.dumps(_create_meta_payload(wamid="wamid.repeat123")).encode() + sig = _compute_hmac(payload, test_config.app_secret) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + # First delivery: processed + resp1 = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp1.status_code == 200 + + # Second delivery (retry with same wamid): dropped + resp2 = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp2.status_code == 200 + assert agent.call_count == 1 # Agent called only once! + + +def test_webhook_multi_user_isolation(test_config): + """Verify that User A and User B have separate, isolated conversation histories.""" + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + # User A speaks + pA = json.dumps(_create_meta_payload(sender="1111", text="Hello from A", wamid="wamid.A1")).encode() + client.post( + "/webhook", + content=pA, + headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _compute_hmac(pA, test_config.app_secret), + }, + ) + + # User B speaks + pB = json.dumps(_create_meta_payload(sender="2222", text="Hello from B", wamid="wamid.B1")).encode() + client.post( + "/webhook", + content=pB, + headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _compute_hmac(pB, test_config.app_secret), + }, + ) + + # Check SQLite store + store = SQLiteSessionStore(db_path=test_config.db_path) + stateA = store.load_session("1111") + stateB = store.load_session("2222") + + assert stateA is not None + assert stateB is not None + + # Verify User A's history contains User A's message, not User B's + contentsA = [m["content"] for m in stateA["memory"]] + contentsB = [m["content"] for m in stateB["memory"]] + + assert "Hello from A" in contentsA + assert "Hello from B" not in contentsA + + assert "Hello from B" in contentsB + assert "Hello from A" not in contentsB + + +def test_agent_serve_whatsapp_method_signature(): + """Verify serve_whatsapp is present on Agent and Server.""" + from sdialog.agents import Agent + from sdialog.server import Server + + assert hasattr(Agent, "serve_whatsapp") + assert callable(Agent.serve_whatsapp) + + assert hasattr(Server, "serve_whatsapp") + assert callable(Server.serve_whatsapp) + + +# --------------------------------------------------------------------------- +# Test 6: Auto-Discovery & Webhook Sync (adapted from WAVoice) +# --------------------------------------------------------------------------- + +def test_meta_discovery_inspect_and_discover(): + import asyncio + from sdialog.whatsapp.discovery import MetaDiscoveryClient + + client = MetaDiscoveryClient(access_token="test_token", app_secret="test_secret") + + async def _test(): + # Mock inspect_token + mock_token_data = { + "app_id": "123456", + "granular_scopes": [ + { + "scope": "whatsapp_business_management", + "target_ids": ["waba_101"], + } + ], + } + + with patch.object(MetaDiscoveryClient, "inspect_token", new_callable=AsyncMock) as mock_inspect: + mock_inspect.return_value = mock_token_data + + waba_id = await client.discover_waba_id() + assert waba_id == "waba_101" + + # Mock list_phone_numbers + with patch.object(MetaDiscoveryClient, "list_phone_numbers", new_callable=AsyncMock) as mock_list: + mock_list.return_value = [{"id": "phone_999", "display_phone_number": "+1555"}] + + phone_id = await client.discover_phone_number_id() + assert phone_id == "phone_999" + + asyncio.run(_test()) + + +def test_meta_discovery_configure_webhooks_mocked(): + import asyncio + from sdialog.whatsapp.discovery import MetaDiscoveryClient + + client = MetaDiscoveryClient(access_token="test_token", app_secret="test_secret") + + async def _test(): + with patch.object(MetaDiscoveryClient, "inspect_token", new_callable=AsyncMock) as mock_inspect: + mock_inspect.return_value = {"app_id": "app_555"} + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"success": True} + mock_post.return_value = mock_resp + + results = await client.configure_webhooks( + callback_url="https://test.ngrok.app/webhook", + verify_token="test_secret", + phone_number_id="phone_777", + waba_id="waba_888", + ) + + assert results["app_level"] is True + assert results["waba_level"] is True + assert results["phone_level"] is True + assert mock_post.call_count == 3 # All 3 tiers synced + + asyncio.run(_test()) From 9fc6c8837ba963e962486d4cac9aa24e3e3dff5f Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:06:01 +0100 Subject: [PATCH 02/10] [sdialog] Add Agent.get_state()/set_state() for conversation persistence Serializable snapshot of an agent's conversation state (memory without the system prompt, finished flag, and orchestrator state) that can be restored on a fresh or reset agent. Uses langchain's message dict serialization so tool calls and ToolMessages round-trip intact. Orchestrators get default no-op get_state()/set_state() hooks; ChangeMindOrchestrator persists its trigger counter so max_times is respected across save/restore. --- src/sdialog/agents.py | 68 ++++++++++++++++++++++++++- src/sdialog/orchestrators/__init__.py | 20 ++++++++ src/sdialog/orchestrators/base.py | 22 +++++++++ tests/test_agents.py | 51 ++++++++++++++++++++ 4 files changed, 160 insertions(+), 1 deletion(-) diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index 954e6e4..b09050a 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -21,7 +21,7 @@ from langchain_core.tools import tool from langchain_core.language_models.base import BaseLanguageModel from langchain_core.messages.base import BaseMessage, messages_to_dict -from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage +from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage, messages_from_dict from .config import config from jinja2 import Template @@ -1050,5 +1050,71 @@ def memory_dump(self, as_dict: bool = False) -> list: """ return messages_to_dict(self.memory) if as_dict else self.memory.copy() + def get_state(self) -> dict: + """ + Returns a JSON-serializable snapshot of the agent's conversation state, which can be restored later + (e.g. on a fresh or reset agent) with :func:`set_state`. + + The snapshot includes the conversation memory (excluding the system prompt, which is part of the agent's + configuration), the finished flag, and the internal state of attached orchestrators. It does not include + the persona, model, or other configuration (see :func:`json` and :func:`clone` for that). + + Example: + + .. code-block:: python + + import json + from sdialog.agents import Agent + + bot = Agent(name="Bot") + bot("Hi, my name is Alice") + + saved = json.dumps(bot.get_state()) + + # Later, e.g. after a restart, continue the same conversation + bot.reset() + bot.set_state(json.loads(saved)) + bot("Do you remember my name?") + + :return: Serializable state dictionary. + :rtype: dict + """ + memory = self._memory[1:] if self._system_prompt is not None else self._memory + return { + "version": 1, + "memory": messages_to_dict(memory), + "finished": self._finished, + "orchestrators": [orchestrator.get_state() for orchestrator in self._orchestrators or []] + } + + def set_state(self, state: dict): + """ + Restores a conversation state previously obtained with :func:`get_state`, replacing the current + conversation memory while keeping the agent's current system prompt. + + Orchestrator states are restored by position; if the number of attached orchestrators differs from the + saved state, orchestrator states are not restored and a warning is logged. + + :param state: State dictionary as returned by :func:`get_state`. + :type state: dict + :raises ValueError: If the state version is not supported. + """ + if state.get("version") != 1: + raise ValueError(f"Unsupported agent state version: {state.get('version')}") + + system_prompt = self._memory[:1] if self._system_prompt is not None else [] + self._memory[:] = system_prompt + messages_from_dict(state.get("memory", [])) + self._stateless_memory = None + self._finished = state.get("finished", False) + + orchestrators = self._orchestrators or [] + orchestrator_states = state.get("orchestrators", []) + if len(orchestrators) == len(orchestrator_states): + for orchestrator, orchestrator_state in zip(orchestrators, orchestrator_states): + orchestrator.set_state(orchestrator_state) + else: + logger.warning(f"Agent has {len(orchestrators)} orchestrator(s) but the state has " + f"{len(orchestrator_states)}; orchestrator states were not restored.") + #: Alias for :func:`Agent.dialog_with`. talk_with = dialog_with diff --git a/src/sdialog/orchestrators/__init__.py b/src/sdialog/orchestrators/__init__.py index 6310996..43f44c7 100644 --- a/src/sdialog/orchestrators/__init__.py +++ b/src/sdialog/orchestrators/__init__.py @@ -187,6 +187,26 @@ def reset(self): """ self.times = 0 + def get_state(self) -> dict: + """ + Return the internal counter of times triggered. + + :return: State dictionary. + :rtype: dict + """ + return {"times": self.times} + + def set_state(self, state: dict): + """ + Restore the internal counter of times triggered. + + :param state: State dictionary as returned by :func:`get_state`. + :type state: dict + :return: None + :rtype: None + """ + self.times = state.get("times", 0) + def instruct(self, dialog: List[Turn], utterance: str) -> str: """ Possibly return a mind-change instruction based on probability and remaining allowance. diff --git a/src/sdialog/orchestrators/base.py b/src/sdialog/orchestrators/base.py index a4a7776..a7941aa 100644 --- a/src/sdialog/orchestrators/base.py +++ b/src/sdialog/orchestrators/base.py @@ -188,6 +188,28 @@ def reset(self): """ pass + def get_state(self) -> dict: + """ + Return a JSON-serializable snapshot of the internal state (overridden in stateful orchestrators). + Used by :func:`sdialog.agents.Agent.get_state`. + + :return: State dictionary (empty for stateless orchestrators). + :rtype: dict + """ + return {} + + def set_state(self, state: dict): + """ + Restore internal state from :func:`get_state` output (overridden in stateful orchestrators). + Used by :func:`sdialog.agents.Agent.set_state`. + + :param state: State dictionary as returned by :func:`get_state`. + :type state: dict + :return: None + :rtype: None + """ + pass + class BasePersistentOrchestrator(BaseOrchestrator): """ diff --git a/tests/test_agents.py b/tests/test_agents.py index 0ef263a..63f4a9f 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -120,3 +120,54 @@ def invoke(self, tool_call): assert any(message.__class__.__name__ == "ToolMessage" and "RuntimeError: boom" in message.content for message in agent.llm.calls[1]) assert any(event.action == "tool" and event.actionLabel == "output" for event in events) + + +def test_agent_get_set_state_roundtrip(): + import json + from langchain_core.messages import HumanMessage, AIMessage, ToolMessage + from sdialog.orchestrators import ChangeMindOrchestrator + + agent = Agent(persona=Persona(name="A"), name="A", model=DummyLLM()) | ChangeMindOrchestrator(max_times=1) + system_prompt = agent.memory[0].content + agent.memory.extend([ + HumanMessage(content="What's the weather?"), + AIMessage(content="", tool_calls=[{"name": "weather", "args": {"city": "Martigny"}, "id": "call-1"}]), + ToolMessage(content="Sunny", name="weather", tool_call_id="call-1"), + AIMessage(content="It's sunny."), + ]) + agent._orchestrators[0].times = 1 + agent._finished = True + + state = json.loads(json.dumps(agent.get_state())) # must be JSON-serializable + assert state["version"] == 1 + assert len(state["memory"]) == 4 # system prompt excluded + + agent.reset() + assert len(agent.memory) == 1 and agent._orchestrators[0].times == 0 and not agent._finished + + agent.set_state(state) + assert agent.memory[0].content == system_prompt + assert [type(m) for m in agent.memory[1:]] == [HumanMessage, AIMessage, ToolMessage, AIMessage] + assert agent.memory[2].tool_calls[0]["args"] == {"city": "Martigny"} + assert agent.memory[3].tool_call_id == "call-1" + assert agent._orchestrators[0].times == 1 + assert agent._finished + + +def test_agent_set_state_keeps_current_system_prompt_and_handles_mismatch(caplog): + import pytest + from langchain_core.messages import HumanMessage + from sdialog.orchestrators import ChangeMindOrchestrator + + source = Agent(persona=Persona(name="A"), name="A", model=DummyLLM()) | ChangeMindOrchestrator() + source.memory.append(HumanMessage(content="Hello")) + state = source.get_state() + + target = Agent(persona=Persona(name="B"), name="B", model=DummyLLM()) + target.set_state(state) + assert target.memory[0].content == target.prompt() + assert target.memory[1].content == "Hello" + assert "orchestrator states were not restored" in caplog.text + + with pytest.raises(ValueError): + target.set_state({"version": 99, "memory": []}) From 780f9391f27ed87625cd6baa8122db5f92d3018f Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:13:58 +0100 Subject: [PATCH 03/10] [sdialog] Fix cross-user memory leak in WhatsApp server A single Agent instance serves all WhatsApp users by swapping each user's state in and out of its memory, but only per-user locks were used, so concurrent users could overwrite each other's memory mid-turn (user B receiving replies based on user A's messages, and histories saved under the wrong user). - Serialize turns with a single FIFO agent lock (created lazily so it binds to the server's event loop); mark-as-read stays outside it - Persist sessions with Agent.get_state()/set_state() and drop the fallback serializer, which lost tool calls and turned ToolMessages into HumanMessages - Start a new conversation after the agent finishes one, instead of silently ignoring the user - Trim stored history at a user-message boundary so tool calls are never separated from their results - Add regression tests for concurrent users and finished sessions --- src/sdialog/whatsapp/server.py | 65 +++++++++++------- src/sdialog/whatsapp/session_store.py | 65 +----------------- tests/test_whatsapp.py | 98 ++++++++++++++++++++------- 3 files changed, 115 insertions(+), 113 deletions(-) diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index a6fa958..b1b16d6 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -11,7 +11,7 @@ import asyncio import logging import threading -from typing import Optional, Dict, Any +from typing import Optional, Dict, Any, List from fastapi import FastAPI, Request, Response, BackgroundTasks, HTTPException, Query from fastapi.responses import JSONResponse @@ -19,11 +19,7 @@ from .config import WhatsAppConfig from .client import MetaWhatsAppClient -from .session_store import ( - SQLiteSessionStore, - restore_agent_state, - dump_agent_state, -) +from .session_store import SQLiteSessionStore logger = logging.getLogger(__name__) @@ -38,14 +34,20 @@ class WhatsAppServer: _agent: Any = None _store: Optional[SQLiteSessionStore] = None _client: Optional[MetaWhatsAppClient] = None - _user_locks: Dict[str, asyncio.Lock] = {} + _agent_lock: Optional[asyncio.Lock] = None @classmethod - def get_user_lock(cls, wa_id: str) -> asyncio.Lock: - """Get or create an async lock for a specific WhatsApp sender ID.""" - if wa_id not in cls._user_locks: - cls._user_locks[wa_id] = asyncio.Lock() - return cls._user_locks[wa_id] + def get_agent_lock(cls) -> asyncio.Lock: + """ + Get the lock guarding the shared agent. + + A single agent instance serves all users, swapping each user's state in and out of its memory, + so turns must be processed one at a time. The lock is FIFO, so messages from the same user are + still processed in arrival order. Created lazily so it binds to the server's event loop. + """ + if cls._agent_lock is None: + cls._agent_lock = asyncio.Lock() + return cls._agent_lock @classmethod def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: @@ -54,7 +56,7 @@ def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: cls._config = config cls._store = SQLiteSessionStore(db_path=config.db_path) cls._client = MetaWhatsAppClient(config) - cls._user_locks.clear() + cls._agent_lock = None app = FastAPI( title="SDialog WhatsApp Agent Server", @@ -149,15 +151,13 @@ async def health(): @classmethod async def _handle_user_turn(cls, sender_id: str, text: str, wamid: str) -> None: - """Process incoming user utterance under a per-user lock and reply via WhatsApp.""" - user_lock = cls.get_user_lock(sender_id) + """Process incoming user utterance under the agent lock and reply via WhatsApp.""" + # Mark message as read (blue checkmark) if configured + if cls._config.mark_as_read and cls._client: + await cls._client.mark_as_read(wamid) - async with user_lock: + async with cls.get_agent_lock(): try: - # Mark message as read (blue checkmark) if configured - if cls._config.mark_as_read and cls._client: - await cls._client.mark_as_read(wamid) - # Check for explicit session reset commands if text.lower() in ("/reset", "/clear", "/restart"): cls._store.reset_session(sender_id) @@ -172,9 +172,14 @@ async def _handle_user_turn(cls, sender_id: str, text: str, wamid: str) -> None: sender_id, timeout_seconds=cls._config.session_timeout ) + # A finished conversation (agent said goodbye) starts over on the next message + if state and state.get("finished"): + state = None + # Reset agent memory and restore user session state cls._agent.reset() - restore_agent_state(cls._agent, state) + if state: + cls._agent.set_state(state) # Run inference in a threadpool so synchronous LLMs don't block the event loop logger.debug(f"[WhatsApp] Processing turn for user {sender_id}: '{text[:30]}...'") @@ -185,9 +190,8 @@ async def _handle_user_turn(cls, sender_id: str, text: str, wamid: str) -> None: return # Save updated user state to SQLite - new_state = dump_agent_state( - cls._agent, max_history_turns=cls._config.max_history_turns - ) + new_state = cls._agent.get_state() + new_state["memory"] = cls._trim_memory(new_state["memory"], cls._config.max_history_turns) cls._store.save_session(sender_id, new_state) # Dispatch reply via Meta WhatsApp Cloud API @@ -213,6 +217,19 @@ async def _handle_unsupported_media(cls, sender_id: str, msg_type: str) -> None: text="I currently only support text messages. Please send your query as text.", ) + @staticmethod + def _trim_memory(memory: List[Dict[str, Any]], max_messages: int) -> List[Dict[str, Any]]: + """ + Keep at most the last ``max_messages`` serialized messages, starting at a user message + so that tool calls are never separated from their tool results. + """ + if len(memory) <= max_messages: + return memory + memory = memory[-max_messages:] + while memory and memory[0].get("type") != "human": + memory = memory[1:] + return memory + @staticmethod def _verify_signature(raw_body: bytes, sig_header: str, app_secret: str) -> bool: """Verify the X-Hub-Signature-256 header using HMAC-SHA256 in constant time.""" diff --git a/src/sdialog/whatsapp/session_store.py b/src/sdialog/whatsapp/session_store.py index 948592a..c22efcf 100644 --- a/src/sdialog/whatsapp/session_store.py +++ b/src/sdialog/whatsapp/session_store.py @@ -10,18 +10,10 @@ import logging import sqlite3 from datetime import datetime -from typing import Optional, Any, Dict, List - -from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage +from typing import Optional, Any, Dict logger = logging.getLogger(__name__) -_TYPE_TO_MSG_CLASS = { - "HumanMessage": HumanMessage, - "AIMessage": AIMessage, - "SystemMessage": SystemMessage, -} - class SQLiteSessionStore: """ @@ -157,58 +149,3 @@ def cleanup_old_records(self, retention_days: int = 7) -> None: conn.execute( f"DELETE FROM processed_messages WHERE received_at < datetime('now', '-{retention_days} days')" ) - - -# --------------------------------------------------------------------------- -# State Bridge: Supports both future Agent.get_state() and interim fallback -# --------------------------------------------------------------------------- - -def restore_agent_state(agent: Any, state: Optional[Dict[str, Any]]) -> None: - """Restore agent state. Prefers native agent.set_state() if present.""" - if not state: - return - - if hasattr(agent, "set_state") and callable(agent.set_state): - agent.set_state(state) - return - - # Fallback bridge before core SDialog implements Agent.set_state(): - memory_data = state.get("memory", []) - restored_msgs: List[BaseMessage] = [] - for item in memory_data: - msg_cls = _TYPE_TO_MSG_CLASS.get(item.get("type"), HumanMessage) - restored_msgs.append(msg_cls(content=item.get("content", ""))) - - if restored_msgs: - # Keep initial system prompt / persona prompt if not present in saved turns - if agent.memory and isinstance(agent.memory[0], SystemMessage): - agent.memory[:] = [agent.memory[0]] + restored_msgs - else: - agent.memory[:] = restored_msgs - - -def dump_agent_state(agent: Any, max_history_turns: int = 40) -> Dict[str, Any]: - """Serialize agent state. Prefers native agent.get_state() if present.""" - if hasattr(agent, "get_state") and callable(agent.get_state): - return agent.get_state() - - # Fallback bridge before core SDialog implements Agent.get_state(): - memory_items = [] - # Skip the primary system message (turn 0) in the dynamic history - history = agent.memory[1:] if (agent.memory and isinstance(agent.memory[0], SystemMessage)) else agent.memory - - # Apply sliding window budget - if len(history) > max_history_turns: - history = history[-max_history_turns:] - - for msg in history: - memory_items.append({ - "type": msg.__class__.__name__, - "content": msg.content - }) - - return { - "version": 1, - "memory": memory_items, - "finished": getattr(agent, "_finished", False) - } diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 7a31ee8..770ce39 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -16,7 +16,7 @@ from unittest.mock import AsyncMock, patch, MagicMock from fastapi.testclient import TestClient -from langchain_core.messages import HumanMessage, AIMessage, SystemMessage +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, messages_to_dict, messages_from_dict from sdialog.whatsapp import ( WhatsAppConfig, @@ -24,29 +24,40 @@ SQLiteSessionStore, MetaWhatsAppClient, ) -from sdialog.whatsapp.session_store import restore_agent_state, dump_agent_state class MockWhatsAppAgent: - """Mock agent for WhatsApp testing.""" + """Mock agent for WhatsApp testing (mirrors Agent's in-place memory reset and state API).""" - def __init__(self, name="TestBot"): + def __init__(self, name="TestBot", latency=0.0): self.name = name + self.latency = latency self.memory = [SystemMessage(content="You are a helpful test bot.")] self.call_count = 0 self.last_input = None + self._finished = False def __call__(self, message: str, **kwargs) -> str: self.call_count += 1 self.last_input = message self.memory.append(HumanMessage(content=message)) - reply = f"Bot reply to: {message}" + time.sleep(self.latency) + seen = [m.content for m in self.memory if isinstance(m, HumanMessage)] + reply = f"Bot reply to: {' | '.join(seen)}" self.memory.append(AIMessage(content=reply)) return reply def reset(self, seed=None): - self.memory = [SystemMessage(content="You are a helpful test bot.")] - self.call_count = 0 + self.memory[:] = self.memory[:1] + self._finished = False + + def get_state(self): + return {"version": 1, "memory": messages_to_dict(self.memory[1:]), + "finished": self._finished, "orchestrators": []} + + def set_state(self, state): + self.memory[:] = self.memory[:1] + messages_from_dict(state["memory"]) + self._finished = state["finished"] @pytest.fixture @@ -150,23 +161,13 @@ def test_session_store_idempotency(temp_db): assert store.is_message_processed(wamid) -def test_state_bridge_restore_and_dump(): - agent = MockWhatsAppAgent() - agent.memory.append(HumanMessage(content="User 1")) - agent.memory.append(AIMessage(content="Bot 1")) - - # Dump state - state = dump_agent_state(agent, max_history_turns=10) - assert len(state["memory"]) == 2 - assert state["memory"][0]["content"] == "User 1" - assert state["memory"][1]["content"] == "Bot 1" +def test_trim_memory_starts_at_user_message(): + memory = [{"type": t} for t in ["human", "ai", "human", "ai", "tool", "ai"]] - # Restore on a fresh agent - new_agent = MockWhatsAppAgent() - restore_agent_state(new_agent, state) - assert len(new_agent.memory) == 3 # SystemMessage + 2 turns - assert isinstance(new_agent.memory[1], HumanMessage) - assert new_agent.memory[1].content == "User 1" + assert WhatsAppServer._trim_memory(memory, 10) == memory + # The last 5 would start with an AI reply, so trimming skips ahead to the next user message + assert WhatsAppServer._trim_memory(memory, 5) == memory[2:] + assert WhatsAppServer._trim_memory(memory, 3) == [] # --------------------------------------------------------------------------- @@ -423,8 +424,8 @@ def test_webhook_multi_user_isolation(test_config): assert stateB is not None # Verify User A's history contains User A's message, not User B's - contentsA = [m["content"] for m in stateA["memory"]] - contentsB = [m["content"] for m in stateB["memory"]] + contentsA = [m["data"]["content"] for m in stateA["memory"]] + contentsB = [m["data"]["content"] for m in stateB["memory"]] assert "Hello from A" in contentsA assert "Hello from B" not in contentsA @@ -433,6 +434,53 @@ def test_webhook_multi_user_isolation(test_config): assert "Hello from A" not in contentsB +def test_concurrent_users_do_not_share_memory(test_config): + """Two users messaging at the same time must not see each other's conversation (shared agent).""" + import asyncio + + WhatsAppServer.create_app(MockWhatsAppAgent(latency=0.3), test_config) + sent = {} + + async def fake_send(to, text): + sent[to] = text + + async def _test(): + async def user(wa_id, text, delay): + await asyncio.sleep(delay) + await WhatsAppServer._handle_user_turn(sender_id=wa_id, text=text, wamid=f"wamid.{wa_id}") + + await asyncio.gather(user("1111", "Alice secret", 0.0), user("2222", "Bob says hi", 0.1)) + + with patch.object(MetaWhatsAppClient, "send_text_message", side_effect=fake_send), \ + patch.object(MetaWhatsAppClient, "mark_as_read", new_callable=AsyncMock): + asyncio.run(_test()) + + assert sent == {"1111": "Bot reply to: Alice secret", "2222": "Bot reply to: Bob says hi"} + + store = SQLiteSessionStore(db_path=test_config.db_path) + assert [m["data"]["content"] for m in store.load_session("1111")["memory"]] == \ + ["Alice secret", "Bot reply to: Alice secret"] + assert [m["data"]["content"] for m in store.load_session("2222")["memory"]] == \ + ["Bob says hi", "Bot reply to: Bob says hi"] + + +def test_finished_session_starts_over(test_config): + """After the agent ends a conversation, the user's next message starts a new one.""" + import asyncio + + agent = MockWhatsAppAgent() + WhatsAppServer.create_app(agent, test_config) + store = SQLiteSessionStore(db_path=test_config.db_path) + store.save_session("1111", {"version": 1, "finished": True, "orchestrators": [], + "memory": messages_to_dict([HumanMessage(content="Bye"), AIMessage(content="Bye!")])}) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock) as mock_send, \ + patch.object(MetaWhatsAppClient, "mark_as_read", new_callable=AsyncMock): + asyncio.run(WhatsAppServer._handle_user_turn(sender_id="1111", text="Hi again", wamid="wamid.x")) + + mock_send.assert_awaited_once_with(to="1111", text="Bot reply to: Hi again") + + def test_agent_serve_whatsapp_method_signature(): """Verify serve_whatsapp is present on Agent and Server.""" from sdialog.agents import Agent From 641772ff9e5b6c010b7197d78ef9fde5a68e51dc Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:18:09 +0100 Subject: [PATCH 04/10] [sdialog] Make WhatsApp server secure by default - Require app_secret: webhook signatures were silently not checked when it was missing, letting anyone who knows the URL make the bot reply to arbitrary numbers. Opting out (enforce_signature=False) now logs a warning. Also validated for serve_in_thread() - Default auto_setup to False and warn when used: it overwrites the webhook URL of the business account and phone number - Replace the hard-coded default verify token with a random one, only logged when generated (it is needed for Meta's webhook settings); compare it in constant time - Fix env var fallbacks being ignored when serve_whatsapp() passes None for unset arguments (resolve them in __post_init__) Co-Authored-By: Claude Opus 5.5 --- src/sdialog/agents.py | 9 +++--- src/sdialog/server.py | 9 +++--- src/sdialog/whatsapp/config.py | 49 +++++++++++++++++++----------- src/sdialog/whatsapp/server.py | 18 +++++++---- tests/test_whatsapp.py | 55 +++++++++++++++++++++++++++++++--- 5 files changed, 105 insertions(+), 35 deletions(-) diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index b09050a..c5b6318 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -681,7 +681,7 @@ def serve_whatsapp(self, webhook_path: str = "/webhook", session_timeout: int = 1800, tunnel: bool = False, - auto_setup: bool = True, + auto_setup: bool = False, log_level: str = "info"): """ Starts a Meta WhatsApp Cloud API webhook server to interact with the agent. @@ -694,9 +694,9 @@ def serve_whatsapp(self, :type phone_number_id: Optional[str] :param access_token: Meta Graph API Access Token (or WHATSAPP_ACCESS_TOKEN env var). :type access_token: Optional[str] - :param verify_token: Webhook verification token (or WHATSAPP_VERIFY_TOKEN env var). + :param verify_token: Webhook verification token (or WHATSAPP_VERIFY_TOKEN env var; random if unset). :type verify_token: Optional[str] - :param app_secret: Meta App Secret for HMAC validation (or WHATSAPP_APP_SECRET env var). + :param app_secret: Meta App Secret for HMAC validation (or WHATSAPP_APP_SECRET env var). Required. :type app_secret: Optional[str] :param db_path: Path to the SQLite session database. :type db_path: str @@ -706,7 +706,8 @@ def serve_whatsapp(self, :type session_timeout: int :param tunnel: If True, attempts to spawn a public ngrok tunnel. :type tunnel: bool - :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp + Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool :param log_level: Logging level for the server. :type log_level: str diff --git a/src/sdialog/server.py b/src/sdialog/server.py index bf0420e..a4a74e9 100644 --- a/src/sdialog/server.py +++ b/src/sdialog/server.py @@ -331,7 +331,7 @@ def serve_whatsapp(cls, webhook_path: str = "/webhook", session_timeout: int = 1800, tunnel: bool = False, - auto_setup: bool = True, + auto_setup: bool = False, log_level: str = "info") -> None: """ Serve an SDialog agent via Meta WhatsApp Cloud API webhook. @@ -346,9 +346,9 @@ def serve_whatsapp(cls, :type phone_number_id: Optional[str] :param access_token: Meta Graph API Access Token. :type access_token: Optional[str] - :param verify_token: Webhook verification token. + :param verify_token: Webhook verification token (or WHATSAPP_VERIFY_TOKEN env var; random if unset). :type verify_token: Optional[str] - :param app_secret: Meta App Secret for HMAC validation. + :param app_secret: Meta App Secret for HMAC validation (or WHATSAPP_APP_SECRET env var). Required. :type app_secret: Optional[str] :param db_path: Path to the SQLite session database. :type db_path: str @@ -358,7 +358,8 @@ def serve_whatsapp(cls, :type session_timeout: int :param tunnel: If True, attempts to spawn a public ngrok tunnel. :type tunnel: bool - :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp + Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool :param log_level: Logging level for the server. :type log_level: str diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py index 9843fc0..4dd7d56 100644 --- a/src/sdialog/whatsapp/config.py +++ b/src/sdialog/whatsapp/config.py @@ -7,7 +7,8 @@ """ import os -from dataclasses import dataclass, field +import secrets +from dataclasses import dataclass from typing import Optional @@ -19,38 +20,50 @@ class WhatsAppConfig: Parameters will fall back to standard environment variables if not provided: - ``WHATSAPP_PHONE_NUMBER_ID`` - ``WHATSAPP_ACCESS_TOKEN`` - - ``WHATSAPP_VERIFY_TOKEN`` - - ``WHATSAPP_APP_SECRET`` + - ``WHATSAPP_VERIFY_TOKEN`` (a random token is generated if not set) + - ``WHATSAPP_APP_SECRET`` (required unless ``enforce_signature=False``) """ port: int = 1234 host: str = "0.0.0.0" - phone_number_id: Optional[str] = field( - default_factory=lambda: os.environ.get("WHATSAPP_PHONE_NUMBER_ID") - ) - access_token: Optional[str] = field( - default_factory=lambda: os.environ.get("WHATSAPP_ACCESS_TOKEN") or os.environ.get("WHATSAPP_TOKEN") - ) - verify_token: Optional[str] = field( - default_factory=lambda: os.environ.get("WHATSAPP_VERIFY_TOKEN") - or os.environ.get("WHATSAPP_WEBHOOK_VERIFY_TOKEN", "sdialog_whatsapp_secret") - ) - app_secret: Optional[str] = field( - default_factory=lambda: os.environ.get("WHATSAPP_APP_SECRET") - ) + phone_number_id: Optional[str] = None + access_token: Optional[str] = None + verify_token: Optional[str] = None + app_secret: Optional[str] = None api_version: str = "v21.0" db_path: str = "whatsapp_sessions.db" webhook_path: str = "/webhook" session_timeout: int = 1800 # 30 minutes idle timeout tunnel: bool = False - auto_setup: bool = True # Automatically discover IDs and configure Meta webhooks + # Overwrites the webhook URL of the business account and phone number, use only with test accounts + auto_setup: bool = False max_history_turns: int = 40 # Context window sliding budget - enforce_signature: bool = True + enforce_signature: bool = True # Reject webhook requests not signed with app_secret mark_as_read: bool = True log_level: str = "info" + def __post_init__(self): + # Resolved here rather than with default_factory so that explicit None values also fall back to env vars + self.phone_number_id = self.phone_number_id or os.environ.get("WHATSAPP_PHONE_NUMBER_ID") + self.access_token = (self.access_token + or os.environ.get("WHATSAPP_ACCESS_TOKEN") or os.environ.get("WHATSAPP_TOKEN")) + self.verify_token = (self.verify_token + or os.environ.get("WHATSAPP_VERIFY_TOKEN") + or os.environ.get("WHATSAPP_WEBHOOK_VERIFY_TOKEN")) + self.app_secret = self.app_secret or os.environ.get("WHATSAPP_APP_SECRET") + + self.verify_token_generated = not self.verify_token + if self.verify_token_generated: + self.verify_token = secrets.token_urlsafe(24) + def validate(self, dry_run: bool = False) -> None: """Validate that required credentials are set when not in dry_run mode.""" + if self.enforce_signature and not self.app_secret: + raise ValueError( + "WhatsApp app_secret is required to verify that webhook requests come from Meta. " + "Pass it to serve_whatsapp() or set the WHATSAPP_APP_SECRET environment variable " + "(or set enforce_signature=False to accept unsigned requests, NOT recommended)." + ) if not dry_run: if not self.phone_number_id: raise ValueError( diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index b1b16d6..462415c 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -52,6 +52,11 @@ def get_agent_lock(cls) -> asyncio.Lock: @classmethod def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: """Create and configure the FastAPI WhatsApp webhook application.""" + config.validate(dry_run=True) + if not config.enforce_signature: + logger.warning("[WhatsApp] enforce_signature=False: webhook requests are NOT verified, anyone who " + "knows the URL can make the agent reply (and spend LLM/WhatsApp credits) on your behalf") + cls._agent = agent cls._config = config cls._store = SQLiteSessionStore(db_path=config.db_path) @@ -71,7 +76,8 @@ async def verify_webhook( hub_challenge: str = Query(None, alias="hub.challenge"), ): """Handle Meta WhatsApp webhook verification challenge.""" - if hub_mode == "subscribe" and hub_verify_token == config.verify_token: + if hub_mode == "subscribe" and hub_verify_token and \ + hmac.compare_digest(hub_verify_token, config.verify_token): logger.info("[WhatsApp] Webhook verification challenge accepted") return Response(content=hub_challenge, media_type="text/plain") logger.warning("[WhatsApp] Webhook verification failed: token mismatch") @@ -88,7 +94,7 @@ async def receive_webhook(request: Request, background_tasks: BackgroundTasks): raw_body = await request.body() # 1. HMAC Signature Verification - if config.app_secret and config.enforce_signature: + if config.enforce_signature: sig_header = request.headers.get("X-Hub-Signature-256", "") if not cls._verify_signature(raw_body, sig_header, config.app_secret): logger.warning("[WhatsApp] Rejected request with invalid HMAC signature") @@ -285,7 +291,8 @@ def serve(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> None: from .discovery import MetaDiscoveryClient discovery = MetaDiscoveryClient(config.access_token, config.app_secret) callback_url = f"{public_url}{config.webhook_path}" - logger.info(f"[WhatsApp] Auto-configuring Meta webhooks to {callback_url}...") + logger.warning(f"[WhatsApp] auto_setup=True: overriding the Meta webhook URL of the WhatsApp Business " + f"Account and phone number to {callback_url} (it is not restored on shutdown)") try: res = asyncio.run(discovery.configure_webhooks( callback_url=callback_url, @@ -300,7 +307,9 @@ def serve(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> None: logger.info(f"[WhatsApp] Starting server on {config.host}:{config.port}") logger.info(f"[WhatsApp] Webhook endpoint: {config.webhook_path}") - logger.info(f"[WhatsApp] Verify Token: {config.verify_token}") + if config.verify_token_generated: + logger.warning(f"[WhatsApp] No verify token set, generated one for this run: {config.verify_token} " + "(enter it in Meta's webhook settings, or set WHATSAPP_VERIFY_TOKEN to keep it fixed)") uvicorn.run(app, host=config.host, port=config.port, log_level=config.log_level) @@ -337,7 +346,6 @@ def _start_tunnel(config: WhatsAppConfig) -> Optional[str]: logger.info("=" * 60) logger.info(f"[WhatsApp] Public Tunnel URL: {public_url}{config.webhook_path}") logger.info(f"[WhatsApp] Meta Callback URL: {public_url}{config.webhook_path}") - logger.info(f"[WhatsApp] Meta Verify Token: {config.verify_token}") logger.info("=" * 60) return public_url except ImportError: diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 770ce39..3a7729f 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -97,20 +97,64 @@ def test_whatsapp_config_defaults(): assert config.max_history_turns == 40 -def test_whatsapp_config_validation(): - config = WhatsAppConfig(phone_number_id=None, access_token=None) +@pytest.fixture +def no_whatsapp_env(monkeypatch): + for var in ("WHATSAPP_PHONE_NUMBER_ID", "WHATSAPP_ACCESS_TOKEN", "WHATSAPP_TOKEN", "WHATSAPP_VERIFY_TOKEN", + "WHATSAPP_WEBHOOK_VERIFY_TOKEN", "WHATSAPP_APP_SECRET"): + monkeypatch.delenv(var, raising=False) + + +def test_whatsapp_config_validation(no_whatsapp_env): + config = WhatsAppConfig(phone_number_id=None, access_token=None, app_secret="secret") with pytest.raises(ValueError, match="phone_number_id is required"): config.validate(dry_run=False) - config = WhatsAppConfig(phone_number_id="123", access_token=None) + config = WhatsAppConfig(phone_number_id="123", access_token=None, app_secret="secret") with pytest.raises(ValueError, match="access_token is required"): config.validate(dry_run=False) # Valid config - config = WhatsAppConfig(phone_number_id="123", access_token="token") + config = WhatsAppConfig(phone_number_id="123", access_token="token", app_secret="secret") config.validate(dry_run=False) # Should not raise +def test_whatsapp_config_requires_app_secret(no_whatsapp_env, temp_db): + config = WhatsAppConfig(phone_number_id="123", access_token="token", db_path=temp_db) + with pytest.raises(ValueError, match="app_secret is required"): + config.validate(dry_run=False) + with pytest.raises(ValueError, match="app_secret is required"): + WhatsAppServer.create_app(MockWhatsAppAgent(), config) + + # Explicit opt-out accepts unsigned requests + config = WhatsAppConfig(phone_number_id="123", access_token="token", db_path=temp_db, + enforce_signature=False, mark_as_read=False) + client = TestClient(WhatsAppServer.create_app(MockWhatsAppAgent(), config)) + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + resp = client.post("/webhook", json=_create_meta_payload()) + assert resp.status_code == 200 + + +def test_whatsapp_config_env_fallback_with_explicit_none(no_whatsapp_env, monkeypatch): + """Agent.serve_whatsapp() passes None for unset arguments; env vars must still be used.""" + monkeypatch.setenv("WHATSAPP_PHONE_NUMBER_ID", "env_phone") + monkeypatch.setenv("WHATSAPP_ACCESS_TOKEN", "env_token") + monkeypatch.setenv("WHATSAPP_APP_SECRET", "env_secret") + monkeypatch.setenv("WHATSAPP_VERIFY_TOKEN", "env_verify") + + config = WhatsAppConfig(phone_number_id=None, access_token=None, app_secret=None, verify_token=None) + assert (config.phone_number_id, config.access_token, config.app_secret, config.verify_token) == \ + ("env_phone", "env_token", "env_secret", "env_verify") + assert not config.verify_token_generated + + +def test_whatsapp_config_generates_random_verify_token(no_whatsapp_env): + config1, config2 = WhatsAppConfig(), WhatsAppConfig() + assert config1.verify_token_generated + assert len(config1.verify_token) >= 32 + assert config1.verify_token != config2.verify_token + assert not WhatsAppConfig().auto_setup + + # --------------------------------------------------------------------------- # Test 2: SQLiteSessionStore # --------------------------------------------------------------------------- @@ -242,6 +286,9 @@ def test_webhook_verification_failure(test_config): ) assert response.status_code == 403 + response = client.get("/webhook", params={"hub.mode": "subscribe", "hub.challenge": "challenge"}) + assert response.status_code == 403 + # --------------------------------------------------------------------------- # Test 5: Inbound Messages & HMAC (POST /webhook) From 15354892f26c19710682dd2d819f589e9251e655 Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:20:20 +0100 Subject: [PATCH 05/10] [sdialog] Fix WhatsApp session store and webhook parsing issues - Close SQLite connections after each operation (the sqlite3 context manager only commits, so connections were leaked) - Pass the idempotency retention cutoff as a query parameter instead of formatting it into the SQL, compute it in local time to match the stored timestamps, and run the cleanup on startup (the table previously grew forever) - Process all entries/changes of a webhook delivery, not only the first one, since Meta may batch them - Add httpx to requirements (it was imported but only installed transitively) - Keep the test suite offline (mark_as_read made real Graph API calls) Co-Authored-By: Claude Opus 5.5 --- requirements.txt | 1 + src/sdialog/whatsapp/server.py | 21 ++++++++++++------- src/sdialog/whatsapp/session_store.py | 30 ++++++++++++++++----------- tests/test_whatsapp.py | 26 +++++++++++++++++++++++ 4 files changed, 58 insertions(+), 20 deletions(-) diff --git a/requirements.txt b/requirements.txt index 3d3d482..d4b29df 100644 --- a/requirements.txt +++ b/requirements.txt @@ -18,6 +18,7 @@ syllables>=1.0.4 # === Web server (optional) === fastapi uvicorn +httpx # === LangChain ecosystem === # LangChain ≥0.3.0 officially targets Python 3.9+, but later releases may require 3.10+. diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index 462415c..7d5b07d 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -60,6 +60,7 @@ def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: cls._agent = agent cls._config = config cls._store = SQLiteSessionStore(db_path=config.db_path) + cls._store.cleanup_old_records() cls._client = MetaWhatsAppClient(config) cls._agent_lock = None @@ -251,17 +252,21 @@ def _verify_signature(raw_body: bytes, sig_header: str, app_secret: str) -> bool @staticmethod def _extract_messages(payload: Dict[str, Any]): - """Safely extract message objects and contact details from Meta webhook payload.""" + """ + Safely extract message objects and contact details from Meta webhook payload. + Meta may batch several entries and changes in a single delivery, so all of them are collected. + """ messages = [] sender_name = None try: - entry = payload.get("entry", [])[0] - change = entry.get("changes", [])[0].get("value", {}) - messages = change.get("messages", []) - contacts = change.get("contacts", []) - if contacts: - sender_name = contacts[0].get("profile", {}).get("name") - except (IndexError, KeyError, AttributeError): + for entry in payload.get("entry", []): + for change in entry.get("changes", []): + value = change.get("value", {}) + messages.extend(value.get("messages", [])) + contacts = value.get("contacts", []) + if contacts and sender_name is None: + sender_name = contacts[0].get("profile", {}).get("name") + except (KeyError, AttributeError, TypeError): pass return messages, sender_name diff --git a/src/sdialog/whatsapp/session_store.py b/src/sdialog/whatsapp/session_store.py index c22efcf..d2da511 100644 --- a/src/sdialog/whatsapp/session_store.py +++ b/src/sdialog/whatsapp/session_store.py @@ -9,8 +9,9 @@ import json import logging import sqlite3 -from datetime import datetime -from typing import Optional, Any, Dict +from contextlib import contextmanager +from datetime import datetime, timedelta +from typing import Optional, Any, Dict, Iterator logger = logging.getLogger(__name__) @@ -29,14 +30,19 @@ def __init__(self, db_path: str = "whatsapp_sessions.db"): self.db_path = db_path self._init_db() - def _get_connection(self) -> sqlite3.Connection: - """Create a new SQLite connection with WAL mode and busy timeout configured.""" + @contextmanager + def _get_connection(self) -> Iterator[sqlite3.Connection]: + """Open a SQLite connection (WAL mode, busy timeout), commit on success and always close it.""" conn = sqlite3.connect(self.db_path, timeout=10.0) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA journal_mode = WAL;") - conn.execute("PRAGMA busy_timeout = 5000;") - conn.execute("PRAGMA synchronous = NORMAL;") - return conn + try: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode = WAL;") + conn.execute("PRAGMA busy_timeout = 5000;") + conn.execute("PRAGMA synchronous = NORMAL;") + with conn: + yield conn + finally: + conn.close() def _init_db(self) -> None: """Initialize database tables and indexes.""" @@ -145,7 +151,7 @@ def mark_message_processed(self, wamid: str) -> None: def cleanup_old_records(self, retention_days: int = 7) -> None: """Purge idempotency records older than retention_days.""" + # received_at is stored in local time, so compute the cutoff the same way (SQLite's 'now' is UTC) + cutoff = (datetime.now() - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S") with self._get_connection() as conn: - conn.execute( - f"DELETE FROM processed_messages WHERE received_at < datetime('now', '-{retention_days} days')" - ) + conn.execute("DELETE FROM processed_messages WHERE received_at < ?", (cutoff,)) diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 3a7729f..0d8c3bf 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -82,6 +82,7 @@ def test_config(temp_db): db_path=temp_db, session_timeout=3600, enforce_signature=True, + mark_as_read=False, # avoid real Graph API calls from background tasks ) @@ -205,6 +206,31 @@ def test_session_store_idempotency(temp_db): assert store.is_message_processed(wamid) +def test_session_store_cleanup_old_records(temp_db): + import sqlite3 + + store = SQLiteSessionStore(db_path=temp_db) + store.mark_message_processed("wamid.new") + conn = sqlite3.connect(temp_db) + with conn: + conn.execute("INSERT INTO processed_messages (wamid, received_at) VALUES ('wamid.old', '2000-01-01 00:00:00')") + conn.close() + + store.cleanup_old_records(retention_days=7) + assert store.is_message_processed("wamid.new") + assert not store.is_message_processed("wamid.old") + + +def test_extract_messages_from_batched_payload(): + payload = _create_meta_payload(sender="1111", wamid="wamid.A") + payload["entry"].append(_create_meta_payload(sender="2222", wamid="wamid.B")["entry"][0]) + + messages, sender_name = WhatsAppServer._extract_messages(payload) + assert [m["id"] for m in messages] == ["wamid.A", "wamid.B"] + assert sender_name == "Alice" + assert WhatsAppServer._extract_messages({"entry": "malformed"}) == ([], None) + + def test_trim_memory_starts_at_user_message(): memory = [{"type": t} for t in ["human", "ai", "human", "ai", "tool", "ai"]] From 019edcfa16265b73faa991635a2435841a8a9334 Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:21:12 +0100 Subject: [PATCH 06/10] [sdialog] Document WhatsApp serving and agent state API Add a "Serving Agents on WhatsApp" docs section (test account setup, example, limitations, privacy and auto_setup caveats) and changelog entries for serve_whatsapp() and get_state()/set_state(). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 +++++++++++ docs/sdialog/index.rst | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d08758..d80d918 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ All notable changes to SDialog will be documented here. --- +## [Unreleased] + +### Added +- **sdialog.agents**: + - `Agent.serve_whatsapp()` (and `Server.serve_whatsapp()`) to chat with agents over WhatsApp through the Meta WhatsApp Cloud API, with one conversation per user saved in SQLite (#135, contributed by @amitbhakt). See [Serving Agents on WhatsApp](https://sdialog.readthedocs.io/en/latest/sdialog/index.html#serving-agents-on-whatsapp). + - `Agent.get_state()` / `Agent.set_state()` to save a conversation (memory, finished flag and orchestrator state) and restore it later, e.g. on a fresh or reset agent. +- **sdialog.orchestrators**: + - `get_state()` / `set_state()` hooks for stateful orchestrators; `ChangeMindOrchestrator` saves its trigger counter. + +--- + ## [0.5.4] 2026-10-26 🩹 ### Fixed diff --git a/docs/sdialog/index.rst b/docs/sdialog/index.rst index 07bd8cc..06fef08 100644 --- a/docs/sdialog/index.rst +++ b/docs/sdialog/index.rst @@ -362,6 +362,42 @@ For serving multiple agents under a single endpoint, use ``Server``'s :meth:`~sd # Serve both agents on port 1333 (select them by name from the GUI) Server.serve([agent1, agent2], port=1333) +.. _serving_agents_whatsapp: + +Serving Agents on WhatsApp +-------------------------- + +Agents can also chat with real users over WhatsApp, through the Meta WhatsApp Cloud API, with :meth:`~sdialog.agents.Agent.serve_whatsapp`. Meta sends each incoming message to a webhook served by SDialog, the agent replies, and the reply is sent back through Meta's API. Each user (phone number) has their own conversation, saved in a local SQLite database and restored on their next message with :meth:`~sdialog.agents.Agent.get_state` / :meth:`~sdialog.agents.Agent.set_state`. Conversations reset after ``session_timeout`` seconds of inactivity (30 minutes by default), or when the user sends ``/reset``. + +To try it with a free Meta test account: + +1. At `developers.facebook.com `_, create an app of type "Business" and add the **WhatsApp** product. Meta provides a test phone number, its **Phone Number ID**, and a temporary **access token**. Add your own phone number as an allowed recipient. +2. Copy the **App Secret** from *App settings → Basic*. It is required, since it is used to check that webhook requests really come from Meta. +3. Expose the server on a public HTTPS URL. For local testing, ``tunnel=True`` opens an `ngrok `_ tunnel (``pip install pyngrok``). +4. In the app's *WhatsApp → Configuration* page, set the callback URL (tunnel URL + ``/webhook``) and the verify token, and subscribe to the ``messages`` field. + +.. code-block:: python + + import os + from sdialog.agents import Agent + + os.environ["WHATSAPP_ACCESS_TOKEN"] = "..." + os.environ["WHATSAPP_PHONE_NUMBER_ID"] = "..." + os.environ["WHATSAPP_APP_SECRET"] = "..." + os.environ["WHATSAPP_VERIFY_TOKEN"] = "any-random-string" # same value as in Meta's webhook settings + + support = Agent(name="Support") + support.serve_whatsapp(tunnel=True) + +Then send a WhatsApp message to the test number from your phone. + +Some things to keep in mind: + +- Only text messages are supported; users sending other media get a short notice. +- A single agent serves all users, so messages are processed one at a time (like :meth:`~sdialog.agents.Agent.serve`). +- Conversations are stored in plain text in ``db_path`` (``whatsapp_sessions.db`` by default), so treat that file as personal data. +- With ``auto_setup=True`` (and ``tunnel=True``), SDialog overwrites the webhook URL of your WhatsApp Business Account and phone number with the tunnel URL. Use it only with test accounts. + ---- From 5ea09ba84cb5097bf2e3e190f7814757310d3c46 Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:37:32 +0100 Subject: [PATCH 07/10] [sdialog] List original author first in WhatsApp file headers --- src/sdialog/whatsapp/__init__.py | 2 +- src/sdialog/whatsapp/client.py | 2 +- src/sdialog/whatsapp/config.py | 2 +- src/sdialog/whatsapp/server.py | 2 +- src/sdialog/whatsapp/session_store.py | 2 +- tests/test_whatsapp.py | 2 +- 6 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/sdialog/whatsapp/__init__.py b/src/sdialog/whatsapp/__init__.py index 6cc40ca..29e8e0f 100644 --- a/src/sdialog/whatsapp/__init__.py +++ b/src/sdialog/whatsapp/__init__.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ WhatsApp integration for SDialog agents. diff --git a/src/sdialog/whatsapp/client.py b/src/sdialog/whatsapp/client.py index b4393af..b392371 100644 --- a/src/sdialog/whatsapp/client.py +++ b/src/sdialog/whatsapp/client.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ Meta WhatsApp Cloud API client. diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py index 4dd7d56..14a2618 100644 --- a/src/sdialog/whatsapp/config.py +++ b/src/sdialog/whatsapp/config.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ Configuration models for SDialog WhatsApp server. diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index 7d5b07d..cf63da1 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ FastAPI webhook server for SDialog WhatsApp agent serving. diff --git a/src/sdialog/whatsapp/session_store.py b/src/sdialog/whatsapp/session_store.py index d2da511..54f4d0d 100644 --- a/src/sdialog/whatsapp/session_store.py +++ b/src/sdialog/whatsapp/session_store.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ SQLite-backed multi-user session and idempotency store for WhatsApp. diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 0d8c3bf..46aafea 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute -# SPDX-FileContributor: Sergio Burdisso # SPDX-FileContributor: Amit Bhakt +# SPDX-FileContributor: Sergio Burdisso # SPDX-License-Identifier: MIT """ Unit and integration tests for SDialog WhatsApp server. From 4b4b4c01a40e5792e8ad343722a8467febe26db2 Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:40:26 +0100 Subject: [PATCH 08/10] [sdialog] Add per-user rate limiting to WhatsApp server Turns are processed one at a time, so a single user flooding the bot would delay everyone else, and anyone who knows the number could run up LLM and WhatsApp costs. Each user can now send up to max_messages_per_minute messages (sliding window, 10 by default, None to disable); extra messages are dropped before reaching the agent, and the user gets a single "slow down" notice per minute. Co-Authored-By: Claude Opus 5.5 --- docs/sdialog/index.rst | 1 + src/sdialog/agents.py | 4 +++ src/sdialog/server.py | 4 +++ src/sdialog/whatsapp/config.py | 1 + src/sdialog/whatsapp/server.py | 49 ++++++++++++++++++++++++++++++++-- tests/test_whatsapp.py | 41 ++++++++++++++++++++++++++++ 6 files changed, 98 insertions(+), 2 deletions(-) diff --git a/docs/sdialog/index.rst b/docs/sdialog/index.rst index 06fef08..e142358 100644 --- a/docs/sdialog/index.rst +++ b/docs/sdialog/index.rst @@ -395,6 +395,7 @@ Some things to keep in mind: - Only text messages are supported; users sending other media get a short notice. - A single agent serves all users, so messages are processed one at a time (like :meth:`~sdialog.agents.Agent.serve`). +- Each user can send up to ``max_messages_per_minute`` messages (10 by default, ``None`` to disable). Extra messages are dropped without calling the LLM, and the user gets a single "slow down" notice. - Conversations are stored in plain text in ``db_path`` (``whatsapp_sessions.db`` by default), so treat that file as personal data. - With ``auto_setup=True`` (and ``tunnel=True``), SDialog overwrites the webhook URL of your WhatsApp Business Account and phone number with the tunnel URL. Use it only with test accounts. diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index c5b6318..9a66506 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -682,6 +682,7 @@ def serve_whatsapp(self, session_timeout: int = 1800, tunnel: bool = False, auto_setup: bool = False, + max_messages_per_minute: Optional[int] = 10, log_level: str = "info"): """ Starts a Meta WhatsApp Cloud API webhook server to interact with the agent. @@ -709,6 +710,8 @@ def serve_whatsapp(self, :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool + :param max_messages_per_minute: Per-user rate limit; extra messages are dropped (None to disable). + :type max_messages_per_minute: Optional[int] :param log_level: Logging level for the server. :type log_level: str """ @@ -726,6 +729,7 @@ def serve_whatsapp(self, session_timeout=session_timeout, tunnel=tunnel, auto_setup=auto_setup, + max_messages_per_minute=max_messages_per_minute, log_level=log_level ) return WhatsAppServer.serve(agent=self, config=config) diff --git a/src/sdialog/server.py b/src/sdialog/server.py index a4a74e9..9ef9995 100644 --- a/src/sdialog/server.py +++ b/src/sdialog/server.py @@ -332,6 +332,7 @@ def serve_whatsapp(cls, session_timeout: int = 1800, tunnel: bool = False, auto_setup: bool = False, + max_messages_per_minute: Optional[int] = 10, log_level: str = "info") -> None: """ Serve an SDialog agent via Meta WhatsApp Cloud API webhook. @@ -361,6 +362,8 @@ def serve_whatsapp(cls, :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool + :param max_messages_per_minute: Per-user rate limit; extra messages are dropped (None to disable). + :type max_messages_per_minute: Optional[int] :param log_level: Logging level for the server. :type log_level: str """ @@ -378,6 +381,7 @@ def serve_whatsapp(cls, session_timeout=session_timeout, tunnel=tunnel, auto_setup=auto_setup, + max_messages_per_minute=max_messages_per_minute, log_level=log_level ) return WhatsAppServer.serve(agent=agent, config=config) diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py index 14a2618..8dcede1 100644 --- a/src/sdialog/whatsapp/config.py +++ b/src/sdialog/whatsapp/config.py @@ -38,6 +38,7 @@ class WhatsAppConfig: # Overwrites the webhook URL of the business account and phone number, use only with test accounts auto_setup: bool = False max_history_turns: int = 40 # Context window sliding budget + max_messages_per_minute: Optional[int] = 10 # Per-user rate limit (None to disable) enforce_signature: bool = True # Reject webhook requests not signed with app_secret mark_as_read: bool = True log_level: str = "info" diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index cf63da1..095dad9 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -11,7 +11,9 @@ import asyncio import logging import threading -from typing import Optional, Dict, Any, List +import time +from collections import deque +from typing import Optional, Dict, Any, List, Deque, Tuple from fastapi import FastAPI, Request, Response, BackgroundTasks, HTTPException, Query from fastapi.responses import JSONResponse @@ -35,6 +37,8 @@ class WhatsAppServer: _store: Optional[SQLiteSessionStore] = None _client: Optional[MetaWhatsAppClient] = None _agent_lock: Optional[asyncio.Lock] = None + _rate_windows: Dict[str, Deque[float]] = {} # sender ID -> arrival times within the last minute + _rate_notified: Dict[str, float] = {} # sender ID -> time of the last "slow down" notice @classmethod def get_agent_lock(cls) -> asyncio.Lock: @@ -63,6 +67,8 @@ def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: cls._store.cleanup_old_records() cls._client = MetaWhatsAppClient(config) cls._agent_lock = None + cls._rate_windows.clear() + cls._rate_notified.clear() app = FastAPI( title="SDialog WhatsApp Agent Server", @@ -128,7 +134,19 @@ async def receive_webhook(request: Request, background_tasks: BackgroundTasks): cls._store.mark_message_processed(wamid) - # 5. Route by message type + # 5. Per-user rate limit, so one user can't flood the (serialized) agent or run up costs + allowed, notify = cls._check_rate_limit(sender_id) + if not allowed: + logger.warning(f"[WhatsApp] Rate limit exceeded for {sender_id}, message {wamid} dropped") + if notify: + background_tasks.add_task( + cls._client.send_text_message, + to=sender_id, + text="You are sending messages too quickly. Please wait a minute and try again.", + ) + continue + + # 6. Route by message type if msg_type == "text": text_body = msg.get("text", {}).get("body", "").strip() if text_body: @@ -224,6 +242,33 @@ async def _handle_unsupported_media(cls, sender_id: str, msg_type: str) -> None: text="I currently only support text messages. Please send your query as text.", ) + @classmethod + def _check_rate_limit(cls, sender_id: str) -> Tuple[bool, bool]: + """ + Sliding one-minute window rate limit per sender. + + :return: ``(allowed, notify)``, where ``notify`` is True for the first rejected message in a minute, + so that the user gets a single notice instead of one reply per dropped message. + """ + limit = cls._config.max_messages_per_minute + if not limit: + return True, False + + now = time.monotonic() + window = cls._rate_windows.setdefault(sender_id, deque()) + while window and now - window[0] >= 60: + window.popleft() + + if len(window) < limit: + window.append(now) + return True, False + + last_notice = cls._rate_notified.get(sender_id) + notify = last_notice is None or now - last_notice >= 60 + if notify: + cls._rate_notified[sender_id] = now + return False, notify + @staticmethod def _trim_memory(memory: List[Dict[str, Any]], max_messages: int) -> List[Dict[str, Any]]: """ diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 46aafea..4fb8957 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -633,3 +633,44 @@ async def _test(): assert mock_post.call_count == 3 # All 3 tiers synced asyncio.run(_test()) + + +# --------------------------------------------------------------------------- +# Test 7: Per-user rate limiting +# --------------------------------------------------------------------------- + +def test_rate_limit_sliding_window(test_config): + test_config.max_messages_per_minute = 3 + WhatsAppServer.create_app(MockWhatsAppAgent(), test_config) + + clock = {"now": 1000.0} + with patch("time.monotonic", side_effect=lambda: clock["now"]): + assert [WhatsAppServer._check_rate_limit("1111") for _ in range(3)] == [(True, False)] * 3 + assert WhatsAppServer._check_rate_limit("1111") == (False, True) # first rejection: notify + assert WhatsAppServer._check_rate_limit("1111") == (False, False) # then drop silently + assert WhatsAppServer._check_rate_limit("2222") == (True, False) # other users unaffected + + clock["now"] += 60 + assert WhatsAppServer._check_rate_limit("1111") == (True, False) # window has slid + + test_config.max_messages_per_minute = None + assert all(WhatsAppServer._check_rate_limit("1111") == (True, False) for _ in range(100)) + + +def test_rate_limit_drops_messages_without_calling_agent(test_config): + test_config.max_messages_per_minute = 2 + agent = MockWhatsAppAgent() + client = TestClient(WhatsAppServer.create_app(agent, test_config)) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock) as mock_send: + for i in range(4): + payload = json.dumps(_create_meta_payload(sender="1111", text=f"msg {i}", wamid=f"wamid.{i}")).encode() + resp = client.post("/webhook", content=payload, headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _compute_hmac(payload, test_config.app_secret), + }) + assert resp.status_code == 200 + + assert agent.call_count == 2 + notices = [c for c in mock_send.await_args_list if "too quickly" in c.kwargs["text"]] + assert len(notices) == 1 From a5cc6a378cde5e16a5989da4fb8731aabb4fdc37 Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 16:44:54 +0100 Subject: [PATCH 09/10] [sdialog] Add hourly WhatsApp rate limit and lower the per-minute one A per-minute limit alone mostly handles bursts: a client staying just under it could still trigger ~14k LLM calls a day. Add max_messages_per_hour (60 by default) for sustained use, and lower max_messages_per_minute from 10 to 6, which still leaves room for users splitting one thought into several quick messages. Each limit sends its own notice at most once per window, and rejected messages do not count towards the limits. Co-Authored-By: Claude Opus 5.5 --- docs/sdialog/index.rst | 2 +- src/sdialog/agents.py | 10 +++++-- src/sdialog/server.py | 10 +++++-- src/sdialog/whatsapp/config.py | 4 ++- src/sdialog/whatsapp/server.py | 55 ++++++++++++++++++---------------- tests/test_whatsapp.py | 40 ++++++++++++++++++++----- 6 files changed, 82 insertions(+), 39 deletions(-) diff --git a/docs/sdialog/index.rst b/docs/sdialog/index.rst index e142358..ad4170b 100644 --- a/docs/sdialog/index.rst +++ b/docs/sdialog/index.rst @@ -395,7 +395,7 @@ Some things to keep in mind: - Only text messages are supported; users sending other media get a short notice. - A single agent serves all users, so messages are processed one at a time (like :meth:`~sdialog.agents.Agent.serve`). -- Each user can send up to ``max_messages_per_minute`` messages (10 by default, ``None`` to disable). Extra messages are dropped without calling the LLM, and the user gets a single "slow down" notice. +- Each user can send up to ``max_messages_per_minute`` messages per minute (6 by default, enough for a few quick messages in a row) and ``max_messages_per_hour`` per hour (60 by default). Extra messages are dropped without calling the LLM, and the user gets a single notice; set either limit to ``None`` to disable it. - Conversations are stored in plain text in ``db_path`` (``whatsapp_sessions.db`` by default), so treat that file as personal data. - With ``auto_setup=True`` (and ``tunnel=True``), SDialog overwrites the webhook URL of your WhatsApp Business Account and phone number with the tunnel URL. Use it only with test accounts. diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index 9a66506..2c836e8 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -682,7 +682,8 @@ def serve_whatsapp(self, session_timeout: int = 1800, tunnel: bool = False, auto_setup: bool = False, - max_messages_per_minute: Optional[int] = 10, + max_messages_per_minute: Optional[int] = 6, + max_messages_per_hour: Optional[int] = 60, log_level: str = "info"): """ Starts a Meta WhatsApp Cloud API webhook server to interact with the agent. @@ -710,8 +711,12 @@ def serve_whatsapp(self, :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool - :param max_messages_per_minute: Per-user rate limit; extra messages are dropped (None to disable). + :param max_messages_per_minute: Per-user limit for bursts of messages; extra messages are dropped + (None to disable). :type max_messages_per_minute: Optional[int] + :param max_messages_per_hour: Per-user limit for sustained use; extra messages are dropped + (None to disable). + :type max_messages_per_hour: Optional[int] :param log_level: Logging level for the server. :type log_level: str """ @@ -730,6 +735,7 @@ def serve_whatsapp(self, tunnel=tunnel, auto_setup=auto_setup, max_messages_per_minute=max_messages_per_minute, + max_messages_per_hour=max_messages_per_hour, log_level=log_level ) return WhatsAppServer.serve(agent=self, config=config) diff --git a/src/sdialog/server.py b/src/sdialog/server.py index 9ef9995..3b53560 100644 --- a/src/sdialog/server.py +++ b/src/sdialog/server.py @@ -332,7 +332,8 @@ def serve_whatsapp(cls, session_timeout: int = 1800, tunnel: bool = False, auto_setup: bool = False, - max_messages_per_minute: Optional[int] = 10, + max_messages_per_minute: Optional[int] = 6, + max_messages_per_hour: Optional[int] = 60, log_level: str = "info") -> None: """ Serve an SDialog agent via Meta WhatsApp Cloud API webhook. @@ -362,8 +363,12 @@ def serve_whatsapp(cls, :param auto_setup: If True (with tunnel=True), overwrites the Meta webhook URL of the WhatsApp Business Account and phone number with the tunnel URL. Use only with test accounts. :type auto_setup: bool - :param max_messages_per_minute: Per-user rate limit; extra messages are dropped (None to disable). + :param max_messages_per_minute: Per-user limit for bursts of messages; extra messages are dropped + (None to disable). :type max_messages_per_minute: Optional[int] + :param max_messages_per_hour: Per-user limit for sustained use; extra messages are dropped + (None to disable). + :type max_messages_per_hour: Optional[int] :param log_level: Logging level for the server. :type log_level: str """ @@ -382,6 +387,7 @@ def serve_whatsapp(cls, tunnel=tunnel, auto_setup=auto_setup, max_messages_per_minute=max_messages_per_minute, + max_messages_per_hour=max_messages_per_hour, log_level=log_level ) return WhatsAppServer.serve(agent=agent, config=config) diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py index 8dcede1..d6e72eb 100644 --- a/src/sdialog/whatsapp/config.py +++ b/src/sdialog/whatsapp/config.py @@ -38,7 +38,9 @@ class WhatsAppConfig: # Overwrites the webhook URL of the business account and phone number, use only with test accounts auto_setup: bool = False max_history_turns: int = 40 # Context window sliding budget - max_messages_per_minute: Optional[int] = 10 # Per-user rate limit (None to disable) + # Per-user rate limits (None to disable): bursts of quick messages, and sustained use + max_messages_per_minute: Optional[int] = 6 + max_messages_per_hour: Optional[int] = 60 enforce_signature: bool = True # Reject webhook requests not signed with app_secret mark_as_read: bool = True log_level: str = "info" diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py index 095dad9..3b198dd 100644 --- a/src/sdialog/whatsapp/server.py +++ b/src/sdialog/whatsapp/server.py @@ -37,8 +37,8 @@ class WhatsAppServer: _store: Optional[SQLiteSessionStore] = None _client: Optional[MetaWhatsAppClient] = None _agent_lock: Optional[asyncio.Lock] = None - _rate_windows: Dict[str, Deque[float]] = {} # sender ID -> arrival times within the last minute - _rate_notified: Dict[str, float] = {} # sender ID -> time of the last "slow down" notice + _rate_windows: Dict[str, Deque[float]] = {} # sender ID -> arrival times of accepted messages (last hour) + _rate_notified: Dict[Tuple[str, str], float] = {} # (sender ID, "minute"/"hour") -> time of last notice @classmethod def get_agent_lock(cls) -> asyncio.Lock: @@ -135,15 +135,11 @@ async def receive_webhook(request: Request, background_tasks: BackgroundTasks): cls._store.mark_message_processed(wamid) # 5. Per-user rate limit, so one user can't flood the (serialized) agent or run up costs - allowed, notify = cls._check_rate_limit(sender_id) + allowed, notice = cls._check_rate_limit(sender_id) if not allowed: logger.warning(f"[WhatsApp] Rate limit exceeded for {sender_id}, message {wamid} dropped") - if notify: - background_tasks.add_task( - cls._client.send_text_message, - to=sender_id, - text="You are sending messages too quickly. Please wait a minute and try again.", - ) + if notice: + background_tasks.add_task(cls._client.send_text_message, to=sender_id, text=notice) continue # 6. Route by message type @@ -243,31 +239,38 @@ async def _handle_unsupported_media(cls, sender_id: str, msg_type: str) -> None: ) @classmethod - def _check_rate_limit(cls, sender_id: str) -> Tuple[bool, bool]: + def _check_rate_limit(cls, sender_id: str) -> Tuple[bool, Optional[str]]: """ - Sliding one-minute window rate limit per sender. + Per-sender sliding window rate limit: a per-minute limit for bursts and a per-hour limit for + sustained use. Rejected messages do not count towards the limits. - :return: ``(allowed, notify)``, where ``notify`` is True for the first rejected message in a minute, - so that the user gets a single notice instead of one reply per dropped message. + :return: ``(allowed, notice)``, where ``notice`` is the text to send back for the first rejected message + in each window (None otherwise), so the user gets a single notice instead of one per message. """ - limit = cls._config.max_messages_per_minute - if not limit: - return True, False + limits = [(limit, seconds, unit) for limit, seconds, unit in [ + (cls._config.max_messages_per_hour, 3600, "hour"), + (cls._config.max_messages_per_minute, 60, "minute"), + ] if limit] + if not limits: + return True, None now = time.monotonic() window = cls._rate_windows.setdefault(sender_id, deque()) - while window and now - window[0] >= 60: + while window and now - window[0] >= max(seconds for _, seconds, _ in limits): window.popleft() - if len(window) < limit: - window.append(now) - return True, False - - last_notice = cls._rate_notified.get(sender_id) - notify = last_notice is None or now - last_notice >= 60 - if notify: - cls._rate_notified[sender_id] = now - return False, notify + for limit, seconds, unit in limits: + if sum(1 for t in window if now - t < seconds) >= limit: + last_notice = cls._rate_notified.get((sender_id, unit)) + if last_notice is not None and now - last_notice < seconds: + return False, None + cls._rate_notified[(sender_id, unit)] = now + if unit == "minute": + return False, "You are sending messages too quickly. Please wait a minute and try again." + return False, "You have reached the message limit for now. Please try again later." + + window.append(now) + return True, None @staticmethod def _trim_memory(memory: List[Dict[str, Any]], max_messages: int) -> List[Dict[str, Any]]: diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py index 4fb8957..d5f7890 100644 --- a/tests/test_whatsapp.py +++ b/tests/test_whatsapp.py @@ -639,22 +639,48 @@ async def _test(): # Test 7: Per-user rate limiting # --------------------------------------------------------------------------- -def test_rate_limit_sliding_window(test_config): +def test_rate_limit_per_minute(test_config): test_config.max_messages_per_minute = 3 WhatsAppServer.create_app(MockWhatsAppAgent(), test_config) clock = {"now": 1000.0} with patch("time.monotonic", side_effect=lambda: clock["now"]): - assert [WhatsAppServer._check_rate_limit("1111") for _ in range(3)] == [(True, False)] * 3 - assert WhatsAppServer._check_rate_limit("1111") == (False, True) # first rejection: notify - assert WhatsAppServer._check_rate_limit("1111") == (False, False) # then drop silently - assert WhatsAppServer._check_rate_limit("2222") == (True, False) # other users unaffected + assert [WhatsAppServer._check_rate_limit("1111") for _ in range(3)] == [(True, None)] * 3 + allowed, notice = WhatsAppServer._check_rate_limit("1111") # first rejection: notify + assert not allowed and "too quickly" in notice + assert WhatsAppServer._check_rate_limit("1111") == (False, None) # then drop silently + assert WhatsAppServer._check_rate_limit("2222") == (True, None) # other users unaffected clock["now"] += 60 - assert WhatsAppServer._check_rate_limit("1111") == (True, False) # window has slid + assert WhatsAppServer._check_rate_limit("1111") == (True, None) # window has slid + +def test_rate_limit_per_hour(test_config): + test_config.max_messages_per_minute = 6 + test_config.max_messages_per_hour = 10 + WhatsAppServer.create_app(MockWhatsAppAgent(), test_config) + + clock = {"now": 1000.0} + with patch("time.monotonic", side_effect=lambda: clock["now"]): + # One message every 30 seconds stays under the per-minute limit... + for _ in range(10): + assert WhatsAppServer._check_rate_limit("1111") == (True, None) + clock["now"] += 30 + # ...but hits the per-hour one + allowed, notice = WhatsAppServer._check_rate_limit("1111") + assert not allowed and "message limit" in notice + clock["now"] += 120 + assert WhatsAppServer._check_rate_limit("1111") == (False, None) # one notice per hour + + clock["now"] += 3600 + assert WhatsAppServer._check_rate_limit("1111") == (True, None) + + +def test_rate_limit_disabled(test_config): test_config.max_messages_per_minute = None - assert all(WhatsAppServer._check_rate_limit("1111") == (True, False) for _ in range(100)) + test_config.max_messages_per_hour = None + WhatsAppServer.create_app(MockWhatsAppAgent(), test_config) + assert all(WhatsAppServer._check_rate_limit("1111") == (True, None) for _ in range(100)) def test_rate_limit_drops_messages_without_calling_agent(test_config): From 5a2257d52ba5a3c7f662352b566983afb64578ae Mon Sep 17 00:00:00 2001 From: Sergio Burdisso Date: Wed, 28 Oct 2026 17:36:54 +0100 Subject: [PATCH 10/10] [sdialog] Pin CI runner to Ubuntu 24.04 to keep Python 3.9 tests ubuntu-latest now resolves to Ubuntu 26.04, for which actions/setup-python has no Python 3.9 builds (3.9 is end-of-life), so the 3.9 job failed to start. Also update actions/checkout to v4 and actions/setup-python to v5. --- .github/workflows/ci.yml | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 58e5247..a1945b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,17 +8,18 @@ on: jobs: build: - runs-on: ubuntu-latest + # Pinned: ubuntu-latest (26.04) has no Python 3.9 builds (3.9 is end-of-life) + runs-on: ubuntu-24.04 strategy: matrix: python-version: ["3.9", "3.10"] steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 - name: Set up Python - uses: actions/setup-python@v4 + uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }}