PubMed MCP Server

Instructions

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,