PubMed MCP Server

Instructions

Implementation Reference

• src/tool_handler.py:587-656 (handler)

  Main execution logic for the compare_articles tool: validates 2-5 PMIDs, fetches article details via PubMedClient, formats comparison of basic info, MeSH terms, and abstracts.
  async def _handle_compare_articles(self, arguments: Dict[str, Any]) -> MCPResponse: """Handle article comparison.""" try: pmids = arguments.get("pmids", []) if len(pmids) < 2 or len(pmids) > 5: return MCPResponse( content=[{"type": "text", "text": "Please provide 2-5 PMIDs for comparison"}], is_error=True, ) comparison_fields = arguments.get( "comparison_fields", ["authors", "methods", "conclusions"] ) # Get article details articles = await self.pubmed_client.get_article_details( pmids=pmids, include_abstracts=True, cache=self.cache ) if len(articles) < 2: return MCPResponse( content=[ {"type": "text", "text": "Not enough valid articles found for comparison"} ], is_error=True, ) content = [] content.append( {"type": "text", "text": f"**Comparison of {len(articles)} Articles**\n"} ) # Basic comparison comparison_text = "**Articles:**\n" for i, article in enumerate(articles, 1): authors_str = format_authors( [f"{a.first_name or a.initials} {a.last_name}" for a in article.authors[:3]] ) comparison_text += f"{i}. {article.title}\n" comparison_text += f" Authors: {authors_str}\n" comparison_text += ( f" Journal: {article.journal.title} ({format_date(article.pub_date)})\n" ) comparison_text += f" PMID: {article.pmid}\n\n" content.append({"type": "text", "text": comparison_text}) # Compare specific fields if "mesh_terms" in comparison_fields: mesh_comparison = "**MeSH Terms Comparison:**\n" for i, article in enumerate(articles, 1): mesh_terms = [term.descriptor_name for term in article.mesh_terms[:5]] mesh_comparison += f"{i}. {', '.join(mesh_terms)}\n" content.append({"type": "text", "text": mesh_comparison}) if "abstracts" in comparison_fields: content.append({"type": "text", "text": "**Abstracts:**\n"}) for i, article in enumerate(articles, 1): abstract_text = truncate_text(article.abstract or "No abstract available", 200) content.append({"type": "text", "text": f"{i}. {abstract_text}\n"}) return MCPResponse(content=content) except Exception as e: logger.error(f"Error in compare_articles: {e}") return MCPResponse( content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True )
• src/tools.py:270-302 (schema)

  JSON schema defining the input parameters for compare_articles: requires array of 2-5 PMIDs, optional comparison_fields enum.
  { "name": "compare_articles", "description": "Compare multiple articles side by side", "inputSchema": { "type": "object", "properties": { "pmids": { "type": "array", "items": {"type": "string"}, "minItems": 2, "maxItems": 5, "description": "List of PMIDs to compare (2-5 articles)", }, "comparison_fields": { "type": "array", "items": { "type": "string", "enum": [ "authors", "methods", "results", "conclusions", "mesh_terms", "citations", ], }, "default": ["authors", "methods", "conclusions"], "description": "Fields to compare", }, }, "required": ["pmids"], }, },
• src/tool_handler.py:65-78 (registration)

  Registration of the _handle_compare_articles handler in the tool routing dictionary used by handle_tool_call method.
  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, }