Files
bruno 2f2a5e9b4a
CI / Lint & Format (push) Failing after 28s
CI / Tests (push) Has been skipped
CI / Security Scan (push) Failing after 8s
CI / Docker Build (push) Has been skipped
Add admin panel, WebSocket support, and API versioning
Introduce an admin portal (React + Nginx), WebSocket routing, and
API versioning middleware with `/api/v1/` prefix deprecation.
Add master API key authentication, new Prometheus metrics for AI
token consumption and active WebSockets, and extend S3 config
with a public endpoint URL. Update test paths and fixtures to
align with the new routing structure.
2026-06-22 11:25:22 -04:00

1.8 KiB

Imago SDK - imago-client

The official Python SDK provides an asynchronous, strongly-typed interface to the Imago API.

Installation

cd sdk
pip install -e .

Basic Usage

The SDK provides an asynchronous Context Manager (HubClient) which handles connection pooling, retries, and authentication.

import asyncio
from imago_client import HubClient

async def main():
    async with HubClient("http://localhost:8000/api/v1", api_key="sk-test-secret-key") as client:
        # 1. Upload an image
        image = await client.images.upload("photo.jpg")
        print(f"Uploaded! UUID: {image.uuid}")

        # 2. Get status
        details = await client.images.get(image.uuid)
        print(f"Status: {details.processing_status}")
        
        # 3. Use AI
        summary = await client.ai.draft_task("https://example.com/article", "Summarize this quickly.")
        print(summary)

if __name__ == "__main__":
    asyncio.run(main())

Advanced Features

1. Resilient Retries

By default, the SDK automatically retries requests on timeouts, TCP connection errors, 429 Too Many Requests, and 5xx Server Errors. It uses exponential backoff with jitter to ensure safe recovery.

2. Real-Time Pipeline Streaming

The SDK natively wraps the WebSockets API. You can async for over events until the pipeline finishes:

async with HubClient(url, api_key) as client:
    async for event in client.images.stream_pipeline_events("IMAGE_UUID"):
        print(event["event"])
        if event["event"] == "pipeline.done":
            break

3. Error Handling

All specific cases throw custom exceptions inherited from ImagoError:

  • AuthError (401)
  • QuotaError (403 or 429 quota exhaustion)
  • NotFoundError (404)
  • PipelineError (WebSockets specific)