cBioPortal MCP Server

Instructions

Implementation Reference

• cbioportal_mcp/endpoints/genes.py:216-322 (handler)

  Core handler function implementing the logic to retrieve mutations in a specific gene for a study and sample list. Fetches the appropriate molecular profile, makes API request to mutations endpoint, handles pagination, limits, sorting, and errors.

  @handle_api_errors("get mutations in gene")
  async def get_mutations_in_gene(
      self,
      gene_id: str,
      study_id: str,
      sample_list_id: str,
      page_number: int = 0,
      page_size: int = 50,
      sort_by: Optional[str] = None,
      direction: str = "ASC",
      limit: Optional[int] = None,
  ) -> Dict:
      """
      Get mutations in a specific gene for a given study and sample list, with pagination support.
      Uses the /molecular-profiles/{molecularProfileId}/mutations endpoint with GET and query parameters.
      The molecularProfileId is dynamically determined based on the studyId.
      """
      try:
          molecular_profiles_response = await self.api_client.make_api_request(
              f"studies/{study_id}/molecular-profiles"
          )
          if (
              isinstance(molecular_profiles_response, dict)
              and "api_error" in molecular_profiles_response
          ):
              return {
                  "error": f"Failed to fetch molecular profiles for study {study_id} to find mutation profile",
                  "details": molecular_profiles_response,
              }
  
          mutation_profile_id = None
          if isinstance(molecular_profiles_response, list):
              for profile in molecular_profiles_response:
                  if profile.get("molecularAlterationType") == "MUTATION_EXTENDED":
                      mutation_profile_id = profile.get("molecularProfileId")
                      break
  
          if not mutation_profile_id:
              return {
                  "error": f"No MUTATION_EXTENDED molecular profile found for study {study_id}"
              }
  
          api_call_params = {
              "studyId": study_id,
              "sampleListId": sample_list_id,
              "pageNumber": page_number,
              "pageSize": page_size,
              "direction": direction,
          }
          if sort_by:
              api_call_params["sortBy"] = sort_by
          if limit == 0:
              api_call_params["pageSize"] = FETCH_ALL_PAGE_SIZE
  
          if str(gene_id).isdigit():
              api_call_params["entrezGeneId"] = gene_id
          else:
              api_call_params["hugoGeneSymbol"] = gene_id
  
          endpoint = f"molecular-profiles/{mutation_profile_id}/mutations"
          mutations_from_api = await self.api_client.make_api_request(
              endpoint, method="GET", params=api_call_params
          )
  
          if (
              isinstance(mutations_from_api, dict)
              and "api_error" in mutations_from_api
          ):
              return {
                  "error": "API error fetching mutations",
                  "details": mutations_from_api,
                  "request_params": api_call_params,
              }
          if not isinstance(mutations_from_api, list):
              return {
                  "error": "Unexpected API response type for mutations (expected list)",
                  "details": mutations_from_api,
                  "request_params": api_call_params,
              }
  
          api_might_have_more = len(mutations_from_api) == api_call_params["pageSize"]
          if (
              api_call_params["pageSize"] == FETCH_ALL_PAGE_SIZE
              and len(mutations_from_api) < FETCH_ALL_PAGE_SIZE
          ):
              api_might_have_more = False
  
          mutations_for_response = mutations_from_api
          if limit and limit > 0 and len(mutations_from_api) > limit:
              mutations_for_response = mutations_from_api[:limit]
  
          total_items_in_response = len(mutations_for_response)
  
          return {
              "mutations": mutations_for_response,
              "pagination": {
                  "page": page_number,
                  "page_size": page_size,
                  "total_found": total_items_in_response,
                  "has_more": api_might_have_more,
              },
          }
      except Exception as e:
          return {
              "error": f"An unexpected error occurred in get_mutations_in_gene: {str(e)}"
          }

• cbioportal_mcp/server.py:96-131 (registration)

  Tool registration in _register_tools method where 'get_mutations_in_gene' is listed in tool_methods and registered via self.mcp.add_tool(self.get_mutations_in_gene).

  def _register_tools(self):
      """Register tool methods as MCP tools."""
      # List of methods to register as tools (explicitly defined)
      tool_methods = [
          # Pagination utilities
          "paginate_results",
          "collect_all_results",
          # Studies endpoints
          "get_cancer_studies",
          "get_cancer_types",
          "search_studies",
          "get_study_details",
          "get_multiple_studies",
          # Genes endpoints
          "search_genes",
          "get_genes",
          "get_multiple_genes",
          "get_mutations_in_gene",
          # Samples endpoints
          "get_samples_in_study",
          "get_sample_list_id",
          # Molecular profiles endpoints
          "get_molecular_profiles",
          "get_clinical_data",
          "get_gene_panels_for_study",
          "get_gene_panel_details",
      ]
  
      for method_name in tool_methods:
          if hasattr(self, method_name):
              method = getattr(self, method_name)
              self.mcp.add_tool(method)
              logger.debug(f"Registered tool: {method_name}")
          else:
              logger.warning(f"Method {method_name} not found for tool registration")

• cbioportal_mcp/server.py:249-270 (handler)

  Wrapper handler method on the main server class that delegates to the GenesEndpoints implementation. This is the method directly registered as the MCP tool.

  async def get_mutations_in_gene(
      self,
      gene_id: str,
      study_id: str,
      sample_list_id: str,
      page_number: int = 0,
      page_size: int = 50,
      sort_by: Optional[str] = None,
      direction: str = "ASC",
      limit: Optional[int] = None,
  ) -> Dict:
      """Get mutations in a specific gene for a given study and sample list, with pagination support."""
      return await self.genes.get_mutations_in_gene(
          gene_id,
          study_id,
          sample_list_id,
          page_number,
          page_size,
          sort_by,
          direction,
          limit,
      )