mcp-pubmed

Instructions

Implementation Reference

• main.py:271-375 (handler)

  The `search_pubmed` function is the core handler for searching PubMed articles, implementing logic to build a query, call the NCBI E-utilities API, and format the results.

  async def search_pubmed(
      query: str,
      max_results: int = 10,
      year_from: Optional[int] = None,
      year_to: Optional[int] = None,
      article_type: Optional[str] = None,
      sort: str = "relevance",
  ) -> str:
      """Search PubMed for articles matching a query.
  
      Args:
          query: Search query. Supports full PubMed syntax:
                 AND / OR / NOT, field tags like [tiab], [MeSH], [au], etc.
                 Examples:
                   "covid-19 vaccine efficacy"
                   "myocardial infarction[MeSH] AND aspirin[tiab]"
          max_results: Number of articles to return (1-100, default 10).
          year_from: Restrict results to articles published from this year.
          year_to:   Restrict results to articles published up to this year.
          article_type: Filter by publication type, e.g. "Review",
                        "Clinical Trial", "Meta-Analysis",
                        "Randomized Controlled Trial".
          sort: "relevance" (default) or "date" (most recent first).
  
      Returns:
          A formatted list of matching articles with PMID, title, authors,
          journal, date, and a short abstract snippet.
          Returns an error message if the query fails or yields no results.
      """
      if not query or not query.strip():
          return _err("Query must not be empty.")
  
      max_results = max(1, min(max_results, 100))
  
      # Build full query with optional filters
      full_query = query.strip()
      if year_from and year_to:
          if year_from > year_to:
              return _err(f"year_from ({year_from}) must be ≤ year_to ({year_to}).")
          full_query += f" AND {year_from}:{year_to}[pdat]"
      elif year_from:
          full_query += f" AND {year_from}:3000[pdat]"
      elif year_to:
          full_query += f" AND 1900:{year_to}[pdat]"
  
      if article_type:
          full_query += f' AND "{article_type.strip()}"[pt]'
  
      sort_param = "pub_date" if sort == "date" else "relevance"
  
      try:
          # Step 1: esearch → PMIDs
          search_resp = await _get(
              "esearch.fcgi",
              {
                  "db": "pubmed",
                  "term": full_query,
                  "retmax": max_results,
                  "retmode": "json",
                  "sort": sort_param,
              },
          )
          esearch = search_resp.json().get("esearchresult", {})
          id_list: list[str] = esearch.get("idlist", [])
          total: str = esearch.get("count", "0")
  
          # Warn if query was corrected/translated
          query_translation = esearch.get("querytranslation", "")
  
          if not id_list:
              return (
                  f"No articles found for query: {query!r}\n"
                  f"(Full query sent: {full_query})"
              )
  
          # Step 2: efetch → article XML
          fetch_resp = await _get(
              "efetch.fcgi",
              {
                  "db": "pubmed",
                  "id": ",".join(id_list),
                  "retmode": "xml",
                  "rettype": "abstract",
              },
          )
          root = _require_xml(fetch_resp, "efetch articles")
          articles = [_parse_article(a) for a in root.findall(".//PubmedArticle")]
  
          if not articles:
              return _err("Received empty article list from NCBI.")
  
          header_parts = [f"Found {total} total result(s). Showing {len(articles)}."]
          if query_translation:
              header_parts.append(f"Query interpreted as: {query_translation}")
          header = "\n".join(header_parts) + "\n"
  
          blocks = [_format_brief(a, i) for i, a in enumerate(articles, 1)]
          return header + "\n\n".join(blocks)
  
      except PubMedError as exc:
          return _err(str(exc))
  
  
  @mcp.tool()
  async def get_article(pmid: str) -> str: