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)