cBioPortal MCP Server

Instructions

Implementation Reference

• cbioportal_mcp/server.py:240-248 (handler)

  MCP tool handler for get_multiple_genes, delegates to GenesEndpoints implementation.

  async def get_multiple_genes(
      self,
      gene_ids: List[str],
      gene_id_type: str = "ENTREZ_GENE_ID",
      projection: str = "SUMMARY",
  ) -> Dict:
      """Get information about multiple genes concurrently."""
      return await self.genes.get_multiple_genes(gene_ids, gene_id_type, projection)

• cbioportal_mcp/endpoints/genes.py:127-214 (helper)

  Core implementation of get_multiple_genes using concurrent batch fetches to the cBioPortal genes/fetch endpoint, with error handling, batching, and metadata.

  async def get_multiple_genes(
      self,
      gene_ids: List[str],
      gene_id_type: str = "ENTREZ_GENE_ID",
      projection: str = "SUMMARY",
  ) -> Dict:
      """
      Get information about multiple genes concurrently.
  
      This method uses concurrency to fetch multiple genes in parallel,
      which is much more efficient than sequential requests for large batches.
  
      Args:
          gene_ids: List of gene IDs (Entrez IDs or Hugo symbols)
          gene_id_type: Type of gene ID provided (ENTREZ_GENE_ID or HUGO_GENE_SYMBOL)
          projection: Level of detail to return (ID, SUMMARY, DETAILED)
  
      Returns:
          Dictionary with gene information and performance metadata
      """
      overall_start_time = time.time()  # Single start time for the whole operation
      if not gene_ids:
          return {
              "genes": {},
              "metadata": {
                  "count": 0,
                  "total_requested": 0,
                  "errors": 0,
                  "concurrent": True,
                  "batches": 0,
                  "execution_time": round(time.time() - overall_start_time, 3),
              },
          }
  
      # For large gene lists, break into smaller batches for API compatibility
      batch_size = (
          self.config.get("api.batch_size.genes", 100) if self.config
          else 100  # Default to 100 if no config provided
      )
      gene_batches = [
          gene_ids[i : i + batch_size] for i in range(0, len(gene_ids), batch_size)
      ]
  
      async def fetch_gene_batch(batch):
          try:
              params = {"geneIdType": gene_id_type, "projection": projection}
              batch_data = await self.api_client.make_api_request(
                  "genes/fetch", method="POST", params=params, json_data=batch
              )
              return {"data": batch_data, "success": True}
          except Exception as e:
              return {"error": str(e), "success": False}
  
      # Create tasks for all batches and run them concurrently
      tasks = [fetch_gene_batch(batch) for batch in gene_batches]
      batch_results = await asyncio.gather(*tasks)
  
      # Process results
      all_genes = []
      error_count = 0
  
      for result in batch_results:
          if result["success"]:
              all_genes.extend(result["data"])
          else:
              error_count += 1
  
      # Convert to dictionary for easier lookup
      genes_dict = {}
      key_field = (
          "hugoGeneSymbol" if gene_id_type == "HUGO_GENE_SYMBOL" else "entrezGeneId"
      )
      for gene in all_genes:
          gene_key_value = gene.get(key_field)
          if gene_key_value:
              genes_dict[str(gene_key_value)] = gene
  
      return {
          "genes": genes_dict,
          "metadata": {
              "count": len(genes_dict),
              "total_requested": len(gene_ids),
              "errors": error_count,
              "execution_time": round(time.time() - overall_start_time, 3),
              "concurrent": True,
              "batches": len(gene_batches),
          },
      }

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

  Tool registration method that includes 'get_multiple_genes' in the tool_methods list and registers it with FastMCP via self.mcp.add_tool.

  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")