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