Skip to content

trading_agent.tools.web_search_tool

The web_search tool: searches the web, or recent news, through the local SearXNG server.

SearXNG asks several search engines at once and merges their answers, so one engine failing does not stop a search. Results are labelled as data from the web, because a page's title or snippet may contain text written to mislead an AI model.

Typical usage example:

tool = WebSearchTool('http://127.0.0.1:4001', result_limit=8)

WebSearchArguments

Bases: BaseModel

The arguments of web_search.

Attributes:

Name Type Description
query str

What to search for.
category Literal['general', 'news']

"news" for news articles, "general" for the whole web.
time_range Literal['day', 'week', 'month', 'year'] | None

Only results from the last day, week, month or year, or None for any time.

WebSearchTool 

WebSearchTool(searxng_base_url, result_limit, transport=None)

Bases: Tool

Searches through SearXNG.

Creates the tool.

Parameters:

Name Type Description Default
searxng_base_url str

The SearXNG server's address.

 required
result_limit int

The most results to return.

 required
transport AsyncBaseTransport | None

What sends the requests; a test passes a fake one, and None means the network.

 None
Source code in src/trading_agent/tools/web_search_tool.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    searxng_base_url: str,
    result_limit: int,
    transport: httpx.AsyncBaseTransport | None = None,
):
    """Creates the tool.

    Args:
        searxng_base_url (str): The SearXNG server's address.
        result_limit (int): The most results to return.
        transport (httpx.AsyncBaseTransport | None): What sends the requests; a test passes a fake one, and None means the network.
    """
    self._searxng_base_url = searxng_base_url.rstrip('/')
    self._result_limit = result_limit
    self._transport = transport

run async 

run(arguments)

Runs the search, adding unfiltered results when the time range finds too few.

A time range makes SearXNG skip every engine that cannot filter by date, so a filtered news search can come back nearly empty while an unfiltered one finds plenty. The extra results are shown newest first, after a notice, so the model checks the dates itself.

Parameters:

Name Type Description Default
arguments WebSearchArguments

The query and filters.

 required

Returns:

Type Description
ToolResult

One numbered entry per result, or an error when SearXNG cannot be reached.
Source code in src/trading_agent/tools/web_search_tool.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def run(self, arguments: WebSearchArguments) -> tool.ToolResult:
    """Runs the search, adding unfiltered results when the time range finds too few.

    A time range makes SearXNG skip every engine that cannot filter by date, so a filtered news search can come back nearly empty while an unfiltered one finds plenty. The extra results are shown newest first, after a notice, so the model checks the dates itself.

    Args:
        arguments (WebSearchArguments): The query and filters.

    Returns:
        tool.ToolResult: One numbered entry per result, or an error when SearXNG cannot be reached.
    """
    try:
        results = await self._search(
            arguments.query,
            arguments.category,
            arguments.time_range,
        )
        notice = ''
        if len(results) < _FEW_RESULTS and arguments.time_range is not None:
            notice = f'[Only {len(results)} result(s) were found within the last {arguments.time_range}, because most engines cannot filter by date. Results from any time follow them, newest first; check each date.]'
            unfiltered = await self._search(
                arguments.query,
                arguments.category,
                None,
            )
            results = self._merged(results, unfiltered)
    except httpx.HTTPError as error:
        return tool.ToolResult.error(
            f'The search server at {self._searxng_base_url} did not answer ({error!r}). It runs in Docker; the user can start it with: docker compose up -d'
        )
    results = results[: self._result_limit]
    if not results:
        return tool.ToolResult.success(
            f'No results for {arguments.query!r}.'
        )
    lines = []
    if notice:
        lines.append(notice)
    for position, result in enumerate(results, start=1):
        published = self._published_text(result)[:10] or 'date unknown'
        snippet = (result.get('content') or '').strip()[
            :_SNIPPET_CHARACTERS
        ]
        lines.append(f'{position}. {result.get("title", "(no title)")}')
        lines.append(f'   {published} · {result.get("url", "")}')
        if snippet:
            lines.append(f'   {snippet}')
    return tool.ToolResult.success(
        untrusted_text.UntrustedText().wrap(
            f'a web search for {arguments.query!r}',
            '\n'.join(lines),
        )
    )