Paper Search MCP

Instructions

Implementation Reference

• paper_search_mcp/server.py:242-325 (handler)

  The 'search_papers' tool handler, which orchestrates searches across various academic platforms based on the user-provided 'sources' parameter.

  async def search_papers(
      query: str,
      max_results_per_source: int = 5,
      sources: str = "all",
      year: Optional[str] = None,
  ) -> Dict[str, Any]:
      """Unified top-level search across all configured academic platforms.
  
      Args:
          query: Search query string.
          max_results_per_source: Max results to fetch from each selected source.
          sources: Comma-separated source names or 'all'.
              Available: arxiv,pubmed,biorxiv,medrxiv,google_scholar,iacr,semantic,crossref,openalex,pmc,core,europepmc,dblp,openaire,citeseerx,doaj,base,zenodo,hal,ssrn,unpaywall
          year: Optional year filter for Semantic Scholar only.
      Returns:
          Aggregated dictionary with per-source stats, errors, and deduplicated papers.
      """
      selected_sources = _parse_sources(sources)
  
      if not selected_sources:
          return {
              "query": query,
              "sources_requested": sources,
              "sources_used": [],
              "source_results": {},
              "errors": {"sources": "No valid sources selected."},
              "papers": [],
              "total": 0,
          }
  
      task_map = {}
      for source in selected_sources:
          if source == "arxiv":
              task_map[source] = search_arxiv(query, max_results_per_source)
          elif source == "pubmed":
              task_map[source] = search_pubmed(query, max_results_per_source)
          elif source == "biorxiv":
              task_map[source] = search_biorxiv(query, max_results_per_source)
          elif source == "medrxiv":
              task_map[source] = search_medrxiv(query, max_results_per_source)
          elif source == "google_scholar":
              task_map[source] = search_google_scholar(query, max_results_per_source)
          elif source == "iacr":
              task_map[source] = search_iacr(query, max_results_per_source, fetch_details=False)
          elif source == "semantic":
              task_map[source] = search_semantic(query, year=year, max_results=max_results_per_source)
          elif source == "crossref":
              task_map[source] = search_crossref(query, max_results=max_results_per_source)
          elif source == "openalex":
              task_map[source] = search_openalex(query, max_results_per_source)
          elif source == "pmc":
              task_map[source] = search_pmc(query, max_results_per_source)
          elif source == "core":
              task_map[source] = search_core(query, max_results_per_source)
          elif source == "europepmc":
              task_map[source] = search_europepmc(query, max_results_per_source)
          elif source == "dblp":
              task_map[source] = search_dblp(query, max_results_per_source)
          elif source == "openaire":
              task_map[source] = search_openaire(query, max_results_per_source)
          elif source == "citeseerx":
              task_map[source] = search_citeseerx(query, max_results_per_source)
          elif source == "doaj":
              task_map[source] = search_doaj(query, max_results_per_source)
          elif source == "base":
              task_map[source] = search_base(query, max_results_per_source)
          elif source == "zenodo":
              task_map[source] = search_zenodo(query, max_results_per_source)
          elif source == "hal":
              task_map[source] = search_hal(query, max_results_per_source)
          elif source == "ssrn":
              task_map[source] = search_ssrn(query, max_results_per_source)
          elif source == "unpaywall":
              task_map[source] = search_unpaywall(query, max_results_per_source)
          elif source == "ieee":
              if ieee_searcher is not None:
                  task_map[source] = async_search(ieee_searcher, query, max_results_per_source)
          elif source == "acm":
              if acm_searcher is not None:
                  task_map[source] = async_search(acm_searcher, query, max_results_per_source)
  
      source_names = list(task_map.keys())
      source_outputs = await asyncio.gather(*task_map.values(), return_exceptions=True)