Add quick_search function and remove citation note in deep_research

Introduced the quick_search function for faster, snippet-based searches and adjusted mcp.run to remove the "sse" argument. Additionally, modified the deep_research docstring to remove the citation requirement note for simpler usage documentation.
This commit is contained in:
ThomasTaroni 2025-04-26 18:46:11 +02:00
parent 7e5a6db0f6
commit 7f7a6083f8

View file

@ -9,7 +9,7 @@ import os
import sys import sys
import uuid import uuid
import logging import logging
from typing import Dict, Any, Optional from typing import Dict, Any, Optional, List
from dotenv import load_dotenv from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from gpt_researcher import GPTResearcher from gpt_researcher import GPTResearcher
@ -91,9 +91,8 @@ async def research_resource(topic: str) -> str:
@mcp.tool() @mcp.tool()
async def deep_research(query: str) -> Dict[str, Any]: async def deep_research(query: str) -> Dict[str, Any]:
""" """
Conduct a deep web research on a given query using GPT Researcher. Conduct a web deep research on a given query using GPT Researcher.
Use this tool when you need time-sensitive, real-time information like stock prices, news, people, specific knowledge, etc. Use this tool when you need time-sensitive, real-time information like stock prices, news, people, specific knowledge, etc.
You must include citations that back your responses when using this tool.
Args: Args:
query: The research query or topic query: The research query or topic
@ -136,6 +135,43 @@ async def deep_research(query: str) -> Dict[str, Any]:
return handle_exception(e, "Research") return handle_exception(e, "Research")
@mcp.tool()
async def quick_search(query: str) -> Dict[str, Any]:
"""
Perform a quick web search on a given query and return search results with snippets.
This optimizes for speed over quality and is useful when an LLM doesn't need in-depth
information on a topic.
Args:
query: The search query
Returns:
Dict containing search results and snippets
"""
logger.info(f"Performing quick search on query: {query}...")
# Generate a unique ID for this search session
search_id = str(uuid.uuid4())
# Initialize GPT Researcher
researcher = GPTResearcher(query, report_type=research_type)
try:
# Perform quick search
search_results = await researcher.quick_search(query=query)
mcp.researchers[search_id] = researcher
logger.info(f"Quick search completed for ID: {search_id}")
return create_success_response({
"search_id": search_id,
"query": query,
"result_count": len(search_results) if search_results else 0,
"search_results": search_results
})
except Exception as e:
return handle_exception(e, "Quick search")
@mcp.tool() @mcp.tool()
async def write_report(research_id: str, custom_prompt: Optional[str] = None) -> Dict[str, Any]: async def write_report(research_id: str, custom_prompt: Optional[str] = None) -> Dict[str, Any]:
""" """
@ -246,7 +282,7 @@ def run_server():
# Let FastMCP handle the event loop # Let FastMCP handle the event loop
try: try:
mcp.run("sse") mcp.run()
# Note: If we reach here, the server has stopped # Note: If we reach here, the server has stopped
logger.info("MCP Server has stopped") logger.info("MCP Server has stopped")
except Exception as e: except Exception as e: