Python SDK

pip install infery

import os

from infery import Infery

client = Infery(api_key=os.environ["INFERY_API_KEY"])

chat = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Say hello in one line."}],
)

print(chat.choices[0]["message"]["content"])

import asyncio
import os

from infery import AsyncInfery


async def main() -> None:
    async with AsyncInfery(api_key=os.environ["INFERY_API_KEY"]) as client:
        chat = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Say hello in one line."}],
        )
        print(chat.choices[0]["message"]["content"])


asyncio.run(main())

for chunk in client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a haiku about the sea."}],
):
    if chunk["choices"]:
        print(chunk["choices"][0]["delta"].get("content", ""), end="")

# `modality` is exactly what GET /v1/models reports as `_infery.modality`,
# so it can come straight off a model you looked up.
result = client.media.generate(
    modality="video",  # "image" | "video" | "audio" | "music" | "object_3d" | "upscale"
    model="veo-3.1",
    prompt="a drone shot over a coastline",
    duration=8,  # model-specific params pass straight through
)

print(result.artifacts[0].url)

def make(job: dict[str, str]) -> list[str | bytes | None]:
    result = client.media.generate(**job)
    return [a.url or a.b64 or a.data for a in result.artifacts]

# Reports.
client.media.generate(
    modality="video",
    model="veo-3.1",
    prompt="a reef",
    on_progress=lambda p: print(p.status, p.progress),
)

# Accepted, never called.
client.media.generate(
    modality="image",
    model="flux-pro",
    prompt="a reef",
    on_progress=lambda p: print(p.status, p.progress),
)

started = client.media.generate(modality="image", model="flux-pro", prompt="a topographic map", background=True)

# `modality` is required here too: video jobs and every other modality's jobs
# live at different endpoints, and a job id does not say which it is.
done = client.media.wait(modality="image", job_id=started.job_id, on_progress=print)

@dataclass(frozen=True)
class MediaResult:
    modality: MediaModality
    artifacts: list[MediaArtifact]
    created: int | None
    credits_used: float | None
    job_id: str | None            # background=True only
    artifacts_expired: bool | None
    raw: Any                      # the untouched per-modality payload

# Waits, and collects the result if the gateway defers. Either way you get an image.
image = client.images.generate(model="dall-e-3", prompt="an isometric dashboard")
print(image.data[0]["url"])

from infery import JobDeferredError

try:
    client.images.generate(model="flux-pro", prompt="a topographic map", background=True)
except JobDeferredError as deferred:
    job = client.jobs.wait(deferred.job_id, interval=5.0, max_wait=600.0)
    print(job.data[0]["url"])

started = client.media.generate(modality="image", model="flux-pro", prompt="a map", background=True)
done = client.media.wait(modality="image", job_id=started.job_id)

video = client.videos.generate(model="veo-3", prompt="a drone shot over a coastline")
# A finished video carries its file on `result`, not on `data[]`.
print(video.result["url"])

run = client.workflows.runs.create(
    workflow_id="wf_abc123",
    input={"topic": "quarterly summary"},
)
print(run.status, run.creditsUsed)

for event in client.workflows.runs.stream(workflow_id="wf_abc123", input={}):
    if event["type"] == "step.started":
        print("running", event["stepId"])
    elif event["type"] == "step.completed":
        print("done", event["stepId"], event["creditsUsed"])
    elif event["type"] == "pipeline.completed":
        print("total", event["creditsUsed"])
    elif event["type"] == "unknown_event":
        print("new event", event["name"])

from infery import APIError, ConflictError, InsufficientCreditsError

try:
    client.chat.completions.create(model="gpt-4o", messages=messages)
except InsufficientCreditsError:
    ...  # 402, or any status with `insufficient_credits`. Top up; do not retry.
except ConflictError as err:
    if err.code == "upload_in_progress":
        ...  # Retry in a moment.
    raise
except APIError as err:
    print(err.status, err.code, err.request_id)

import httpx

from infery import Infery

pool = httpx.Client(
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=10),
    proxy="http://proxy.internal:3128",
)

client = Infery(api_key=os.environ["INFERY_API_KEY"], http_client=pool)

async with AsyncInfery(api_key=key) as client:
    ...
# The connection pool is closed here. Reusing `client` now fails.

client = AsyncInfery(api_key=key)
try:
    await serve(client)
finally:
    await client.aclose()