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