Skip to main content

​
Installation

pip install chatads-sdk

​
Quick Start

from chatads_sdk import ChatAdsClient

# Use context manager for automatic cleanup
with ChatAdsClient(api_key="cak_your_api_key") as client:
    response = client.analyze_message(
        message="What are the best noise-cancelling headphones?",
        country="US"
    )

    if response.error is None and response.data and response.data.returned > 0:
        # Access first offer
        offer = response.data.offers[0]
        print(f"link_text: {offer.link_text}")
        print(f"url: {offer.url}")

        # Or iterate all offers
        for i, offer in enumerate(response.data.offers):
            print(f"Offer {i + 1}: {offer.link_text} - {offer.url}")

​
Configuration

from chatads_sdk import ChatAdsClient

# Basic usage with context manager (recommended)
with ChatAdsClient(
    api_key="cak_your_api_key",          # Required
    base_url="https://...",               # Optional: custom API URL
    raise_on_failure=True,                # Optional: raise exceptions on API errors
    max_retries=2,                        # Optional: retry failed requests
    retry_backoff_factor=0.75,            # Optional: backoff multiplier between retries
    debug=False,                          # Optional: enable debug logging
) as client:
    # Use client here
    pass

​
Methods

​
analyze_message

Analyze a message for affiliate opportunities. 
response = client.analyze_message(
    message="Looking for a good yoga mat",
    country="US",
    quality="standard"              # fast | standard | best
)

​
Response Structure

# response.data.status: str                 # "filled", "partial_fill", "no_offers_found", or "internal_error"
# response.data.offers: List[Offer]         # Array of affiliate offers (only contains filled offers)
# response.data.requested: int              # Number requested
# response.data.returned: int               # Actual number returned
# response.error: ChatAdsError | None       # Error details (None on success)
# response.meta.request_id: str
# response.meta.usage.monthly_requests: int
# response.meta.usage.daily_requests: int

# Each offer in response.data.offers has:
# offer.link_text: str                       # Text to use for affiliate link
# offer.url: str                             # Affiliate URL (always populated)
# offer.confidence_level: str                # Confidence classification (low, medium, high)
# offer.product: Product                     # Product metadata (optional)

​
Error Handling

The SDK provides two exception types for different error scenarios:
  • ChatAdsAPIError - Raised for API-level errors (HTTP 4xx/5xx responses)
  • ChatAdsSDKError - Raised for SDK-level issues (validation, configuration, transport)
from chatads_sdk import ChatAdsClient, ChatAdsAPIError, ChatAdsSDKError

with ChatAdsClient(api_key="cak_your_api_key", raise_on_failure=True) as client:
    try:
        response = client.analyze_message(message="...")

        if response.error is not None:
            if response.error.code in ("DAILY_LIMIT_EXCEEDED", "MONTHLY_LIMIT_EXCEEDED"):
                # Rate limit hit - wait for reset
                pass
            elif response.error.code == "UNAUTHORIZED":
                # Check API key
                pass
            else:
                print(f"Error: {response.error.message}")

    except ChatAdsAPIError as e:
        # API returned an error response (4xx/5xx)
        print(f"API error: {e.status_code} - {e.payload}")
    except ChatAdsSDKError as e:
        # SDK-level error (validation, config, transport)
        print(f"SDK error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

​
Framework Examples

​
FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI
from chatads_sdk import AsyncChatAdsClient
import os

# Use lifespan for proper client lifecycle management
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.chatads = AsyncChatAdsClient(api_key=os.environ["CHATADS_API_KEY"])
    yield
    await app.state.chatads.close()

app = FastAPI(lifespan=lifespan)

@app.post("/analyze")
async def analyze(message: str):
    response = await app.state.chatads.analyze_message(
        message=message,
        country="US"
    )
    return response.raw

​
Flask

from flask import Flask, request, jsonify, g
from chatads_sdk import ChatAdsClient
import os

app = Flask(__name__)

def get_chatads_client():
    if 'chatads' not in g:
        g.chatads = ChatAdsClient(api_key=os.environ["CHATADS_API_KEY"])
    return g.chatads

@app.teardown_appcontext
def close_chatads(error):
    client = g.pop('chatads', None)
    if client is not None:
        client.close()

@app.route("/analyze", methods=["POST"])
def analyze():
    data = request.get_json()
    client = get_chatads_client()
    response = client.analyze_message(
        message=data["message"]
    )
    return jsonify(response.raw)

​
Django

# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from chatads_sdk import ChatAdsClient
import json
import os

@require_POST
def analyze(request):
    data = json.loads(request.body)
    # Use context manager for automatic cleanup
    with ChatAdsClient(api_key=os.environ["CHATADS_API_KEY"]) as client:
        response = client.analyze_message(
            message=data["message"]
        )
        return JsonResponse(response.raw)

​
Environment Variables

We recommend storing your API key in environment variables: 
import os
from chatads_sdk import ChatAdsClient

with ChatAdsClient(api_key=os.environ["CHATADS_API_KEY"]) as client:
    response = client.analyze_message(message="...")

export CHATADS_API_KEY=cak_your_api_key

​
Using FunctionItemPayload

For more control over request parameters, use FunctionItemPayload: 
from chatads_sdk import ChatAdsClient, FunctionItemPayload

with ChatAdsClient(api_key="cak_your_api_key") as client:
    payload = FunctionItemPayload(
        message="Looking for a CRM to close more deals",
        ip="1.2.3.4",
        country="US",
        quality="standard"            # fast | standard | best
    )

    result = client.analyze(payload)

    if result.error is None and result.data.returned > 0:
        print(result.data.offers[0])
    else:
        print("No offers found")

​
Async Client

For asynchronous workflows, use AsyncChatAdsClient: 
import asyncio
from chatads_sdk import AsyncChatAdsClient, FunctionItemPayload

async def main():
    async with AsyncChatAdsClient(
        api_key="cak_your_api_key",
        max_retries=3,
        debug=True,
    ) as client:
        # Using analyze_message shorthand
        result = await client.analyze_message(
            message="Need data warehousing ideas",
            country="US",
            ip="1.2.3.4",
        )

        print(result.raw)

        # Or using FunctionItemPayload for full control
        payload = FunctionItemPayload(
            message="Best standing desk for home office",
            country="US",
        )
        result = await client.analyze(payload)

        if result.error is None and result.data.returned > 0:
            print(f"Found: {result.data.offers[0].link_text}")

asyncio.run(main())

​
Concurrent Requests

Process multiple messages concurrently: 
import asyncio
from chatads_sdk import AsyncChatAdsClient

async def analyze_batch(messages: list[str]):
    async with AsyncChatAdsClient(api_key="cak_your_api_key") as client:
        tasks = [
            client.analyze_message(message=msg, country="US")
            for msg in messages
        ]
        results = await asyncio.gather(*tasks)
        return results

# Usage
messages = [
    "Best noise-cancelling headphones",
    "Top rated standing desks",
    "Ergonomic office chairs",
]
results = asyncio.run(analyze_batch(messages))