Skip to main content

Async Usage

The AsyncFerroClient provides the same API as FerroClient using httpx.AsyncClient under the hood. Use it in async frameworks like FastAPI, Starlette, or any asyncio application.

Setup​

from ferrolabsai import AsyncFerroClient

client = AsyncFerroClient(
    api_key="sk-ferro-...",
    base_url="http://localhost:8080",
)

Basic request​

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello from async!"}],
)
print(response.content)

Async streaming​

stream = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Count from 1 to 10"}],
    stream=True,
)

async for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="", flush=True)

Async context manager​

async with AsyncFerroClient() as client:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}],
    )
    print(response.content)
# httpx.AsyncClient is closed automatically

Async embeddings​

response = await client.embeddings.create(
    model="text-embedding-3-small",
    input=["async embedding request"],
)
print(len(response.data[0].embedding))

FastAPI example​

from fastapi import FastAPI
from ferrolabsai import AsyncFerroClient

app = FastAPI()
client = AsyncFerroClient()

@app.post("/chat")
async def chat(message: str):
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}],
    )
    return {
        "reply": response.content,
        "provider": response.provider,
        "cost": response.usage.cost_usd,
    }

Parallel requests​

import asyncio
from ferrolabsai import AsyncFerroClient

async def main():
    async with AsyncFerroClient() as client:
        tasks = [
            client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": f"What is {i} * {i}?"}],
            )
            for i in range(5)
        ]
        results = await asyncio.gather(*tasks)

        for r in results:
            print(f"{r.content} (via {r.provider})")

asyncio.run(main())

Current async coverage​

ResourceAsync support
chat.completions✅ Full (including streaming)
embeddings✅ Full
images✅ Full
models✅ Full
admin✅ Full
info

AsyncFerroClient exposes fully native async resources across every namespace — chat, embeddings, images, models, and admin — all backed by httpx.AsyncClient. await every call and stream as shown above.

Next steps​