Skip to content

WhatsApp Reference

This section documents the WhatsApp notification subsystem, configuration classes, REST client transport, and the two-turn verbal confirmation tool.

WhatsAppConfig

edgevoiceai.notifications.whatsapp

WhatsApp Notification Subsystem for Voice AI Framework. Provides developer configuration, async client transport, and two-turn verbal confirmation tool. Compatible with generic REST gateways, Meta Cloud API, and webhook providers on Python 3.10+.

WhatsAppConfig dataclass

Configuration options for WhatsApp messaging in the Voice AI Framework. Natively supports: 1. Official Meta / Facebook WhatsApp Business Cloud API (token, phone_number_id) 2. Custom REST gateways / webhook relays (service_url, api_key)

Source code in edgevoiceai/notifications/whatsapp.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@dataclass(slots=True)
class WhatsAppConfig:
    """
    Configuration options for WhatsApp messaging in the Voice AI Framework.
    Natively supports:
      1. Official Meta / Facebook WhatsApp Business Cloud API (token, phone_number_id)
      2. Custom REST gateways / webhook relays (service_url, api_key)
    """
    token: str = ""  # Official Meta WhatsApp Cloud API Access Token
    phone_number_id: str = ""  # Official Meta Phone Number ID (sender ID)
    business_account_id: str = ""  # Meta WABA / Business Account ID (optional)
    api_version: str = "v21.0"  # Meta Graph API version

    # Generic or custom gateway parameters
    api_key: str = ""  # API key for generic/custom gateway (or alias for token)
    service_url: str = ""  # Custom service URL (auto-generated if phone_number_id is provided)
    provider: str = "auto"  # "auto" | "meta" | "generic" | "webhook"

    # Messaging defaults
    default_template: str = ""
    default_language: str = "en"
    phone_prefix: str = "91"  # Default country calling code
    timeout: float = 15.0

    # Custom auth header overrides
    auth_header_name: str | None = None  # e.g., "Authorization" or "API-KEY"
    auth_header_prefix: str | None = None  # e.g., "Bearer "

    # Voice AI safety & guardrails
    require_two_turn_confirmation: bool = True

    # Localized / customized conversational messages
    invalid_phone_message: str | None = None
    confirmation_prompt_message: str | None = None
    success_message: str | None = None
    prompt_instruction: str | None = None


    def __post_init__(self) -> None:
        # Standardize token and api_key aliases
        if not self.token and self.api_key:
            object.__setattr__(self, "token", self.api_key)
        elif not self.api_key and self.token:
            object.__setattr__(self, "api_key", self.token)

        # Resolve provider
        if self.provider == "auto":
            if self.phone_number_id:
                object.__setattr__(self, "provider", "meta")
            elif self.service_url:
                object.__setattr__(self, "provider", "generic")
            else:
                object.__setattr__(self, "provider", "generic")

        # Auto-construct Meta Graph API endpoint URL if phone_number_id is provided
        if self.provider == "meta" and self.phone_number_id:
            constructed_url = f"https://graph.facebook.com/{self.api_version}/{self.phone_number_id}/messages"
            object.__setattr__(self, "service_url", constructed_url)


    @classmethod
    def from_env(cls) -> WhatsAppConfig:
        """
        Loads configuration from standard environment variables:
        Supports both Meta Cloud API conventions (WHATSAPP_TOKEN, WHATSAPP_PHONE_NUMBER_ID)
        and custom gateway conventions (WHATSAPP_SERVICE_URL, WHATSAPP_API_KEY).
        """
        token = (
            os.environ.get("WHATSAPP_TOKEN")
            or os.environ.get("WHATSAPP_ACCESS_TOKEN")
            or os.environ.get("WHATSAPP_API_KEY")
            or ""
        ).strip()

        phone_number_id = (
            os.environ.get("WHATSAPP_PHONE_NUMBER_ID")
            or os.environ.get("WHATSAPP_MOBILE_NUMBER_ID")
            or os.environ.get("WHATSAPP_SENDER_PHONE_ID")
            or ""
        ).strip()

        business_account_id = (
            os.environ.get("WHATSAPP_BUSINESS_ACCOUNT_ID")
            or os.environ.get("WHATSAPP_WABA_ID")
            or ""
        ).strip()

        service_url = os.environ.get("WHATSAPP_SERVICE_URL", "").strip()
        provider = os.environ.get("WHATSAPP_PROVIDER", "auto").strip()
        api_version = os.environ.get("WHATSAPP_API_VERSION", "v21.0").strip()

        default_template = (
            os.environ.get("WHATSAPP_DEFAULT_TEMPLATE")
            or os.environ.get("WHATSAPP_TEMPLATE_NAME")
            or ""
        ).strip()

        default_language = (
            os.environ.get("WHATSAPP_DEFAULT_LANGUAGE")
            or os.environ.get("WHATSAPP_LANGUAGE")
            or "en"
        ).strip()

        phone_prefix = os.environ.get("WHATSAPP_PHONE_PREFIX", "91").strip()
        timeout = float(os.environ.get("WHATSAPP_TIMEOUT", "15.0"))

        auth_header_name = os.environ.get("WHATSAPP_AUTH_HEADER_NAME")
        auth_header_prefix = os.environ.get("WHATSAPP_AUTH_HEADER_PREFIX")

        return cls(
            token=token,
            api_key=token,
            phone_number_id=phone_number_id,
            business_account_id=business_account_id,
            service_url=service_url,
            provider=provider,
            api_version=api_version,
            default_template=default_template,
            default_language=default_language,
            phone_prefix=phone_prefix,
            timeout=timeout,
            auth_header_name=auth_header_name.strip() if auth_header_name else None,
            auth_header_prefix=auth_header_prefix if auth_header_prefix is not None else None,
        )

from_env classmethod 

from_env()

Loads configuration from standard environment variables: Supports both Meta Cloud API conventions (WHATSAPP_TOKEN, WHATSAPP_PHONE_NUMBER_ID) and custom gateway conventions (WHATSAPP_SERVICE_URL, WHATSAPP_API_KEY).

Source code in edgevoiceai/notifications/whatsapp.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@classmethod
def from_env(cls) -> WhatsAppConfig:
    """
    Loads configuration from standard environment variables:
    Supports both Meta Cloud API conventions (WHATSAPP_TOKEN, WHATSAPP_PHONE_NUMBER_ID)
    and custom gateway conventions (WHATSAPP_SERVICE_URL, WHATSAPP_API_KEY).
    """
    token = (
        os.environ.get("WHATSAPP_TOKEN")
        or os.environ.get("WHATSAPP_ACCESS_TOKEN")
        or os.environ.get("WHATSAPP_API_KEY")
        or ""
    ).strip()

    phone_number_id = (
        os.environ.get("WHATSAPP_PHONE_NUMBER_ID")
        or os.environ.get("WHATSAPP_MOBILE_NUMBER_ID")
        or os.environ.get("WHATSAPP_SENDER_PHONE_ID")
        or ""
    ).strip()

    business_account_id = (
        os.environ.get("WHATSAPP_BUSINESS_ACCOUNT_ID")
        or os.environ.get("WHATSAPP_WABA_ID")
        or ""
    ).strip()

    service_url = os.environ.get("WHATSAPP_SERVICE_URL", "").strip()
    provider = os.environ.get("WHATSAPP_PROVIDER", "auto").strip()
    api_version = os.environ.get("WHATSAPP_API_VERSION", "v21.0").strip()

    default_template = (
        os.environ.get("WHATSAPP_DEFAULT_TEMPLATE")
        or os.environ.get("WHATSAPP_TEMPLATE_NAME")
        or ""
    ).strip()

    default_language = (
        os.environ.get("WHATSAPP_DEFAULT_LANGUAGE")
        or os.environ.get("WHATSAPP_LANGUAGE")
        or "en"
    ).strip()

    phone_prefix = os.environ.get("WHATSAPP_PHONE_PREFIX", "91").strip()
    timeout = float(os.environ.get("WHATSAPP_TIMEOUT", "15.0"))

    auth_header_name = os.environ.get("WHATSAPP_AUTH_HEADER_NAME")
    auth_header_prefix = os.environ.get("WHATSAPP_AUTH_HEADER_PREFIX")

    return cls(
        token=token,
        api_key=token,
        phone_number_id=phone_number_id,
        business_account_id=business_account_id,
        service_url=service_url,
        provider=provider,
        api_version=api_version,
        default_template=default_template,
        default_language=default_language,
        phone_prefix=phone_prefix,
        timeout=timeout,
        auth_header_name=auth_header_name.strip() if auth_header_name else None,
        auth_header_prefix=auth_header_prefix if auth_header_prefix is not None else None,
    )

WhatsAppClient

Asynchronous and synchronous transport client for WhatsApp notification gateways. Seamlessly dispatches to Meta WhatsApp Business Cloud API or custom HTTP REST gateways.

Source code in edgevoiceai/notifications/whatsapp.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
class WhatsAppClient:
    """
    Asynchronous and synchronous transport client for WhatsApp notification gateways.
    Seamlessly dispatches to Meta WhatsApp Business Cloud API or custom HTTP REST gateways.
    """

    def __init__(self, config: WhatsAppConfig | None = None) -> None:
        self.config = config or WhatsAppConfig.from_env()

    def normalize_phone(self, raw: str) -> str:
        """
        Validates and standardizes a phone number using the configured country prefix.
        Filters out dummy, invalid, or repeating phone sequences.
        """
        if not raw:
            return ""
        digits = re.sub(r"\D", "", raw)
        prefix = self.config.phone_prefix

        # If already starts with prefix (e.g., 919876543210 -> 9876543210)
        if prefix and digits.startswith(prefix) and len(digits) == 10 + len(prefix):
            digits = digits[len(prefix):]
        elif digits.startswith("0") and len(digits) == 11:
            digits = digits[1:]

        # Must be standard 10 mobile digits
        if len(digits) != 10:
            return ""
        # Filter trivial / dummy numbers
        if len(set(digits)) <= 2:
            return ""
        if digits in ("1234567890", "0123456789", "9876543210", "0000000000"):
            return ""
        # Valid starting digits
        if digits[0] not in "56789":
            return ""

        return (prefix + digits) if prefix else digits

    def _get_headers(self) -> dict[str, str]:
        headers = {"Content-Type": "application/json"}
        token = self.config.token or self.config.api_key
        if not token:
            return headers

        if self.config.provider == "meta":
            h_name = self.config.auth_header_name or "Authorization"
            h_prefix = self.config.auth_header_prefix if self.config.auth_header_prefix is not None else "Bearer "
            headers[h_name] = f"{h_prefix}{token}".strip()
        else:
            h_name = self.config.auth_header_name or "API-KEY"
            h_prefix = self.config.auth_header_prefix or ""
            headers[h_name] = f"{h_prefix}{token}".strip()
        return headers

    def _build_payload(
        self,
        to: str | list[str],
        body_params: list[str] | None = None,
        button_params: list[str] | None = None,
        template_name: str | None = None,
        language: str | None = None,
        raw_text: str | None = None,
    ) -> dict[str, Any]:
        recipients = [to] if isinstance(to, str) else list(to)
        normalized = [self.normalize_phone(r) for r in recipients if self.normalize_phone(r)]
        if not normalized:
            raise ValueError("No valid recipient phone numbers provided.")

        tpl = template_name or self.config.default_template
        lang = language or self.config.default_language

        if self.config.provider == "meta":
            # Meta Graph Cloud API payload structure
            if tpl:
                params_obj = [{"type": "text", "text": p} for p in (body_params or [])]
                return {
                    "messaging_product": "whatsapp",
                    "recipient_type": "individual",
                    "to": normalized[0],
                    "type": "template",
                    "template": {
                        "name": tpl,
                        "language": {"code": lang},
                        "components": [{"type": "body", "parameters": params_obj}],
                    },
                }
            else:
                return {
                    "messaging_product": "whatsapp",
                    "recipient_type": "individual",
                    "to": normalized[0],
                    "type": "text",
                    "text": {
                        "preview_url": False,
                        "body": raw_text or "",
                    },
                }

        # Generic REST gateway payload
        payload: dict[str, Any] = {
            "api_key": self.config.token or self.config.api_key,
            "to": normalized,
            "type": "template" if tpl else "text",
            "template_name": tpl,
            "language": lang,
            "body_params": body_params or [],
        }
        if raw_text:
            payload["message"] = raw_text
        if button_params:
            payload["button_params"] = button_params
        return payload

    async def send_template_async(
        self,
        to: str | list[str],
        body_params: list[str] | None = None,
        button_params: list[str] | None = None,
        template_name: str | None = None,
        language: str | None = None,
    ) -> dict[str, Any]:
        """Sends a WhatsApp template message asynchronously via HTTP POST."""
        if not HTTPX_AVAILABLE or httpx is None:
            raise RuntimeError("httpx is required for WhatsApp messaging. Run: pip install httpx")
        service_url = self.config.service_url
        if not service_url:
            raise ValueError(
                "WhatsApp service_url or phone_number_id is not configured. "
                "For Meta Cloud API, provide 'phone_number_id' and 'token'. "
                "For custom gateways, provide 'service_url' and 'api_key'."
            )

        payload = self._build_payload(
            to=to,
            body_params=body_params,
            button_params=button_params,
            template_name=template_name,
            language=language,
        )
        headers = self._get_headers()

        try:
            async with httpx.AsyncClient(timeout=self.config.timeout) as client:
                resp = await client.post(service_url, json=payload, headers=headers)
                resp.raise_for_status()
                try:
                    return resp.json()
                except Exception:
                    return {"status": "success", "raw": resp.text}
        except httpx.HTTPStatusError as exc:
            err_text = exc.response.text
            try:
                err_json = exc.response.json()
                if "error" in err_json and isinstance(err_json["error"], dict):
                    err_text = err_json["error"].get("message", err_text)
            except Exception:
                pass
            return {
                "status": "error",
                "error": f"HTTP {exc.response.status_code}: {err_text}",
                "status_code": exc.response.status_code,
            }
        except Exception as exc:
            return {"status": "error", "error": str(exc)}

    async def send_text_async(
        self,
        to: str | list[str],
        text: str,
    ) -> dict[str, Any]:
        """Sends a direct text WhatsApp message asynchronously."""
        if not HTTPX_AVAILABLE or httpx is None:
            raise RuntimeError("httpx is required for WhatsApp messaging. Run: pip install httpx")
        service_url = self.config.service_url
        if not service_url:
            raise ValueError(
                "WhatsApp service_url or phone_number_id is not configured. "
                "For Meta Cloud API, provide 'phone_number_id' and 'token'. "
                "For custom gateways, provide 'service_url' and 'api_key'."
            )

        payload = self._build_payload(to=to, raw_text=text)
        headers = self._get_headers()

        try:
            async with httpx.AsyncClient(timeout=self.config.timeout) as client:
                resp = await client.post(service_url, json=payload, headers=headers)
                resp.raise_for_status()
                try:
                    return resp.json()
                except Exception:
                    return {"status": "success", "raw": resp.text}
        except httpx.HTTPStatusError as exc:
            err_text = exc.response.text
            try:
                err_json = exc.response.json()
                if "error" in err_json and isinstance(err_json["error"], dict):
                    err_text = err_json["error"].get("message", err_text)
            except Exception:
                pass
            return {
                "status": "error",
                "error": f"HTTP {exc.response.status_code}: {err_text}",
                "status_code": exc.response.status_code,
            }
        except Exception as exc:
            return {"status": "error", "error": str(exc)}

normalize_phone 

normalize_phone(raw)

Validates and standardizes a phone number using the configured country prefix. Filters out dummy, invalid, or repeating phone sequences.

Source code in edgevoiceai/notifications/whatsapp.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def normalize_phone(self, raw: str) -> str:
    """
    Validates and standardizes a phone number using the configured country prefix.
    Filters out dummy, invalid, or repeating phone sequences.
    """
    if not raw:
        return ""
    digits = re.sub(r"\D", "", raw)
    prefix = self.config.phone_prefix

    # If already starts with prefix (e.g., 919876543210 -> 9876543210)
    if prefix and digits.startswith(prefix) and len(digits) == 10 + len(prefix):
        digits = digits[len(prefix):]
    elif digits.startswith("0") and len(digits) == 11:
        digits = digits[1:]

    # Must be standard 10 mobile digits
    if len(digits) != 10:
        return ""
    # Filter trivial / dummy numbers
    if len(set(digits)) <= 2:
        return ""
    if digits in ("1234567890", "0123456789", "9876543210", "0000000000"):
        return ""
    # Valid starting digits
    if digits[0] not in "56789":
        return ""

    return (prefix + digits) if prefix else digits

send_template_async async 

send_template_async(to, body_params=None, button_params=None, template_name=None, language=None)

Sends a WhatsApp template message asynchronously via HTTP POST.

Source code in edgevoiceai/notifications/whatsapp.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
async def send_template_async(
    self,
    to: str | list[str],
    body_params: list[str] | None = None,
    button_params: list[str] | None = None,
    template_name: str | None = None,
    language: str | None = None,
) -> dict[str, Any]:
    """Sends a WhatsApp template message asynchronously via HTTP POST."""
    if not HTTPX_AVAILABLE or httpx is None:
        raise RuntimeError("httpx is required for WhatsApp messaging. Run: pip install httpx")
    service_url = self.config.service_url
    if not service_url:
        raise ValueError(
            "WhatsApp service_url or phone_number_id is not configured. "
            "For Meta Cloud API, provide 'phone_number_id' and 'token'. "
            "For custom gateways, provide 'service_url' and 'api_key'."
        )

    payload = self._build_payload(
        to=to,
        body_params=body_params,
        button_params=button_params,
        template_name=template_name,
        language=language,
    )
    headers = self._get_headers()

    try:
        async with httpx.AsyncClient(timeout=self.config.timeout) as client:
            resp = await client.post(service_url, json=payload, headers=headers)
            resp.raise_for_status()
            try:
                return resp.json()
            except Exception:
                return {"status": "success", "raw": resp.text}
    except httpx.HTTPStatusError as exc:
        err_text = exc.response.text
        try:
            err_json = exc.response.json()
            if "error" in err_json and isinstance(err_json["error"], dict):
                err_text = err_json["error"].get("message", err_text)
        except Exception:
            pass
        return {
            "status": "error",
            "error": f"HTTP {exc.response.status_code}: {err_text}",
            "status_code": exc.response.status_code,
        }
    except Exception as exc:
        return {"status": "error", "error": str(exc)}

send_text_async async 

send_text_async(to, text)

Sends a direct text WhatsApp message asynchronously.

Source code in edgevoiceai/notifications/whatsapp.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
async def send_text_async(
    self,
    to: str | list[str],
    text: str,
) -> dict[str, Any]:
    """Sends a direct text WhatsApp message asynchronously."""
    if not HTTPX_AVAILABLE or httpx is None:
        raise RuntimeError("httpx is required for WhatsApp messaging. Run: pip install httpx")
    service_url = self.config.service_url
    if not service_url:
        raise ValueError(
            "WhatsApp service_url or phone_number_id is not configured. "
            "For Meta Cloud API, provide 'phone_number_id' and 'token'. "
            "For custom gateways, provide 'service_url' and 'api_key'."
        )

    payload = self._build_payload(to=to, raw_text=text)
    headers = self._get_headers()

    try:
        async with httpx.AsyncClient(timeout=self.config.timeout) as client:
            resp = await client.post(service_url, json=payload, headers=headers)
            resp.raise_for_status()
            try:
                return resp.json()
            except Exception:
                return {"status": "success", "raw": resp.text}
    except httpx.HTTPStatusError as exc:
        err_text = exc.response.text
        try:
            err_json = exc.response.json()
            if "error" in err_json and isinstance(err_json["error"], dict):
                err_text = err_json["error"].get("message", err_text)
        except Exception:
            pass
        return {
            "status": "error",
            "error": f"HTTP {exc.response.status_code}: {err_text}",
            "status_code": exc.response.status_code,
        }
    except Exception as exc:
        return {"status": "error", "error": str(exc)}

TwoTurnConfirmationNotificationTool

edgevoiceai.tools.notification_tool

Two-turn verbal confirmation notification tool for Voice AI. Prevents hallucinated or unsolicited messaging by strictly verifying recipient contact information across conversational turns before sending.

TwoTurnConfirmationNotificationTool

Bases: BaseTool

Guarded notification tool requiring explicit multi-turn confirmation before dispatch.

Source code in edgevoiceai/tools/notification_tool.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class TwoTurnConfirmationNotificationTool(BaseTool):
    """
    Guarded notification tool requiring explicit multi-turn confirmation before dispatch.
    """

    def __init__(
        self,
        name: str,
        description: str,
        parameters_schema: Dict[str, Any],
        sender_func: Callable[[str, str], Coroutine[Any, Any, Dict[str, Any]]],
        normalize_phone_func: Callable[[str], str] = default_normalize_phone,
        format_text_func: Optional[Callable[[str], str]] = None,
        phone_param: str = "phone_number",
        text_param: str = "answer_text",
        invalid_phone_message_builder: Optional[Callable[[str], str]] = None,
        confirmation_needed_message_builder: Optional[Callable[[str, str], str]] = None,
        success_message_builder: Optional[Callable[[str], str]] = None,
        pending_confirmations_dict: Optional[Dict[str, Dict[str, Any]]] = None,
        turn_counter_dict: Optional[Dict[str, int]] = None,
        prompt_instruction: Optional[str] = None,
    ):
        self._name = name
        self._description = description
        self._parameters_schema = parameters_schema
        self.sender_func = sender_func
        self.normalize_phone_func = normalize_phone_func
        self.format_text_func = format_text_func or (lambda t: t)
        self.phone_param = phone_param
        self.text_param = text_param
        self.invalid_phone_message_builder = invalid_phone_message_builder
        self.confirmation_needed_message_builder = confirmation_needed_message_builder
        self.success_message_builder = success_message_builder
        self._prompt_instruction = prompt_instruction or (
            f"• NOTIFICATION & MESSAGING PROTOCOL (`{self._name}`):\n"
            f"  1. When the user asks to send an answer or provides their phone number, DO NOT call `{self._name}` in that turn.\n"
            f"  2. First say verbally: 'Your phone number is [phone_number], correct?' and stop speaking.\n"
            f"  3. Only in the subsequent turn, after the user confirms (e.g. 'yes/correct'), call `{self._name}`.\n"
            "  4. Never invent or use dummy phone numbers like '0000000000'."
        )

        # In-memory tracking of pending confirmations: session_id -> confirmation dict
        self._pending_confirmations = pending_confirmations_dict if pending_confirmations_dict is not None else {}
        self.turn_counter_dict = turn_counter_dict

    @property
    def name(self) -> str:
        return self._name

    @property
    def description(self) -> str:
        return self._description

    @property
    def parameters_schema(self) -> Dict[str, Any]:
        return self._parameters_schema

    @property
    def prompt_instruction(self) -> Optional[str]:
        return self._prompt_instruction

    @property
    def pending_confirmations(self) -> Dict[str, Dict[str, Any]]:
        return self._pending_confirmations


    def get_pending(self, session_id: str) -> Optional[Dict[str, Any]]:
        return self._pending_confirmations.get(session_id)

    def clear_pending(self, session_id: str) -> None:
        self._pending_confirmations.pop(session_id, None)

    async def execute(self, args: Dict[str, Any], context: ToolExecutionContext) -> ToolResult:
        phone_raw = args.get(self.phone_param, "")
        text_raw = args.get(self.text_param, "")
        formatted_text = self.format_text_func(text_raw)

        phone = self.normalize_phone_func(phone_raw)
        if not phone:
            print(f"📲 [{self.name}] ❌ Blocked invalid/dummy phone: {phone_raw!r}")
            if self.invalid_phone_message_builder:
                msg = self.invalid_phone_message_builder(phone_raw)
            else:
                msg = (
                    f"Invalid or missing phone number: '{phone_raw}'. "
                    "A valid 10-digit mobile number was not found. "
                    "Please ask the user for their 10-digit phone number, "
                    "and verbally confirm it with them before sending. "
                    "Never use fictional or dummy numbers."
                )
            return ToolResult(status="error", message=msg)

        curr_turn = context.current_turn
        if self.turn_counter_dict is not None and context.session_id in self.turn_counter_dict:
            curr_turn = self.turn_counter_dict[context.session_id]

        pending = self._pending_confirmations.get(context.session_id)

        # Strict 2-turn confirmation gate:
        # Message is ONLY dispatched if confirmation was previously requested
        # in an earlier turn (< curr_turn) and the phone matches.
        is_confirmed = False
        if pending and pending.get("phone") == phone:
            if pending.get("asked_at_turn", -1) < curr_turn:
                is_confirmed = True

        if not is_confirmed:
            self._pending_confirmations[context.session_id] = {
                "phone": phone,
                "answer_text": formatted_text,
                "asked_at_turn": curr_turn,
                "timestamp": time.time(),
            }
            print(f"📲 [{self.name} Gate] ⏳ Confirmation requested for {phone} (Turn {curr_turn}). Holding send...")
            if self.confirmation_needed_message_builder:
                conf_msg = self.confirmation_needed_message_builder(phone, phone_raw)
            else:
                conf_msg = (
                    f"Phone number {phone} recorded, but notification is not sent yet. "
                    f"Please say to the user: 'Your phone number is {phone_raw}, right?' and stop speaking. "
                    "Wait for the user to confirm in the next turn before calling this tool again."
                )
            return ToolResult(status="confirmation_required", message=conf_msg)

        # Confirmed! Proceed to send
        self._pending_confirmations.pop(context.session_id, None)
        try:
            send_res = await self.sender_func(phone, formatted_text)
            if send_res.get("status") == "error":
                err_msg = f"Notification send failed: {send_res.get('error', 'Unknown error')}"
                print(f"📲 [{self.name}] Send failed to {phone}: {send_res}")
                return ToolResult(status="error", message=err_msg)
            else:
                if self.success_message_builder:
                    succ_msg = self.success_message_builder(phone)
                else:
                    succ_msg = f"Notification successfully sent to {phone}."
                print(f"📲 [{self.name}] Sent to {phone}: {formatted_text[:60]}...")
                return ToolResult(status="sent", message=succ_msg)

        except Exception as exc:
            print(f"📲 [{self.name}] Exception while sending: {exc}")
            return ToolResult(status="error", message=f"Notification error: {exc}")