cuery.search#

Async helpers for grounded / live web search via OpenAI, Google Gemini and xAI Grok.

Provides a convenience search function that executes multiple web searches concurrently (rate limited via a semaphore) and preserves input order. Logic mirrors cuery.call.gather_calls() but is simplified for plain string prompts.

Currently supported providers (client argument):

  • "openai" – Uses the OpenAI Responses API with the web_search_preview tool.

  • "gemini" – Uses the Google Gemini (google-genai) API with Google Search grounding

    (google_search for Gemini 2.x, google_search_retrieval for 1.5 models).

Both return a SearchResult with extracted plain text plus a list of Reference objects (title + URL) when available. If parsing fails a ValueError is raised (no silent fallbacks here – upstream caller can decide how to handle).

Attributes#

OAIResponse

VALID_MODELS

__md

LLMS

FORMATTED_SEARCH_PROMPT

SUPPORT_COUNTRY

Classes#

Source

Single search reference with title and URL.

SearchResult

Search result with extracted text and references.

Functions#

unmark_element(element[, stream])

unmark(text)

Convert Markdown text to plain text, stripping all formatting.

resolve_redirect(redirect_url[, timeout])

Resolve a URL redirect to its final destination URL.

validate_openai(response[, plain])

Convert a raw web search response into a SearchResult instance.

validate_gemini(response[, plain])

Convert a Gemini grounded search raw response into SearchResult.

validate_xai(response[, plain])

Convert an xAI Grok Live Search chat completion into SearchResult.

query_openai(prompt[, country, city, context_size, ...])

Call OpenAI Responses API with Web Search enabled (async).

coords(country)

query_gemini(prompt[, model, country, language, ...])

Call Gemini with Google Search grounding (supports Gemini > 2.0 only).

query_xai(prompt[, model, timeout, mode, ...])

Call xAI Grok with Live Search (async).

query_deepseek(prompt[, model, validate, use_search])

DeepSeek doesn't support search yet, so we simply execute the prompt as-is.

validate_model(client, model)

Validate if the given client and model are supported.

search_with_format(prompt, model, response_format, **kwds)

Perform a web search using the specified model and return structured response.

gather(prompts[, model, use_search, country, ...])

Simplified gather mirroring hasdata.gather using all_with_policies.

Module Contents#

cuery.search.OAIResponse#
cuery.search.VALID_MODELS#
cuery.search.unmark_element(element, stream=None)#
cuery.search.__md#
cuery.search.unmark(text)#

Convert Markdown text to plain text, stripping all formatting.

class cuery.search.Source(/, **data)#

Bases: cuery.response.Response

Single search reference with title and URL.

Parameters:

data (Any)

title: str#
url: str#
property domain: str | None#

Extract domain from URL.

Return type:

str | None

class cuery.search.SearchResult(/, **data)#

Bases: cuery.response.Response

Search result with extracted text and references.

Parameters:

data (Any)

answer: str#
sources: list[Source]#
cuery.search.resolve_redirect(redirect_url, timeout=2)#

Resolve a URL redirect to its final destination URL.

Gemini grounding chunk URIs are things like: https://vertexaisearch.cloud.google.com/grounding-api-redirect/AUZIYQFXCboQq0LRg6hUOP…

Parameters:

  • redirect_url (str)

  • timeout (int)

Return type:

str

cuery.search.validate_openai(response, plain=False)#

Convert a raw web search response into a SearchResult instance.

Parameters:

plain (bool)

Return type:

SearchResult

cuery.search.validate_gemini(response, plain=False)#

Convert a Gemini grounded search raw response into SearchResult.

Parameters:

  • response (pydantic.BaseModel)

  • plain (bool)

Return type:

SearchResult

cuery.search.validate_xai(response, plain=False)#

Convert an xAI Grok Live Search chat completion into SearchResult.

Citations (if return_citations enabled – default true) are exposed as a list of URL strings on response.citations. We map each into a Reference using the URL as both title and URL (title data not currently provided).

Parameters:

plain (bool)

Return type:

SearchResult

async cuery.search.query_openai(prompt, country=None, city=None, context_size='medium', reasoning_effort='low', model='gpt-4.1-mini', use_search=True, validate=True, response_format=None)#

Call OpenAI Responses API with Web Search enabled (async).

API Docs: - https://platform.openai.com/docs/guides/tools-web-search?api-mode=responses

Parameters:

  • prompt (str)

  • country (str | None)

  • city (str | None)

  • context_size (Literal['low', 'medium', 'high'] | str)

  • reasoning_effort (Literal['low', 'medium', 'high'] | str)

  • model (str)

  • use_search (bool)

  • validate (bool)

  • response_format (cuery.response.Response | None)

Return type:

SearchResult | OAIResponse

cuery.search.coords(country)#
Parameters:

country (str)

Return type:

google.genai.types.LatLng | None

async cuery.search.query_gemini(prompt, model='gemini-2.0-flash', country=None, language=None, thinking_budget=-1, use_search=True, validate=True)#

Call Gemini with Google Search grounding (supports Gemini > 2.0 only).

API Docs: - https://ai.google.dev/gemini-api/docs/google-search - https://ai.google.dev/gemini-api/docs/thinking

Parameters:

  • prompt (str)

  • model (str)

  • country (str | None)

  • language (str | None)

  • thinking_budget (int)

  • use_search (bool)

  • validate (bool)

Return type:

SearchResult | google.genai.types.GenerateContentResponse

async cuery.search.query_xai(prompt, model='grok-4', timeout=None, mode='on', max_search_results=15, sources=None, country=None, use_search=True, validate=True)#

Call xAI Grok with Live Search (async).

API Docs: - https://docs.x.ai/docs/guides/live-search

Parameters:

  • prompt (str)

  • model (str)

  • timeout (int | None)

  • mode (Literal['auto', 'on', 'off'])

  • max_search_results (int)

  • sources (list[Literal['web', 'news', 'x']] | None)

  • country (str | None)

  • use_search (bool)

  • validate (bool)

Return type:

SearchResult | xai_sdk.chat.Response

async cuery.search.query_deepseek(prompt, model='deepseek-chat', validate=True, use_search=True, **kwds)#

DeepSeek doesn’t support search yet, so we simply execute the prompt as-is.

Parameters:

  • prompt (str)

  • model (str)

  • validate (bool)

  • use_search (bool)

Return type:

Any

cuery.search.LLMS#
cuery.search.validate_model(client, model)#

Validate if the given client and model are supported.

Parameters:

  • client (str)

  • model (str)

Return type:

None

cuery.search.FORMATTED_SEARCH_PROMPT = ''#
async cuery.search.search_with_format(prompt, model, response_format, **kwds)#

Perform a web search using the specified model and return structured response.

Parameters:
Return type:

cuery.response.Response

cuery.search.SUPPORT_COUNTRY = ['openai', 'xai', 'google']#
async cuery.search.gather(prompts, model='openai/gpt-4.1-mini', use_search=True, country=None, validate=True, policies=None, execute=True, **kwds)#

Simplified gather mirroring hasdata.gather using all_with_policies.

Creates one coroutine per prompt for the selected model/provider with optional policies (timeout, retries, semaphore, fallback, progress). When execute is False the wrapped coroutine objects are returned for the caller to schedule.

Parameters:

  • prompts (str | list[str])

  • model (str)

  • use_search (bool)

  • country (str | None)

  • validate (bool)

  • policies (dict[str, Any] | None)

  • execute (bool)

Return type:

list[collections.abc.Coroutine] | collections.abc.Coroutine | cuery.response.ResponseSet | SearchResult