The WSAPI Python SDK provides a typed client (Pydantic v2) for sending WhatsApp messages, managing groups and chats, and receiving real-time events via webhooks or Server-Sent Events (SSE).
Installation
Requirements
- Python 3.9 or later
- Valid WSAPI API key and instance ID
Quick start
from wsapi_client import WSApiClient, ApiException
from wsapi_client.models.requests.messages import MessageSendTextRequest
client = WSApiClient(api_key="<your-api-key>", instance_id="<instance-id>")
req = MessageSendTextRequest(
to="1234567890@s.whatsapp.net",
text="Hello from Python!",
)
result = client.messages.send_text(req)
print("Message id:", result.id)
client.close()
Always call client.close() when done, or use the client as a context manager.
Error handling
The SDK provides two patterns for every operation:
Exception-based
Try-based (non-raising)
Standard methods raise ApiException on failure:from wsapi_client import WSApiClient, ApiException
from wsapi_client.models.requests.messages import MessageSendTextRequest
client = WSApiClient(api_key="<your-api-key>", instance_id="<instance-id>")
try:
req = MessageSendTextRequest(
to="1234567890@s.whatsapp.net", text="Hello!"
)
created = client.messages.send_text(req)
print("Message id:", created.id)
except ApiException as ex:
print("Send failed:", ex.problem.detail)
Methods prefixed with try_ never raise. They return an ApiResponse[T]:resp = client.messages.try_send_text(req)
if resp.is_success and resp.result:
print("Message id:", resp.result.id)
else:
print("Send failed:", resp.error.detail if resp.error else "unknown")
ApiResponse[T] object contains:
result — data when successful (None if failed)error — ProblemDetails when not successful (None if success)is_success — convenience boolean
Available resources
# Messages
client.messages.send_text(req)
client.messages.send_image(req)
client.messages.send_video(req)
client.messages.send_audio(req)
client.messages.send_document(req)
client.messages.send_sticker(req)
client.messages.send_contact(req)
client.messages.send_location(req)
client.messages.send_link(req)
client.messages.send_reaction(req)
client.messages.edit_text(req)
client.messages.mark_as_read(req)
client.messages.star(req)
client.messages.delete(req)
# Users, Instance, Contacts, Groups, Chats, Calls, Media
client.users.get_by_id("5511999999999")
client.instance.get_settings()
client.contacts.get_all()
client.groups.get_all()
client.chats.get_all()
client.calls.reject(call_id, req)
client.media.download(media_id)
Receiving events via SSE
from wsapi_client import SSEClient, SSEConnectionState
from wsapi_client.models.events.messages import MessageEvent
sse = SSEClient(api_key="<your-api-key>", instance_id="<instance-id>")
def on_event(evt):
if isinstance(evt, MessageEvent) and evt.text:
print("[sse]", evt.sender_name, ":", evt.text)
def on_state(state, exc):
print("[sse] Connection state:", state)
if exc:
print("[sse] Error:", exc)
sse.on_event = on_event
sse.on_connection_state_changed = on_state
sse.start()
# ... later
sse.stop()
Handling webhooks
from flask import Flask, request, jsonify
from wsapi_client.events.factory import parse_event
from wsapi_client.models.events.messages import MessageEvent
app = Flask(__name__)
WEBHOOK_AUTH_TOKEN = "your-secret-token"
@app.route("/wsapi/webhook", methods=["POST"])
def handle_webhook():
auth_header = request.headers.get("X-Webhook-Auth")
if auth_header != WEBHOOK_AUTH_TOKEN:
return jsonify({"error": "Unauthorized"}), 401
try:
parsed_event = parse_event(request.get_data(as_text=True))
if isinstance(parsed_event, MessageEvent) and parsed_event.text:
print(f"New message from {parsed_event.sender_name}: {parsed_event.text}")
return jsonify({"status": "success"}), 200
except Exception as e:
print(f"Error processing event: {e}")
return jsonify({"error": "Processing failed"}), 500
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
from fastapi import FastAPI, Request, HTTPException, Header
from wsapi_client.events.factory import parse_event
from wsapi_client.models.events.messages import MessageEvent
from typing import Optional
app = FastAPI()
WEBHOOK_AUTH_TOKEN = "your-secret-token"
@app.post("/wsapi/webhook")
async def handle_webhook(
request: Request,
x_webhook_auth: Optional[str] = Header(None),
):
if x_webhook_auth != WEBHOOK_AUTH_TOKEN:
raise HTTPException(status_code=401, detail="Unauthorized")
try:
body = await request.body()
parsed_event = parse_event(body.decode("utf-8"))
if isinstance(parsed_event, MessageEvent) and parsed_event.text:
print(f"New message: {parsed_event.text}")
return {"status": "success"}
except Exception as e:
print(f"Error: {e}")
raise HTTPException(status_code=500, detail="Processing failed")
Available event types
| Category | Events |
|---|
| Session | SessionLoggedInEvent, SessionLoggedOutEvent, SessionLoggedErrorEvent |
| Messages | MessageEvent, MessageDeleteEvent, MessageReadEvent, MessageStarEvent, MessageHistorySyncEvent |
| Chats | ChatPresenceEvent, ChatSettingEvent |
| Contacts | ContactEvent |
| Users | UserPushNameEvent, UserPictureEvent, UserPresenceEvent, UserStatusEvent |
| Calls | CallOfferEvent, CallAcceptEvent, CallTerminateEvent |
GitHub
github.com/wsapi-chat/wsapi-python 
