gushwork-argus

Python client for Argus, Gushwork's image ingestion and search API. You give Argus pages and images; it finds every image, describes it, and makes it searchable by meaning.

• One client, two settings: an API key and the API host.
• Argus (blocking) and AsyncArgus (asyncio) with the same methods.
• Typed responses (Pydantic v2), typed errors, retries with backoff.
• Python 3.10+. Depends only on httpx and pydantic.

Install

pip install gushwork-argus
# or
uv add gushwork-argus

Pin a version in services (gushwork-argus==0.1.0); see CHANGELOG.md.

Configure

from gushwork_argus import Argus

argus = Argus(api_key="...", base_url="https://<argus host>")

Or set ARGUS_API_KEY and ARGUS_BASE_URL and call Argus(). The Argus team gives you both: a key with the scopes your service needs, and the host for each environment.

Optional: timeout=30.0 (seconds per request), max_retries=2. Create one client per process and reuse it; it keeps a connection pool. Close it with argus.close() or use with Argus(...) as argus:.

Use

from pathlib import Path
from gushwork_argus import Argus, ConflictError

argus = Argus()

# 1. Create a namespace (admin key). Safe to retry with the same fields.
namespace = argus.namespaces.create(
    slug="acme-dental", name="Acme Dental", site_name="Acme Dental", domain="acme.com"
)

# 2. Get it back, by id or slug
namespace = argus.namespaces.get(namespace.id)
namespace = argus.namespaces.find_by_slug("acme-dental")  # None when there is none

# 3. Add a page or an image (ingest scope). Files go to S3 through a presigned upload.
page = argus.pages.add(namespace.id, Path("about.md"), page_url="https://acme.com/about")
page = argus.pages.add(namespace.id, html_text, type="html", page_url="https://acme.com/")
image = argus.images.add(namespace.id, file=Path("crown.jpg"), alt="Dentist fitting a crown")
image = argus.images.add(namespace.id, url="https://cdn.acme.com/crown.jpg")  # no upload

# 7. Check an ingestion, or wait until it is complete
status = argus.ingestions.get(page.ingestion_id)
status = argus.ingestions.wait(page.ingestion_id, timeout=600)
print(status.status, status.images.searchable, "of", status.images.total, "searchable")

# 4. Search (retrieve scope)
response = argus.images.search(
    namespace.id, "dentist fitting a crown", top_k=5, filters={"orientation": "landscape"}
)
for result in response.results:
    print(result.rank, result.match, result.image.url, result.description)

# 5. List namespaces: one page, or all of them
page_of_namespaces = argus.namespaces.list(limit=50)  # CursorPage: .items, .next_cursor
for ns in argus.namespaces.list_all():
    print(ns.slug)

# 6. List images, newest first
for img in argus.images.list_all(namespace.id, status="RETRIEVAL_READY"):
    print(img.media_id, img.description, img.file.url if img.file else None)

Async

from gushwork_argus import AsyncArgus

async with AsyncArgus() as argus:
    namespace = await argus.namespaces.get(namespace_id)
    page = await argus.pages.add(namespace.id, Path("about.md"))
    async for img in argus.images.list_all(namespace.id):
        ...

Pages and images, in detail

• pages.add(namespace_id, content, *, type=None, page_url=None, page_title=None, markdown=None) content is a Path (type read from .md/.html), or the page itself as str or bytes with type="md" or type="html". A str is always page text, never a file name. Send page_url whenever you have it: it resolves relative image links and tells the describer where each image appears. For HTML, markdown (the same page converted to markdown, if your scraper has it) improves captions. A page is identified by its bytes: sending the same bytes again returns the first ingestion with duplicate=True, and any new page_url, page_title or markdown is ignored.
• images.add(namespace_id, *, file=None, url=None, alt=None, caption=None, page_url=None) Exactly one of file (a Path, bytes or a file opened "rb") or url.
• An ingestion is complete when the page is parsed (or failed) and every image on it is searchable or failed. Complete is not success: check status, error and images.failed.
• Signed image links (url, thumbnail_url) expire after an hour. Store media_id, not the link.

Errors

Every runtime error is an ArgusError. Bad arguments raise ValueError before any request is sent.

Every APIStatusError has status_code, detail (the server's message, or a list of field errors) and body.

try:
    argus.namespaces.create(slug="acme-dental", name="Acme")
except ConflictError as exc:
    namespace = argus.namespaces.get(exc.namespace_id)

Retries

Writes are safe to repeat: namespaces are keyed by slug, pages by their content hash, images by their source and context. So every call except search is retried on connection errors, timeouts, 429, 502, 503 and 504, with exponential backoff and jitter (max_retries=2 by default; 0 turns retries off).

A search runs paid model calls each time, so it is retried only when it certainly never reached Argus (a refused connection, or 429). A search that times out raises APITimeoutError instead of running again; raise timeout if accurate searches on large namespaces need longer. ingestions.wait() keeps polling through connection errors and 5xx until its own timeout.

Retries are logged at INFO on the gushwork_argus logger.

S3 uploads go through a separate HTTP client that never carries the API key. It does not use a custom http_client; like any httpx client it reads HTTPS_PROXY and SSL_CERT_FILE from the environment.

API keys and scopes

Any key can read namespaces and list images. A key bound to a namespace sees only that namespace.

Develop

See CLAUDE.md for the layout, conventions and the release process.

cd sdk/python
uv sync
uv run pytest
uv run ruff check . && uv run ruff format --check . && uv run pyright