Python

pip install httpx

The client

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

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


class QuotaExhausted(RuntimeError):
    def __init__(self, reset_seconds: int):
        super().__init__(f"monthly quota exhausted; resets in {reset_seconds}s")
        self.reset_seconds = reset_seconds


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:
                # Two conditions share this status. The quota one is not
                # transient, and retrying just burns the rest of the day.
                if response.headers.get("x-quota-remaining") == "0":
                    raise QuotaExhausted(int(response.headers.get("retry-after", 0)))
                # retry-after is 1: the rate bucket refills continuously, so a
                # second is enough. 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

            response.raise_for_status()
            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 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")

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

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, 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})
            # One content block; its text is the same JSON /stories returns.
            print(result.content[0].text)

asyncio.run(main())

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