get_trending_topics
Identify trending medical research topics and emerging areas within specific categories by analyzing recent PubMed publications over a defined time period.
Instructions
Get trending medical topics and research areas
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| category | No | Medical category (e.g., 'cardiology', 'oncology', 'neurology') | |
| days | No | Number of days to analyze for trends |
Implementation Reference
- src/tool_handler.py:423-501 (handler)The main handler function that implements get_trending_topics by querying recent PubMed articles with trending keywords, grouping by article keywords, and returning top trending topics.async def _handle_get_trending_topics(self, arguments: Dict[str, Any]) -> MCPResponse: """Handle trending topics analysis.""" try: category = arguments.get("category", "") days = arguments.get("days", 7) # Calculate date range for trending analysis end_date = datetime.now() start_date = end_date - timedelta(days=days) date_from = start_date.strftime("%Y/%m/%d") date_to = end_date.strftime("%Y/%m/%d") # Build query for trending topics if category: query = f'{category} AND ("trending" OR "emerging" OR "new" OR "novel")' else: query = ( '("trending" OR "emerging" OR "breakthrough" OR "novel") ' "AND (medicine OR medical)" ) search_result = await self.pubmed_client.search_articles( query=query, max_results=30, sort_order=SortOrder.PUBLICATION_DATE, date_from=date_from, date_to=date_to, cache=self.cache, ) content = [] content.append( { "type": "text", "text": ( f"**Trending Topics in {category or 'Medicine'} " f"(Last {days} days)**\n\n" f"Found: {search_result.returned_results} recent articles\n" ), } ) # Group by topics/keywords topics = {} for article_data in search_result.articles: # Handle Article objects - access keywords attribute directly keywords = ( getattr(article_data, "keywords", []) if hasattr(article_data, "keywords") else [] ) for keyword in keywords[:3]: # Top 3 keywords if keyword not in topics: topics[keyword] = [] topics[keyword].append(article_data) # Show top topics sorted_topics = sorted(topics.items(), key=lambda x: len(x[1]), reverse=True)[:5] for topic, articles in sorted_topics: articles_list = "\n".join( [f"• {article.title} (PMID: {article.pmid})" for article in articles[:3]] ) content.append( { "type": "text", "text": f"\n**{topic}** ({len(articles)} articles)\n{articles_list}", } ) return MCPResponse(content=content) except Exception as e: logger.error(f"Error in get_trending_topics: {e}") return MCPResponse( content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True )
- src/tools.py:225-246 (schema)Defines the tool schema, description, and input parameters (category and days) for get_trending_topics.{ "name": "get_trending_topics", "description": "Get trending medical topics and research areas", "inputSchema": { "type": "object", "properties": { "category": { "type": "string", "description": ( "Medical category (e.g., 'cardiology', 'oncology', " "'neurology')" ), }, "days": { "type": "integer", "minimum": 1, "maximum": 30, "default": 7, "description": "Number of days to analyze for trends", }, }, }, },
- src/tool_handler.py:73-73 (registration)Registers the tool name to its handler method in the handler_map dictionary used by handle_tool_call."get_trending_topics": self._handle_get_trending_topics,