Coverage for agentos/channels/adapters/whatsapp.py: 0%
120 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2WhatsApp Channel Adapter — WhatsApp Business Cloud API.
4Meta Developer App → Phone Number ID + Access Token → webhook → ChannelMessage.
5"""
7from __future__ import annotations
9import json
10from typing import Optional
12from agentos.channels.base import BaseChannelAdapter, ChannelConfig, ReplyResult
13from agentos.channels.message import ChannelMessage, ChannelType, MessageType
16class WhatsAppAdapter(BaseChannelAdapter):
17 """WhatsApp Business Cloud API adapter.
19 Config fields:
20 access_token: Meta permanent page access token
21 phone_number_id: WhatsApp Business phone number ID
22 verify_token: Webhook verify token (for Meta handshake)
23 app_secret: Meta App secret (for payload signing, optional)
24 business_id: WhatsApp Business Account ID
25 """
27 CHANNEL_TYPE = ChannelType.WHATSAPP
28 API_BASE = "https://graph.facebook.com/v19.0"
30 def __init__(self, config: ChannelConfig):
31 super().__init__(config)
32 self._access_token = config.extra.get("access_token", "")
33 self._phone_number_id = config.extra.get("phone_number_id", "")
34 self._verify_token = config.extra.get("verify_token", "")
35 self._app_secret = config.extra.get("app_secret", "")
36 self._business_id = config.extra.get("business_id", "")
38 # ── Webhook verification ──
40 def verify_webhook(self, query_params: dict) -> tuple[bool, str]:
41 """Handle Meta webhook verification challenge.
43 Returns (verified, challenge_token).
44 """
45 mode = query_params.get("hub.mode", "")
46 token = query_params.get("hub.verify_token", "")
47 challenge = query_params.get("hub.challenge", "")
49 if mode == "subscribe" and token == self._verify_token:
50 return True, challenge
51 return False, ""
53 # ── Message parsing ──
55 async def parse_incoming(self, payload: dict) -> Optional[ChannelMessage]:
56 """Parse WhatsApp webhook payload into ChannelMessage."""
57 entries = payload.get("entry", [])
59 for entry in entries:
60 changes = entry.get("changes", [])
61 for change in changes:
62 value = change.get("value", {})
64 # Messages
65 messages = value.get("messages", [])
66 for msg in messages:
67 return self._parse_message(msg, value)
69 # Status updates
70 statuses = value.get("statuses", [])
71 for status in statuses:
72 return ChannelMessage(
73 channel_type=ChannelType.WHATSAPP,
74 channel_id=status.get("recipient_id", ""),
75 user_id=status.get("recipient_id", ""),
76 content=f"Message status: {status.get('status', 'unknown')}",
77 message_type=MessageType.SYSTEM,
78 raw=status,
79 metadata={"status": status.get("status")},
80 )
82 return None
84 def _parse_message(self, msg: dict, value: dict) -> Optional[ChannelMessage]:
85 """Parse a WhatsApp message object."""
86 msg_type = msg.get("type", "text")
87 user_phone = msg.get("from", "")
88 msg_id = msg.get("id", "")
90 content = ""
91 mtype = MessageType.TEXT
92 metadata = {
93 "phone": user_phone,
94 "display_name": value.get("contacts", [{}])[0].get("profile", {}).get("name", ""),
95 }
97 if msg_type == "text":
98 content = msg.get("text", {}).get("body", "")
99 elif msg_type == "image":
100 content = msg.get("image", {}).get("caption", "[Image]")
101 mtype = MessageType.IMAGE
102 metadata["media_id"] = msg.get("image", {}).get("id", "")
103 elif msg_type == "audio":
104 content = "[Voice message]"
105 mtype = MessageType.VOICE
106 metadata["media_id"] = msg.get("audio", {}).get("id", "")
107 elif msg_type == "video":
108 content = msg.get("video", {}).get("caption", "[Video]")
109 mtype = MessageType.VIDEO
110 elif msg_type == "document":
111 content = msg.get("document", {}).get("caption", "[Document]")
112 mtype = MessageType.FILE
113 metadata["file_name"] = msg.get("document", {}).get("filename", "")
114 elif msg_type == "location":
115 loc = msg.get("location", {})
116 content = f"[Location: {loc.get('latitude')}, {loc.get('longitude')}]"
117 mtype = MessageType.LOCATION
118 elif msg_type == "button":
119 content = msg.get("button", {}).get("text", "")
120 mtype = MessageType.INTERACTIVE
121 elif msg_type == "interactive":
122 interactive = msg.get("interactive", {})
123 if interactive.get("type") == "button_reply":
124 content = interactive.get("button_reply", {}).get("id", "")
125 else:
126 content = interactive.get("list_reply", {}).get("id", "")
127 mtype = MessageType.INTERACTIVE
128 else:
129 content = f"[{msg_type}]"
131 return ChannelMessage(
132 channel_type=ChannelType.WHATSAPP,
133 channel_id=user_phone,
134 user_id=user_phone,
135 content=content,
136 message_type=mtype,
137 raw=msg,
138 reply_token=msg_id,
139 metadata=metadata,
140 )
142 # ── Reply ──
144 async def reply(self, channel_id: str, content: str, **kwargs) -> ReplyResult:
145 """Send a text message via WhatsApp Cloud API."""
146 return await self._send_msg(channel_id, {
147 "type": "text",
148 "text": {"body": content, "preview_url": False},
149 })
151 async def reply_template(
152 self, channel_id: str, template_name: str,
153 language_code: str = "en", components: list = None, **kwargs,
154 ) -> ReplyResult:
155 """Send a WhatsApp message template."""
156 body = {
157 "type": "template",
158 "template": {
159 "name": template_name,
160 "language": {"code": language_code},
161 },
162 }
163 if components:
164 body["template"]["components"] = components
165 return await self._send_msg(channel_id, body)
167 async def reply_interactive(
168 self, channel_id: str, body_text: str,
169 buttons: list[dict], **kwargs,
170 ) -> ReplyResult:
171 """Send an interactive message with reply buttons.
173 buttons = [{"id": "yes", "title": "Yes"}, ...]
174 """
175 button_list = [
176 {"type": "reply", "reply": {"id": b["id"], "title": b["title"]}}
177 for b in buttons[:3]
178 ]
179 return await self._send_msg(channel_id, {
180 "type": "interactive",
181 "interactive": {
182 "type": "button",
183 "body": {"text": body_text},
184 "action": {"buttons": button_list},
185 },
186 })
188 async def reply_image(
189 self, channel_id: str, image_url: str, caption: str = "", **kwargs
190 ) -> ReplyResult:
191 """Send an image."""
192 return await self._send_msg(channel_id, {
193 "type": "image",
194 "image": {"link": image_url, "caption": caption},
195 })
197 # ── API Helper ──
199 async def _send_msg(self, to: str, msg_data: dict) -> ReplyResult:
200 """Send message via WhatsApp Cloud API."""
201 url = f"{self.API_BASE}/{self._phone_number_id}/messages"
202 headers = {
203 "Authorization": f"Bearer {self._access_token}",
204 "Content-Type": "application/json",
205 }
206 body = {
207 "messaging_product": "whatsapp",
208 "recipient_type": "individual",
209 "to": to,
210 **msg_data,
211 }
213 try:
214 import aiohttp
215 async with aiohttp.ClientSession() as session:
216 async with session.post(url, headers=headers, json=body) as resp:
217 data = await resp.json()
218 wa_id = data.get("messages", [{}])[0].get("id", "")
219 if wa_id:
220 return ReplyResult(success=True, message_id=wa_id)
221 return ReplyResult(
222 success=False,
223 error=data.get("error", {}).get("message", "unknown"),
224 )
225 except ImportError:
226 import urllib.request
227 req = urllib.request.Request(
228 url, data=json.dumps(body).encode(), headers=headers
229 )
230 with urllib.request.urlopen(req) as resp:
231 data = json.loads(resp.read())
232 wa_id = data.get("messages", [{}])[0].get("id", "")
233 return ReplyResult(success=True, message_id=wa_id)
235 async def mark_as_read(self, message_id: str) -> bool:
236 """Mark a message as read."""
237 url = f"{self.API_BASE}/{self._phone_number_id}/messages"
238 headers = {
239 "Authorization": f"Bearer {self._access_token}",
240 "Content-Type": "application/json",
241 }
242 body = {
243 "messaging_product": "whatsapp",
244 "status": "read",
245 "message_id": message_id,
246 }
247 try:
248 import aiohttp
249 async with aiohttp.ClientSession() as session:
250 async with session.post(url, headers=headers, json=body) as resp:
251 return resp.status == 200
252 except ImportError:
253 import urllib.request
254 req = urllib.request.Request(
255 url, data=json.dumps(body).encode(), headers=headers
256 )
257 with urllib.request.urlopen(req) as resp:
258 return resp.status == 200