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 }}
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/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..ad4170b 100644
--- a/docs/sdialog/index.rst
+++ b/docs/sdialog/index.rst
@@ -362,6 +362,43 @@ 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`).
+- 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/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/agents.py b/src/sdialog/agents.py
index 92abbbe..2c836e8 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
@@ -670,6 +670,76 @@ 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 = False,
+ 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.
+
+ :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; random if unset).
+ :type verify_token: Optional[str]
+ :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
+ :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 (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 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
+ """
+ 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,
+ 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)
+
def response_lookahead(self, message: str = None):
"""
Generates a response without updating the agent's memory.
@@ -991,5 +1061,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/src/sdialog/server.py b/src/sdialog/server.py
index 5119000..3b53560 100644
--- a/src/sdialog/server.py
+++ b/src/sdialog/server.py
@@ -318,6 +318,80 @@ 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 = False,
+ 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.
+
+ :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 (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). Required.
+ :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 (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 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
+ """
+ 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,
+ 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)
+
@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..29e8e0f
--- /dev/null
+++ b/src/sdialog/whatsapp/__init__.py
@@ -0,0 +1,25 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# 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..b392371
--- /dev/null
+++ b/src/sdialog/whatsapp/client.py
@@ -0,0 +1,121 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# 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..d6e72eb
--- /dev/null
+++ b/src/sdialog/whatsapp/config.py
@@ -0,0 +1,80 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# SPDX-License-Identifier: MIT
+"""
+Configuration models for SDialog WhatsApp server.
+"""
+
+import os
+import secrets
+from dataclasses import dataclass
+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`` (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] = 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
+ # 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
+ # 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"
+
+ 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(
+ "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..3b198dd
--- /dev/null
+++ b/src/sdialog/whatsapp/server.py
@@ -0,0 +1,409 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# SPDX-License-Identifier: MIT
+"""
+FastAPI webhook server for SDialog WhatsApp agent serving.
+"""
+
+import hmac
+import hashlib
+import asyncio
+import logging
+import threading
+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
+import uvicorn
+
+from .config import WhatsAppConfig
+from .client import MetaWhatsAppClient
+from .session_store import SQLiteSessionStore
+
+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
+ _agent_lock: Optional[asyncio.Lock] = None
+ _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:
+ """
+ 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:
+ """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)
+ 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",
+ 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 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")
+ 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.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. Per-user rate limit, so one user can't flood the (serialized) agent or run up costs
+ 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 notice:
+ background_tasks.add_task(cls._client.send_text_message, to=sender_id, text=notice)
+ continue
+
+ # 6. 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 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 cls.get_agent_lock():
+ try:
+ # 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
+ )
+
+ # 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()
+ 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]}...'")
+ 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 = 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
+ 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.",
+ )
+
+ @classmethod
+ def _check_rate_limit(cls, sender_id: str) -> Tuple[bool, Optional[str]]:
+ """
+ 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, 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.
+ """
+ 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] >= max(seconds for _, seconds, _ in limits):
+ window.popleft()
+
+ 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]]:
+ """
+ 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."""
+ 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.
+ Meta may batch several entries and changes in a single delivery, so all of them are collected.
+ """
+ messages = []
+ sender_name = None
+ try:
+ 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
+
+ @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.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,
+ 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}")
+ 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)
+
+ @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("=" * 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..54f4d0d
--- /dev/null
+++ b/src/sdialog/whatsapp/session_store.py
@@ -0,0 +1,157 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# SPDX-License-Identifier: MIT
+"""
+SQLite-backed multi-user session and idempotency store for WhatsApp.
+"""
+
+import json
+import logging
+import sqlite3
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+from typing import Optional, Any, Dict, Iterator
+
+logger = logging.getLogger(__name__)
+
+
+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()
+
+ @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)
+ 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."""
+ 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."""
+ # 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("DELETE FROM processed_messages WHERE received_at < ?", (cutoff,))
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": []})
diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py
new file mode 100644
index 0000000..d5f7890
--- /dev/null
+++ b/tests/test_whatsapp.py
@@ -0,0 +1,702 @@
+# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute
+# SPDX-FileContributor: Amit Bhakt
+# SPDX-FileContributor: Sergio Burdisso
+# 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, messages_to_dict, messages_from_dict
+
+from sdialog.whatsapp import (
+ WhatsAppConfig,
+ WhatsAppServer,
+ SQLiteSessionStore,
+ MetaWhatsAppClient,
+)
+
+
+class MockWhatsAppAgent:
+ """Mock agent for WhatsApp testing (mirrors Agent's in-place memory reset and state API)."""
+
+ 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))
+ 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[:] = 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
+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,
+ mark_as_read=False, # avoid real Graph API calls from background tasks
+ )
+
+
+# ---------------------------------------------------------------------------
+# 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
+
+
+@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, 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", 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
+# ---------------------------------------------------------------------------
+
+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_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"]]
+
+ 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) == []
+
+
+# ---------------------------------------------------------------------------
+# 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
+
+ response = client.get("/webhook", params={"hub.mode": "subscribe", "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["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
+
+ assert "Hello from B" in contentsB
+ 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 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())
+
+
+# ---------------------------------------------------------------------------
+# Test 7: Per-user rate limiting
+# ---------------------------------------------------------------------------
+
+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, 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, 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
+ 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):
+ 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