Python

pip install httpx

The client

import os, time
from typing import Any, Iterator
import httpx

BASE = "https://api.unzoi.com"


class OutOfCredits(RuntimeError):
    def __init__(self, reset_seconds: int):
        super().__init__(f"out of monthly credits; resets in {reset_seconds}s")
        self.reset_seconds = reset_seconds


class ApiError(RuntimeError):
    def __init__(self, status: int, code: str, detail: str):
        super().__init__(f"{status} {code}: {detail}")
        self.status, self.code = status, code


class Unzoi:
    def __init__(self, key: str | None = None, timeout: float = 30.0):
        self._client = httpx.Client(
            base_url=BASE,
            headers={"x-api-key": key or os.environ["UNZOI_KEY"]},
            timeout=timeout,
        )

    def _get(self, path: str, **params: Any) -> dict:
        params = {k: v for k, v in params.items() if v is not None}
        for attempt in range(6):
            response = self._client.get(path, params=params)

            if response.status_code == 429:
                # Three conditions share this status, and the body's code says
                # which. Out of credits for the month is not transient.
                if response.json().get("code") == "insufficient_credits":
                    raise OutOfCredits(int(response.headers.get("retry-after", 0)))
                # retry-after is when this call's credits refill, usually a
                # second or two. Exponential backoff here wastes allowance you
                # already paid for.
                time.sleep(float(response.headers.get("retry-after", 1)))
                continue

            # No shard answered. Deliberately not reported as zero results —
            # a narrower time range makes it likelier to succeed.
            if response.status_code == 503 and attempt < 3:
                time.sleep(2**attempt)
                continue

            if not response.is_success:
                # Every error body is JSON with a stable code; surface it
                # rather than the status alone.
                body = response.json()
                raise ApiError(response.status_code, body["code"], body["detail"])
            return response.json()
        raise RuntimeError("giving up after repeated rate limiting")

    def search(self, **params: Any) -> dict:
        return self._get("/search", **params)

    def stories(self, **params: Any) -> dict:
        return self._get("/stories", **params)

    def headlines(self, **params: Any) -> dict:
        return self._get("/top-headlines", **params)

    def story(self, story_id: str, limit: int = 20) -> dict:
        return self._get(f"/stories/{story_id}", limit=limit)

    def entities(self, q: str, type_: str, **params: Any) -> dict:
        return self._get("/entities", q=q, type=type_, **params)

    def article(self, article_id: str) -> dict:
        return self._get(f"/doc/{article_id}")

    def related(self, article_id: str, limit: int = 10) -> dict:
        return self._get(f"/similar/{article_id}", limit=limit)

    def account(self) -> dict:
        return self._get("/account")

    def walk(self, **params: Any) -> Iterator[dict]:
        """Every result, 100 at a time — each page is a billed request."""
        offset = 0
        while True:
            page = self._get("/search", offset=offset, limit=100, **params)
            yield from page["results"]
            if not page["has_more"]:
                return
            offset += 100

Using it

unzoi = Unzoi()

result = unzoi.stories(q="semiconductor export controls", from_="2026-08-01", limit=10)

# A plan boundary and a corpus with no coverage look identical without this.
if result["from_clamped"]:
    print(f"window narrowed to the last {result['history_days']} days by your plan")

# So do a degraded index and a quiet news day.
if result["partial"] or result["coverage"] != "complete":
    print(f"incomplete answer ({result['coverage']}); do not treat a short list as the answer")

for story in result["stories"]:
    print(f"{story['count']:>3} articles / {story['outlets']:>2} outlets  {story['title']}")

# One call for the whole event, no re-search.
detail = unzoi.story(result["stories"][0]["story_id"], limit=5)
print(detail["first_seen"], detail["last_seen"], [a["source"] for a in detail["articles"]])

Async

import asyncio, httpx

class AsyncUnzoi:
    def __init__(self, key: str | None = None):
        self._client = httpx.AsyncClient(
            base_url=BASE,
            headers={"x-api-key": key or os.environ["UNZOI_KEY"]},
            timeout=30.0,
        )

    async def stories(self, **params) -> dict:
        response = await self._client.get("/stories", params=params)
        response.raise_for_status()
        return response.json()

    async def aclose(self) -> None:
        await self._client.aclose()


async def main():
    unzoi = AsyncUnzoi()
    try:
        # Keep concurrency small. Twenty parallel requests hit the per-minute
        # cap in the first second and spend the rest of it retrying.
        limiter = asyncio.Semaphore(4)

        async def one(topic: str):
            async with limiter:
                return topic, await unzoi.stories(q=topic, limit=5)

        for topic, result in await asyncio.gather(
            *(one(t) for t in ["lithium supply", "port congestion", "grid outage"])
        ):
            print(topic, result["total"])
    finally:
        await unzoi.aclose()

asyncio.run(main())

Calling MCP instead

# pip install mcp
import os, json, asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    headers = {"x-api-key": os.environ["UNZOI_KEY"]}
    async with streamablehttp_client("https://api.unzoi.com/mcp", headers=headers) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("list_stories", {"q": "port congestion", "limit": 5})
            # structuredContent is the typed payload, the same JSON /stories
            # returns, in the shape the tool's outputSchema declares; the text
            # block holds the same JSON for older clients. Read one, never both.
            body = result.structuredContent or json.loads(result.content[0].text)
            print(body["total"], "stories")

asyncio.run(main())

For an agent rather than a script, LangChain and LlamaIndex both turn these into native tools in a few lines.

Connection lifecycle

Neither client above closes its httpx connection explicitly in the short examples, and that is fine for a script that runs once and exits — the process teardown reclaims the socket. It stops being fine the moment Unzoi is constructed inside a request handler or a loop: each instance opens its own connection pool, and a new one per call defeats the keep-alive httpx already gives you for free. Construct one Unzoi per process — a module-level singleton or a value threaded through your app's context — and reuse it. For the sync client, wrap it as a context manager (httpx.Client already supports __enter__/__exit__; delegate to it) so a short-lived script still cleans up on an exception rather than leaking a socket until garbage collection gets around to it.

Choosing sync or async

The sync client is the right default for anything that makes one Unzoi call and does something else with the result — a script, a Celery task, a Django view outside an async path. Reach for AsyncUnzoi only when you are already inside an event loop and specifically want to overlap several requests, as the concurrent example above does for three topics at once. Mixing them — calling the sync client from inside async def — blocks the whole event loop for the duration of the HTTP round trip, which is worse than not being async at all, because every other coroutine scheduled on that loop stalls with it rather than merely your own code waiting.

Typing the response beyond dict

Both clients return a plain dict from response.json(), which is honest about what httpx actually gives you but loses autocomplete and lets a renamed field fail silently at the call site instead of at import time. A TypedDict per response shape — SearchResponse, StoriesResponse, matching the field names in the REST reference — costs nothing at runtime and turns a typo like result["totall"] into a type-checker error under mypy or pyright, rather than a KeyError the first time that code path actually runs in production.

Cancelling a slow call without cancelling everything

asyncio.timeout() (3.11+) scopes a deadline to one call rather than to the whole client, which matters once AsyncUnzoi is shared across a process instead of constructed per-request: a single slow response should not force every other in-flight call to inherit the same budget.

async def stories_or_none(unzoi: AsyncUnzoi, **params) -> dict | None:
    try:
        async with asyncio.timeout(5):
            return await unzoi.stories(**params)
    except TimeoutError:
        return None  # the caller decides whether "no answer yet" is retryable

This composes with the semaphore in the concurrent example above rather than replacing it — the semaphore bounds how many requests run at once, and the timeout bounds how long any one of them is allowed to hold its slot. Without the timeout, one hung connection can occupy a semaphore permit indefinitely and quietly reduce your effective concurrency to whatever is left, which looks like the API slowing down when it is really your own pool starving itself.

Size the semaphore against the same per-minute rate the retry loop already respects — a limiter set higher than the plan's rate_per_min just relocates the wait into _get's own 429 handling instead of avoiding it.