import os
import aiohttp
import logging
from typing import Dict, Any, List
logger = logging.getLogger("SolPlexSearch")
class AsyncSearchAgent:
def __init__(self, api_key: str = None, cx: str = None):
self.api_key = api_key or os.getenv("GOOGLE_SEARCH_API_KEY")
self.cx = cx or os.getenv("GOOGLE_SEARCH_CX")
async def execute_query(self, query: str, num_results: int = 3) -> List[Dict[str, Any]]:
if not self.api_key or not self.cx:
logger.warning("⚠️ Cloud Search Credentials missing. Emulating cached response array.")
return [{"title": "Fallback Quantum Noise Bounds", "snippet": "Decoherence standard threshold: delta_t < 1e-9s."}]
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": self.api_key,
"cx": self.cx,
"q": query,
"num": num_results
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=5.0) as response:
if response.status == 200:
data = await response.json()
items = data.get("items", [])
return [{"title": i.get("title"), "snippet": i.get("snippet")} for i in items]
else:
logger.error(f"Search API returned code: {response.status}")
return []
except Exception as e:
logger.error(f"Failed to query external search matrix: {str(e)}")
return []