unlob Docs
Console
Browse documentation

Python

A small client with the retry policy the API actually needs.

There is no unlob package to install. The API is HTTP and JSON, so a client is one file.

pip install httpx

The client

import os, random, time
import httpx

class Unlob:
    """A client for the unlob search API.

    Small on purpose. The only parts worth writing are the retry policy — which
    has one non-obvious rule — and surfacing `partial`, which is a 200 that means
    the answer is incomplete.
    """

    RETRYABLE = {408, 500, 502, 503, 504}

    def __init__(self, api_key: str | None = None, *, timeout: float = 20.0):
        self._c = httpx.Client(
            base_url="https://api.unlob.com",
            headers={"x-api-key": api_key or os.environ["UNLOB_API_KEY"]},
            timeout=timeout,
        )

    def _get(self, path: str, params: dict, attempts: int = 4) -> dict:
        for attempt in range(attempts):
            r = self._c.get(path, params={k: v for k, v in params.items() if v is not None})

            if r.status_code == 429:
                after = r.headers.get("retry-after")
                if after is None:
                    # THE non-obvious rule: a 429 without retry-after is a
                    # hard-capped monthly quota, not a rate limit. Retrying it is a
                    # loop that cannot succeed.
                    raise QuotaExhausted(r.text)
                time.sleep(int(after))
                continue

            if r.status_code in self.RETRYABLE and attempt < attempts - 1:
                # Full jitter: without it, every client that failed together
                # retries together and re-creates the spike.
                time.sleep(random.uniform(0, 2 ** attempt))
                continue

            r.raise_for_status()
            return r.json()
        raise RuntimeError("exhausted retries")

    def search(self, q: str, **filters) -> dict:
        """Search. Filters are the query parameters, verbatim — see the reference.

        IN-lists take a comma-separated string: tld="gov,edu".
        """
        return self._get("/search", {"q": q, **filters})

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

    def corroborate(self, passage_id: str, limit: int = 10) -> dict:
        return self._get("/corroborate", {"id": passage_id, "limit": limit})

    def dossier(self, entity: str, limit: int = 10) -> dict:
        return self._get("/dossier", {"entity": entity, "limit": limit})

    def context(self, question: str, budget: int = 4000) -> dict:
        return self._get("/assemble_context", {"q": question, "budget": budget})

    def why_not(self, url: str) -> dict:
        return self._get("/why_not", {"url": url})

class QuotaExhausted(RuntimeError):
    """The monthly quota is spent and this key hard-caps. Not retryable."""

Using it

unlob = Unlob()

body = unlob.search("how does tokio schedule tasks", vertical="code", limit=5)

if body.get("partial"):
    # A 200 that means the answer is incomplete: part of the corpus was
    # unreachable. Short here is a symptom, not a finding.
    raise RuntimeError("incomplete results — retry")

for hit in body["results"]:
    print(hit["title"], hit["url"])
    print(hit["snippet"])          # the passage text — usually the answer
    print(hit.get("independent_sources", 0), "independent sources\n")

Filters

Every query parameter, verbatim. IN-lists are comma-separated strings, because a URL query string cannot carry an array:

unlob.search(
    "vitamin d dosage",
    tld="gov,edu",
    min_host_rank=0.6,
    min_independent_sources=2,
    collapse="story",
    published_from=1704067200,
    sort="centrality",
    limit=10,
)

Async

Same shape with httpx.AsyncClient. Worth it if you are fanning out across entities:

import asyncio, httpx

async def dossiers(entities: list[str], key: str) -> list[dict]:
    async with httpx.AsyncClient(
        base_url="https://api.unlob.com",
        headers={"x-api-key": key},
        timeout=20,
    ) as c:
        rs = await asyncio.gather(*(c.get("/dossier", params={"entity": e}) for e in entities))
        return [r.json() for r in rs]

Mind the per-minute rate limit when you do — x-ratelimit-remaining is on every response.

Reading the capability catalog

Rather than hard-coding the filter list:

import httpx
caps = httpx.get("https://api.unlob.com/describe").json()   # no key needed
print({name for name, _ in caps["filters"]})
print(caps["verticals"])                                     # what this deployment holds

Next