WhatsApp Business 接入 Claude

WhatsApp 有 20 亿活跃用户——把 Claude 接入你的业务 WhatsApp，让客户直接在最熟悉的 App 里获得 AI 支持。

WhatsApp Business API 架构

WhatsApp Business + Claude 完整实现

"""
WhatsApp Business API + Claude 智能客服系统
依赖: pip install anthropic flask requests
API: Meta WhatsApp Cloud API (免费层: 1000 条/月)
"""
import os
import json
import hashlib
import hmac
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import anthropic
class MessageType(Enum):
TEXT = "text"
IMAGE = "image"
AUDIO = "audio"
DOCUMENT = "document"
INTERACTIVE = "interactive"
TEMPLATE = "template"
class ConversationStage(Enum):
GREETING = "greeting"
INQUIRY = "inquiry"
SUPPORT = "support"
ESCALATION = "escalation"
RESOLVED = "resolved"
@dataclass
class WhatsAppMessage:
"""WhatsApp 消息对象"""
message_id: str
from_number: str
display_name: str
msg_type: MessageType
content: str
timestamp: int
media_id: str = ""
@dataclass
class CustomerSession:
"""客户会话状态"""
phone: str
name: str
stage: ConversationStage = ConversationStage.GREETING
messages: list[dict] = field(default_factory=list)
context: dict = field(default_factory=dict)  # 业务上下文 (订单号、产品等)
message_count: int = 0
escalated: bool = False
SYSTEM_PROMPT = """你是一个专业的客服助手。你的职责：
1. 热情、简洁地回答客户问题（用客户的语言）
2. 帮助客户查询订单状态、产品信息、退款政策
3. 复杂问题收集足够信息后转人工
4. 每次回复控制在 150 字以内（WhatsApp 适合短消息）
公司背景：[在此填写你的公司信息和常见问题]
退款政策：7天无理由退换，联系客服处理。
"""
def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self.message_count += 1
# 保留最近 10 条对话
if len(self.messages) > 10:
self.messages = self.messages[-10:]
class WhatsAppClaudeBot:
"""WhatsApp Business + Claude 智能客服"""
# 意图关键词 → 处理策略
INTENT_PATTERNS = {
"订单查询": ["订单", "order", "发货", "物流", "快递", "tracking"],
"退款退货": ["退款", "退货", "refund", "return", "换货"],
"产品咨询": ["产品", "价格", "规格", "怎么用", "功能"],
"投诉": ["投诉", "不满意", "差评", "问题", "损坏", "错误"],
"人工服务": ["人工", "客服", "真人", "转接", "agent"],
}
# 快速回复模板（WhatsApp Interactive Buttons）
QUICK_REPLIES = {
"greeting": ["查询订单", "申请退款", "产品咨询", "联系人工"],
"after_resolve": ["问题已解决 ✅", "还有其他问题", "联系人工客服"],
}
def __init__(self, claude_api_key: str, wa_token: str, wa_phone_id: str):
self.claude = anthropic.Anthropic(api_key=claude_api_key)
self.wa_token = wa_token
self.wa_phone_id = wa_phone_id
self.sessions: dict[str, CustomerSession] = defaultdict(
lambda: CustomerSession(phone="", name="")
)
def detect_intent(self, text: str) -> str:
"""检测消息意图"""
text_lower = text.lower()
for intent, keywords in self.INTENT_PATTERNS.items():
if any(kw in text_lower for kw in keywords):
return intent
return "general"
def should_escalate(self, session: CustomerSession, intent: str) -> bool:
"""判断是否需要转人工"""
return (
intent in ("投诉", "人工服务")
or session.message_count > 8          # 对话过长
or session.context.get("frustrated")  # 情绪标记
)
def call_claude(self, session: CustomerSession, user_msg: str) -> str:
"""调用 Claude 生成回复"""
session.add_message("user", user_msg)
try:
response = self.claude.messages.create(
model="claude-haiku-4-5-20251001",   # Haiku 更快、更经济，适合客服
max_tokens=300,
system=session.SYSTEM_PROMPT,
messages=session.messages,
)
reply = response.content[0].text
session.add_message("assistant", reply)
return reply
except anthropic.APIError as e:
return f"抱歉，系统暂时繁忙，请稍后再试。（错误：{type(e).__name__}）"
def process_message(self, wa_msg: WhatsAppMessage) -> dict:
"""处理消息，返回回复内容"""
session = self.sessions[wa_msg.from_number]
session.phone = wa_msg.from_number
session.name = wa_msg.display_name
# 图片/文件消息
if wa_msg.msg_type == MessageType.IMAGE:
return self._text_reply(
wa_msg.from_number,
"收到图片！请描述您的问题，我来帮您处理。"
)
# 文字消息处理
text = wa_msg.content.strip()
intent = self.detect_intent(text)
# 首次接触 — 发送欢迎菜单
if session.message_count == 0:
session.stage = ConversationStage.GREETING
return self._interactive_reply(
wa_msg.from_number,
f"您好 {wa_msg.display_name}！欢迎联系客服 😊\n请问有什么可以帮到您？",
self.QUICK_REPLIES["greeting"],
)
# 需要转人工
if self.should_escalate(session, intent):
session.escalated = True
session.stage = ConversationStage.ESCALATION
return self._text_reply(
wa_msg.from_number,
"已为您转接人工客服，预计等待时间 3-5 分钟。\n"
"您也可以通过邮件联系我们：support@company.com"
)
# Claude 处理
session.stage = ConversationStage.SUPPORT
reply = self.call_claude(session, text)
# 记录意图到上下文
session.context["last_intent"] = intent
return self._text_reply(wa_msg.from_number, reply)
def _text_reply(self, to: str, body: str) -> dict:
return {
"messaging_product": "whatsapp",
"to": to,
"type": "text",
"text": {"body": body},
}
def _interactive_reply(self, to: str, body: str, buttons: list[str]) -> dict:
return {
"messaging_product": "whatsapp",
"to": to,
"type": "interactive",
"interactive": {
"type": "button",
"body": {"text": body},
"action": {
"buttons": [
{"type": "reply", "reply": {"id": f"btn_{i}", "title": btn}}
for i, btn in enumerate(buttons[:3])  # WhatsApp 最多 3 个按钮
]
},
},
}
@staticmethod
def verify_webhook(payload: bytes, signature: str, app_secret: str) -> bool:
"""验证 Meta Webhook 签名（安全必须项）"""
expected = "sha256=" + hmac.new(
app_secret.encode(), payload, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
@staticmethod
def parse_webhook_message(payload: dict) -> WhatsAppMessage | None:
"""解析 Meta Webhook 消息格式"""
try:
entry = payload["entry"][0]["changes"][0]["value"]
msg = entry["messages"][0]
contact = entry["contacts"][0]
msg_type = MessageType(msg["type"])
content = ""
if msg_type == MessageType.TEXT:
content = msg["text"]["body"]
elif msg_type == MessageType.IMAGE:
content = msg.get("image", {}).get("caption", "")
elif msg_type == MessageType.INTERACTIVE:
content = msg["interactive"]["button_reply"]["title"]
return WhatsAppMessage(
message_id=msg["id"],
from_number=msg["from"],
display_name=contact["profile"]["name"],
msg_type=msg_type,
content=content,
timestamp=int(msg["timestamp"]),
)
except (KeyError, IndexError, ValueError):
return None
# 演示
def demo():
bot = WhatsAppClaudeBot(
claude_api_key=os.environ.get("ANTHROPIC_API_KEY", "demo"),
wa_token="YOUR_WA_TOKEN",
wa_phone_id="YOUR_PHONE_ID",
)
print("=== WhatsApp Business Bot 模拟演示 ===\n")
# 模拟客户对话
conversations = [
("首次接触", "hello"),
("查询订单", "我的订单 #12345 发货了吗？"),
("申请退款", "我想退款"),
("补充说明", "产品坏了，质量很差，我非常不满意！"),
]
phone = "+60123456789"
for stage, msg_text in conversations:
wa_msg = WhatsAppMessage(
message_id=f"msg_{hash(msg_text)}",
from_number=phone,
display_name="张伟",
msg_type=MessageType.TEXT,
content=msg_text,
timestamp=1711180800,
)
intent = bot.detect_intent(msg_text)
reply_payload = bot.process_message(wa_msg)
session = bot.sessions[phone]
print(f"[{stage}] 客户: {msg_text}")
print(f"  意图: {intent} | 阶段: {session.stage.value} | 消息数: {session.message_count}")
if reply_payload["type"] == "text":
body = reply_payload["text"]["body"]
print(f"  Bot: {body[:100]}{'...' if len(body) > 100 else ''}")
else:
buttons = [b["reply"]["title"] for b in reply_payload["interactive"]["action"]["buttons"]]
print(f"  Bot: [交互菜单] 选项: {buttons}")
if session.escalated:
print("  ⚠️  已触发人工转接")
print()
demo()

WhatsApp API 接入层级对比

行动清单

• [ ] 在 Meta for Developers 创建应用，开通 WhatsApp Business API
• [ ] 部署 Webhook 服务器（Railway 免费层即可），配置 /webhook 端点验证
• [ ] 用上方代码搭建基础客服 Bot，设置公司背景和退款政策到 SYSTEM_PROMPT
• [ ] 配置意图关键词列表，覆盖你的业务高频问题（订单/退款/产品）
• [ ] 测试人工转接流程：确保复杂问题能平滑转接到真人客服
• [ ] 监控月度消息量，超过 1000 条考虑付费套餐（约 $0.005–0.08/条）

下一节：03-Webhook与多渠道统一入口 — 用一套代码同时支持 Telegram + WhatsApp + 网站聊天窗口。