src/tools/workflows/web_search.py

#
# SPDX-FileCopyrightText: Hadad <[email protected]>
# SPDX-License-Identifier: Apache-2.0
#

import aiohttp
from config import (
    SEARXNG,
    FORMAT,
    TIMEOUT,
    AIOHTTP,
    HEADERS,
    REMINDERS
)

async def web_search(query):
    try:
        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(**AIOHTTP),
            timeout=aiohttp.ClientTimeout(total=TIMEOUT),
            headers=HEADERS
        ) as session:
            async with session.get(f"{SEARXNG}?q={query}&format={FORMAT}") as response:
                response.raise_for_status()
                content = await response.text()
        return content + "\n\n\n" + REMINDERS
    except Exception as error:
        return f"Error during web search: {str(error)}"