PubMed MCP Server

Instructions

Implementation Reference

• src/tool_handler.py:96-178 (handler)

  The main handler function _handle_search_pubmed that processes the tool call arguments, performs validation, calls PubMedClient.search_articles with the parameters, formats the search results into an MCPResponse with summaries and article details.
  async def _handle_search_pubmed(self, arguments: Dict[str, Any]) -> MCPResponse: """Handle PubMed search with advanced filtering.""" try: # Parse arguments query = arguments.get("query", "") if not query: return MCPResponse( content=[{"type": "text", "text": "Query parameter is required"}], is_error=True ) max_results = arguments.get("max_results", 20) # Handle negative max_results if max_results < 0: max_results = 0 sort_order = SortOrder(arguments.get("sort_order", "relevance")) date_from = arguments.get("date_from") date_to = arguments.get("date_to") date_range = ( DateRange(arguments.get("date_range")) if arguments.get("date_range") else None ) # Parse article types article_types = None if arguments.get("article_types"): article_types = [ArticleType(at) for at in arguments["article_types"]] # Perform search search_result = await self.pubmed_client.search_articles( query=query, max_results=max_results, sort_order=sort_order, date_from=date_from, date_to=date_to, date_range=date_range, article_types=article_types, authors=arguments.get("authors"), journals=arguments.get("journals"), mesh_terms=arguments.get("mesh_terms"), language=arguments.get("language"), has_abstract=arguments.get("has_abstract"), has_full_text=arguments.get("has_full_text"), humans_only=arguments.get("humans_only"), cache=self.cache, ) # Format response content = [] # Summary content.append( { "type": "text", "text": f"**PubMed Search Results**\n\n" f"Query: {query}\n" f"Total Results: {search_result.total_results:,}\n" f"Returned: {search_result.returned_results}\n" f"Search Time: {search_result.search_time:.2f}s\n", } ) # Articles if search_result.articles: 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}) else: content.append({"type": "text", "text": "No articles found for this query."}) return MCPResponse( content=content, metadata={ "total_results": search_result.total_results, "search_time": search_result.search_time, }, ) except Exception as e: logger.error(f"Error in search_pubmed: {e}") return MCPResponse( content=[{"type": "text", "text": f"Search error: {str(e)}"}], is_error=True )
• src/tools.py:11-93 (schema)

  The schema definition for the 'search_pubmed' tool, including inputSchema with all parameters, types, descriptions, and required fields. Part of the TOOL_DEFINITIONS list used by get_tools().
  { "name": "search_pubmed", "description": ("Search PubMed for articles with advanced filtering options"), "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query using PubMed syntax"}, "max_results": { "type": "integer", "minimum": 1, "maximum": 200, "default": 20, "description": "Maximum number of results to return", }, "sort_order": { "type": "string", "enum": ["relevance", "pub_date", "author", "journal", "title"], "default": "relevance", "description": "Sort order for results", }, "date_from": { "type": "string", "description": "Start date (YYYY/MM/DD, YYYY/MM, or YYYY)", }, "date_to": { "type": "string", "description": "End date (YYYY/MM/DD, YYYY/MM, or YYYY)", }, "date_range": { "type": "string", "enum": ["1y", "5y", "10y", "all"], "description": "Predefined date range", }, "article_types": { "type": "array", "items": { "type": "string", "enum": [ "Journal Article", "Review", "Systematic Review", "Meta-Analysis", "Clinical Trial", "Randomized Controlled Trial", "Case Reports", "Letter", "Editorial", "Comment", ], }, "description": "Filter by article types", }, "authors": { "type": "array", "items": {"type": "string"}, "description": "Filter by author names", }, "journals": { "type": "array", "items": {"type": "string"}, "description": "Filter by journal names", }, "mesh_terms": { "type": "array", "items": {"type": "string"}, "description": "Filter by MeSH terms", }, "language": { "type": "string", "description": ("Language filter (e.g., 'eng', 'fre', 'ger')"), }, "has_abstract": { "type": "boolean", "description": "Only include articles with abstracts", }, "has_full_text": { "type": "boolean", "description": ("Only include articles with full text available"), }, "humans_only": {"type": "boolean", "description": "Only include human studies"}, }, "required": ["query"], },
• src/tool_handler.py:65-78 (registration)

  The handler_map dictionary in handle_tool_call method that registers and routes 'search_pubmed' tool calls to the _handle_search_pubmed function.
  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/pubmed_client.py:71-206 (helper)

  Core helper method PubMedClient.search_articles that executes the PubMed API search using ESearch, fetches details with EFetch, handles caching, and returns a SearchResult. Directly called by the tool handler.
  async def search_articles( self, query: str, max_results: int = 20, sort_order: SortOrder = SortOrder.RELEVANCE, date_from: Optional[str] = None, date_to: Optional[str] = None, date_range: Optional[DateRange] = None, article_types: Optional[List[ArticleType]] = None, authors: Optional[List[str]] = None, journals: Optional[List[str]] = None, mesh_terms: Optional[List[str]] = None, language: Optional[str] = None, has_abstract: Optional[bool] = None, has_full_text: Optional[bool] = None, humans_only: Optional[bool] = None, cache: Optional[CacheManager] = None, ) -> SearchResult: """ Search PubMed with advanced filtering. Args: query: Search query max_results: Maximum results to return sort_order: Sort order for results date_from: Start date filter date_to: End date filter date_range: Predefined date range article_types: Article type filters authors: Author filters journals: Journal filters mesh_terms: MeSH term filters language: Language filter has_abstract: Only articles with abstracts has_full_text: Only articles with full text humans_only: Only human studies cache: Cache manager instance Returns: SearchResult containing articles and metadata """ start_time = time.time() # Check cache first if cache: cache_key = cache.generate_key( "search", query=query, max_results=max_results, sort_order=sort_order.value, date_from=date_from, date_to=date_to, date_range=date_range.value if date_range else None, article_types=[at.value for at in article_types] if article_types else None, authors=authors, journals=journals, mesh_terms=mesh_terms, language=language, has_abstract=has_abstract, has_full_text=has_full_text, humans_only=humans_only, ) 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) # Handle date range shortcuts if date_range and not (date_from or date_to): date_to = datetime.now().strftime("%Y/%m/%d") if date_range == DateRange.LAST_YEAR: date_from = (datetime.now() - timedelta(days=365)).strftime("%Y/%m/%d") elif date_range == DateRange.LAST_5_YEARS: date_from = (datetime.now() - timedelta(days=365 * 5)).strftime("%Y/%m/%d") elif date_range == DateRange.LAST_10_YEARS: date_from = (datetime.now() - timedelta(days=365 * 10)).strftime("%Y/%m/%d") # Build complex search query search_query = build_search_query( query, authors=authors, journals=journals, mesh_terms=mesh_terms, article_types=([at.value for at in article_types] if article_types else None), date_from=date_from, date_to=date_to, language=language, has_abstract=has_abstract, has_full_text=has_full_text, humans_only=humans_only, ) logger.info(f"Executing PubMed search: {search_query}") # Search for article IDs search_params = self._build_params( db="pubmed", term=search_query, retmax=str(max_results), retmode="json", sort=sort_order.value, ) search_response = await self._make_request("esearch.fcgi", search_params) search_data = search_response.json() search_result = search_data.get("esearchresult", {}) pmids = search_result.get("idlist", []) total_results = int(search_result.get("count", 0)) # Get detailed article information articles = [] if pmids: articles = await self._fetch_article_details(pmids, include_full_details=True) # Build result result_data = { "query": query, "total_results": total_results, "returned_results": len(articles), "articles": articles, # Store Article objects directly "search_time": time.time() - start_time, "suggestions": [], # Could implement spelling suggestions } # Cache the result (store as dicts for serialization) if cache: cache_data = {**result_data, "articles": [article.model_dump() for article in articles]} cache.set(cache_key, cache_data) return SearchResult(**result_data)