Web tools — pluggable search + fetch for autonomous agents¶

Autonomous Orqest agents that decide to investigate a question on the open web need two reliable primitives: a ranked search, and a content-aware page fetch. orqest.tools.web provides both as async functions. No provider-specific SDK is imported in the calling code, and the provider is selectable via environment variables so ops can swap backends without touching agent logic.

Providers¶

Provider	Env value (`ORQEST_WEB_PROVIDER`)	API key env	Best for
Tavily	`tavily` (default)	`ORQEST_WEB_API_KEY`	General research; default
Exa	`exa`	same	Long-form / neural search
Brave Search	`brave`	same	Broad web coverage
Serper (Google)	`serper`	same	Google ranking
Disabled	`none`	—	CI / offline

All providers are called through httpx — no SDK dependencies. Unknown providers degrade gracefully: web_search returns an empty response with a disabled_reason explaining why.

Minimal example¶

import asyncio

from orqest.tools.web import web_fetch, web_search


async def main() -> None:
    # Search (provider + key from env)
    resp = await web_search("k-omega SST turbulence model choice Re=1M")
    for hit in resp.results:
        print(hit.title, hit.url)

    # Fetch the top hit for deeper context
    if resp.results:
        page = await web_fetch(resp.results[0].url, max_chars=4000)
        print(page.text)


asyncio.run(main())

Graceful degradation¶

Missing ORQEST_WEB_API_KEY? web_search returns:

WebSearchResponse(
    query="...",
    results=[],
    provider="tavily",
    disabled_reason="ORQEST_WEB_API_KEY not set",
)

This matters for autonomous investigation tools: a _investigate compound tool wraps web_search but must not fail when the researcher can't reach the web. The caller sees an empty results list with a clear reason and can fall back to memory recall or the agent's own knowledge.

web_fetch does not require an API key — it's a plain follow_redirects=True GET with a User-Agent: orqest-web-fetch/1.0 header. If a page is over max_chars the body is truncated and result.truncated = True.

Reference¶

web_search `async` ¶

web_search(
    query: str,
    *,
    k: int = DEFAULT_RESULTS,
    provider: str | None = None,
    api_key: str | None = None,
    timeout_s: float = DEFAULT_TIMEOUT_S,
) -> WebSearchResponse

Search the web and return up to k ranked hits.

Selects the provider from the provider argument, then from ORQEST_WEB_PROVIDER, then defaults to "tavily". Degrades gracefully when no API key is available — returns an empty :class:WebSearchResponse with disabled_reason explaining why.

Parameters:

Name	Type	Description	Default
`query`	`str`	Free-text search query.	required
`k`	`int`	Maximum number of results.	`DEFAULT_RESULTS`
`provider`	`str \| None`	Override of the env-selected provider.	`None`
`api_key`	`str \| None`	Override of the env-supplied API key.	`None`
`timeout_s`	`float`	HTTP timeout in seconds.	`DEFAULT_TIMEOUT_S`

Source code in orqest/tools/web.py

async def web_search(
    query: str,
    *,
    k: int = DEFAULT_RESULTS,
    provider: str | None = None,
    api_key: str | None = None,
    timeout_s: float = DEFAULT_TIMEOUT_S,
) -> WebSearchResponse:
    """Search the web and return up to *k* ranked hits.

    Selects the provider from the ``provider`` argument, then from
    ``ORQEST_WEB_PROVIDER``, then defaults to ``"tavily"``. Degrades
    gracefully when no API key is available — returns an empty
    :class:`WebSearchResponse` with ``disabled_reason`` explaining why.

    Args:
        query: Free-text search query.
        k: Maximum number of results.
        provider: Override of the env-selected provider.
        api_key: Override of the env-supplied API key.
        timeout_s: HTTP timeout in seconds.
    """
    resolved_provider = (provider or os.getenv(_PROVIDER_ENV, "tavily")).lower()
    resolved_key = api_key or os.getenv(_API_KEY_ENV)

    if resolved_provider == "none":
        return WebSearchResponse(
            query=query, provider="none", disabled_reason="provider disabled",
        )
    if not resolved_key:
        return WebSearchResponse(
            query=query,
            provider=resolved_provider,
            disabled_reason=f"{_API_KEY_ENV} not set",
        )

    async with httpx.AsyncClient(timeout=timeout_s) as client:
        if resolved_provider == "tavily":
            return await _search_tavily(client, query, k, resolved_key)
        if resolved_provider == "exa":
            return await _search_exa(client, query, k, resolved_key)
        if resolved_provider == "brave":
            return await _search_brave(client, query, k, resolved_key)
        if resolved_provider == "serper":
            return await _search_serper(client, query, k, resolved_key)

    return WebSearchResponse(
        query=query,
        provider=resolved_provider,
        disabled_reason=f"unknown provider '{resolved_provider}'",
    )

web_fetch `async` ¶

web_fetch(
    url: str,
    *,
    max_chars: int = MAX_FETCH_CHARS,
    timeout_s: float = DEFAULT_TIMEOUT_S,
) -> WebFetchResult

Fetch url and return its (truncated) body as text.

No API key needed — this is a plain GET. The body is truncated to max_chars to keep LLM context bounded; consumers that need the full body can pass max_chars=-1 or zero to disable truncation.