""" title: Gemini Search author: sgpsonnet description: Search using Gemini API version: 0.1.2 """ import json import aiohttp from typing import Optional from pydantic import BaseModel, Field class Tools: def __init__(self): self.valves = self.Valves() class Valves(BaseModel): api_key: str = Field("", description="Gemini API Key") api_baseurl: str = Field( "https://generativelanguage.googleapis.com", description="Gemini API Base URL", ) async def search(self, query: str, __event_emitter__=None) -> str: """ Search using Gemini API :param query: Search query """ if not self.valves.api_key: return "API key not configured" try: # Emit searching status if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "Searching...", "done": False}, } ) # Make API request async with aiohttp.ClientSession() as session: async with session.post( f"{self.valves.api_baseurl}/v1beta/models/gemini-2.0-flash:generateContent", params={"key": self.valves.api_key}, headers={ "Content-Type": "application/json", "User-Agent": "Apifox/1.0.0 (https://apifox.com)", }, json={ "contents": [{"parts": [{"text": query}]}], "tools": [{"googleSearch": {}}], }, ) as response: result = await response.json() if "candidates" in result and len(result["candidates"]) > 0: text = result["candidates"][0]["content"]["parts"][0]["text"] return text return "No results found" except Exception as e: if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": f"Error: {str(e)}", "done": True}, } ) return f"Search failed: {str(e)}" finally: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Search complete", "done": True, "hidden": True, }, } )