Metadata-Version: 2.5
Name: synquic-slide
Version: 0.3.0
Summary: Official Python SDK for the Slide API
Project-URL: Homepage, https://slide.synquic.com
Project-URL: Documentation, https://slide.synquic.com
Author-email: Synquic <p@synquic.com>
License: MIT
Keywords: api,crm,email,otp,rcs,sdk,slide,sms,synquic,voice,whatsapp
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx>=0.27.0
Description-Content-Type: text/markdown

# sdk-python

`synquic-slide` is the official Python SDK for the **Slide by Synquic** API, a multi-channel customer engagement platform. One client covers WhatsApp Business messaging, SMS, RCS, OTP verification, transactional email, Instagram DMs, one send call across those channels, AI voice calls, Shopify lookups, unified contacts, automations and outbound webhooks. It targets **server-side Python 3.9+** and ships a synchronous client and an identical async one, both built on `httpx`. It authenticates with a secret API key, so it belongs on your backend, never in a client application.

## Installation

```bash
pip install synquic-slide
```

## Quick start

Responses are plain dictionaries (typed as `TypedDict`), so read them with `["key"]`.

```python
import os
from synquic_slide import SlideClient

slide = SlideClient(api_key=os.environ["SLIDE_API_KEY"])

# WhatsApp template message
msg = slide.whatsapp.send_template(
    to="+919876543210",
    template_name="order_shipped",
    language_code="en",
    components=[{"type": "body", "parameters": [{"type": "text", "text": "#1042"}]}],
)
print(msg["wamid"], msg["conversationId"])

# Transactional email from a saved template
slide.email.send(
    recipient={"to": "customer@example.com", "firstName": "Priya"},
    template_id="tmpl_order_confirm",
    from_name="Acme Store",
    from_email="orders@acme.com",
    variables={"orderId": "#1042"},
)

# Paginated reads always return {"data": [...], "meta": {...}}
page = slide.contacts.list(page=1, limit=50, search="priya")
print(len(page["data"]), page["meta"]["totalPages"])

# Outbound AI voice call. Failures raise SlideValidationError, SlideAuthError, etc.
call = slide.voice.initiate_outbound_call(
    agent_id="agent_abc123",
    to_number="+919876543210",
    call_context={"orderId": "#1042"},
)
print(call["id"], call["status"])

slide.close()   # or use `with SlideClient(api_key=...) as slide:`
```

The async client exposes exactly the same resources and method names:

```python
import asyncio, os
from synquic_slide import AsyncSlideClient

async def main():
    async with AsyncSlideClient(api_key=os.environ["SLIDE_API_KEY"]) as slide:
        agents = await slide.voice.list_agents(page=1, limit=20)
        print([a["name"] for a in agents["data"]])

asyncio.run(main())
```

Verifying an incoming webhook delivery. Pass the **raw** body, never a re-serialised dict:

```python
import os
from flask import request
from synquic_slide import construct_event, SlideSignatureVerificationError

@app.post("/webhooks/slide")
def slide_webhook():
    try:
        event = construct_event(
            request.get_data(),                        # not request.json
            request.headers["X-Slide-Signature"],
            os.environ["SLIDE_WEBHOOK_SECRET"],
        )
    except SlideSignatureVerificationError:
        return "bad signature", 400
    enqueue(event)          # event["id"] is stable across retries: dedupe on it
    return "", 200
```

### One send call, contacts and automations

```python
# One send call for any channel, with a fallback
sent = slide.messages.send(
    to="+919876543210",
    channel="whatsapp",
    fallback_channels=["sms"],
    message="Your order #1042 has shipped.",
    template={"name": "order_shipped", "languageCode": "en"},
    sms={"senderId": "ACMEIN"},
)
print(sent["channelUsed"], sent["messageId"], sent["attempts"])

# A new automation is always created paused. Going live is its own call.
catalog = slide.automations.node_catalog()   # trigger types, node types, their settings
automation = slide.automations.create(
    name="Order follow-up",
    trigger={"type": "WEBHOOK"},
    steps=[{"order": 0, "type": "SMART_DELAY", "delayMinutes": 5}],
)
slide.automations.activate(automation["id"])

# The AI models your workspace can use, with their prices
models = slide.ai.list_models()["data"]
```

```python
# Create a contact, or update the one with the same phone or email.
# The API key needs the permission contacts:write.
contact = slide.contacts.upsert(
    phone="+919876543210",
    email="priya@example.com",
    name="Priya Sharma",
    tags=["Newsletter"],
)
print(contact["id"], contact["created"])

# Find one person: phone and email match exactly, search is a "contains" match
found = slide.contacts.list(email="priya@example.com")
same = slide.contacts.get(contact["id"])
```

The SDK makes every call once and never retries on its own. Do not retry a send blindly either: a timeout does not mean the message was not sent, and a second send reaches a real person twice.

`contacts.upsert` needs an API key with the permission `contacts:write` (and `contacts.list` / `contacts.get` need `contacts:read`). A key that only has the email list permission cannot create contacts: it is refused with `SlideScopeError` (403). A key's permissions cannot be edited, so create a new API key that has `contacts:write`.

Sending the same `contacts.upsert` again after it has answered, also after a timeout or a server error, gives the same contact and creates nothing twice. Two calls for the same person made at the same moment can be refused with `SlideError` status 409, and `error.body["code"]` says what to do: on `CONTACT_UPSERT_IN_PROGRESS` nothing was written, so wait `error.body["retryAfter"]` seconds and send it again; on `PHONE_IN_USE` send it again once.

### Chat inside your own app

Your server makes one call: a short-lived chat token for a user it has already signed in. Your app opens the chat with that token. The API key never leaves your server.

Pilot: `slide.chat` is not in the version of this package on the registry yet. During the pilot you receive a build that has it with your pilot access; until then the token call is one plain `POST /v1/chat/tokens` with your key.

```python
import os
from flask import Flask, jsonify, request
from synquic_slide import SlideClient, construct_event, SlideSignatureVerificationError

app = Flask(__name__)
slide = SlideClient(api_key=os.environ["SLIDE_API_KEY"])   # needs the scope chat:tokens:write

# 1. Your app asks YOUR server for a token, behind your own sign-in.
@app.get("/me/chat-token")
@require_sign_in
def chat_token():
    result = slide.chat.create_token(
        user_id=current_user.id,      # your own stable id for the user
        name=current_user.name,       # optional: how the team sees them
        role="patient",               # optional
        app="patient",                # optional: only when you have more than one app
    )
    return jsonify(token=result["token"])   # to refresh, the app simply asks again

# 2. A reply is waiting for a user who has not seen it: send your own push.
@app.post("/webhooks/slide-chat")
def chat_webhook():
    try:
        event = construct_event(
            request.get_data(),
            request.headers.get("X-Slide-Signature", ""),
            os.environ["SLIDE_CHAT_WEBHOOK_SECRET"],
        )
    except SlideSignatureVerificationError:
        return "bad signature", 400
    if event["type"] == "chat.message.unread":
        data = event["data"]      # never carries the text; de-duplicate on data["messageId"]
        enqueue_push(data["userId"], data["app"], badge=data["unreadCount"])
    return "", 200
```

Optional, when a user signs out or asks to be erased: `slide.chat.revoke_user(user_id, app=...)` and `slide.chat.delete_user(user_id, app=...)` (scope `chat:users:write`).

## API reference

Every method below exists twice: on `SlideClient` (blocking) and on `AsyncSlideClient` (awaitable, same name). Arguments are keyword-only unless shown positionally. List methods also accept `page` (default 1) and `limit` (default varies by endpoint, 10 to 50).

| Method | Parameters | Returns |
|---|---|---|
| `SlideClient(...)` / `AsyncSlideClient(...)` | `api_key: str`, `base_url: str = "https://slide.synquic.com/api"`, `timeout: float = 30.0` | client instance, raises `ValueError` if `api_key` is empty |
| `client.close()` | none | `None`, awaitable on the async client. Both clients are context managers |
| `contacts.list(...)` | `search: str \| None` ("contains" match), `phone: str \| None` and `email: str \| None` (exact match), `include_blocked: bool \| None` | `PaginatedResponse[Contact]`, contacts of the contacts hub |
| `contacts.upsert(...)` | `phone=None`, `email=None` (at least one of the two), `name=None`, `first_name=None`, `last_name=None`, `company=None`, `tags: list[str] \| None`, `custom_fields: dict \| None` | `UpsertContactResponse`: the `Contact` plus `created`, `emailContact`, `emailUnsubscribed`, `unmatchedCustomFields`, `warnings`. Matched by phone, then email. Needs `contacts:write` |
| `contacts.get(contact_id)` | positional id | `Contact` |
| `email.send(...)` | `recipient: dict`, `template_id`, `from_name`, `from_email`, `subject=None`, `reply_to=None`, `variables: dict \| None` | `SendEmailResponse` |
| `email.list_templates()` / `email.get_template(template_id)` | none / positional `template_id` | `dict` / `EmailTemplate` |
| `email.list_contacts(...)` | `search: str \| None` | `PaginatedResponse[EmailContact]` |
| `email.upsert_contact(...)` | `email`, `first_name=None`, `last_name=None`, `phone=None`, `company=None`, `tags: list \| None`, `custom_fields: dict \| None` | `EmailContact`. Email list only: to create or update a contact use `contacts.upsert` |
| `whatsapp.send_template(...)` | `to`, `template_name`, `language_code`, `components: list \| None` | `SendWhatsAppTemplateResponse` |
| `whatsapp.upload_header_media(...)` | `file: bytes`, `filename`, `mime_type`, `expected_format: "IMAGE" \| "VIDEO" \| "DOCUMENT" \| None` | `dict` |
| `whatsapp.list_logs(...)` | `direction: "inbound" \| "outbound" \| None`, `status=None`, `from_date=None`, `to_date=None` (ISO dates, not addresses) | `PaginatedResponse[WhatsAppLog]` |
| `whatsapp.list_templates(...)` | `status=None`, `category=None` | `PaginatedResponse[WhatsAppTemplate]` |
| `whatsapp.get_conversation_by_phone(...)` | `phone`, `limit=50`, oldest message first | `WhatsAppConversationByPhone` |
| `whatsapp.list_campaigns(...)` | `status=None` | `PaginatedResponse[dict]` |
| `whatsapp.get_campaign(campaign_id)` / `whatsapp.get_campaign_analytics(campaign_id)` | positional id | `dict` |
| `whatsapp.create_campaign(...)` | `template_id`, `phones: list[str]`, `name=None`, `template_components=None`, `variable_mapping=None`, `scheduled_at=None`, `schedule_timezone=None` | `dict` |
| `whatsapp.launch_campaign(campaign_id)` / `whatsapp.cancel_campaign(campaign_id)` | positional id | `dict` |
| `instagram.get_profile()` | none | `InstagramProfile` |
| `instagram.list_conversations(...)` | pagination only | `PaginatedResponse[InstagramConversation]` |
| `instagram.send_message(...)` | `recipient_igsid`, `message` | `SendInstagramMessageResponse` |
| `instagram.get_insights(...)` | `period=None`, `since=None`, `until=None` | `dict` |
| `shopify.search_products(...)` | `q`, `limit=10` | `dict` |
| `shopify.get_product_types()` / `shopify.list_collections()` | none | `dict` |
| `shopify.get_product(product_id)` | positional id | `ShopifyProduct` |
| `shopify.get_collection_products(collection_id, ...)` | positional id, `limit=50` | `dict` |
| `shopify.get_order_status(...)` | `identifier` (order number or id) | `ShopifyOrder` or `list[ShopifyOrder]` |
| `shopify.get_customer_orders(...)` | `identifier` (phone or email) | `dict` |
| `shopify.validate_discount(...)` | `code` | `DiscountValidation` |
| `agents.list(...)` / `agents.get(agent_id)` | `q=None`, `audience=None`, pagination / positional id | `dict`. Needs `agents:read` |
| `agents.performance(agent_id, days=None)` | `days` 1 to 180, default 30 | `dict`: results its tools completed (bookings, orders, payments), cancellations apart, `conversionRate` (a percentage, `None` over no conversations), trend, cost per result in paise |
| `agents.list_runs(agent_id, ...)` / `agents.get_run(run_id)` | `status`, `outcome`, `channel`, `version_id`, `from_`, `to`, pagination | `dict`. Never what a customer wrote, what the agent replied or who the customer is |
| `agents.tool_catalog()` | none | `dict` |
| `agents.create(name=...)` / `agents.create_draft(agent_id)` / `agents.update_draft(agent_id, version_id, changes)` | keyword `name`, `description`, `audience` / id / dict of changes | `dict`. Needs `agents:write` |
| `agents.publish(agent_id, version_id)` / `agents.rollback(agent_id, version_id)` | ids | `dict`. Needs `agents:publish` |
| `voice.list_agents(...)` / `voice.get_agent(agent_id)` | pagination only / positional id | `PaginatedResponse[VoiceAgent]` / `VoiceAgent` |
| `voice.list_calls(...)` | `status=None`, `direction=None` | `PaginatedResponse[VoiceCall]` |
| `voice.initiate_outbound_call(...)` | `agent_id`, `to_number`, `call_context: dict \| None` | `InitiateOutboundCallResponse` |
| `voice.get_call(call_id)` / `voice.get_call_recording(call_id)` | positional id | `VoiceCall` / `dict` |
| `voice.get_analytics(...)` | `from_date`, `to_date` | `VoiceAnalytics` |
| `sms.send(...)` | `sender_id`, `to: list[str]`, `message`, `template_id=None`, `route=None` | `SendSmsResponse` |
| `sms.list_templates(...)` / `sms.list_senders(...)` | `status=None` / pagination only | `PaginatedResponse[dict]` |
| `sms.list_campaigns(...)` / `sms.get_campaign(campaign_id)` | pagination only / positional id | `PaginatedResponse[dict]` / `dict` |
| `sms.create_campaign(...)` | `name`, `message_body`, `sender_id=None`, `template_id=None`, `route=None`, `audience_type=None`, `audience_criteria=None`, `scheduled_at=None` | `dict` |
| `sms.launch_campaign(campaign_id)` / `sms.cancel_campaign(campaign_id)` | positional id | `SmsCampaignActionResponse` |
| `sms.list_logs(...)` | `direction=None`, `status=None`, `start=None`, `end=None` | `PaginatedResponse[dict]` |
| `sms.get_stats(...)` | `start=None`, `end=None` | `SmsStats` |
| `rcs.send(...)` | `to`, `template_id`, `rcs_variables: dict \| None`, `sms_fallback: dict \| None`, `ttl: int \| None` | `dict` |
| `rcs.list_bots()` / `rcs.list_templates(...)` | none / pagination only | `list[dict]` / `PaginatedResponse[dict]` |
| `rcs.send_campaign(...)` | `bot_id`, `template_id`, `campaign_name=None`, `numbers=None`, `country=None`, `remove_duplicate=None`, `ttl=None`, `schedule_campaign=None`, `scheduled_at=None`, `fallback=None`, `fallback_message=None` | `dict` |
| `rcs.list_campaigns(...)` / `rcs.list_logs(...)` | pagination only | `PaginatedResponse[dict]` |
| `rcs.get_stats(...)` | `start=None`, `end=None` | `RcsStats` |
| `otp.send(...)` | `widget_id`, `identifier` (a phone number) | `SendOtpResponse` (`requestId`, `channel`) |
| `otp.retry(...)` | `request_id`, `channel=None` (`"sms"` or `"whatsapp"`) | `SendOtpResponse` (`requestId`, `channel`) |
| `otp.verify(...)` / `otp.verify_token(...)` | `request_id` and `otp` / `access_token` | `VerifyOtpResponse` / `VerifyOtpTokenResponse` |
| `otp.list_logs(...)` | `widget_id=None`, `status=None` | `PaginatedResponse[dict]` |
| `otp.get_analytics(...)` | `widget_id=None` | `dict` |
| `messages.send(...)` | `to`, `channel` (`"whatsapp"`, `"sms"`, `"rcs"`, `"email"` or `"instagram"`), `fallback_channels: list \| None` (`"whatsapp"`, `"sms"`, `"rcs"`), `message=None`, `template: dict \| None` (`name`, `id`, `languageCode`, `variables`, `components`), `sms: dict \| None` (`senderId`, `route`), `email: dict \| None` (`subject`, `fromName`, `fromEmail`, `replyTo`, `firstName`, `lastName`) | `SendMessageResponse` (`status`, `channelUsed`, `messageId`, `attempts`). Made once, never retried |
| `automations.list(...)` / `automations.get(automation_id)` | pagination only / positional id | `PaginatedResponse[AutomationSummary]` / `Automation`, with its trigger and steps |
| `automations.node_catalog()` | none | `AutomationNodeCatalog`, every trigger type and node type with the settings each node reads |
| `automations.create(...)` | `name`, `description=None`, `trigger: dict \| None`, `steps: list[dict] \| None`, `keywords: list[dict] \| None`, `posts: list[dict] \| None`, `check_follower=None`, `node_layout: dict \| None` | `CreateAutomationResponse`, ALWAYS created paused |
| `automations.update(automation_id, ...)` | positional id, any create argument, plus `active=None` | `Automation`. `trigger`, `steps`, `keywords` and `posts` replace what is stored |
| `automations.activate(automation_id)` / `automations.pause(automation_id)` | positional id | `Automation` |
| `automations.delete(automation_id)` | positional id | `None` |
| `automations.list_executions(automation_id, ...)` | positional id, pagination only | `PaginatedResponse[AutomationExecution]`, newest run first |
| `automations.rerun_execution(automation_id, execution_id)` | both positional | `dict` (`{"ok": True}`), the steps run again for real |
| `ai.list_models()` | none | `dict` (`{"data": [AiModel, ...]}`), each model with the price your workspace is charged |
| `webhooks.list_event_types()` | none | `dict`, the catalogue plus a sample payload each |
| `webhooks.list(...)` / `webhooks.get(endpoint_id)` | pagination only / positional id | `PaginatedResponse[WebhookEndpoint]` / `WebhookEndpoint` |
| `webhooks.create(...)` | `url`, `events: list[str]` (`["*"]` for all), `description=None`, `is_active=None` | `dict` including `secret`, returned once only |
| `webhooks.update(endpoint_id, ...)` | positional id, `url=None`, `events=None`, `description=None`, `is_active=None` | `WebhookEndpoint` |
| `webhooks.delete(endpoint_id)` / `webhooks.rotate_secret(endpoint_id)` | positional id | `None` / `dict`, the previous secret stops working |
| `webhooks.send_test(endpoint_id, ...)` | positional id, `event_type=None` | `dict` |
| `webhooks.list_deliveries(endpoint_id, ...)` | positional id, `status=None` | `PaginatedResponse[WebhookDelivery]` |
| `webhooks.get_delivery(delivery_id)` / `webhooks.replay_delivery(delivery_id)` | positional id | `dict`, a replay reuses the original event id |
| `chat.create_token(...)` | `user_id`, `name=None`, `role=None`, `app=None`, `ttl_seconds=None` (300 to 86400, default 3600) | `ChatToken` (`token`, `expiresAt`, `user`). Call it again to refresh |
| `chat.revoke_user(user_id, ...)` | positional id, `app=None` | `ChatUserRevoked` (`revoked`, `notBefore`), every earlier token of that user stops working |
| `chat.delete_user(user_id, ...)` | positional id, `app=None` | `ChatUserDeleted` (`deleted`, `conversations`), erases the user's conversations; safe to call twice |

### Module-level exports

| Export | Signature |
|---|---|
| `verify_webhook_signature` | `(raw_body: str \| bytes, signature_header: str, secret: str, tolerance_seconds: int = 300) -> bool`, constant-time, accepts several `v1` values during a rotation |
| `construct_event` | `(raw_body, signature_header, secret, tolerance_seconds=300) -> dict`, verifies then parses, raises rather than returning unverified data |
| `DEFAULT_TOLERANCE_SECONDS` | `300`, the maximum timestamp drift accepted |
| `SlideSignatureVerificationError` | raised by `construct_event` |

Failed requests raise `SlideValidationError` (400), `SlideAuthError` (401), `SlideScopeError` (403), `SlideNotFoundError` (404) or the base `SlideError` for any other status. All carry `.status_code`, `.body` and the API's message. Response `TypedDict`s such as `Contact`, `VoiceCall` and `WhatsAppLog` are importable from the package root.

## Configuration

| Argument | Type | Default | Notes |
|---|---|---|---|
| `api_key` | `str`, required, keyword-only | none | Sent as `Authorization: Bearer <key>`. Read it from an environment variable such as `SLIDE_API_KEY`, never hardcode it. To create or update contacts, create a new API key with `contacts:write` (and `contacts:read` to list or read them): a key that only has the email list permission cannot create contacts, and a key's permissions cannot be edited |
| `base_url` | `str` | `https://slide.synquic.com/api` | Trailing slash is normalised, and every resource path is appended as `v1/...` |
| `timeout` | `float` | `30.0` | Seconds, applied by the underlying HTTP client to every request |

Webhook signing secrets are separate from the API key: `webhooks.create()` returns one once, and that is what `construct_event` and `verify_webhook_signature` expect (for example from `SLIDE_WEBHOOK_SECRET`).

## Requirements

- Python **>= 3.9**
- `httpx >= 0.27.0`, the only declared dependency
- `typing_extensions` is imported by `synquic_slide.types`, and normally arrives transitively with `httpx`. Install it explicitly if your environment prunes transitive dependencies

## Development

The project builds with the hatchling backend declared in `pyproject.toml`:

```bash
pip install -e .
python -m build      # produces the wheel and sdist
```

The `chat` resource and the `chat.message.unread` event have tests in `tests/test_chat.py`, on the standard library's `unittest` with the HTTP layer replaced, so nothing leaves the machine:

```bash
python3 -B -m unittest discover -s tests -v
```

The `messages`, `automations`, `ai` and `contacts` resources are covered the same way (`tests/test_messages.py`, `tests/test_automations.py`, `tests/test_ai.py`, `tests/test_contacts.py`): every method is checked on both clients for its HTTP method, path, query and body.

No linter or formatter configuration is checked in to this repository.

## Licence

MIT.
