Web Tools API

web

Pluggable web search + fetch tools for Orqest agents.

Autonomous agents that decide to investigate a question on the open web need two reliable primitives: a ranked search, and a content-aware page fetch. web_search and web_fetch provide both as async functions that agents call directly — no provider-specific SDKs in the calling code, no hard dependency on any single search vendor.

Provider selection is driven by env vars so ops can swap backends without touching agent code:

• ORQEST_WEB_PROVIDER — one of "tavily" (default), "exa", "brave", "serper", or "none" (disabled).
• ORQEST_WEB_API_KEY — the API key for the selected provider.

When ORQEST_WEB_API_KEY is unset or the provider is "none" the tools degrade gracefully: web_search returns an empty result list with a disabled_reason note, web_fetch still works for public URLs (no key required for a plain GET). This lets investigation tools stay functional in offline / CI settings without tripping on missing credentials.

WebSearchResult dataclass

WebSearchResult(
    title: str,
    url: str,
    snippet: str,
    score: float | None = None,
)

One hit returned by :func:web_search.

WebSearchResponse dataclass

WebSearchResponse(
    query: str,
    results: list[WebSearchResult] = list(),
    provider: str = "none",
    disabled_reason: str | None = None,
)

Full response envelope returned by :func:web_search.

WebFetchResult dataclass

WebFetchResult(
    url: str,
    status_code: int,
    content_type: str,
    text: str,
    truncated: bool,
)

Content fetched by :func:web_fetch.

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.

Source code in orqest/tools/web.py

async def 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.
    """
    async with httpx.AsyncClient(timeout=timeout_s, follow_redirects=True) as client:
        response = await client.get(url, headers={"User-Agent": "orqest-web-fetch/1.0"})

    text = response.text
    truncated = bool(max_chars > 0 and len(text) > max_chars)
    if truncated:
        text = text[:max_chars]

    return WebFetchResult(
        url=str(response.url),
        status_code=response.status_code,
        content_type=response.headers.get("content-type", ""),
        text=text,
        truncated=truncated,
    )