Remove unused research resource and related utilities

Eliminated the `research://{topic}` resource API, associated utilities, and the `research_store`. These components were redundant due to existing alternatives using the `conduct_research` tool. This cleanup reduces complexity and improves maintainability.
This commit is contained in:
ThomasTaroni 2025-05-03 11:10:56 +02:00
parent eec1b34517
commit 47c036a973
2 changed files with 1 additions and 94 deletions

View file

@ -18,13 +18,10 @@ from gpt_researcher import GPTResearcher
load_dotenv() load_dotenv()
from utils import ( from utils import (
research_store,
create_success_response, create_success_response,
handle_exception, handle_exception,
get_researcher_by_id, get_researcher_by_id,
format_sources_for_response, format_sources_for_response,
format_context_with_sources,
store_research_results,
create_research_prompt create_research_prompt
) )
@ -54,50 +51,6 @@ class CustomLogsHandler:
self.logs.append(data) # Append data to logs self.logs.append(data) # Append data to logs
print(f"MCP Log: {data}") # For demonstration, print the log print(f"MCP Log: {data}") # For demonstration, print the log
@mcp.resource("research://{topic}")
async def research_resource(topic: str) -> str:
"""
Provide research context for a given topic directly as a resource.
This allows LLMs to access web-sourced information without explicit function calls.
Args:
topic: The research topic or query
Returns:
String containing the research context with source information
"""
# Check if we've already researched this topic
if topic in research_store:
logger.info(f"Returning cached research for topic: {topic}")
return research_store[topic]["context"]
# If not, conduct the research
logger.info(f"Conducting new research for resource on topic: {topic}")
custom_logs_handler = CustomLogsHandler()
# Initialize GPT Researcher
researcher = GPTResearcher(query=topic, report_type=research_type, websocket=custom_logs_handler)
try:
# Conduct the research
await researcher.conduct_research()
# Get the context and sources
context = researcher.get_research_context()
sources = researcher.get_research_sources()
source_urls = researcher.get_source_urls()
# Format with sources included
formatted_context = format_context_with_sources(topic, context, sources)
# Store for future use
store_research_results(topic, context, sources, source_urls, formatted_context)
return formatted_context
except Exception as e:
return f"Error conducting research on '{topic}': {str(e)}"
@mcp.tool() @mcp.tool()
async def deep_research(query: str) -> Dict[str, Any]: async def deep_research(query: str) -> Dict[str, Any]:
@ -132,9 +85,6 @@ async def deep_research(query: str) -> Dict[str, Any]:
sources = researcher.get_research_sources() sources = researcher.get_research_sources()
source_urls = researcher.get_source_urls() source_urls = researcher.get_source_urls()
# Store in the research store for the resource API
store_research_results(query, context, sources, source_urls)
return create_success_response({ return create_success_response({
"research_id": research_id, "research_id": research_id,
"query": query, "query": query,

View file

@ -11,8 +11,6 @@ from loguru import logger
# Configure logging for console only (no file logging) # Configure logging for console only (no file logging)
logger.configure(handlers=[{"sink": sys.stderr, "level": "INFO"}]) logger.configure(handlers=[{"sink": sys.stderr, "level": "INFO"}])
# Research store to track ongoing research topics and contexts
research_store = {}
# API Response Utilities # API Response Utilities
def create_error_response(message: str) -> Dict[str, Any]: def create_error_response(message: str) -> Dict[str, Any]:
@ -68,44 +66,6 @@ def format_sources_for_response(sources: List[Dict[str, Any]]) -> List[Dict[str,
] ]
def format_context_with_sources(topic: str, context: str, sources: List[Dict[str, Any]]) -> str:
"""
Format research context with sources for display.
Args:
topic: Research topic
context: Research context
sources: List of sources
Returns:
Formatted context string with sources
"""
formatted_context = f"## Research: {topic}\n\n{context}\n\n"
formatted_context += "## Sources:\n"
for i, source in enumerate(sources):
formatted_context += f"{i+1}. {source.get('title', 'Unknown')}: {source.get('url', '')}\n"
return formatted_context
def store_research_results(topic: str, context: str, sources: List[Dict[str, Any]],
source_urls: List[str], formatted_context: Optional[str] = None):
"""
Store research results in the research store.
Args:
topic: Research topic
context: Research context
sources: List of sources
source_urls: List of source URLs
formatted_context: Optional pre-formatted context
"""
research_store[topic] = {
"context": formatted_context or context,
"sources": sources,
"source_urls": source_urls
}
def create_research_prompt(topic: str, goal: str, report_format: str = "research_report") -> str: def create_research_prompt(topic: str, goal: str, report_format: str = "research_report") -> str:
""" """
Create a research query prompt for GPT Researcher. Create a research query prompt for GPT Researcher.
@ -125,10 +85,7 @@ def create_research_prompt(topic: str, goal: str, report_format: str = "research
You have two methods to access web-sourced information: You have two methods to access web-sourced information:
1. Use the "research://{topic}" resource to directly access context about this topic if it exists Use the conduct_research tool to perform new research and get a research_id for later use.
or if you want to get straight to the information without tracking a research ID.
2. Use the conduct_research tool to perform new research and get a research_id for later use.
This tool also returns the context directly in its response, which you can use immediately. This tool also returns the context directly in its response, which you can use immediately.
After getting context, you can: After getting context, you can: