PubMed MCP Server

Instructions

Implementation Reference

• src/tool_handler.py:220-259 (handler)

  Handler function _handle_search_by_author that parses arguments, calls PubMedClient.search_by_author, formats and returns MCPResponse with search results.
  async def _handle_search_by_author(self, arguments: Dict[str, Any]) -> MCPResponse: """Handle author-based search.""" try: author_name = arguments.get("author_name", "") if not author_name: return MCPResponse( content=[{"type": "text", "text": "Author name is required"}], is_error=True ) max_results = arguments.get("max_results", 20) include_coauthors = arguments.get("include_coauthors", True) search_result = await self.pubmed_client.search_by_author( author_name=author_name, max_results=max_results, include_coauthors=include_coauthors, cache=self.cache, ) content = [] content.append( { "type": "text", "text": f"**Publications by {author_name}**\n\n" f"Total Articles: {search_result.total_results}\n" f"Showing: {search_result.returned_results}\n", } ) 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}) return MCPResponse(content=content) except Exception as e: logger.error(f"Error in search_by_author: {e}") return MCPResponse( content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True )
• src/tools.py:120-142 (schema)

  Input schema definition for the search_by_author tool, specifying parameters like author_name (required), max_results, and include_coauthors.
  { "name": "search_by_author", "description": "Search for articles by a specific author", "inputSchema": { "type": "object", "properties": { "author_name": {"type": "string", "description": "Author name to search for"}, "max_results": { "type": "integer", "minimum": 1, "maximum": 100, "default": 20, "description": "Maximum number of results", }, "include_coauthors": { "type": "boolean", "default": True, "description": "Include co-author information", }, }, "required": ["author_name"], }, },
• src/tool_handler.py:65-78 (registration)

  Registration of search_by_author tool in the handler_map dictionary, mapping it to the _handle_search_by_author method.
  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:258-316 (helper)

  PubMedClient.search_by_author method that performs the actual PubMed API search using author query, fetches details, handles caching, and returns SearchResult.
  async def search_by_author( self, author_name: str, max_results: int = 20, include_coauthors: bool = True, cache: Optional[CacheManager] = None, ) -> SearchResult: """Search articles by author name.""" start_time = time.time() if cache: cache_key = cache.generate_key( "author_search", author=author_name, max_results=max_results, include_coauthors=include_coauthors, ) 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) # Build author search query search_query = f'"{author_name}"[Author]' search_params = self._build_params( db="pubmed", term=search_query, retmax=str(max_results), retmode="json", sort="pub_date" ) 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)) articles = [] if pmids: articles = await self._fetch_article_details(pmids, include_full_details=True) result_data = { "query": f"Author: {author_name}", "total_results": total_results, "returned_results": len(articles), "articles": articles, # Store Article objects directly "search_time": time.time() - start_time, "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)