ResearchMCP

Instructions

Implementation Reference

• src/server.py:156-207 (handler)

  The primary handler function for the 'get_citations' tool. Decorated with @mcp.tool() for automatic registration and schema inference in FastMCP. Fetches paper details, retrieves citing papers and/or references via PaperFetcher, formats them into a markdown report.

  @mcp.tool()
  def get_citations(paper_id: str, direction: str = "both", max_results: int = 10) -> str:
      """
      Get citation network for a paper - see what cites it and what it cites.
  
      Args:
          paper_id: The OpenAlex paper ID
          direction: "cited_by" (papers citing this), "references" (papers this cites), or "both"
          max_results: Maximum number of citations to return per direction (default: 10)
  
      Returns:
          Citation network information with paper details
      """
      paper = fetcher.fetch_paper_by_id(paper_id)
  
      if "error" in paper:
          return paper["error"]
  
      result = f"**Citation Network for:** {paper['title']}\n"
      result += f"**Authors:** {paper['authors']}\n"
      result += f"**Year:** {paper['publication_year']}\n"
      result += f"**Total Citations:** {paper['cited_by_count']}\n\n"
  
      if direction in ["cited_by", "both"]:
          cited_by_papers = fetcher.get_cited_by_papers(paper_id, max_results)
          
          if cited_by_papers and "error" not in cited_by_papers[0]:
              result += f"**📈 Papers Citing This Work ({len(cited_by_papers)} shown):**\n\n"
              for i, citing_paper in enumerate(cited_by_papers, 1):
                  result += f"{i}. **{citing_paper['title']}**\n"
                  result += f"   Authors: {citing_paper['authors']}\n"
                  result += f"   Year: {citing_paper['publication_year']}\n"
                  result += f"   Citations: {citing_paper['cited_by_count']}\n"
                  result += f"   ID: {citing_paper['id']}\n\n"
          else:
              result += "No citing papers found or error fetching citations.\n\n"
  
      if direction in ["references", "both"]:
          references = fetcher.get_references(paper_id, max_results)
          
          if references and "error" not in references[0]:
              result += f"**📚 Papers This Work References ({len(references)} shown):**\n\n"
              for i, ref_paper in enumerate(references, 1):
                  result += f"{i}. **{ref_paper['title']}**\n"
                  result += f"   Authors: {ref_paper['authors']}\n"
                  result += f"   Year: {ref_paper['publication_year']}\n"
                  result += f"   Citations: {ref_paper['cited_by_count']}\n"
                  result += f"   ID: {ref_paper['id']}\n\n"
          else:
              result += "No references found or error fetching references.\n\n"
  
      return result

• src/tools/paper_fetcher.py:141-181 (helper)

  Helper method in PaperFetcher class that queries OpenAlex API for papers citing the given paper (forward citations), parses results using _parse_paper.

  def get_cited_by_papers(self, paper_id: str, max_results: int = 10) -> List[Dict]:
      """
      Get papers that cite this paper (forward citations).
  
      Args:
          paper_id: The OpenAlex paper ID
          max_results: Maximum number of citing papers to return
  
      Returns:
          List of papers that cite this paper
      """
      try:
          paper_id = paper_id.strip()
          if "openalex.org/" in paper_id:
              work_id = paper_id.split("/")[-1]
          else:
              work_id = paper_id
  
          url = f"{self.BASE_URL}/works"
          params = {
              "filter": f"cites:{work_id}",
              "per_page": max_results,
              "sort": "cited_by_count:desc"
          }
  
          response = self.client.get(url, params=params)
          
          if response.status_code != 200:
              return [{"error": f"Failed to fetch citations: Status {response.status_code}"}]
  
          data = response.json()
          papers = []
          for work in data.get("results", []):
              paper = self._parse_paper(work)
              papers.append(paper)
  
          return papers
  
      except Exception as e:
          return [{"error": f"Error fetching citations: {str(e)}"}]

• src/tools/paper_fetcher.py:182-222 (helper)

  Helper method in PaperFetcher class that queries OpenAlex API for papers referenced by the given paper (backward citations), parses results using _parse_paper.

  def get_references(self, paper_id: str, max_results: int = 10) -> List[Dict]:
      """
      Get papers that this paper references (backward citations).
  
      Args:
          paper_id: The OpenAlex paper ID
          max_results: Maximum number of references to return
  
      Returns:
          List of papers this paper references
      """
      try:
          paper_id = paper_id.strip()
          if "openalex.org/" in paper_id:
              work_id = paper_id.split("/")[-1]
          else:
              work_id = paper_id
  
          url = f"{self.BASE_URL}/works"
          params = {
              "filter": f"cited_by:{work_id}",
              "per_page": max_results,
              "sort": "cited_by_count:desc"
          }
  
          response = self.client.get(url, params=params)
          
          if response.status_code != 200:
              return [{"error": f"Failed to fetch references: Status {response.status_code}"}]
  
          data = response.json()
          papers = []
          for work in data.get("results", []):
              paper = self._parse_paper(work)
              papers.append(paper)
  
          return papers
  
      except Exception as e:
          return [{"error": f"Error fetching references: {str(e)}"}]

• src/tools/paper_fetcher.py:82-116 (helper)

  Helper method used by get_citations to fetch the main paper details by ID before getting citations.

  def fetch_paper_by_id(self, paper_id: str) -> Dict:
      """
      Fetch a specific paper's details by its OpenAlex ID.
  
      Args:
          paper_id: The OpenAlex paper ID (e.g., "https://openalex.org/W123456")
  
      Returns:
          Dictionary with paper details including abstract
      """
      try:
          paper_id = paper_id.strip()
  
          if "openalex.org/" in paper_id:
              work_id = paper_id.split("/")[-1]
          else:
              work_id = paper_id
  
          api_url = f"{self.BASE_URL}/works/{work_id}"
          response = self.client.get(api_url)
  
          if response.status_code != 200:
              return {"error": f"OpenAlex returned status {response.status_code}"}
  
          if not response.text:
              return {"error": "OpenAlex returned empty response"}
  
          work = response.json()
          return self._parse_paper(work)
  
      except httpx.HTTPError as e:
          return {"error": f"HTTP Error: {str(e)}"}
      except Exception as e:
          return {"error": f"Error: {type(e).__name__}: {str(e)}"}