Best Exa Alternatives For AI Agents

from decodo_exa import Exa  # was: from exa_py import Exa

exa = Exa()  # reads DECODO_AUTH_TOKEN
results = exa.search("tutorial first steps", num_results=5,
                     include_domains=["fastapi.tiangolo.com"], contents={"text": True})
for r in results.results:
    print(r.title, r.url, len(r.text or ""))

page = exa.get_contents(["https://books.toscrape.com/"], text=True).results[0]
print(page.url, len(page.text), "characters of markdown")

urls = exa.map("https://docs.docker.com/")  # A site's URLs in one call
print(len(urls), "URLs found by map()")

pip install httpx
export DECODO_AUTH_TOKEN="your-api-key"  # Windows PowerShell: $env:DECODO_AUTH_TOKEN="your-api-key"

"""An Exa-compatible client backed by Decodo's Web Data API.

Change one import, from `from exa_py import Exa` to `from decodo_exa import Exa`, and calls to search(),
get_contents(), and the agent tool helpers in exa.anthropic and exa.openai keep working. map() and crawl()
add site discovery and crawling.
"""
import json
import os
import re
import time
import warnings
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import date

import httpx

BLOCK = re.compile(r"(?i)captcha|access denied|verify you are human|enable javascript|unusual traffic")
LINK_TARGETS = re.compile(r"!\[[^\]]*\]\([^)]*\)|(?<=\])\([^)]*\)")  # Images and link URLs, which aren't readable text
H1 = re.compile(r"(?m)^# \S")
MIN_PAGE_CHARS = 2_000  # Below this, the site served a snippet, an error page, or a page that needs rendering
MAX_TOOL_CHARS = 10_000  # Page text per result sent to the model, the same default as exa_py's tools
RETRY_STATUS = {429, 500, 502, 503, 504}
SEARCH_SCHEMA = {"type": "object", "required": ["query"], "properties": {
    "query": {"type": "string", "description": "Natural language description of the pages to find."}}}
CONTENTS_SCHEMA = {"type": "object", "required": ["urls"], "properties": {
    "urls": {"type": "array", "items": {"type": "string"}, "description": "Absolute URLs of the pages to read."}}}


@dataclass
class Result:
    url: str
    id: str = ""
    title: str = ""
    text: str | None = None
    highlights: list[str] = field(default_factory=list)
    published_date: str | None = None


@dataclass
class ContentStatus:
    id: str
    status: str  # "success", or "error" when no complete page was returned


@dataclass
class SearchResponse:
    results: list[Result]
    statuses: list[ContentStatus] = field(default_factory=list)


def page_is_complete(text: str) -> bool:
    # Count readable text only, because a page shell full of menu links can be long and still have no content
    return not BLOCK.search(text[:5_000]) and len(LINK_TARGETS.sub("", text)) >= MIN_PAGE_CHARS


def main_content(text: str | None) -> str | None:
    # Many docs and article pages put the site menu before the H1 heading, so start the text at the first H1
    match = H1.search(text or "")
    return text[match.start():] if match else text


def _warn_ignored(options: dict) -> None:
    if options:
        warnings.warn(f"decodo_exa doesn't use these options: {', '.join(sorted(options))}", stacklevel=3)


def _get(obj, name: str, default=None):
    # Read a field from an SDK response object or from a plain dict
    return obj.get(name, default) if isinstance(obj, dict) else getattr(obj, name, default)


def _format(response: SearchResponse) -> str:
    blocks = []
    for r in response.results:
        if r.text:
            body = f"Text: {r.text[:MAX_TOOL_CHARS]}"
        else:
            body = "Highlights:\n" + "\n".join(r.highlights) if r.highlights else "Text: N/A"
        blocks.append(f"Title: {r.title or 'N/A'}\nURL: {r.url}\n{body}")
    return "\n\n---\n\n".join(blocks) or "No results found."


class AgentTools:
    """Tool definitions and tool-call handling for agent loops, with the method names of exa_py's helpers."""

    def __init__(self, exa: "Exa", style: str, handlers: dict | None = None):
        self.exa, self.style = exa, style
        self.handlers = {} if handlers is None else handlers

    @property
    def responses(self) -> "AgentTools":
        """Tools in the format of OpenAI's Responses API, as in exa.openai.responses."""
        return AgentTools(self.exa, "responses", self.handlers)

    def web_search(self, name: str = "web_search", num_results: int = 10, **_) -> dict:
        return self._tool(name, "Search the web for up-to-date, relevant information.", SEARCH_SCHEMA,
                          lambda args: self.exa.search(args["query"], num_results, contents={"highlights": True}))

    def get_contents(self, name: str = "get_contents", **_) -> dict:
        return self._tool(name, "Read the full contents of web pages you already have URLs for.", CONTENTS_SCHEMA,
                          lambda args: self.exa.get_contents(args["urls"]))

    def handle_tool_use(self, message) -> list[dict]:
        """Anthropic Messages API: answer each tool_use block with a tool_result block."""
        return [{"type": "tool_result", "tool_use_id": _get(b, "id"),
                 "content": self.run(_get(b, "name"), _get(b, "input"))}
                for b in _get(message, "content", []) if _get(b, "type") == "tool_use"]

    def handle_tool_calls(self, message) -> list[dict]:
        """OpenAI: answer each tool call, from a Chat Completions message or a Responses API response."""
        items = message if isinstance(message, list) else _get(message, "output")
        if items is not None:
            return [{"type": "function_call_output", "call_id": _get(i, "call_id"),
                     "output": self.run(_get(i, "name"), _get(i, "arguments"))}
                    for i in items if _get(i, "type") == "function_call"]
        calls = [(c, _get(c, "function")) for c in _get(message, "tool_calls") or []]
        return [{"role": "tool", "tool_call_id": _get(c, "id"),
                 "content": self.run(_get(f, "name"), _get(f, "arguments"))} for c, f in calls]

    def run(self, name: str, args) -> str:
        if name not in self.handlers:
            return f'Error: unknown tool "{name}"'
        try:
            return _format(self.handlers[name](json.loads(args) if isinstance(args, str) else args))
        except Exception as exc:  # Send the error to the model as text, so the agent loop continues
            return f"Error: {exc}"

    def _tool(self, name: str, description: str, schema: dict, handler) -> dict:
        self.handlers[name] = handler
        if self.style == "anthropic":
            return {"name": name, "description": description, "input_schema": schema}
        if self.style == "responses":
            return {"type": "function", "name": name, "description": description, "parameters": schema, "strict": False}
        return {"type": "function", "function": {"name": name, "description": description, "parameters": schema}}


class Exa:
    """Same method names as exa_py.Exa for search(), get_contents(), and the agent tool helpers."""

    def __init__(self, api_key: str | None = None, geo: str = "United States", timeout: float = 150.0):
        key = api_key or os.environ.get("DECODO_AUTH_TOKEN")
        if not key:
            raise RuntimeError("Set DECODO_AUTH_TOKEN to your Web Data API key. The free plan includes one: "
                               "https://decodo.com/web-data-api")
        self.geo = geo
        self.http = httpx.Client(base_url="https://data.decodo.com/v1", timeout=timeout,
                                 headers={"Authorization": f"Bearer {key}"})
        self.anthropic, self.openai = AgentTools(self, "anthropic"), AgentTools(self, "openai")

    def _post(self, path: str, body: dict, attempts: int = 3) -> dict:
        return self._request("POST", path, body, attempts)

    def _request(self, method: str, path: str, body: dict | None = None, attempts: int = 3) -> dict:
        for attempt in range(1, attempts + 1):
            try:
                resp = self.http.request(method, path, json=body)
                resp.raise_for_status()
                data = resp.json()
                if data.get("status") in ("failed", "faulted"):
                    raise RuntimeError(f"{path} failed: {data.get('message')}")
                return data
            except (httpx.TransportError, httpx.HTTPStatusError, RuntimeError) as exc:
                if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 401:
                    raise PermissionError("The Web Data API rejected the key. Check DECODO_AUTH_TOKEN.") from exc
                if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code not in RETRY_STATUS:
                    raise  # Don't retry a bad request or a bad key
                if attempt == attempts:
                    raise
                time.sleep(2 ** attempt)  # Wait 2 s, then 4 s

    def search(self, query: str, num_results: int = 10, include_domains: list[str] | None = None,
               exclude_domains: list[str] | None = None, start_published_date: str | None = None,
               end_published_date: str | None = None, contents: dict | None = None, **options) -> SearchResponse:
        _warn_ignored(options)
        if include_domains:
            query += " (" + " OR ".join(f"site:{d}" for d in include_domains) + ")"
        query += "".join(f" -site:{d}" for d in exclude_domains or [])
        body = {"target": "google_search", "query": query, "parse": True, "geo": self.geo, "google_nfpr": True}
        tbs = ["cdr:1"]  # Google's custom date range
        for edge, value in (("min", start_published_date), ("max", end_published_date)):
            if value:
                d = date.fromisoformat(value[:10])
                tbs.append(f"cd_{edge}:{d.month}/{d.day}/{d.year}")
        if len(tbs) > 1:
            body["google_tbs"] = ",".join(tbs)
        results, page = [], 1
        while len(results) < num_results and page <= 10:
            data = self._post("/scrape", {**body, "page_from": page})
            organic = data["results"][0]["content"]["results"]["results"].get("organic", [])
            if not organic:
                break
            results += [Result(url=o["url"], id=o["url"], title=o.get("title", ""),
                               highlights=[o.get("desc", "").removesuffix("Read more")]) for o in organic]
            page += 1
        results = results[:num_results]
        if contents is None or (contents and contents.get("text")):  # Like exa_py, omitting contents returns text
            for r, (page_text, _) in zip(results, self._fetch_many([r.url for r in results])):
                r.text = page_text
        return SearchResponse(results)

    def get_contents(self, urls: list[str] | str, text: bool = True, **options) -> SearchResponse:
        _warn_ignored(options)
        urls = [urls] if isinstance(urls, str) else urls
        pages = self._fetch_many(urls)
        return SearchResponse([Result(url=u, id=u, text=t) for u, (t, _) in zip(urls, pages)],
                              [ContentStatus(u, "success" if ok else "error") for u, (_, ok) in zip(urls, pages)])

    def map(self, url: str, **params) -> list[str]:
        """A site's URLs from its sitemap, in one call."""
        data = self._post("/map", {"url": url, **params})
        return [link["url"] for link in data["results"][0]["content"].get("links", [])]

    def crawl(self, url: str, limit: int = 100, poll_seconds: int = 5, timeout_seconds: int = 1800,
              **params) -> list[Result]:
        """Crawl a site and return each page as markdown. Add callback_url to params to get notified instead."""
        job = self._post("/crawl", {"url": url, "limit": limit, "output": "markdown", "JS_retry": True, **params})
        deadline = time.monotonic() + timeout_seconds
        while self._request("GET", f"/crawl/{job['crawl_id']}/status").get("status") != "done":
            if time.monotonic() > deadline:
                raise TimeoutError(f"crawl {job['crawl_id']} is still running after {timeout_seconds} s")
            time.sleep(poll_seconds)
        pages, url = [], f"/crawl/{job['crawl_id']}?type=content"
        while url:  # Results arrive in batches, each with a link to the next one
            batch = self._request("GET", url)
            pages += batch.get("url", [])
            url = batch.get("next")
        return [Result(url=p["url"], id=p["url"], text=main_content(p.get("content")))
                for p in pages if p.get("status") == "done"]

    def _fetch_many(self, urls: list[str]) -> list[tuple[str, bool]]:
        with ThreadPoolExecutor(5) as pool:
            return list(pool.map(self._fetch, urls))

    def _fetch(self, url: str) -> tuple[str, bool]:
        # Try the lowest-cost settings first, and add JavaScript rendering only for pages that need it
        steps = [{"proxy_pool": "standard"}, {"proxy_pool": "premium"}, {"proxy_pool": "premium", "headless": "html"}]
        text = ""
        for i, options in enumerate(steps):
            body = {"url": url, "markdown": True, "geo": self.geo, **options}
            try:
                data = self._post("/scrape", body, attempts=3 if i == len(steps) - 1 else 1)
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code != 400:
                    continue
                body.pop("geo")
                try:
                    data = self._post("/scrape", body, attempts=1)
                except (httpx.HTTPError, RuntimeError):
                    continue
            except (httpx.TransportError, RuntimeError):
                continue
            text = data["results"][0].get("content") or ""
            if page_is_complete(text):
                return main_content(text), True
        return main_content(text), False

"""Same calls as with exa_py, handled by Decodo's Web Data API."""
from decodo_exa import Exa  # was: from exa_py import Exa

exa = Exa()

results = exa.search("tutorial first steps", num_results=5,
                     include_domains=["fastapi.tiangolo.com"], contents={"text": True})
for r in results.results:
    print(r.title, r.url, len(r.text or ""), sep=" | ")

pages = exa.get_contents(["https://books.toscrape.com/", "https://www.walmart.com/search?q=laptop"], text=True)
for p in pages.results:
    print(p.url, len(p.text or ""), "characters")

urls = exa.map("https://docs.docker.com/")
print(len(urls), "URLs found by map()")
for page in exa.crawl("https://www.mkdocs.org/", limit=5):
    print(page.url, len(page.text or ""))

First Steps | https://fastapi.tiangolo.com/tutorial/first-steps/ | 15192
Tutorial - User Guide | https://fastapi.tiangolo.com/tutorial/ | 7661
Security - First Steps | https://fastapi.tiangolo.com/tutorial/security/first-steps/ | 10655
Dependencies | https://fastapi.tiangolo.com/tutorial/dependencies/ | 13389
Request Body | https://fastapi.tiangolo.com/tutorial/body/ | 8841
https://books.toscrape.com/ 7478 characters
https://www.walmart.com/search?q=laptop 35717 characters
2153 URLs found by map()
https://www.mkdocs.org/ 2270
https://www.mkdocs.org/getting-started 7232
https://www.mkdocs.org/about/contributing 8673
https://www.mkdocs.org/about/license 2165
https://www.mkdocs.org/about/release-notes 157598