wsapi-client 2.0.0

Python SDK for WSApi: send messages, manage chats/groups, and receive events via webhooks or SSE.

WSApi Python SDK

A Python SDK for integrating with the WSApi platform. Send WhatsApp messages, manage chats/contacts, and receive real-time events via webhooks or Server‑Sent Events (SSE). This is an independent API and is not affiliated with META or WhatsApp

Features

• HTTP Client: Simple, typed models (Pydantic v2) for all API operations
• Dual API Pattern: Exception-based and try_ methods for error handling
• Resource Clients: Messages, Instance, Media, Contacts, Groups, Chats, Users, Calls
• Event Handling: Webhooks and SSE client for real-time events with parsed models

---

Getting Started

Prerequisites

• Python 3.9+ (as specified in pyproject.toml)
• WSApi instance credentials (API Key and Instance Id)

Installation

From PyPI (recommended):

pip install wsapi-client

📦 Package page: https://pypi.org/project/wsapi-client/

From source (development):

1. Clone this repository
2. Create and activate a virtual environment (recommended)
3. Install in editable mode:

pip install -e .

Dependencies: httpx, pydantic (automatically installed)

---

🔧 API Client Usage

Basic Client Setup

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>")

Sending Messages

req = MessageSendTextRequest(to="[e-mail]", text="Hello from Python!")

# Style 1: Exception-based (raises on error)
try:
    created = client.messages.send_text(req)
    print("Message id:", created.id)
except ApiException as ex:
    # ProblemDetails available via ex.problem
    print("Send failed:", ex.problem.detail)

# Style 2: ApiResponse-based (never raises)
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 error")

client.close()

Advanced Message Options

# Reply to a message
reply = MessageSendTextRequest(
    to="[e-mail]",
    text="This is a reply!",
    reply_to="MESSAGE_ID_TO_REPLY_TO",
    reply_to_sender_id="SENDER_JID"  # Required for group replies
)
client.messages.send_text(reply)

# Send with ephemeral expiration (disappearing message)
ephemeral = MessageSendTextRequest(
    to="[e-mail]",
    text="This message will disappear!",
    ephemeral_expiration="24h"  # Options: "24h", "7d", "90d"
)
client.messages.send_text(ephemeral)

Other API Operations

# Users
user = client.users.get_by_id("5511999999999")  # Just phone number for user lookup
print(user.status)

# Instance settings
settings = client.instance.get_settings()
print(f"Instance: {settings.name}")

# Calls (reject)
from wsapi_client.models.requests.calls import RejectCallRequest
client.calls.reject("call_123", RejectCallRequest(caller="[phone]"))

Client Reference

WSApiClient Resources:

• messages: send text/image/video/audio/voice/sticker/document/contact/location/link/reaction, edit_text; mark_as_read, star, delete, delete_for_me (supports replyTo, ephemeralExpiration)
• account: get_info, update_name, update_status, update_picture, update_presence
• session: get_status, get_qr_code, get_pair_code, logout
• instance: get_settings, update_settings
• media: download files
• contacts: list, get, create, update, delete, get_picture, get_business_profile
• groups: list, get, create, delete, update_name/description/picture, manage participants, invite links
• chats: list, get, delete, update_read/archive/pin/mute/ephemeral/presence, get_picture
• users: get_by_id/try_get_by_id
• calls: reject/try_reject

ApiResponse[T] Pattern

Each method has an exception-based and a try_ variant:

• result: T | None — data when successful
• error: ProblemDetails | None — error when not successful
• is_success: bool — convenience flag

Important Notes

• Close the client when you're done: client.close()
• No-content endpoints (204) return None on success
• Models use Pydantic v2 with aliases to match WSApi's JSON
• Timeouts and transport errors are surfaced as ApiException (exception path) or as ProblemDetails with status 408/500 in ApiResponse (try_ path)

---

📡 Event Handling

There are two ways to receive real-time events from the WSApi: Webhooks (recommended) and Server-Sent Events (SSE).

Option 1: Webhooks (Recommended)

Configure a webhook endpoint in your WSApi instance settings to receive events via HTTP POST requests. This is the most reliable and scalable approach.

Setup Process:

1. Configure webhook URL and signing secret in your WSApi instance settings
2. Create an endpoint in your web application to receive events
3. Verify the webhook signature using verify_signature()
4. Use the same event parsing functionality as SSE

Example using Flask:

from flask import Flask, request, jsonify
from wsapi_client import verify_signature
from wsapi_client.events.factory import parse_event
from wsapi_client.models.events.messages import MessageEvent

app = Flask(__name__)

# Configure this in your WSApi instance settings
SIGNING_SECRET = "your-signing-secret"

@app.route('/wsapi/webhook', methods=['POST'])
def handle_webhook():
    # Read the raw body before parsing — signature is computed over raw bytes
    raw_body = request.get_data()

    # Verify the HMAC-SHA256 webhook signature
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_signature(raw_body, SIGNING_SECRET, signature):
        return jsonify({"error": "Invalid signature"}), 401

    try:
        # Parse using the same factory as SSE client
        parsed_event = parse_event(raw_body.decode("utf-8"))

        # Handle the event based on type
        if isinstance(parsed_event, MessageEvent) and parsed_event.text:
            print(f"[webhook] New message from {parsed_event.sender_name}: {parsed_event.text}")

            # Example: Auto-reply to messages
            # client = WSApiClient(api_key="...", instance_id="...")
            # client.messages.send_text(MessageSendTextRequest(
            #     to=parsed_event.chat_id,
            #     text="Thanks for your message!"
            # ))

        return jsonify({"status": "success"}), 200

    except Exception as e:
        print(f"[webhook] Error processing event: {e}")
        return jsonify({"error": "Processing failed"}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Example using FastAPI:

from fastapi import FastAPI, Request, HTTPException, Header
from wsapi_client import verify_signature
from wsapi_client.events.factory import parse_event
from wsapi_client.models.events.messages import MessageEvent
from typing import Optional

app = FastAPI()

SIGNING_SECRET = "your-signing-secret"

@app.post("/wsapi/webhook")
async def handle_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None),
):
    # Read the raw body before parsing — signature is computed over raw bytes
    body = await request.body()

    # Verify the HMAC-SHA256 webhook signature
    if not verify_signature(body, SIGNING_SECRET, x_webhook_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid signature")

    try:
        # Parse the event
        parsed_event = parse_event(body.decode('utf-8'))

        # Handle different event types
        if isinstance(parsed_event, MessageEvent) and parsed_event.text:
            print(f"[webhook] New message: {parsed_event.text}")
            # Process message...

        return {"status": "success"}

    except Exception as e:
        print(f"[webhook] Error: {e}")
        raise HTTPException(status_code=500, detail="Processing failed")

Option 2: Server-Sent Events (SSE)

For scenarios where webhooks aren't suitable, use the built-in SSE client for a persistent connection.

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):
    # evt is one of the typed event models (e.g., MessageEvent)
    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()

Available Event Types

Both webhook and SSE approaches support the same event types:

• Session: SessionLoggedInEvent, SessionLoggedOutEvent, SessionLoggedErrorEvent
• Messages: MessageEvent, MessageDeleteEvent, MessageReadEvent, MessageStarEvent, MessageHistorySyncEvent
• Chats: ChatPresenceEvent, ChatSettingEvent
• Contacts: ContactEvent
• Groups: GroupEvent (participant changes, settings updates)
• Users: UserPushNameEvent, UserPictureEvent, UserPresenceEvent, UserStatusEvent
• Calls: CallOfferEvent, CallAcceptEvent, CallTerminateEvent

All events are parsed by wsapi_client.events.factory.parse_event which converts raw JSON into strongly-typed Python objects.

---

Troubleshooting

• If you see JSON parsing errors, confirm the SDK version matches the WSApi server you're targeting.
• For webhook issues, check that your endpoint is accessible and returns proper HTTP status codes.
• For SSE connection issues, verify your API credentials and network connectivity.