Files
bruno d323ed98e5
CI / Lint & Format (push) Failing after 10s
CI / Tests (push) Has been skipped
CI / Security Scan (push) Failing after 8s
CI / Docker Build (push) Has been skipped
T1-T4: tests de resilience, SDK PyPI ready, worker docker-compose, fix compat tests
2026-06-22 16:49:39 -04:00
..

Imago Python SDK

The official Python client for interacting with the Imago backend API.

Installation

pip install imago-client

For development:

cd sdk
pip install -e ".[dev]"

Publishing to PyPI

cd sdk
pip install build twine
python -m build
twine check dist/*
twine upload dist/*

Basic Usage

import asyncio
from imago_client import HubClient

async def main():
    async with HubClient("http://localhost:8000", api_key="your-api-key") as client:
        # Upload an image
        with open("photo.jpg", "rb") as f:
            image = await client.images.upload(f.read(), "photo.jpg")
            print(f"Uploaded! ID: {image.id}")

            # Stream real-time pipeline events
            stream = client.images.stream_pipeline(image.id, "your-api-key")
            async for event in stream.stream_events():
                print(f"Event: {event}")

        # Get processed data
        processed = await client.images.get(image.id)
        if processed.ai:
            print("AI Description:", processed.ai.description)

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

API Reference

HubClient

client = HubClient(
    base_url="http://localhost:8000",  # Imago API URL
    api_key="your-api-key",            # Imago API key
    timeout=30.0,                      # Request timeout (seconds)
)

Images Resource

# Upload
image = await client.images.upload(data, filename, content_type="image/jpeg")

# Get
detail = await client.images.get(image_id)

# List
images = await client.images.list(page=1, page_size=20, tag="nature")

# Delete
await client.images.delete(image_id)

# Reprocess
await client.images.reprocess(image_id)

# WebSocket pipeline stream
stream = client.images.stream_pipeline(image_id, api_key)
async for event in stream.stream_events():
    print(event)

AI Resource

# Summarize URL
result = await client.ai.summarize("https://example.com")

# Draft task
task = await client.ai.draft_task("Organiser les photos de vacances")