PubMed MCP Server

Instructions

Implementation Reference

• src/tools.py:143-160 (schema)

  Tool schema definition with input validation for 'pmid' and optional 'max_results'.

  {
      "name": "find_related_articles",
      "description": "Find articles related to a specific PMID",
      "inputSchema": {
          "type": "object",
          "properties": {
              "pmid": {"type": "string", "description": "PMID of the reference article"},
              "max_results": {
                  "type": "integer",
                  "minimum": 1,
                  "maximum": 50,
                  "default": 10,
                  "description": "Maximum number of related articles",
              },
          },
          "required": ["pmid"],
      },
  },

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

  Tool registration in handler_map dictionary mapping 'find_related_articles' to its handler 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,
  }

• src/tool_handler.py:261-296 (handler)

  MCP tool handler method that validates input, calls PubMed client, formats response as MCPResponse.

  async def _handle_find_related_articles(self, arguments: Dict[str, Any]) -> MCPResponse:
      """Handle finding related articles."""
      try:
          pmid = arguments.get("pmid", "")
          if not pmid:
              return MCPResponse(
                  content=[{"type": "text", "text": "PMID is required"}], is_error=True
              )
  
          max_results = arguments.get("max_results", 10)
  
          search_result = await self.pubmed_client.find_related_articles(
              pmid=pmid, max_results=max_results, cache=self.cache
          )
  
          content = []
          content.append(
              {
                  "type": "text",
                  "text": f"**Articles Related to PMID: {pmid}**\n\n"
                  f"Found: {search_result.returned_results} related articles\n",
              }
          )
  
          for i, article_data in enumerate(search_result.articles, 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 find_related_articles: {e}")
          return MCPResponse(
              content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True
          )

• src/pubmed_client.py:318-375 (helper)

  Core implementation using PubMed ELink API to fetch related articles by PMID, with caching and parsing.

  async def find_related_articles(
      self, pmid: str, max_results: int = 10, cache: Optional[CacheManager] = None
  ) -> SearchResult:
      """Find articles related to a specific PMID."""
      start_time = time.time()
  
      if not validate_pmid(pmid):
          raise ValueError(f"Invalid PMID: {pmid}")
  
      if cache:
          cache_key = cache.generate_key("related", pmid=pmid, max_results=max_results)
          cached_result = cache.get(cache_key)
          if cached_result:
              # Convert cached article dicts back to Article objects
              cached_articles = [
                  Article(**article_data) for article_data in cached_result["articles"]
              ]
              cached_result["articles"] = cached_articles
              return SearchResult(**cached_result)
  
      # Use elink to find related articles
      link_params = self._build_params(
          dbfrom="pubmed",
          db="pubmed",
          id=pmid,
          linkname="pubmed_pubmed",
          retmode="json",
      )
  
      link_response = await self._make_request("elink.fcgi", link_params)
      link_data = link_response.json()
  
      related_pmids = []
      linksets = link_data.get("linksets", [])
      if linksets and "linksetdbs" in linksets[0]:
          for linksetdb in linksets[0]["linksetdbs"]:
              if linksetdb.get("linkname") == "pubmed_pubmed":
                  related_pmids = linksetdb.get("links", [])[:max_results]
                  break
  
      articles = []
      if related_pmids:
          articles = await self._fetch_article_details(related_pmids, include_full_details=True)
  
      result_data = {
          "query": f"Related to PMID: {pmid}",
          "total_results": len(related_pmids),
          "returned_results": len(articles),
          "articles": articles,  # Store Article objects directly
          "search_time": time.time() - start_time,
          "suggestions": [],
      }
  
      if cache:
          cache_data = {**result_data, "articles": [article.model_dump() for article in articles]}
          cache.set(cache_key, cache_data)
  
      return SearchResult(**result_data)