ccc_rank_aggregate

Instructions

Implementation Reference

• src/scmcp/tool/ccc.py:109-141 (handler)

  Handler function run_ccc_func that executes ccc_rank_aggregate by dispatching to li.mt.rank_aggregate(adata, **filtered_kwargs) in the 'else' branch.

  def run_ccc_func(ads, func, arguments):
      
      if func not in ccc_func:
          raise ValueError(f"不支持的函数: {func}")
      run_func = ccc_func[func]
      adata = ads.adata_dic[ads.active]
      try:
          logger.info(f"Running function {func} with arguments {arguments}")
          
          if func == "ls_ccc_method":
              res = run_func()
          elif func == "ccc":
              # Extract method from arguments and pass remaining args
              method = arguments.get("method", "cellphonedb")
              method_args = {k: v for k, v in arguments.items() if k != "method"}
              res = run_func(adata, method, **method_args)
              
          elif "plot" in func:
              from ..util import savefig
              ax = run_func(adata, **arguments)
              fig_path = Path(os.getcwd()) / f"figures/{func}.png"
              res = savefig(ax, fig_path, format="png")
              add_op_log(adata, run_func, arguments)  # 
          else:   
              parameters = inspect.signature(run_func).parameters
              kwargs = {k: arguments.get(k) for k in parameters if k in arguments}
              res = run_func(adata, **kwargs)
              add_op_log(adata, run_func, kwargs)
          return res
      except Exception as e:
          logger.error(f"Error running function {func}: {e}")
          raise e

• src/scmcp/schema/ccc.py:11-73 (schema)

  Pydantic model defining the input schema for the ccc_rank_aggregate tool, used via model_json_schema().

  class RankAggregateModel(JSONParsingModel):
      """Input schema for LIANA's rank_aggregate method for cell-cell communication analysis."""
      
      groupby: str = Field(
          ...,  # Required field
          description="Key to be used for grouping or clustering cells (e.g., cell type annotations)."
      )
      
      resource_name: str = Field(
          default="consensus",
          description="Name of the resource to be used for ligand-receptor inference. See `li.rs.show_resources()` for available resources."
      )
      
      expr_prop: float = Field(
          default=0.1,
          description="Minimum expression proportion for the ligands and receptors in the corresponding cell identities. Set to 0 to return unfiltered results."
      )
      
      min_cells: int = Field(
          default=5,
          description="Minimum cells per cell identity to be considered for downstream analysis."
      )
      
      base: float = Field(
          default=2.718281828459045,  # e
          description="Exponent base used to reverse the log-transformation of the matrix. Relevant only for the `logfc` method."
      )
      
      aggregate_method: Literal["rra", "mean"] = Field(
          default="rra",
          description="Method aggregation approach: 'mean' for mean rank, 'rra' for RobustRankAggregate."
      )
      
      return_all_lrs: bool = Field(
          default=False,
          description="Whether to return all ligand-receptor pairs, or only those that surpass the expr_prop threshold."
      )
      
      key_added: str = Field(
          default="liana_res",
          description="Key under which the results will be stored in adata.uns."
      )
      
      use_raw: Optional[bool] = Field(
          default=True,
          description="Use raw attribute of adata if present."
      )
      
      layer: Optional[str] = Field(
          default=None,
          description="Layer in AnnData.layers to use. If None, use AnnData.X."
      )
      
      de_method: str = Field(
          default="t-test",
          description="Differential expression method used to rank genes according to 1vsRest."
      )
      
      n_perms: int = Field(
          default=1000,
          description="Number of permutations for permutation-based methods. If None, no permutation testing is performed."
      )

• src/scmcp/tool/ccc.py:22-26 (registration)

  Creates the MCP Tool object for ccc_rank_aggregate with name, description, and input schema.

  rank_aggregate_tool = types.Tool(
      name="ccc_rank_aggregate",
      description="Get an aggregate of ligand-receptor scores from multiple  Cell-cell communication methods. ",
      inputSchema=RankAggregateModel.model_json_schema(),
  )

• src/scmcp/tool/ccc.py:92-98 (helper)

  Maps tool name 'ccc_rank_aggregate' to the actual function li.mt.rank_aggregate for execution.

  ccc_func = {
      "ls_ccc_method": ls_ccc_method,
      "ccc_rank_aggregate": li.mt.rank_aggregate,
      "ccc_circle_plot": plot_circleplot,
      "ccc_dot_plot": plot_dotplot,
      "ccc": run_ccc,
  }

• src/scmcp/tool/ccc.py:100-106 (registration)

  Registers the rank_aggregate_tool in the ccc_tools dictionary, which is used by server.list_tools() and call_tool() dispatch.

  ccc_tools = {
      "ls_ccc_method": ls_ccc_method_tool,
      "ccc_rank_aggregate": rank_aggregate_tool,
      "ccc_circle_plot": circle_plot_tool,
      "ccc_dot_plot": dot_plot_tool,
      "ccc": ccc_tool,
  }