NIH RePORTER MCP

Instructions

Search for publications linked to NIH projects

Args:
    pmids: Comma-separated list of PubMed IDs
    core_project_nums: Comma-separated list of NIH core project numbers
    limit: Maximum number of results to return (default: 10, max: 50)

Implementation Reference

• mcp-nih-reporter.py:503-548 (handler)

  Main handler for the 'search_publications' tool. Processes input parameters to build API criteria, fetches data using NIHReporterClient.get_publications, and formats output with format_publication_results. The @mcp.tool() decorator registers it with the MCP server.

  @mcp.tool()
  async def search_publications(
      pmids: Optional[str] = None,
      core_project_nums: Optional[str] = None,
      limit: Optional[int] = 10
  ) -> str:
      """
      Search for publications linked to NIH projects
      
      Args:
          pmids: Comma-separated list of PubMed IDs
          core_project_nums: Comma-separated list of NIH core project numbers
          limit: Maximum number of results to return (default: 10, max: 50)
      """
      try:
          logger.info(f"Publication search request received with parameters: {locals()}")
          
          criteria = {}
          
          # Handle PMIDs
          if pmids:
              pmid_list = [pmid.strip() for pmid in pmids.split(",")]
              criteria["pmids"] = pmid_list
          
          # Handle core project numbers
          if core_project_nums:
              logger.info(f"Processing core_project_nums input: {core_project_nums}")
              # Clean the input string of any quotes
              clean_input = core_project_nums.strip().strip('"').strip("'")
              logger.info(f"Cleaned input: {clean_input}")
              proj_list = [num.strip() for num in clean_input.split(",")]
              logger.info(f"Created project list: {proj_list}")
              criteria["core_project_nums"] = proj_list
          
          # Ensure limit is within bounds
          criteria["limit"] = min(max(1, limit), 50)
          
          logger.info(f"Constructed publication search criteria: {json.dumps(criteria, indent=2)}")
          
          results = await api_client.get_publications({"criteria": criteria})
          return api_client.format_publication_results(results)
          
      except Exception as e:
          logger.error(f"Publication search failed: {str(e)}", exc_info=True)
          return f"Publication search failed: {str(e)}\nPlease check the logs for more details."

• mcp-nih-reporter.py:215-278 (helper)

  Core helper method in NIHReporterClient class that performs the actual API call to NIH RePORTER publications/search endpoint, optionally fetches additional PubMed details for PMIDs, and returns structured data.

  async def get_publications(self, criteria: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """Get publications from NIH RePORTER API"""
      logger.info(f"Fetching publications from NIH RePORTER with criteria: {criteria}")
      async with httpx.AsyncClient() as client:
          # Construct the payload according to API specification
          payload = {
              "criteria": criteria.get("criteria", {}),
              "limit": criteria.get("limit", 50),
              "offset": criteria.get("offset", 0),
              "sort_field": criteria.get("sort_field", "core_project_nums"),
              "sort_order": criteria.get("sort_order", "desc")
          }
          
          # Add publication years if specified
          if "publication_years" in criteria.get("criteria", {}):
              payload["criteria"]["publication_years"] = criteria["criteria"]["publication_years"]
          
          logger.debug(f"Sending payload to NIH Publications API: {json.dumps(payload, indent=2)}")
          
          try:
              response = await client.post(
                  f"{API_BASE}/publications/search",
                  headers=self.headers,
                  json=payload
              )
              response.raise_for_status()
              response_data = response.json()
              
              # If we got PMIDs, fetch the full publication details from PubMed
              if response_data.get("results"):
                  pmids = [str(result.get("pmid")) for result in response_data["results"] if result.get("pmid")]
                  if pmids:
                      async with httpx.AsyncClient() as pubmed_client:
                          # Use E-utilities to get full publication details
                          pubmed_url = f"https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi?db=pubmed&id={','.join(pmids)}&retmode=json"
                          pubmed_response = await pubmed_client.get(pubmed_url)
                          pubmed_data = pubmed_response.json()
                          
                          # Update our results with PubMed data
                          for result in response_data["results"]:
                              if result.get("pmid"):
                                  pmid = str(result["pmid"])
                                  if pmid in pubmed_data.get("result", {}):
                                      pub_details = pubmed_data["result"][pmid]
                                      result.update({
                                          "title": pub_details.get("title", ""),
                                          "authors": [author.get("name", "") for author in pub_details.get("authors", [])],
                                          "journal_title": pub_details.get("fulljournalname", ""),
                                          "publication_year": pub_details.get("pubdate", "").split()[0] if pub_details.get("pubdate") else None
                                      })
              
              logger.debug(f"Received response: {json.dumps(response_data, indent=2)}")
              return response_data
          except httpx.HTTPStatusError as e:
              logger.error(f"HTTP error occurred: {e.response.status_code} - {e.response.text}")
              raise
          except json.JSONDecodeError as e:
              logger.error(f"Failed to parse API response: {e}")
              logger.error(f"Raw response: {response.text}")
              raise
          except Exception as e:
              logger.error(f"Unexpected error during API call: {str(e)}")
              raise

• mcp-nih-reporter.py:279-332 (helper)

  Helper method that converts raw publication API results into user-friendly Markdown formatted string, including links to PubMed and DOIs.

  def format_publication_results(self, results: Dict[str, Any], include_projects: bool = False) -> str:
      """Format publication results into markdown string with optional project links"""
      logger.debug(f"Formatting publication results: {json.dumps(results, indent=2)}")
      
      if not results.get("results"):
          logger.info("No publications found in API response")
          return "No publications found."
      
      try:
          formatted_results = []
          for pub in results["results"]:
              # Format authors safely
              authors = pub.get('authors', [])
              author_str = ", ".join(authors) if authors else "N/A"
              
              pub_info = [
                  f"### {pub.get('title', 'Untitled Publication')}",
                  "",
                  f"**Authors:** {author_str}",
                  f"**PMID:** `{pub.get('pmid', 'N/A')}`",
                  f"**Core Project Number:** `{pub.get('core_project_num', 'N/A')}`"
              ]
              
              # Add publication year if available
              if pub.get('publication_year'):
                  pub_info.append(f"**Publication Year:** {pub['publication_year']}")
              
              # Add journal info if available
              if pub.get('journal_title'):
                  pub_info.append(f"**Journal:** {pub['journal_title']}")
              
              # Add DOI if available
              if pub.get('doi'):
                  pub_info.append(f"**DOI:** [{pub['doi']}](https://doi.org/{pub['doi']})")
              
              # Add project links if available
              if pub.get('core_project_num'):
                  pub_info.extend([
                      "",
                      "#### Related NIH Projects",
                      f"- Core Project: `{pub['core_project_num']}`"
                  ])
              
              pub_info.extend(["", "---", ""])
              formatted_results.append("\n".join(filter(None, pub_info)))
          
          total = f"# NIH RePORTER Publication Results\n\n**Total matching publications:** {results.get('meta', {}).get('total', 0)}"
          return f"{total}\n\n" + "\n".join(formatted_results)
          
      except Exception as e:
          logger.error(f"Error formatting publication results: {str(e)}")
          logger.error(f"Results that caused error: {json.dumps(results, indent=2)}")
          raise

• mcp-nih-reporter.py:509-516 (schema)

  Input schema defined in the tool's docstring, describing parameters and their types/usage. Function signature provides type hints (Optional[str], Optional[int]).

  """
  Search for publications linked to NIH projects
  
  Args:
      pmids: Comma-separated list of PubMed IDs
      core_project_nums: Comma-separated list of NIH core project numbers
      limit: Maximum number of results to return (default: 10, max: 50)
  """

• mcp-nih-reporter.py:503-503 (registration)

  The @mcp.tool() decorator registers the search_publications function as an MCP tool.

  @mcp.tool()