Paper Search MCP

Instructions

Implementation Reference

• paper_search_mcp/server.py:666-691 (handler)

  The handler function `search_crossref` which defines the MCP tool interface and calls the underlying `async_search` helper.

  async def search_crossref(
      query: str,
      max_results: int = 10,
      filter: Optional[str] = None,
      sort: Optional[str] = None,
      order: Optional[str] = None,
  ) -> List[Dict]:
      """Search academic papers from CrossRef database.
      
      CrossRef is a scholarly infrastructure organization that provides 
      persistent identifiers (DOIs) for scholarly content and metadata.
      It's one of the largest citation databases covering millions of 
      academic papers, journals, books, and other scholarly content.
  
      Args:
          query: Search query string (e.g., 'machine learning', 'climate change').
          max_results: Maximum number of papers to return (default: 10, max: 1000).
          filter: CrossRef filter string (e.g., 'has-full-text:true,from-pub-date:2020').
          sort: Sort field ('relevance', 'published', 'updated', 'deposited', etc.).
          order: Sort order ('asc' or 'desc').
      Returns:
          List of paper metadata in dictionary format.
      """
      extra = {k: v for k, v in {'filter': filter, 'sort': sort, 'order': order}.items() if v is not None}
      papers = await async_search(crossref_searcher, query, max_results, **extra)
      return papers if papers else []

• paper_search_mcp/server.py:69-76 (helper)

  The `async_search` helper function that bridges the asynchronous MCP tool handler and the synchronous searcher implementation.

  async def async_search(searcher, query: str, max_results: int, **kwargs) -> List[Dict]:
      if 'year' in kwargs:
          papers = await asyncio.to_thread(searcher.search, query, max_results=max_results, year=kwargs['year'])
      elif kwargs:
          papers = await asyncio.to_thread(searcher.search, query, max_results=max_results, **kwargs)
      else:
          papers = await asyncio.to_thread(searcher.search, query, max_results=max_results)
      return [paper.to_dict() for paper in papers]

• paper_search_mcp/academic_platforms/crossref.py:28-64 (helper)

  The `CrossRefSearcher.search` method which performs the actual synchronous API request to the CrossRef database.

  def search(self, query: str, max_results: int = 10, **kwargs) -> List[Paper]:
      """
      Search CrossRef database for papers.
      
      Args:
          query: Search query string
          max_results: Maximum number of results to return (default: 10)
          **kwargs: Additional parameters like filters, sort, etc.
          
      Returns:
          List of Paper objects
      """
      try:
          params = {
              'query': query,
              'rows': min(max_results, 1000),  # CrossRef API max is 1000
              'sort': 'relevance',
              'order': 'desc'
          }
          
          # Add any additional filters from kwargs
          if 'filter' in kwargs:
              params['filter'] = kwargs['filter']
          if 'sort' in kwargs:
              params['sort'] = kwargs['sort']
          if 'order' in kwargs:
              params['order'] = kwargs['order']
              
          # Add polite pool parameter
          params['mailto'] = 'paper-search@example.org'
          
          url = f"{self.BASE_URL}/works"
          response = self.session.get(url, params=params, timeout=30)
          
          if response.status_code == 429:
              # Rate limited - wait and retry once
              logger.warning("Rate limited by CrossRef API, waiting 2 seconds...")