dpt

Instructions

Implementation Reference

• src/scmcp/tool/tl.py:164-177 (handler)

  Handler function that executes all tl tools, including 'dpt', by dispatching to scanpy.tl.dpt with validated arguments from the input schema.

  def run_tl_func(ads, func, arguments):
      adata = ads.adata_dic[ads.active]
      if func not in tl_func:
          raise ValueError(f"Unsupported function: {func}")
      run_func = tl_func[func]
      parameters = inspect.signature(run_func).parameters
      kwargs = {k: arguments.get(k) for k in parameters if k in arguments}    
      try:
          res = run_func(adata, **kwargs)
          add_op_log(adata, run_func, kwargs)
      except Exception as e:
          logger.error(f"Error running function {func}: {e}")
          raise
      return 

• src/scmcp/schema/tl.py:481-529 (schema)

  Pydantic model defining the input schema and validation for the 'dpt' tool.

  class DPTModel(JSONParsingModel):
      """Input schema for the Diffusion Pseudotime (DPT) tool."""
      
      n_dcs: int = Field(
          default=10,
          description="The number of diffusion components to use.",
          gt=0
      )
      n_branchings: int = Field(
          default=0,
          description="Number of branchings to detect.",
          ge=0
      )
      min_group_size: float = Field(
          default=0.01,
          description="During recursive splitting of branches, do not consider groups that contain less than min_group_size data points. If a float, refers to a fraction of the total number of data points.",
          gt=0,
          le=1.0
      )
      allow_kendall_tau_shift: bool = Field(
          default=True,
          description="If a very small branch is detected upon splitting, shift away from maximum correlation in Kendall tau criterion to stabilize the splitting."
      )
      neighbors_key: Optional[str] = Field(
          default=None,
          description="If specified, dpt looks .uns[neighbors_key] for neighbors settings and uses the corresponding connectivities and distances."
      )
      
      @field_validator('n_dcs')
      def validate_n_dcs(cls, v: int) -> int:
          """Validate n_dcs is positive"""
          if v <= 0:
              raise ValueError("n_dcs must be a positive integer")
          return v
      
      @field_validator('n_branchings')
      def validate_n_branchings(cls, v: int) -> int:
          """Validate n_branchings is non-negative"""
          if v < 0:
              raise ValueError("n_branchings must be a non-negative integer")
          return v
      
      @field_validator('min_group_size')
      def validate_min_group_size(cls, v: float) -> float:
          """Validate min_group_size is between 0 and 1"""
          if v <= 0 or v > 1:
              raise ValueError("min_group_size must be between 0 and 1")
          return v

• src/scmcp/tool/tl.py:68-73 (registration)

  Creates and defines the MCP Tool object for 'dpt', including name, description, and input schema reference.

  # Add dpt tool
  dpt_tool = types.Tool(
      name="dpt",
      description="Diffusion Pseudotime (DPT) analysis",
      inputSchema=DPTModel.model_json_schema(),
  )

• src/scmcp/server.py:35-56 (registration)

  MCP server registration handler that lists tools, including 'dpt' from tl_tools when MODULE=='tl' or 'all'.

  @server.list_tools()
  async def list_tools() -> list[types.Tool]:
      if MODULE == "io":
          tools = io_tools.values()
      elif MODULE == "pp":
          tools = pp_tools.values()
      elif MODULE == "tl":
          tools = tl_tools.values()
      elif MODULE == "pl":
          tools = pl_tools.values()
      elif MODULE == "util":
          tools = util_tools.values()
      else:
          tools = [
              *io_tools.values(),
              *pp_tools.values(),
              *tl_tools.values(),
              *pl_tools.values(),
              *util_tools.values(),
              *ccc_tools.values(),
          ]
      return tools

• src/scmcp/tool/tl.py:125-142 (helper)

  Helper dictionary that maps the 'dpt' tool name to the underlying scanpy.tl.dpt function for execution.

  tl_func = {
      "tsne": sc.tl.tsne,
      "umap": sc.tl.umap,
      "draw_graph": sc.tl.draw_graph,
      "diffmap": sc.tl.diffmap,
      "embedding_density": sc.tl.embedding_density,
      "leiden": sc.tl.leiden,
      "louvain": sc.tl.louvain,
      "dendrogram": sc.tl.dendrogram,
      "dpt": sc.tl.dpt,
      "paga": sc.tl.paga,
      "ingest": sc.tl.ingest,
      "rank_genes_groups": sc.tl.rank_genes_groups,
      "filter_rank_genes_groups": sc.tl.filter_rank_genes_groups,
      "marker_gene_overlap": sc.tl.marker_gene_overlap,
      "score_genes": sc.tl.score_genes,
      "score_genes_cell_cycle": sc.tl.score_genes_cell_cycle,
  }