pubmed_search

Instructions

Implementation Reference

• trends_mcp.py:574-640 (handler)

  The pubmed_search async function that executes the PubMed search logic. It calls esearch.fcgi for PMIDs, esummary.fcgi for metadata, and efetch.fcgi for abstracts, then formats results as markdown or JSON.

  async def pubmed_search(
      query: str,
      days: int | None = None,
      max_results: int = 20,
      response_format: ResponseFormat = ResponseFormat.MARKDOWN,
  ) -> str:
      try:
          args = PubMedSearchInput(
              query=query,
              days=days,
              max_results=max_results,
              response_format=response_format,
          )
          api_key = os.environ.get("NCBI_API_KEY")
          term = args.query
          if args.days:
              term = f"({term}) AND (\"last {args.days} days\"[PDat])"
  
          common: dict[str, Any] = {"db": "pubmed", "retmode": "json"}
          if api_key:
              common["api_key"] = api_key
  
          # Step 1: esearch -> PMID list
          esearch_params = {**common, "term": term, "retmax": args.max_results, "sort": "pub_date"}
          es_data = await _http_get_json(
              f"{PUBMED_BASE}/esearch.fcgi", params=esearch_params, ttl=TTL_STATIC
          )
          pmids: list[str] = es_data.get("esearchresult", {}).get("idlist", [])
          if not pmids:
              header = f"PubMed `{args.query}` (0건)"
              return _format([], args.response_format, render_md=lambda x: _render_pubmed_md(x, header))
  
          # Step 2: esummary -> metadata
          esum_params = {**common, "id": ",".join(pmids)}
          es2_data = await _http_get_json(
              f"{PUBMED_BASE}/esummary.fcgi", params=esum_params, ttl=TTL_STATIC
          )
          result = es2_data.get("result", {})
          uids = result.get("uids", pmids)
  
          items: list[dict[str, Any]] = []
          for uid in uids:
              r = result.get(uid)
              if not r:
                  continue
              authors = [a.get("name", "") for a in r.get("authors", []) if a.get("name")]
              items.append(
                  {
                      "pmid": uid,
                      "url": f"https://pubmed.ncbi.nlm.nih.gov/{uid}/",
                      "title": r.get("title", "").rstrip("."),
                      "journal": r.get("fulljournalname") or r.get("source") or "",
                      "pubdate": r.get("pubdate") or r.get("epubdate") or "",
                      "authors": authors,
                  }
              )
  
          # Step 3: efetch -> abstracts (one batched call). Best-effort: if it
          # fails or an article has no abstract, we just skip the field.
          abstracts = await _pubmed_fetch_abstracts([it["pmid"] for it in items])
          for it in items:
              it["abstract"] = abstracts.get(it["pmid"], "")
  
          header = f"PubMed `{args.query}` ({len(items)}건)"
          return _format(items, args.response_format, render_md=lambda x: _render_pubmed_md(x, header))
      except Exception as e:
          return _handle_error(e, "pubmed_search")

• trends_mcp.py:484-490 (schema)

  PubMedSearchInput Pydantic model defining input validation schema for pubmed_search: query (required), days, max_results, and response_format.

  class PubMedSearchInput(BaseModel):
      model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
      query: str = Field(..., min_length=1, max_length=500)
      days: int | None = Field(None, ge=1, le=3650)
      max_results: int = Field(20, ge=1, le=50)
      response_format: ResponseFormat = ResponseFormat.MARKDOWN

• trends_mcp.py:558-572 (registration)

  Registration of pubmed_search via @_maybe_tool decorator with source='pubmed', name='pubmed_search', description, and annotations. The tool is only registered if 'pubmed' is in ENABLED_SOURCES.

  @_maybe_tool(
      source="pubmed",
      name="pubmed_search",
      description=(
          "Search PubMed for biomedical publications. Plain keywords work; for "
          "advanced queries use MeSH and field tags: `mammography[MeSH]`, "
          "`smith[Author]`, `2025[PDat]`. Combine with `AND`/`OR`. "
          "`days` filters by publication date (`PDat` field)."
      ),
      annotations={
          "readOnlyHint": True,
          "destructiveHint": False,
          "openWorldHint": True,
          "idempotentHint": True,
      },

• trends_mcp.py:492-536 (helper)

  _pubmed_fetch_abstracts helper function that fetches abstract text for given PMIDs via efetch.fcgi (XML). Returns {pmid: abstract_text} dictionary.

  async def _pubmed_fetch_abstracts(pmids: list[str]) -> dict[str, str]:
      """Fetch abstract text for each PMID via efetch.fcgi (XML).
  
      Returns {pmid: abstract_text}. Empty values for PMIDs without abstracts
      (e.g. editorials, letters). Failures are swallowed — caller falls back
      to no-abstract rendering.
      """
      if not pmids:
          return {}
      params: dict[str, Any] = {
          "db": "pubmed",
          "id": ",".join(pmids),
          "retmode": "xml",
      }
      api_key = os.environ.get("NCBI_API_KEY")
      if api_key:
          params["api_key"] = api_key
      try:
          text = await _http_get_text(
              f"{PUBMED_BASE}/efetch.fcgi", params=params, ttl=TTL_STATIC
          )
      except Exception:
          return {}
      out: dict[str, str] = {}
      try:
          root = ET.fromstring(text)
      except ET.ParseError:
          return {}
      for article in root.findall(".//PubmedArticle"):
          pmid_el = article.find(".//MedlineCitation/PMID")
          if pmid_el is None or not pmid_el.text:
              continue
          pmid = pmid_el.text.strip()
          # AbstractText can be split into Background/Methods/Results/Conclusions
          # via the `Label` attribute; concatenate with labels for readability.
          parts: list[str] = []
          for ab in article.findall(".//Abstract/AbstractText"):
              label = ab.attrib.get("Label", "").strip()
              content = "".join(ab.itertext()).strip()
              if not content:
                  continue
              parts.append(f"{label}: {content}" if label else content)
          if parts:
              out[pmid] = " ".join(parts)
      return out

• trends_mcp.py:539-555 (helper)

  _render_pubmed_md helper function that renders a list of PubMed items into Markdown format with title/URL, PMID, journal, date, authors, and abstract.

  def _render_pubmed_md(items: list[dict[str, Any]], header: str) -> str:
      if not items:
          return f"# {header}\n\n_결과 없음_"
      lines = [f"# {header}", f"_총 {len(items)}건_", ""]
      for i, it in enumerate(items, 1):
          authors = ", ".join(it["authors"][:4])
          if len(it["authors"]) > 4:
              authors += f" 외 {len(it['authors']) - 4}명"
          block = (
              f"## {i}. [{it['title']}]({it['url']})\n"
              f"- PMID `{it['pmid']}` · {it['journal']} · {it['pubdate']}\n"
              f"- 저자: {authors}\n"
          )
          if it.get("abstract"):
              block += f"- {_trim(it['abstract'], 500)}\n"
          lines.append(block)
      return "\n".join(lines)