Skip to Content
IntegrationsPython

Python

Use Convoy’s REST API directly from Python with httpx or requests. This guide provides ready-to-use patterns for submitting prompts, receiving results via callbacks, and polling for status.

Testing tip — Build and verify this integration end-to-end with convoy-mock first. It’s a free, synthetic model that returns a callback in ~60 seconds and is never billed. Once your wiring works, swap "model": "convoy-mock" for a production model like claude-3-haiku.

Quick Start

Submit a prompt and receive the result via callback in under 10 lines:

import httpx
 
response = httpx.post(
    "https://api.cnvy.ai/cargo/load",
    headers={"X-API-Key": "convoy_sk_your_key_here"},
    json={
        "params": {
            "model": "claude-3-haiku",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Explain quantum computing in 3 sentences."}]
        },
        "callback_url": "https://your-server.com/webhook/convoy"
    }
)
 
data = response.json()
print(f"Submitted! Cargo ID: {data['cargo_id']}")

Convoy processes requests in cost-efficient batches. Results arrive at your callback_url within minutes to hours, depending on queue volume and batch size (up to 24-hour SLA). For the callback receiver pattern, see the Webhooks guide.

Installation

# Using pip
pip install httpx
 
# Or using uv (recommended)
uv add httpx

We use httpx in this guide because it supports both sync and async patterns. You can substitute requests for the synchronous examples — the API is nearly identical.

Convoy Client Class

Here’s a reusable client class that wraps the Convoy API:

"""convoy_client.py — Lightweight Convoy API wrapper."""
 
import time
from dataclasses import dataclass
from typing import Optional
 
import httpx
 
 
@dataclass
class CargoResult:
    """Result from a cargo tracking request."""
    cargo_id: str
    status: str
    response: Optional[dict] = None
    error: Optional[str] = None
 
    @property
    def is_complete(self) -> bool:
        return self.status == "COMPLETED"
 
    @property
    def is_failed(self) -> bool:
        return self.status == "FAILED"
 
    @property
    def text(self) -> Optional[str]:
        """Extract the generated text from the response."""
        if self.response and self.response.get("content"):
            return self.response["content"][0]["text"]
        return None
 
 
class ConvoyClient:
    """Client for the Convoy batch AI inference API."""
 
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.cnvy.ai",
        timeout: float = 30.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.client = httpx.Client(
            base_url=self.base_url,
            headers={
                "X-API-Key": api_key,
                "Content-Type": "application/json",
            },
            timeout=timeout,
        )
 
    def load(
        self,
        messages: list[dict],
        model: str = "claude-3-haiku",
        max_tokens: int = 1024,
        callback_url: Optional[str] = None,
        system: Optional[str] = None,
        temperature: Optional[float] = None,
    ) -> str:
        """
        Submit a prompt for batch processing.
 
        Returns the cargo_id for tracking.
        """
        params = {
            "model": model,
            "max_tokens": max_tokens,
            "messages": messages,
        }
        if system:
            params["system"] = system
        if temperature is not None:
            params["temperature"] = temperature
 
        body = {"params": params}
        if callback_url:
            body["callback_url"] = callback_url
 
        response = self.client.post("/cargo/load", json=body)
        response.raise_for_status()
        return response.json()["cargo_id"]
 
    def track(self, cargo_id: str) -> CargoResult:
        """Check the status of a submitted cargo request."""
        response = self.client.get(f"/cargo/{cargo_id}/tracking")
        response.raise_for_status()
        data = response.json()
        return CargoResult(
            cargo_id=cargo_id,
            status=data["status"],
            response=data.get("response"),
            error=data.get("error"),
        )
 
    def wait_for_result(
        self,
        cargo_id: str,
        poll_interval: float = 10.0,
        timeout: float = 1800.0,
    ) -> CargoResult:
        """
        Poll until the cargo is complete or failed.
 
        Args:
            cargo_id: The cargo ID to track.
            poll_interval: Seconds between polls (default 10s).
            timeout: Maximum seconds to wait (default 30 minutes).
 
        Returns:
            CargoResult with the final status and response.
 
        Raises:
            TimeoutError: If the result isn't ready within the timeout.
        """
        start = time.time()
        while time.time() - start < timeout:
            result = self.track(cargo_id)
            if result.is_complete or result.is_failed:
                return result
            time.sleep(poll_interval)
        raise TimeoutError(
            f"Cargo {cargo_id} did not complete within {timeout}s"
        )
 
    def models(self) -> list[dict]:
        """List available models."""
        response = self.client.get("/models")
        response.raise_for_status()
        return response.json()
 
    def close(self):
        """Close the HTTP client."""
        self.client.close()
 
    def __enter__(self):
        return self
 
    def __exit__(self, *args):
        self.close()

Usage Examples

Basic: Submit and Poll

The simplest pattern — submit a prompt and poll until the result is ready:

from convoy_client import ConvoyClient
 
client = ConvoyClient(api_key="convoy_sk_your_key_here")
 
# Submit a prompt
cargo_id = client.load(
    messages=[{"role": "user", "content": "Write a haiku about Python programming."}],
    model="claude-3-haiku",
    max_tokens=256,
)
print(f"Submitted: {cargo_id}")
 
# Wait for the result (polls every 10 seconds)
result = client.wait_for_result(cargo_id)
 
if result.is_complete:
    print(f"Result:\n{result.text}")
else:
    print(f"Failed: {result.error}")

With System Prompt

cargo_id = client.load(
    messages=[{"role": "user", "content": "Explain our Q4 revenue growth to investors."}],
    model="claude-3-sonnet",
    max_tokens=2048,
    system="You are a senior financial analyst. Write in a professional, concise tone suitable for investor communications.",
    temperature=0.3,
)

With Callback URL

For production use, callbacks are more efficient than polling:

cargo_id = client.load(
    messages=[{"role": "user", "content": "Generate a product description for our new widget."}],
    model="claude-3-sonnet",
    max_tokens=1024,
    callback_url="https://your-server.com/webhook/convoy",
)
print(f"Submitted: {cargo_id} — result will be delivered to your webhook")

Batch Submit Multiple Prompts

Process many prompts efficiently:

import time
from convoy_client import ConvoyClient
 
prompts = [
    "Write a tweet about our new AI feature",
    "Generate a subject line for our newsletter",
    "Create a one-paragraph product description",
    "Write a customer testimonial request email",
    "Draft a changelog entry for version 2.0",
]
 
client = ConvoyClient(api_key="convoy_sk_your_key_here")
 
# Submit all prompts
cargo_ids = []
for prompt in prompts:
    cargo_id = client.load(
        messages=[{"role": "user", "content": prompt}],
        model="claude-3-haiku",
        max_tokens=512,
        system="You are a marketing copywriter. Be concise and engaging.",
    )
    cargo_ids.append(cargo_id)
    print(f"Submitted: {cargo_id}")
    time.sleep(0.1)  # Small delay to avoid rate limits
 
print(f"\nSubmitted {len(cargo_ids)} prompts. Waiting for results...")
 
# Wait for all results
for cargo_id in cargo_ids:
    result = client.wait_for_result(cargo_id, poll_interval=15.0)
    if result.is_complete:
        print(f"\n[{cargo_id}] ✅\n{result.text}\n{'─' * 40}")
    else:
        print(f"\n[{cargo_id}] ❌ {result.error}")

Convoy allows 60 requests per minute and 500 per hour per project. When submitting many prompts, add a small delay between requests or use the batch callback pattern.

Async Client

For high-throughput applications, use the async version:

"""convoy_async.py — Async Convoy API wrapper."""
 
import asyncio
from typing import Optional
 
import httpx
 
from convoy_client import CargoResult
 
 
class AsyncConvoyClient:
    """Async client for the Convoy batch AI inference API."""
 
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.cnvy.ai",
        timeout: float = 30.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "X-API-Key": api_key,
                "Content-Type": "application/json",
            },
            timeout=timeout,
        )
 
    async def load(
        self,
        messages: list[dict],
        model: str = "claude-3-haiku",
        max_tokens: int = 1024,
        callback_url: Optional[str] = None,
        system: Optional[str] = None,
        temperature: Optional[float] = None,
    ) -> str:
        """Submit a prompt for batch processing. Returns cargo_id."""
        params = {
            "model": model,
            "max_tokens": max_tokens,
            "messages": messages,
        }
        if system:
            params["system"] = system
        if temperature is not None:
            params["temperature"] = temperature
 
        body = {"params": params}
        if callback_url:
            body["callback_url"] = callback_url
 
        response = await self.client.post("/cargo/load", json=body)
        response.raise_for_status()
        return response.json()["cargo_id"]
 
    async def track(self, cargo_id: str) -> CargoResult:
        """Check the status of a submitted cargo request."""
        response = await self.client.get(f"/cargo/{cargo_id}/tracking")
        response.raise_for_status()
        data = response.json()
        return CargoResult(
            cargo_id=cargo_id,
            status=data["status"],
            response=data.get("response"),
            error=data.get("error"),
        )
 
    async def wait_for_result(
        self,
        cargo_id: str,
        poll_interval: float = 10.0,
        timeout: float = 1800.0,
    ) -> CargoResult:
        """Poll until complete or failed."""
        import time
        start = time.time()
        while time.time() - start < timeout:
            result = await self.track(cargo_id)
            if result.is_complete or result.is_failed:
                return result
            await asyncio.sleep(poll_interval)
        raise TimeoutError(f"Cargo {cargo_id} did not complete within {timeout}s")
 
    async def close(self):
        await self.client.aclose()
 
    async def __aenter__(self):
        return self
 
    async def __aexit__(self, *args):
        await self.close()

Async Usage

import asyncio
from convoy_async import AsyncConvoyClient
 
async def main():
    async with AsyncConvoyClient(api_key="convoy_sk_your_key_here") as client:
        # Submit multiple prompts concurrently
        tasks = [
            client.load(
                messages=[{"role": "user", "content": f"Write tip #{i} for Python developers."}],
                model="claude-3-haiku",
                max_tokens=256,
            )
            for i in range(1, 6)
        ]
        cargo_ids = await asyncio.gather(*tasks)
        print(f"Submitted {len(cargo_ids)} prompts")
 
        # Wait for all results concurrently
        results = await asyncio.gather(
            *[client.wait_for_result(cid) for cid in cargo_ids]
        )
 
        for result in results:
            if result.is_complete:
                print(f"✅ {result.text[:80]}...")
            else:
                print(f"❌ {result.error}")
 
asyncio.run(main())

Environment Configuration

Keep your API key out of source code:

import os
from convoy_client import ConvoyClient
 
client = ConvoyClient(
    api_key=os.environ["CONVOY_API_KEY"],
    base_url=os.environ.get("CONVOY_BASE_URL", "https://api.cnvy.ai"),
)

.env file:

CONVOY_API_KEY=convoy_sk_your_key_here
CONVOY_BASE_URL=https://api.cnvy.ai  # or http://localhost:8000 for local dev

Never commit API keys to version control. Use environment variables or a secrets manager.

Error Handling

The client raises httpx.HTTPStatusError for API errors. Handle common cases:

import httpx
from convoy_client import ConvoyClient
 
client = ConvoyClient(api_key="convoy_sk_your_key_here")
 
try:
    cargo_id = client.load(
        messages=[{"role": "user", "content": "Hello!"}],
        model="claude-3-haiku",
    )
except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Invalid API key. Check your CONVOY_API_KEY.")
    elif e.response.status_code == 422:
        print(f"Invalid request: {e.response.json()}")
    elif e.response.status_code == 429:
        print("Rate limited. Wait and retry.")
    else:
        print(f"API error {e.response.status_code}: {e.response.text}")
except httpx.ConnectError:
    print("Cannot reach Convoy API. Check your network or base_url.")

Retry with Backoff

For production reliability, add automatic retries:

import time
import httpx
from convoy_client import ConvoyClient
 
 
def load_with_retry(
    client: ConvoyClient,
    messages: list[dict],
    max_retries: int = 3,
    **kwargs,
) -> str:
    """Submit a prompt with automatic retry on transient failures."""
    for attempt in range(max_retries + 1):
        try:
            return client.load(messages=messages, **kwargs)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429 and attempt < max_retries:
                wait = 2 ** attempt  # 1s, 2s, 4s
                print(f"Rate limited. Retrying in {wait}s...")
                time.sleep(wait)
            elif e.response.status_code >= 500 and attempt < max_retries:
                wait = 2 ** attempt
                print(f"Server error. Retrying in {wait}s...")
                time.sleep(wait)
            else:
                raise
    raise RuntimeError("Max retries exceeded")

Listing Available Models

from convoy_client import ConvoyClient
 
client = ConvoyClient(api_key="convoy_sk_your_key_here")
 
models = client.models()
for model in models:
    print(f"  {model['id']} — {model.get('description', 'No description')}")

Integration Patterns

Pattern 1: Script with CSV Input/Output

Process a CSV of prompts and write results to a new file:

import csv
from convoy_client import ConvoyClient
 
client = ConvoyClient(api_key="convoy_sk_your_key_here")
 
# Read prompts from CSV
with open("prompts.csv") as f:
    reader = csv.DictReader(f)
    rows = list(reader)
 
# Submit all prompts
for row in rows:
    row["cargo_id"] = client.load(
        messages=[{"role": "user", "content": row["prompt"]}],
        model=row.get("model", "claude-3-haiku"),
        max_tokens=int(row.get("max_tokens", 1024)),
    )
    print(f"Submitted: {row['cargo_id']}")
 
# Wait and collect results
results = []
for row in rows:
    result = client.wait_for_result(row["cargo_id"])
    results.append({
        "prompt": row["prompt"],
        "cargo_id": row["cargo_id"],
        "status": result.status,
        "response": result.text or result.error,
    })
 
# Write results
with open("results.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["prompt", "cargo_id", "status", "response"])
    writer.writeheader()
    writer.writerows(results)
 
print(f"Done! {len(results)} results written to results.csv")

Pattern 2: Django/Flask Background Job

Submit from a web request, process the callback asynchronously:

# views.py (Django example)
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from convoy_client import ConvoyClient
 
client = ConvoyClient(api_key=settings.CONVOY_API_KEY)
 
def submit_prompt(request):
    """User submits a prompt via the web UI."""
    prompt = request.POST["prompt"]
 
    cargo_id = client.load(
        messages=[{"role": "user", "content": prompt}],
        model="claude-3-sonnet",
        callback_url=f"{settings.BASE_URL}/api/webhook/convoy/",
    )
 
    # Save to database for tracking
    PromptJob.objects.create(
        user=request.user,
        cargo_id=cargo_id,
        prompt=prompt,
        status="pending",
    )
 
    return JsonResponse({"cargo_id": cargo_id, "status": "submitted"})
 
 
@csrf_exempt
def convoy_webhook(request):
    """Receive results from Convoy."""
    import json
    payload = json.loads(request.body)
 
    job = PromptJob.objects.get(cargo_id=payload["cargo_id"])
    if payload["success"]:
        job.status = "completed"
        job.result = payload["response"]["content"][0]["text"]
    else:
        job.status = "failed"
        job.error = payload["error"]
    job.save()
 
    # Notify the user (email, websocket, etc.)
    notify_user(job.user, job)
 
    return JsonResponse({"received": True})

Pattern 3: CLI Tool

A simple command-line tool for quick prompts:

#!/usr/bin/env python3
"""convoy-cli — Submit prompts to Convoy from the command line."""
 
import argparse
import os
import sys
 
from convoy_client import ConvoyClient
 
 
def main():
    parser = argparse.ArgumentParser(description="Submit a prompt to Convoy")
    parser.add_argument("prompt", help="The prompt to send")
    parser.add_argument("--model", default="claude-3-haiku", help="Model to use")
    parser.add_argument("--max-tokens", type=int, default=1024)
    parser.add_argument("--system", help="System prompt")
    parser.add_argument("--poll-interval", type=float, default=10.0)
    args = parser.parse_args()
 
    api_key = os.environ.get("CONVOY_API_KEY")
    if not api_key:
        print("Error: Set CONVOY_API_KEY environment variable", file=sys.stderr)
        sys.exit(1)
 
    client = ConvoyClient(api_key=api_key)
 
    print(f"Submitting to {args.model}...", file=sys.stderr)
    cargo_id = client.load(
        messages=[{"role": "user", "content": args.prompt}],
        model=args.model,
        max_tokens=args.max_tokens,
        system=args.system,
    )
    print(f"Cargo ID: {cargo_id}", file=sys.stderr)
    print("Waiting for result...", file=sys.stderr)
 
    result = client.wait_for_result(cargo_id, poll_interval=args.poll_interval)
 
    if result.is_complete:
        print(result.text)
    else:
        print(f"Error: {result.error}", file=sys.stderr)
        sys.exit(1)
 
 
if __name__ == "__main__":
    main()

Usage:

export CONVOY_API_KEY=convoy_sk_your_key_here
 
# Simple prompt
python convoy-cli.py "Write a haiku about containers"
 
# With options
python convoy-cli.py "Explain Kubernetes" --model claude-3-sonnet --system "Explain like I'm 5"
 
# Pipe output
python convoy-cli.py "Generate 10 test email addresses as JSON" | jq .

Troubleshooting

httpx.ConnectError: Connection refused

  • Check that CONVOY_BASE_URL is correct
  • For local development, ensure the Convoy API is running (make up)
  • Default local URL: http://localhost:8000

401 Unauthorized

  • Verify your API key starts with convoy_sk_
  • Check the key hasn’t been revoked in the dashboard
  • Ensure the header is X-API-Key (case-sensitive)

422 Unprocessable Entity

  • Check the request body format — messages must be a list of {"role": "...", "content": "..."}
  • Verify the model ID is valid (use client.models() to list)
  • Ensure max_tokens is a positive integer

429 Too Many Requests

  • You’ve hit the rate limit (60/min or 500/hour)
  • Add delays between requests or implement retry with backoff
  • Consider using callbacks instead of polling to reduce API calls

Polling timeout

  • Batch processing can take minutes to hours depending on queue volume
  • Increase the timeout parameter in wait_for_result()
  • For production, use callbacks instead of polling
Last updated on
WebhooksNode.js / TypeScript