PubMed MCP Server

Instructions

Implementation Reference

• src/tool_handler.py:658-727 (handler)

  The core handler function that implements the get_journal_metrics tool: validates journal_name input, searches PubMed for current year articles from the journal, computes metrics (article count, type distribution), optionally lists recent articles, and returns formatted MCPResponse.

  async def _handle_get_journal_metrics(self, arguments: Dict[str, Any]) -> MCPResponse:
      """Handle journal metrics request."""
      try:
          journal_name = arguments.get("journal_name", "")
          if not journal_name:
              return MCPResponse(
                  content=[{"type": "text", "text": "Journal name is required"}], is_error=True
              )
  
          include_recent_articles = arguments.get("include_recent_articles", True)
  
          # Get recent articles from the journal
          current_year = datetime.now().year
          search_result = await self.pubmed_client.search_articles(
              query=f'"{journal_name}"[Journal]',
              max_results=50,
              date_from=f"{current_year}/01/01",
              sort_order=SortOrder.PUBLICATION_DATE,
              cache=self.cache,
          )
  
          content = []
          content.append(
              {
                  "type": "text",
                  "text": f"**Journal Metrics: {journal_name}**\n\n"
                  f"Articles in {current_year}: {search_result.total_results:,}\n"
                  f"Sample Size: {search_result.returned_results}\n",
              }
          )
  
          # Analyze article types
          if search_result.articles:
              article_types = {}
              for article_data in search_result.articles:
                  # Handle Article objects - access article_types attribute directly
                  types = (
                      getattr(article_data, "article_types", [])
                      if hasattr(article_data, "article_types")
                      else []
                  )
                  for article_type in types:
                      article_types[article_type] = article_types.get(article_type, 0) + 1
  
              if article_types:
                  types_text = "**Article Types Distribution:**\n"
                  for article_type, count in sorted(
                      article_types.items(), key=lambda x: x[1], reverse=True
                  )[:5]:
                      percentage = (count / len(search_result.articles)) * 100
                      types_text += f"• {article_type}: {count} ({percentage:.1f}%)\n"
  
                  content.append({"type": "text", "text": types_text})
  
          # Show recent notable articles
          if include_recent_articles and search_result.articles:
              content.append({"type": "text", "text": "\n**Recent Articles:**\n"})
  
              for i, article_data in enumerate(search_result.articles[:5], 1):
                  article_text = self._format_article_summary(article_data, i)
                  content.append({"type": "text", "text": article_text})
  
          return MCPResponse(content=content)
  
      except Exception as e:
          logger.error(f"Error in get_journal_metrics: {e}")
          return MCPResponse(
              content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True
          )

• src/tools.py:303-318 (schema)

  The tool schema definition in TOOL_DEFINITIONS list, including name, description, and inputSchema specifying journal_name as required parameter.

  {
      "name": "get_journal_metrics",
      "description": "Get metrics and information about a specific journal",
      "inputSchema": {
          "type": "object",
          "properties": {
              "journal_name": {"type": "string", "description": "Journal name or abbreviation"},
              "include_recent_articles": {
                  "type": "boolean",
                  "default": True,
                  "description": "Include recent notable articles",
              },
          },
          "required": ["journal_name"],
      },
  },

• src/tool_handler.py:65-78 (registration)

  The handler_map in ToolHandler.handle_tool_call that registers 'get_journal_metrics' to route to _handle_get_journal_metrics function.

  handler_map = {
      "search_pubmed": self._handle_search_pubmed,
      "get_article_details": self._handle_get_article_details,
      "search_by_author": self._handle_search_by_author,
      "find_related_articles": self._handle_find_related_articles,
      "export_citations": self._handle_export_citations,
      "search_mesh_terms": self._handle_search_mesh_terms,
      "search_by_journal": self._handle_search_by_journal,
      "get_trending_topics": self._handle_get_trending_topics,
      "analyze_research_trends": self._handle_analyze_research_trends,
      "compare_articles": self._handle_compare_articles,
      "get_journal_metrics": self._handle_get_journal_metrics,
      "advanced_search": self._handle_advanced_search,
  }

• src/tool_handler.py:34-37 (registration)

  The get_tools method that returns the TOOL_DEFINITIONS list, registering the get_journal_metrics tool schema with the MCP protocol.

  def get_tools(self) -> List[Dict[str, Any]]:
      """Get list of available tools."""
      return TOOL_DEFINITIONS