search_pubmed

Instructions

Implementation Reference

• aria_mcp_server/server.py:32-80 (registration)

  MCP tool registration via @mcp.tool decorator with description and output_schema. The server-side handler function that delegates to the core implementation in tools.py.

  @mcp.tool(
      description=(
          "Search PubMed for peer-reviewed biomedical literature. "
          "Read-only operation. No authentication required. "
          "Uses NCBI E-utilities public API (rate limit: ~3 requests/sec). "
          "Returns up to 10 results per call. No pagination. "
          "Returns 'No papers found.' if no results match. "
          "Use for: research papers, drug mechanisms, clinical outcomes, disease studies, "
          "safety/efficacy data, biomarkers, diagnostics, and any scientific question."
      ),
      output_schema={
          "type": "object",
          "properties": {
              "result": {
                  "type": "string",
                  "description": "Formatted list of papers with title, authors, journal, year, PMID, and abstract. Returns 'no results' message if nothing found."
              }
          },
          "required": ["result"]
      }
  )
  def search_pubmed(
      query: Annotated[str, "Search query e.g. 'velarixin pediatric epilepsy phase 2'"],
      max_results: Annotated[int, "Number of papers to return, between 1 and 10"] = 5,
  ) -> str:
      """
      Search PubMed for peer-reviewed biomedical literature.
  
      Use for: research papers, drug mechanisms, clinical outcomes, disease studies,
      safety/efficacy data, biomarkers, diagnostics, and any scientific question.
  
      Args:
          query: Search query (e.g. "velarixin pediatric epilepsy phase 2")
          max_results: Number of papers to return (1-10, default 5)
  
      Returns:
          Formatted string with title, authors, journal, year, PMID, and abstract
          for each paper. Returns a "no results" message if nothing is found.
          Handles API errors gracefully with descriptive error messages.
  
      Notes:
          - Results are sorted by relevance
          - max_results is clamped to 1-10 regardless of input
          - Requires no API key; uses NCBI E-utilities public API
      """
      from aria_mcp_server.tools import search_pubmed as _search, format_results_for_claude as _fmt
      max_results = max(1, min(max_results, 10))
      papers = _search(query=query, max_results=max_results)
      return _fmt(papers)

• aria_mcp_server/server.py:42-51 (schema)

  Output schema for the search_pubmed tool, defining that it returns a string result.

  output_schema={
      "type": "object",
      "properties": {
          "result": {
              "type": "string",
              "description": "Formatted list of papers with title, authors, journal, year, PMID, and abstract. Returns 'no results' message if nothing found."
          }
      },
      "required": ["result"]
  }

• aria_mcp_server/server.py:16-19 (helper)

  Import of search_pubmed (and other tools) from tools.py into the server module.

  from aria_mcp_server.tools import (
      search_pubmed, search_clinical_trials, format_results_for_claude, format_trials_for_claude,
      search_isrctn, format_isrctn_for_claude
  )

• aria_mcp_server/tools.py:99-141 (handler)

  Core implementation of search_pubmed: performs NCBI E-utilities esearch for IDs, then efetch for article details, parses XML, and returns a list of dicts with pmid, title, authors, journal, year, abstract, and url.

  def search_pubmed(query: str, max_results: int = 5) -> list[dict]:
      """Search PubMed via NCBI E-utilities. No API key required."""
      if not query or not query.strip():
          return []
      query = query.strip()
      max_results = max(1, min(max_results, 100))
      try:
          r = requests.get(
              f"{PUBMED_BASE}/esearch.fcgi",
              params={"db": "pubmed", "term": query, "retmax": max_results, "retmode": "xml"},
              timeout=15,
          )
          r.raise_for_status()
          data = xmltodict.parse(r.content)
      except Exception as e:
          raise RuntimeError(f"PubMed search failed: {e}") from e
  
      id_list = (data.get("eSearchResult") or {}).get("IdList") or {}
      id_el = id_list.get("Id") if isinstance(id_list, dict) else None
      if not id_el:
          return []
      pmids = [id_el] if isinstance(id_el, str) else list(id_el)[:max_results]
      if not pmids:
          return []
  
      try:
          r2 = requests.get(
              f"{PUBMED_BASE}/efetch.fcgi",
              params={"db": "pubmed", "id": ",".join(pmids), "rettype": "xml"},
              timeout=20,
          )
          r2.raise_for_status()
          fetch_data = xmltodict.parse(r2.content)
      except Exception as e:
          raise RuntimeError(f"PubMed fetch failed: {e}") from e
  
      root = fetch_data.get("PubmedArticleSet") or fetch_data
      articles = root.get("PubmedArticle") or root.get("PubmedData")
      if not articles:
          return []
      if isinstance(articles, dict):
          articles = [articles]
      return [p for p in (_parse_article(a) for a in articles) if p]

• aria_mcp_server/tools.py:47-96 (helper)

  _parse_article helper: parses a single PubMed article XML dict into a structured result dict with pmid, title, authors, journal, year, abstract, and url.

  def _parse_article(article_xml: dict) -> dict | None:
      try:
          medline = article_xml.get("MedlineCitation") or article_xml
          if not medline:
              return None
          pmid_el = medline.get("PMID")
          pmid = _get_text(pmid_el) if isinstance(pmid_el, dict) else str(pmid_el or "").strip()
          if not pmid:
              return None
          article = (medline.get("Article") or {}) if isinstance(medline, dict) else {}
          if isinstance(article, str):
              article = {}
          title = _get_text(article.get("ArticleTitle"))
          author_list = article.get("AuthorList") or {}
          authors = _extract_authors(author_list)
          journal = ""
          year = ""
          journal_el = article.get("Journal")
          if isinstance(journal_el, dict):
              journal = _get_text(journal_el.get("Title"))
              issue = journal_el.get("JournalIssue") or {}
              if isinstance(issue, dict):
                  pub_date = issue.get("PubDate") or {}
                  if isinstance(pub_date, dict):
                      year = _get_text(pub_date.get("Year"))
          abstract_el = article.get("Abstract")
          abstract = ""
          if isinstance(abstract_el, dict):
              abstract_parts = abstract_el.get("AbstractText")
              if isinstance(abstract_parts, list):
                  abstract = " ".join(_get_text(p.get("#text") if isinstance(p, dict) else p) for p in abstract_parts)
              elif isinstance(abstract_parts, dict):
                  abstract = _get_text(abstract_parts.get("#text") or abstract_parts)
              else:
                  abstract = _get_text(abstract_parts)
          else:
              abstract = _get_text(abstract_el)
          if len(abstract) > 500:
              abstract = abstract[:497] + "..."
          return {
              "pmid": pmid,
              "title": title,
              "authors": authors,
              "journal": journal,
              "year": year,
              "abstract": abstract,
              "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/",
          }
      except (KeyError, TypeError, AttributeError):
          return None