dpt
Analyze single-cell RNA sequencing data to compute diffusion pseudotime, revealing cellular trajectories and branching patterns in development or disease progression.
Instructions
Diffusion Pseudotime (DPT) analysis
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| n_dcs | No | The number of diffusion components to use. | |
| n_branchings | No | Number of branchings to detect. | |
| min_group_size | No | 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. | |
| allow_kendall_tau_shift | No | If a very small branch is detected upon splitting, shift away from maximum correlation in Kendall tau criterion to stabilize the splitting. | |
| neighbors_key | No | If specified, dpt looks .uns[neighbors_key] for neighbors settings and uses the corresponding connectivities and distances. |
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, }