Skip to main content

​
Installation

Install the EnviaAI SDK using pip: 
pip install enviaai
Or with Poetry: 
poetry add enviaai

​
Quick Start

from enviaai import EnviaAI

client = EnviaAI(api_key="your-api-key")

# Send a message
response = client.messages.send(
    instance_id="inst_abc123",
    to="5511999999999",
    message="Hello from EnviaAI!"
)

print(f"Message ID: {response.message_id}")

​
Configuration

​
Client Options

from enviaai import EnviaAI

client = EnviaAI(
    api_key="your-api-key",           # Required
    base_url="https://api.enviaai.app",  # Optional
    timeout=30.0,                      # Optional, timeout in seconds
    max_retries=3                      # Optional
)

​
Environment Variables

The SDK reads from environment variables: 
export ENVIAAI_API_KEY=your-api-key
export ENVIAAI_BASE_URL=https://api.enviaai.app

# API key is read from ENVIAAI_API_KEY
client = EnviaAI()

​
Messages API

​
Send Text Message

response = client.messages.send(
    instance_id="inst_abc123",
    to="5511999999999",
    message="Hello! How can I help you?"
)

​
Send Media Message

# Image
image_response = client.messages.send(
    instance_id="inst_abc123",
    to="5511999999999",
    media_url="https://example.com/image.jpg",
    media_type="image",
    caption="Check this out!"
)

# Document
doc_response = client.messages.send(
    instance_id="inst_abc123",
    to="5511999999999",
    media_url="https://example.com/document.pdf",
    media_type="document",
    filename="invoice.pdf"
)

# Audio
audio_response = client.messages.send(
    instance_id="inst_abc123",
    to="5511999999999",
    media_url="https://example.com/audio.mp3",
    media_type="audio"
)

​
Get Message Status

status = client.messages.get_status("msg_xyz789")
print(status.status)  # 'sent', 'delivered', 'read', 'failed'

​
List Chats

chats = client.messages.get_chats(
    instance_id="inst_abc123",
    limit=20,
    unread_only=True
)

for chat in chats.data:
    print(f"{chat.contact_name}: {chat.unread_count} unread")

​
Get Chat Messages

messages = client.messages.get_chat_messages(
    instance_id="inst_abc123",
    chat_id="chat_def456",
    limit=50
)

​
Instances API

​
List Instances

instances = client.instances.list()

for instance in instances.data:
    print(f"{instance.name}: {instance.status}")

​
Get Instance Status

status = client.instances.get_status("inst_abc123")

if status.data.status == "connected":
    print(f"Connected as {status.data.phone}")
elif status.data.status == "qr_pending":
    print(f"Scan QR: {status.data.qr_code}")

​
Connect Instance

client.instances.connect("inst_abc123")

# Poll for QR code
status = client.instances.get_status("inst_abc123")
if status.data.qr_code:
    # Display QR code to user
    pass

​
Disconnect Instance

client.instances.disconnect("inst_abc123")

​
Webhooks API

​
Create Webhook

webhook = client.webhooks.create(
    url="https://myapp.com/webhooks/enviaai",
    events=["message.received", "message.delivered"],
    secret="my-webhook-secret"
)

​
List Webhooks

webhooks = client.webhooks.list()

​
Update Webhook

client.webhooks.update(
    "wh_abc123",
    events=["message.received", "message.delivered", "message.read"]
)

​
Delete Webhook

client.webhooks.delete("wh_abc123")

​
Verify Webhook Signature

from enviaai import EnviaAI

is_valid = EnviaAI.webhooks.verify_signature(
    payload=payload,
    signature=signature,
    secret="your-webhook-secret"
)

​
Async Support

The SDK provides an async client for use with asyncio: 
import asyncio
from enviaai import AsyncEnviaAI

async def main():
    client = AsyncEnviaAI(api_key="your-api-key")

    # Send message asynchronously
    response = await client.messages.send(
        instance_id="inst_abc123",
        to="5511999999999",
        message="Hello async!"
    )

    print(f"Message ID: {response.message_id}")

    # Close client when done
    await client.close()

asyncio.run(main())

​
Using Context Manager

async def main():
    async with AsyncEnviaAI(api_key="your-api-key") as client:
        response = await client.messages.send(
            instance_id="inst_abc123",
            to="5511999999999",
            message="Hello!"
        )

​
Type Hints

The SDK is fully typed with Python type hints: 
from enviaai import EnviaAI
from enviaai.types import Message, Instance, SendMessageOptions

client = EnviaAI(api_key="your-key")

options: SendMessageOptions = {
    "instance_id": "inst_abc123",
    "to": "5511999999999",
    "message": "Hello!"
}

response: Message = client.messages.send(**options)

​
Error Handling

from enviaai import EnviaAI
from enviaai.exceptions import (
    EnviaAIError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError
)

client = EnviaAI()

try:
    client.messages.send(
        instance_id="inst_abc123",
        to="5511999999999",
        message="Hello!"
    )
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except NotFoundError:
    print("Resource not found")
except EnviaAIError as e:
    print(f"API Error: {e.code} - {e.message}")

​
Error Types

Exception ClassDescription
EnviaAIErrorBase exception for all API errors
AuthenticationErrorInvalid or missing API key
RateLimitErrorRate limit exceeded
ValidationErrorInvalid request parameters
NotFoundErrorResource not found
ServerErrorInternal server error

​
Pagination

# Manual pagination
offset = 0
limit = 50
all_chats = []

while True:
    response = client.messages.get_chats(
        instance_id="inst_abc123",
        limit=limit,
        offset=offset
    )

    all_chats.extend(response.data)

    if not response.pagination.has_more:
        break
    offset += limit

# Using iterator
for chat in client.messages.iter_chats(instance_id="inst_abc123"):
    print(chat.contact_name)

​
Logging

Enable debug logging: 
import logging

logging.basicConfig(level=logging.DEBUG)

# Or for just the SDK
logging.getLogger("enviaai").setLevel(logging.DEBUG)
Or via environment variable: 
export ENVIAAI_DEBUG=true

​
Examples

​
Flask Webhook Handler

from flask import Flask, request, jsonify
from enviaai import EnviaAI

app = Flask(__name__)
client = EnviaAI()

@app.route("/webhooks/enviaai", methods=["POST"])
def webhook():
    signature = request.headers.get("X-EnviaAI-Signature")

    if not EnviaAI.webhooks.verify_signature(
        request.json,
        signature,
        "your-webhook-secret"
    ):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.json

    if event["event"] == "message.received":
        # Auto-reply
        client.messages.send(
            instance_id=event["data"]["instanceId"],
            to=event["data"]["from"],
            message="Thanks for your message! We'll get back to you soon."
        )

    return "OK", 200

if __name__ == "__main__":
    app.run(port=3000)

​
FastAPI Webhook Handler (Async)

from fastapi import FastAPI, Request, HTTPException
from enviaai import AsyncEnviaAI

app = FastAPI()
client = AsyncEnviaAI()

@app.post("/webhooks/enviaai")
async def webhook(request: Request):
    body = await request.json()
    signature = request.headers.get("X-EnviaAI-Signature")

    if not AsyncEnviaAI.webhooks.verify_signature(
        body, signature, "your-webhook-secret"
    ):
        raise HTTPException(status_code=401, detail="Invalid signature")

    if body["event"] == "message.received":
        await client.messages.send(
            instance_id=body["data"]["instanceId"],
            to=body["data"]["from"],
            message="Thanks for your message!"
        )

    return {"status": "ok"}

​
Bulk Message Sending

import asyncio
from enviaai import AsyncEnviaAI

async def send_bulk_messages():
    async with AsyncEnviaAI() as client:
        contacts = [
            {"phone": "5511999999999", "name": "John"},
            {"phone": "5521988888888", "name": "Jane"},
            # ... more contacts
        ]

        semaphore = asyncio.Semaphore(10)  # Max 10 concurrent

        async def send_with_limit(contact):
            async with semaphore:
                return await client.messages.send(
                    instance_id="inst_abc123",
                    to=contact["phone"],
                    message=f"Hello {contact['name']}! Check out our products."
                )

        results = await asyncio.gather(
            *[send_with_limit(c) for c in contacts],
            return_exceptions=True
        )

        success = sum(1 for r in results if not isinstance(r, Exception))
        print(f"Sent {success}/{len(contacts)} messages successfully")

asyncio.run(send_bulk_messages())

​
Django Integration

# settings.py
ENVIAAI_API_KEY = os.environ.get("ENVIAAI_API_KEY")

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings
from enviaai import EnviaAI

client = EnviaAI(api_key=settings.ENVIAAI_API_KEY)

@csrf_exempt
def webhook(request):
    if request.method != "POST":
        return JsonResponse({"error": "Method not allowed"}, status=405)

    import json
    body = json.loads(request.body)

    if body["event"] == "message.received":
        client.messages.send(
            instance_id=body["data"]["instanceId"],
            to=body["data"]["from"],
            message="Thanks for contacting us!"
        )

    return JsonResponse({"status": "ok"})

​
Resources