Skip to content

Adapters

WhatsApp Business

adapters.whatsapp_business.client

get_text_message_input 

get_text_message_input(to, body, preview_url=False)
Source code in src/adapters/whatsapp_business/client.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_text_message_input(
    to: str, body: str, preview_url: bool = False
) -> Dict[str, Any]:
    safe_body = (body or "").strip() or " "
    if len(safe_body) > 4000:
        safe_body = safe_body[:4000] + "…"
    return {
        "messaging_product": "whatsapp",
        "recipient_type": "individual",
        "to": _normalize_to(to),
        "type": "text",
        "text": {"preview_url": preview_url, "body": safe_body},
    }

send_message async 

send_message(
    payload, *, timeout=30.0, retries=2, backoff=1.5
)

Envía un payload ya construido (dict). Devuelve la respuesta JSON (o lanza excepción con logging).

Source code in src/adapters/whatsapp_business/client.py 
 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
async def send_message(
    payload: Dict[str, Any],
    *,
    timeout: float = 30.0,
    retries: int = 2,
    backoff: float = 1.5,
) -> Dict[str, Any]:
    """
    Envía un payload ya construido (dict).
    Devuelve la respuesta JSON (o lanza excepción con logging).
    """
    url, headers = _endpoint(), _headers()

    # Debug de request
    try:
        to_dbg = payload.get("to") or "<no-to>"
        msg_type = payload.get("type")
        body_dbg = ""
        if msg_type == "text":
            body_dbg = _shorten(payload.get("text", {}).get("body", ""))
        logging.info(
            "WA SEND -> %s to=%s type=%s body=%r", url, to_dbg, msg_type, body_dbg
        )
    except Exception:
        logging.info("WA SEND -> %s payload_keys=%s", url, list(payload.keys()))

    attempt = 0
    last_exc: Exception | None = None
    async with httpx.AsyncClient(timeout=timeout) as client:
        while attempt <= retries:
            attempt += 1
            try:
                resp = await client.post(url, headers=headers, json=payload)
                try:
                    data = resp.json()
                except Exception:
                    data = {"raw": resp.text}

                if resp.status_code >= 400:
                    logging.error("WA RESP <- %s %s", resp.status_code, data)
                    if attempt <= retries and _should_retry(resp.status_code):
                        await asyncio.sleep(backoff ** (attempt - 1))
                        continue
                    resp.raise_for_status()
                else:
                    logging.info("WA RESP <- %s %s", resp.status_code, data)
                    return data

            except (httpx.TimeoutException, httpx.ReadTimeout) as e:
                logging.error("WA TIMEOUT (%d/%d): %s", attempt, retries, e)
                last_exc = e
                if attempt <= retries:
                    await asyncio.sleep(backoff ** (attempt - 1))
                    continue
                raise
            except httpx.RequestError as e:
                logging.error("WA REQUEST ERROR (%d/%d): %s", attempt, retries, e)
                last_exc = e
                if attempt <= retries:
                    await asyncio.sleep(backoff ** (attempt - 1))
                    continue
                raise

    if last_exc:
        raise last_exc
    raise RuntimeError("send_message: unknown error")

send_catalog_message async 

send_catalog_message(msg, **kwargs)

Envía un mensaje definido con el catálogo (OutgoingMessage + component).

Source code in src/adapters/whatsapp_business/client.py 
139
140
141
142
143
144
async def send_catalog_message(msg: OutgoingMessage, **kwargs) -> Dict[str, Any]:
    """
    Envía un mensaje definido con el catálogo (OutgoingMessage + component).
    """
    payload = msg.build()
    return await send_message(payload, **kwargs)

adapters.whatsapp_business.cards

dummy_card_message 

dummy_card_message(to)

Crea una tarjeta dummy de ejemplo, útil para pruebas rápidas.

Source code in src/adapters/whatsapp_business/cards.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def dummy_card_message(to: str) -> OutgoingMessage:
    """
    Crea una tarjeta dummy de ejemplo, útil para pruebas rápidas.
    """
    body_text = (
        "👋 Hola!\n"
        "Soy una *tarjeta de prueba*.\n\n"
        "Este es un mensaje dummy para validar el envío de cards."
    )

    header = MediaHeader(
        kind="image",
        link="https://placekitten.com/400/300",  # Imagen de gatito dummy
    )

    buttons = [
        ReplyButton(id="dummy_ok", title="👍 Funciona"),
        ReplyButton(id="dummy_more", title="ℹ️ Más info"),
    ]

    footer_text = "Mensaje generado desde el dummy card"

    return OutgoingMessage(
        to=to,
        component=ButtonsMessage(
            body_text=body_text,
            header=header,
            buttons=buttons,
            footer_text=footer_text,
        ),
    )

adapters.whatsapp_business.catalog

MessageComponent

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
10
11
12
class MessageComponent(BaseModel):
    def to_payload(self) -> Dict[str, Any]:
        raise NotImplementedError

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
11
12
def to_payload(self) -> Dict[str, Any]:
    raise NotImplementedError

TextMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
18
19
20
21
22
23
24
25
26
class TextMessage(MessageComponent):
    body: str
    preview_url: bool = False

    def to_payload(self) -> Dict[str, Any]:
        return {
            "type": "text",
            "text": {"body": self.body, "preview_url": self.preview_url},
        }

body instance-attribute 

body

preview_url class-attribute instance-attribute 

preview_url = False

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
22
23
24
25
26
def to_payload(self) -> Dict[str, Any]:
    return {
        "type": "text",
        "text": {"body": self.body, "preview_url": self.preview_url},
    }

ImageMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
32
33
34
35
36
37
38
39
40
class ImageMessage(MessageComponent):
    link: HttpUrl
    caption: Optional[str] = None

    def to_payload(self) -> Dict[str, Any]:
        p: Dict[str, Any] = {"type": "image", "image": {"link": str(self.link)}}
        if self.caption:
            p["image"]["caption"] = self.caption
        return p

link

caption class-attribute instance-attribute 

caption = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
36
37
38
39
40
def to_payload(self) -> Dict[str, Any]:
    p: Dict[str, Any] = {"type": "image", "image": {"link": str(self.link)}}
    if self.caption:
        p["image"]["caption"] = self.caption
    return p

DocumentMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
46
47
48
49
50
51
52
53
54
55
56
57
class DocumentMessage(MessageComponent):
    link: HttpUrl
    filename: Optional[str] = None
    caption: Optional[str] = None

    def to_payload(self) -> Dict[str, Any]:
        doc: Dict[str, Any] = {"link": str(self.link)}
        if self.filename:
            doc["filename"] = self.filename
        if self.caption:
            doc["caption"] = self.caption
        return {"type": "document", "document": doc}

link

filename class-attribute instance-attribute 

filename = None

caption class-attribute instance-attribute 

caption = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
51
52
53
54
55
56
57
def to_payload(self) -> Dict[str, Any]:
    doc: Dict[str, Any] = {"link": str(self.link)}
    if self.filename:
        doc["filename"] = self.filename
    if self.caption:
        doc["caption"] = self.caption
    return {"type": "document", "document": doc}

ReplyButton

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
63
64
65
66
67
68
69
70
71
72
class ReplyButton(BaseModel):
    id: str
    title: str

    @field_validator("title")
    @classmethod
    def _title_len(cls, v: str) -> str:
        if len(v) > 20:
            raise ValueError("El título del botón debe tener ≤ 20 caracteres.")
        return v

id instance-attribute 

id

title instance-attribute 

title

MediaHeader

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class MediaHeader(BaseModel):
    kind: Literal["text", "image", "video", "document"]
    text: Optional[str] = None
    link: Optional[HttpUrl] = None

    @field_validator("text")
    @classmethod
    def _text_required_for_text_kind(cls, v, info):
        if info.data.get("kind") == "text" and not v:
            raise ValueError("Header 'text' requiere 'text'.")
        return v

    @field_validator("link")
    @classmethod
    def _link_required_for_media_kind(cls, v, info):
        if info.data.get("kind") in ("image", "video", "document") and not v:
            raise ValueError("Header media requiere 'link'.")
        return v

kind instance-attribute 

kind

text class-attribute instance-attribute 

text = None

link = None

ButtonsMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
 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
class ButtonsMessage(MessageComponent):
    body_text: str
    footer_text: Optional[str] = None
    buttons: List[ReplyButton]
    header: Optional[MediaHeader] = None

    def to_payload(self) -> Dict[str, Any]:
        buttons_payload = [
            {"type": "reply", "reply": b.model_dump()} for b in self.buttons
        ]
        interactive: Dict[str, Any] = {
            "type": "button",
            "body": {"text": self.body_text},
            "action": {"buttons": buttons_payload},
        }
        if self.footer_text:
            interactive["footer"] = {"text": self.footer_text}
        if self.header:
            if self.header.kind == "text":
                interactive["header"] = {"type": "text", "text": self.header.text}
            elif self.header.kind == "image":
                interactive["header"] = {
                    "type": "image",
                    "image": {"link": str(self.header.link)},
                }
            elif self.header.kind == "video":
                interactive["header"] = {
                    "type": "video",
                    "video": {"link": str(self.header.link)},
                }
            elif self.header.kind == "document":
                interactive["header"] = {
                    "type": "document",
                    "document": {"link": str(self.header.link)},
                }
        return {"type": "interactive", "interactive": interactive}

body_text instance-attribute 

body_text

footer_text class-attribute instance-attribute 

footer_text = None

buttons instance-attribute 

buttons

header class-attribute instance-attribute 

header = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
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
def to_payload(self) -> Dict[str, Any]:
    buttons_payload = [
        {"type": "reply", "reply": b.model_dump()} for b in self.buttons
    ]
    interactive: Dict[str, Any] = {
        "type": "button",
        "body": {"text": self.body_text},
        "action": {"buttons": buttons_payload},
    }
    if self.footer_text:
        interactive["footer"] = {"text": self.footer_text}
    if self.header:
        if self.header.kind == "text":
            interactive["header"] = {"type": "text", "text": self.header.text}
        elif self.header.kind == "image":
            interactive["header"] = {
                "type": "image",
                "image": {"link": str(self.header.link)},
            }
        elif self.header.kind == "video":
            interactive["header"] = {
                "type": "video",
                "video": {"link": str(self.header.link)},
            }
        elif self.header.kind == "document":
            interactive["header"] = {
                "type": "document",
                "document": {"link": str(self.header.link)},
            }
    return {"type": "interactive", "interactive": interactive}

VideoMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
136
137
138
139
140
141
142
143
144
class VideoMessage(MessageComponent):
    link: HttpUrl
    caption: Optional[str] = None  # opcional; no confundir con header de interactivos

    def to_payload(self) -> Dict[str, Any]:
        p: Dict[str, Any] = {"type": "video", "video": {"link": str(self.link)}}
        if self.caption:
            p["video"]["caption"] = self.caption
        return p

link

caption class-attribute instance-attribute 

caption = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
140
141
142
143
144
def to_payload(self) -> Dict[str, Any]:
    p: Dict[str, Any] = {"type": "video", "video": {"link": str(self.link)}}
    if self.caption:
        p["video"]["caption"] = self.caption
    return p

AudioMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
150
151
152
153
154
class AudioMessage(MessageComponent):
    link: HttpUrl  # MP3/OGG/OPUS soportados por WA Cloud

    def to_payload(self) -> Dict[str, Any]:
        return {"type": "audio", "audio": {"link": str(self.link)}}

link

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
153
154
def to_payload(self) -> Dict[str, Any]:
    return {"type": "audio", "audio": {"link": str(self.link)}}

StickerMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
160
161
162
163
164
class StickerMessage(MessageComponent):
    link: HttpUrl  # webp público

    def to_payload(self) -> Dict[str, Any]:
        return {"type": "sticker", "sticker": {"link": str(self.link)}}

link

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
163
164
def to_payload(self) -> Dict[str, Any]:
    return {"type": "sticker", "sticker": {"link": str(self.link)}}

LocationMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
class LocationMessage(MessageComponent):
    latitude: float
    longitude: float
    name: Optional[str] = None
    address: Optional[str] = None

    def to_payload(self) -> Dict[str, Any]:
        loc: Dict[str, Any] = {"latitude": self.latitude, "longitude": self.longitude}
        if self.name:
            loc["name"] = self.name
        if self.address:
            loc["address"] = self.address
        return {"type": "location", "location": loc}

latitude instance-attribute 

latitude

longitude instance-attribute 

longitude

name class-attribute instance-attribute 

name = None

address class-attribute instance-attribute 

address = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
176
177
178
179
180
181
182
def to_payload(self) -> Dict[str, Any]:
    loc: Dict[str, Any] = {"latitude": self.latitude, "longitude": self.longitude}
    if self.name:
        loc["name"] = self.name
    if self.address:
        loc["address"] = self.address
    return {"type": "location", "location": loc}

ContactPhone

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
188
189
190
191
class ContactPhone(BaseModel):
    phone: str
    type: Optional[Literal["CELL", "WORK", "HOME"]] = None
    wa_id: Optional[str] = None  # si ya es usuario de WA, opcional

phone instance-attribute 

phone

type class-attribute instance-attribute 

type = None

wa_id class-attribute instance-attribute 

wa_id = None

ContactName

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
194
195
196
197
class ContactName(BaseModel):
    formatted_name: str
    first_name: Optional[str] = None
    last_name: Optional[str] = None

formatted_name instance-attribute 

formatted_name

first_name class-attribute instance-attribute 

first_name = None

last_name class-attribute instance-attribute 

last_name = None

Contact

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
200
201
202
class Contact(BaseModel):
    name: ContactName
    phones: List[ContactPhone]

name instance-attribute 

name

phones instance-attribute 

phones

ContactsMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
205
206
207
208
209
210
211
212
class ContactsMessage(MessageComponent):
    contacts: List[Contact]

    def to_payload(self) -> Dict[str, Any]:
        return {
            "type": "contacts",
            "contacts": [c.model_dump(exclude_none=True) for c in self.contacts],
        }

contacts instance-attribute 

contacts

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
208
209
210
211
212
def to_payload(self) -> Dict[str, Any]:
    return {
        "type": "contacts",
        "contacts": [c.model_dump(exclude_none=True) for c in self.contacts],
    }

ReactionMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
218
219
220
221
222
223
224
225
226
class ReactionMessage(MessageComponent):
    message_id: str  # wamid del mensaje al que reaccionas
    emoji: str  # p.ej. "👍", "❤️"

    def to_payload(self) -> Dict[str, Any]:
        return {
            "type": "reaction",
            "reaction": {"message_id": self.message_id, "emoji": self.emoji},
        }

message_id instance-attribute 

message_id

emoji instance-attribute 

emoji

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
222
223
224
225
226
def to_payload(self) -> Dict[str, Any]:
    return {
        "type": "reaction",
        "reaction": {"message_id": self.message_id, "emoji": self.emoji},
    }

ListSectionRow

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
232
233
234
235
class ListSectionRow(BaseModel):
    id: str
    title: str
    description: Optional[str] = None

id instance-attribute 

id

title instance-attribute 

title

description class-attribute instance-attribute 

description = None

ListSection

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
238
239
240
class ListSection(BaseModel):
    title: str
    rows: List[ListSectionRow]

title instance-attribute 

title

rows instance-attribute 

rows

ListMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
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
class ListMessage(MessageComponent):
    body_text: str
    button_text: str
    sections: List[ListSection]
    header_text: Optional[str] = None
    footer_text: Optional[str] = None

    @field_validator("button_text")
    @classmethod
    def _btn_len(cls, v: str) -> str:
        if len(v) > 20:
            raise ValueError("El texto del botón de lista debe tener ≤ 20 caracteres.")
        return v

    def to_payload(self) -> Dict[str, Any]:
        interactive: Dict[str, Any] = {
            "type": "list",
            "body": {"text": self.body_text},
            "action": {
                "button": self.button_text,
                "sections": [
                    {
                        "title": s.title,
                        "rows": [r.model_dump(exclude_none=True) for r in s.rows],
                    }
                    for s in self.sections
                ],
            },
        }
        if self.header_text:
            interactive["header"] = {"type": "text", "text": self.header_text}
        if self.footer_text:
            interactive["footer"] = {"text": self.footer_text}
        return {"type": "interactive", "interactive": interactive}

body_text instance-attribute 

body_text

button_text instance-attribute 

button_text

sections instance-attribute 

sections

header_text class-attribute instance-attribute 

header_text = None

footer_text class-attribute instance-attribute 

footer_text = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def to_payload(self) -> Dict[str, Any]:
    interactive: Dict[str, Any] = {
        "type": "list",
        "body": {"text": self.body_text},
        "action": {
            "button": self.button_text,
            "sections": [
                {
                    "title": s.title,
                    "rows": [r.model_dump(exclude_none=True) for r in s.rows],
                }
                for s in self.sections
            ],
        },
    }
    if self.header_text:
        interactive["header"] = {"type": "text", "text": self.header_text}
    if self.footer_text:
        interactive["footer"] = {"text": self.footer_text}
    return {"type": "interactive", "interactive": interactive}

TemplateMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
class TemplateMessage(MessageComponent):
    name: str
    language_code: str = "en_US"
    components: Optional[List[Dict[str, Any]]] = None

    def to_payload(self) -> Dict[str, Any]:
        t: Dict[str, Any] = {
            "type": "template",
            "template": {"name": self.name, "language": {"code": self.language_code}},
        }
        if self.components:
            t["template"]["components"] = self.components
        return t

name instance-attribute 

name

language_code class-attribute instance-attribute 

language_code = 'en_US'

components class-attribute instance-attribute 

components = None

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
287
288
289
290
291
292
293
294
def to_payload(self) -> Dict[str, Any]:
    t: Dict[str, Any] = {
        "type": "template",
        "template": {"name": self.name, "language": {"code": self.language_code}},
    }
    if self.components:
        t["template"]["components"] = self.components
    return t

BusinessCardMessage

Bases: MessageComponent

Source code in src/adapters/whatsapp_business/catalog.py 
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
class BusinessCardMessage(MessageComponent):
    name: str
    address: str
    phone: str
    url: str
    image: HttpUrl

    def to_payload(self) -> Dict[str, Any]:
        # This is a simplified representation.
        # WhatsApp doesn't have a dedicated "business card" message type.
        # We can use a text message with a formatted body and a header image.

        # Create a formatted text body
        body = f"*{self.name}*\n{self.address}\n{self.phone}\n{self.url}"

        # Use a ButtonsMessage with a header image and a link button
        interactive = {
            "type": "button",
            "header": {"type": "image", "image": {"link": str(self.image)}},
            "body": {"text": body},
            "action": {
                "buttons": [
                    {
                        "type": "reply",
                        "reply": {"id": "view_website", "title": "Visit Website"},
                    }
                ]
            },
        }
        return {"type": "interactive", "interactive": interactive}

name instance-attribute 

name

address instance-attribute 

address

phone instance-attribute 

phone

url instance-attribute 

url

image instance-attribute 

image

to_payload 

to_payload()
Source code in src/adapters/whatsapp_business/catalog.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def to_payload(self) -> Dict[str, Any]:
    # This is a simplified representation.
    # WhatsApp doesn't have a dedicated "business card" message type.
    # We can use a text message with a formatted body and a header image.

    # Create a formatted text body
    body = f"*{self.name}*\n{self.address}\n{self.phone}\n{self.url}"

    # Use a ButtonsMessage with a header image and a link button
    interactive = {
        "type": "button",
        "header": {"type": "image", "image": {"link": str(self.image)}},
        "body": {"text": body},
        "action": {
            "buttons": [
                {
                    "type": "reply",
                    "reply": {"id": "view_website", "title": "Visit Website"},
                }
            ]
        },
    }
    return {"type": "interactive", "interactive": interactive}

OutgoingMessage

Bases: BaseModel

Source code in src/adapters/whatsapp_business/catalog.py 
332
333
334
335
336
337
338
339
340
341
342
class OutgoingMessage(BaseModel):
    to: str
    component: MessageComponent
    context_message_id: Optional[str] = None

    def build(self) -> Dict[str, Any]:
        root: Dict[str, Any] = {"messaging_product": "whatsapp", "to": self.to}
        root.update(self.component.to_payload())
        if self.context_message_id:
            root["context"] = {"message_id": self.context_message_id}
        return root

to instance-attribute 

to

component instance-attribute 

component

context_message_id class-attribute instance-attribute 

context_message_id = None

build 

build()
Source code in src/adapters/whatsapp_business/catalog.py 
337
338
339
340
341
342
def build(self) -> Dict[str, Any]:
    root: Dict[str, Any] = {"messaging_product": "whatsapp", "to": self.to}
    root.update(self.component.to_payload())
    if self.context_message_id:
        root["context"] = {"message_id": self.context_message_id}
    return root

adapters.whatsapp_business.models

WhatsAppBase

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
5
6
class WhatsAppBase(BaseModel):
    object: str

object instance-attribute 

object

Change

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
 9
10
11
class Change(BaseModel):
    value: dict
    field: str

value instance-attribute 

value

field instance-attribute 

field

Entry

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
14
15
16
class Entry(BaseModel):
    id: str
    changes: List[Change]

id instance-attribute 

id

changes instance-attribute 

changes

Webhook

Bases: WhatsAppBase

Source code in src/adapters/whatsapp_business/models.py 
19
20
class Webhook(WhatsAppBase):
    entry: List[Entry]

entry instance-attribute 

entry

object instance-attribute 

object

Message

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
23
24
25
26
27
28
class Message(BaseModel):
    from_number: str = Field(..., alias="from")
    id: str
    timestamp: str
    text: dict
    type: str

from_number class-attribute instance-attribute 

from_number = Field(..., alias='from')

id instance-attribute 

id

timestamp instance-attribute 

timestamp

text instance-attribute 

text

type instance-attribute 

type

Metadata

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
31
32
33
class Metadata(BaseModel):
    display_phone_number: str
    phone_number_id: str

display_phone_number instance-attribute 

display_phone_number

phone_number_id instance-attribute 

phone_number_id

Value

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
36
37
38
39
40
class Value(BaseModel):
    messaging_product: str
    metadata: Metadata
    contacts: List[dict]
    messages: List[Message]

messaging_product instance-attribute 

messaging_product

metadata instance-attribute 

metadata

contacts instance-attribute 

contacts

messages instance-attribute 

messages

OutgoingMessage

Bases: BaseModel

Source code in src/adapters/whatsapp_business/models.py 
43
44
45
46
47
class OutgoingMessage(BaseModel):
    messaging_product: str = "whatsapp"
    to: str
    type: str
    text: Optional[dict] = None

messaging_product class-attribute instance-attribute 

messaging_product = 'whatsapp'

to instance-attribute 

to

type instance-attribute 

type

text class-attribute instance-attribute 

text = None

adapters.whatsapp_business.sender

WhatsAppCatalogSender

Bases: CatalogSender

Source code in src/adapters/whatsapp_business/sender.py 
6
7
8
class WhatsAppCatalogSender(CatalogSender):
    async def send_catalog_message(self, payload: OutgoingMessage) -> None:
        await _send(payload)

send_catalog_message async 

send_catalog_message(payload)
Source code in src/adapters/whatsapp_business/sender.py 
7
8
async def send_catalog_message(self, payload: OutgoingMessage) -> None:
    await _send(payload)

adapters.whatsapp_business.handler

router module-attribute 

router = APIRouter()

get_memory_manager 

get_memory_manager()
Source code in src/adapters/whatsapp_business/handler.py 
28
29
30
31
32
33
34
35
def get_memory_manager() -> MemoryManager:
    base_store = get_memory_store()
    stores: list[HistoryStore] = []
    if isinstance(base_store, list):
        stores.extend(base_store)
    elif base_store is not None:
        stores.append(base_store)
    return MemoryManager(store=stores)

verify_whatsapp_signature 

verify_whatsapp_signature(
    app_secret, raw_body, signature_header
)
Source code in src/adapters/whatsapp_business/handler.py 
42
43
44
45
46
47
48
49
50
def verify_whatsapp_signature(
    app_secret: str, raw_body: bytes, signature_header: str
) -> tuple[bool, str, str]:
    try:
        provided_sig = (signature_header or "").split("=", 1)[1]
    except Exception:
        return False, "", ""
    expected = _calc_sig(app_secret, raw_body)
    return hmac.compare_digest(expected, provided_sig), provided_sig, expected

verify_signature async 

verify_signature(request)
Source code in src/adapters/whatsapp_business/handler.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def verify_signature(request: Request):
    raw = await request.body()
    sig_hdr = request.headers.get("X-Hub-Signature-256", "")

    if not settings.meta_app_secret:
        logging.error("META_APP_SECRET no configurado")
        raise HTTPException(
            status_code=400, detail="Server misconfigured (no app secret)"
        )

    ok, provided, expected = verify_whatsapp_signature(
        settings.meta_app_secret, raw, sig_hdr
    )
    if not ok:
        logging.error(
            "Firma inválida: provided=%s expected=%s raw_len=%d first16=%s hdr=%r",
            (provided[:12] + "...") if provided else "<none>",
            (expected[:12] + "...") if expected else "<none>",
            len(raw),
            binascii.hexlify(raw[:16]).decode(),
            sig_hdr,
        )
        raise HTTPException(status_code=400, detail="Invalid signature")

pick_value 

pick_value(body)
Source code in src/adapters/whatsapp_business/handler.py 
78
79
80
81
82
def pick_value(body: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    try:
        return body["entry"][0]["changes"][0]["value"]
    except Exception:
        return None

pick_first_message 

pick_first_message(value)
Source code in src/adapters/whatsapp_business/handler.py 
85
86
87
def pick_first_message(value: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    msgs = value.get("messages") or []
    return msgs[0] if msgs else None

pick_contact 

pick_contact(value)
Source code in src/adapters/whatsapp_business/handler.py 
90
91
92
def pick_contact(value: Dict[str, Any]) -> Dict[str, Any]:
    contacts = value.get("contacts") or [{}]
    return contacts[0]

extract_user_text 

extract_user_text(msg)
Source code in src/adapters/whatsapp_business/handler.py 
 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
def extract_user_text(msg: Dict[str, Any]) -> str:
    t = msg.get("type")
    if t == "text":
        return msg["text"]["body"]

    if t == "interactive":
        inter = msg.get("interactive") or {}
        if "button_reply" in inter:
            r = inter["button_reply"]
            return f"[button:{r.get('id')}] {r.get('title')}"
        if "list_reply" in inter:
            r = inter["list_reply"]
            return f"[list:{r.get('id')}] {r.get('title')}"
        return "[interactive]"

    if t == "image":
        cap = (msg.get("image") or {}).get("caption")
        return cap or "[image]"

    if t == "reaction":
        return f"[reaction] {msg.get('reaction', {}).get('emoji')}"

    if t == "location":
        loc = msg.get("location") or {}
        return f"[location] {loc.get('latitude')},{loc.get('longitude')}"

    return f"[{t}]"

webhook_post async 

webhook_post(request, mm=Depends(get_memory_manager))
Source code in src/adapters/whatsapp_business/handler.py 
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
178
179
180
@router.post("/webhook", dependencies=[Depends(verify_signature)])
async def webhook_post(
    request: Request, mm: MemoryManager = Depends(get_memory_manager)
):
    try:
        body = await request.json()
    except Exception:
        logging.warning("POST /webhook: cuerpo no JSON o vacío")
        return Response(status_code=200)

    value = pick_value(body)
    if not value:
        logging.warning("POST /webhook: sin 'value' utilizable en payload")
        return Response(status_code=200)

    msg = pick_first_message(value)
    if not msg:
        logging.info("POST /webhook: 'messages' vacío")
        return Response(status_code=200)

    contact = pick_contact(value)
    wa_id = contact.get("wa_id")

    # Ensure wa_id is not None
    if not wa_id:
        logging.warning("POST /webhook: missing wa_id in contact")
        return Response(status_code=200)

    user_text = extract_user_text(msg)
    logging.info("IN: wa_id=%s kind=%s text=%r", wa_id, msg.get("type"), user_text)

    # Ejecuta tu pipeline
    try:
        reply_text, _ctx = await run_with_memory(
            graph,
            deps,
            mm,
            session_id=wa_id,
            user_text=user_text,
        )
    except Exception as e:
        logging.exception("run_graph_with_memory failed: %s", e)
        reply_text = "Ha ocurrido un error momentáneo. Intenta de nuevo."

    # Construir mensaje con el catálogo
    out_msg = OutgoingMessage(to=wa_id, component=TextMessage(body=reply_text))

    try:
        out = await send_catalog_message(out_msg)
        logging.info("OUT ok: %s", out)
    except Exception as e:
        logging.exception("send_message failed: %s", e)

    return Response(status_code=200)

webhook_get 

webhook_get(request)
Source code in src/adapters/whatsapp_business/handler.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@router.get("/webhook")
def webhook_get(request: Request):
    mode = request.query_params.get("hub.mode")
    token = request.query_params.get("hub.verify_token")
    challenge = request.query_params.get("hub.challenge")

    if mode == "subscribe" and token == settings.whatsapp_verify_token:
        logging.info("WEBHOOK_VERIFIED")
        return Response(content=challenge, status_code=200)

    logging.info(
        "VERIFICATION_FAILED: mode=%r token_ok=%s",
        mode,
        token == settings.whatsapp_verify_token,
    )
    raise HTTPException(status_code=403, detail="Verification failed")

Telegram

adapters.telegram.handler

Telegram webhook handler (APIRouter). Opción A mínima.

  • Config vía os.environ (migrar a settings en Opción B).
  • Grafo creado en import (migrar a carga perezosa en Opción B).
  • Envío directo a la API de Telegram (migrar a sender.py + puerto MessageSender en Opción B).

Si no hay TELEGRAM_BOT_TOKEN: - El router puede seguir montado (recibe webhooks), pero el envío será NO-OP (no envía). - Recomendado además: en infrastructure.server montar el router solo si hay token.

router module-attribute 

router = APIRouter()

TELEGRAM_BOT_TOKEN module-attribute 

TELEGRAM_BOT_TOKEN = get('TELEGRAM_BOT_TOKEN')

TELEGRAM_SECRET_TOKEN module-attribute 

TELEGRAM_SECRET_TOKEN = get('TELEGRAM_SECRET_TOKEN')

TELEGRAM_API_BASE module-attribute 

TELEGRAM_API_BASE

TGUpdate

Bases: BaseModel

Subset del Update de Telegram que nos interesa en Opción A.

Source code in src/adapters/telegram/handler.py 
41
42
43
44
45
46
class TGUpdate(BaseModel):
    """Subset del Update de Telegram que nos interesa en Opción A."""
    update_id: int | None = None
    message: dict | None = None
    edited_message: dict | None = None
    channel_post: dict | None = None

update_id class-attribute instance-attribute 

update_id = None

message class-attribute instance-attribute 

message = None

edited_message class-attribute instance-attribute 

edited_message = None

channel_post class-attribute instance-attribute 

channel_post = None

verify_tg_secret async 

verify_tg_secret(
    x_telegram_bot_api_secret_token=Header(None),
)

Verifica el secret header del webhook de Telegram (si está configurado).

Configúralo al registrar el webhook con Telegram: setWebhook?url=...&secret_token=TU_SECRETO

Telegram enviará ese valor en el header X-Telegram-Bot-Api-Secret-Token.

Source code in src/adapters/telegram/handler.py 
50
51
52
53
54
55
56
57
58
59
60
61
async def verify_tg_secret(x_telegram_bot_api_secret_token: str | None = Header(None)):
    """Verifica el secret header del webhook de Telegram (si está configurado).

    Configúralo al registrar el webhook con Telegram:
    setWebhook?url=...&secret_token=TU_SECRETO

    Telegram enviará ese valor en el header `X-Telegram-Bot-Api-Secret-Token`.
    """
    if not TELEGRAM_SECRET_TOKEN:
        return  # auth deshabilitada si no configuraste secreto
    if x_telegram_bot_api_secret_token != TELEGRAM_SECRET_TOKEN:
        raise HTTPException(status_code=401, detail="Invalid Telegram secret token")

get_memory_manager 

get_memory_manager()

Crear MemoryManager con el/los store(s) configurados.

Source code in src/adapters/telegram/handler.py 
64
65
66
67
68
69
70
71
72
def get_memory_manager() -> MemoryManager:
    """Crear MemoryManager con el/los store(s) configurados."""
    base_store = get_memory_store()
    stores: list[HistoryStore] = []
    if isinstance(base_store, list):
        stores.extend(base_store)
    elif base_store is not None:
        stores.append(base_store)
    return MemoryManager(store=stores)

tg_send_text async 

tg_send_text(chat_id, text)

Opción A: envío directo vía API de Telegram.

👉 Opción B: mover a adapters/telegram/sender.py e implementar el puerto MessageSender.

Source code in src/adapters/telegram/handler.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def tg_send_text(chat_id: int | str, text: str) -> None:
    """Opción A: envío directo vía API de Telegram.

    👉 Opción B: mover a adapters/telegram/sender.py e implementar el puerto MessageSender.
    """
    if not TELEGRAM_API_BASE:
        # NO-OP si no hay token; el servidor sigue funcionando.
        # (Recomendado además: no montar este router si no hay token.)
        print("⚠️ Telegram disabled: no TELEGRAM_BOT_TOKEN configured; skipping send.")
        return

    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.post(
            f"{TELEGRAM_API_BASE}/sendMessage",
            json={"chat_id": chat_id, "text": text},
        )
        r.raise_for_status()

telegram_webhook async 

telegram_webhook(
    payload,
    _=Depends(verify_tg_secret),
    mm=Depends(get_memory_manager),
)

Procesa un Update de Telegram y responde usando tu grafo.

Source code in src/adapters/telegram/handler.py 
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
@router.post("/webhook")
async def telegram_webhook(
    payload: TGUpdate,
    _=Depends(verify_tg_secret),
    mm: MemoryManager = Depends(get_memory_manager),
):
    """Procesa un Update de Telegram y responde usando tu grafo."""
    chat_id, text = _extract_chat_and_text(payload)
    if not chat_id or not text:
        # Nada que procesar: confirmamos para que Telegram no reintente en bucle.
        return {"ok": True}

    # Comandos simples (ejemplo mínimo)
    if text.strip().lower() in {"/start", "start"}:
        await tg_send_text(chat_id, "¡Hola! Envía tu consulta.")
        return {"ok": True}

    # Usamos el chat_id como session_id para mantener memoria por conversación
    session_id = str(chat_id)

    reply, _history = await run_with_memory(graph, deps, mm, session_id, text)

    # En Opción A respondemos enviando un mensaje posterior vía API (NO-OP si no hay token).
    await tg_send_text(chat_id, reply)

    # Telegram espera 200 OK; devolver JSON es opcional
    return {"ok": True}

Webhook genérico

adapters.generic_webhook.handler

router module-attribute 

router = APIRouter()

GENERIC_HEADER module-attribute 

GENERIC_HEADER = 'x-api-key'

WebhookIn

Bases: BaseModel

Source code in src/adapters/generic_webhook/handler.py 
 9
10
11
class WebhookIn(BaseModel):
    session_id: str
    message: str

session_id instance-attribute 

session_id

message instance-attribute 

message

WebhookOut

Bases: BaseModel

Source code in src/adapters/generic_webhook/handler.py 
13
14
class WebhookOut(BaseModel):
    response: str

response instance-attribute 

response

generic_webhook async 

generic_webhook(
    payload, x_api_key=Header(None, alias=GENERIC_HEADER)
)
Source code in src/adapters/generic_webhook/handler.py 
21
22
23
24
25
@router.post("/generic-webhook", response_model=WebhookOut)
async def generic_webhook(payload: WebhookIn, x_api_key: str | None = Header(None, alias=GENERIC_HEADER)):
    _check_api_key(x_api_key)
    # Respuesta mínima para smoke tests
    return WebhookOut(response=f"Agente dummy recibió: {payload.message}")