deep_search
Perform advanced web research using Google Gemini models and Google Search to investigate queries with citation-rich answers and adjustable effort levels.
Instructions
Perform a deep search on a given query using an advanced web research agent.
Args: query: The research question or topic to investigate. effort: The amount of effect for the research, low, medium or hight (default: low).
Returns: A dictionary containing the file path to a JSON file with the answer and sources.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | Search query string | |
| effort | No | Search effort | low |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/gemini_deepsearch_mcp/app.py:15-80 (handler)Asynchronous handler for the 'deep_search' MCP tool. Registers the tool with FastMCP, handles input validation via Annotated types, configures the research graph based on effort level, invokes it asynchronously, and returns answer and sources.
@mcp.tool() async def deep_search( query: Annotated[str, Field(description="Search query string")], effort: Annotated[ Literal["low", "medium", "high"], Field(description="Search effort") ] = "low", ) -> dict: """Perform a deep search on a given query using an advanced web research agent. Args: query: The research question or topic to investigate. effort: The amount of effect for the research, low, medium or hight (default: low). Returns: A dictionary containing the answer to the query and a list of sources used. """ # Set search query count, research loops and reasoning model based on effort level if effort == "low": initial_search_query_count = 1 max_research_loops = 1 reasoning_model = "gemini-2.5-flash" elif effort == "medium": initial_search_query_count = 3 max_research_loops = 2 reasoning_model = "gemini-2.5-flash" else: # high effort initial_search_query_count = 5 max_research_loops = 3 reasoning_model = "gemini-2.5-pro" # Prepare the input state with the user's query input_state = { "messages": [HumanMessage(content=query)], "search_query": [], "web_research_result": [], "sources_gathered": [], "initial_search_query_count": initial_search_query_count, "max_research_loops": max_research_loops, "reasoning_model": reasoning_model, } query_generator_model: str = "gemini-2.5-flash" web_search_model: str = "gemini-2.5-flash-lite-preview-06-17" reflection_model: str = "gemini-2.5-flash" answer_model: str = "gemini-2.5-pro" # Configuration for the agent config = { "configurable": { "query_generator_model": query_generator_model, "web_search_model": web_search_model, "reflection_model": reflection_model, "answer_model": answer_model, } } # Run the agent graph to process the query in a separate thread to avoid blocking result = await asyncio.to_thread(graph.invoke, input_state, config) # Extract the final answer and sources from the result answer = ( result["messages"][-1].content if result["messages"] else "No answer generated." ) sources = result["sources_gathered"] return {"answer": answer, "sources": sources} - src/gemini_deepsearch_mcp/main.py:20-93 (handler)Synchronous handler for the 'deep_search' MCP tool (stdio server). Similar to app.py version but synchronous, writes result to a temporary JSON file, and returns the file path.
@mcp.tool() def deep_search( query: Annotated[str, Field(description="Search query string")], effort: Annotated[ Literal["low", "medium", "high"], Field(description="Search effort") ] = "low", ) -> dict: """Perform a deep search on a given query using an advanced web research agent. Args: query: The research question or topic to investigate. effort: The amount of effect for the research, low, medium or hight (default: low). Returns: A dictionary containing the file path to a JSON file with the answer and sources. """ # Set search query count, research loops and reasoning model based on effort level if effort == "low": initial_search_query_count = 1 max_research_loops = 1 reasoning_model = "gemini-2.5-flash-preview-05-20" elif effort == "medium": initial_search_query_count = 3 max_research_loops = 2 reasoning_model = "gemini-2.5-flash-preview-05-20" else: # high effort initial_search_query_count = 5 max_research_loops = 3 reasoning_model = "gemini-2.5-pro-preview-06-05" # Prepare the input state with the user's query input_state = { "messages": [HumanMessage(content=query)], "search_query": [], "web_research_result": [], "sources_gathered": [], "initial_search_query_count": initial_search_query_count, "max_research_loops": max_research_loops, "reasoning_model": reasoning_model, } query_generator_model: str = "gemini-2.5-flash-preview-05-20" reflection_model: str = "gemini-2.5-flash-preview-05-20" answer_model: str = "gemini-2.5-pro-preview-06-05" # Configuration for the agent config = { "configurable": { "query_generator_model": query_generator_model, "reflection_model": reflection_model, "answer_model": answer_model, } } # Run the agent graph to process the query in a separate thread to avoid blocking result = graph.invoke(input_state, config) # Extract the final answer and sources from the result answer = ( result["messages"][-1].content if result["messages"] else "No answer generated." ) sources = result["sources_gathered"] # Create filename from first few characters of query (spaces replaced with underscores) sanitized_query = re.sub(r'[^\w\s-]', '', query)[:20] # Remove special chars, keep first 20 chars filename = re.sub(r'\s+', '_', sanitized_query.strip()) + '.json' file_path = os.path.join(tempfile.gettempdir(), filename) # Write answer and sources to JSON file result_data = {"answer": answer, "sources": sources} with open(file_path, 'w', encoding='utf-8') as f: json.dump(result_data, f, ensure_ascii=False, indent=2) return {"file_path": file_path} - Pydantic schemas used within the research graph for structured LLM outputs: SearchQueryList for generating search queries and Reflection for evaluating research sufficiency.
class SearchQueryList(BaseModel): query: List[str] = Field( description="A list of search queries to be used for web research." ) rationale: str = Field( description="A brief explanation of why these queries are relevant to the research topic." ) class Reflection(BaseModel): is_sufficient: bool = Field( description="Whether the provided summaries are sufficient to answer the user's question." ) knowledge_gap: str = Field( description="A description of what information is missing or needs clarification." ) follow_up_queries: List[str] = Field( description="A list of follow-up queries to address the knowledge gap." ) - LangGraph definition invoked by deep_search handlers. Defines nodes for query generation, web research, reflection, and answer finalization, with edges controlling the research loop.
builder = StateGraph(OverallState, config_schema=Configuration) # Define the nodes we will cycle between builder.add_node("generate_query", generate_query) builder.add_node("web_research", web_research) builder.add_node("reflection", reflection) builder.add_node("finalize_answer", finalize_answer) # Set the entrypoint as `generate_query` # This means that this node is the first one called builder.add_edge(START, "generate_query") # Add conditional edge to continue with search queries in a parallel branch builder.add_conditional_edges( "generate_query", continue_to_web_research, ["web_research"] ) # Reflect on the web research builder.add_edge("web_research", "reflection") # Evaluate the research builder.add_conditional_edges( "reflection", evaluate_research, ["web_research", "finalize_answer"] ) # Finalize the answer builder.add_edge("finalize_answer", END) graph = builder.compile(name="pro-search-agent")