""" MCP tools for bulk expression activity scoring and group comparison. Tools ----- dataset_score_bulk_samples Score pre-normalised bulk expression samples using PROGENy, CollecTRI, or Hallmark prior knowledge via decoupleR ULM/MLM/zscore. dataset_compare_activity_by_group Compare per-sample activity scores between two metadata-defined groups (Welch's t-test + BH FDR). Use after dataset_score_bulk_samples when the user asks whether TF/pathway/hallmark activity differs between groups. Design rules ------------ - No dataset-specific logic. Pass any samples × genes expression file. - Scoring logic lives in src/workflows/activity_scoring.py. - Comparison logic lives in src/workflows/activity_stats.py. - All return values are JSON-serialisable (DataFrames are saved, not returned). """ from __future__ import annotations import sys from datetime import datetime from pathlib import Path from typing import Annotated, Literal import pandas as pd from fastmcp import FastMCP _PROJECT_ROOT = Path(__file__).parent.parent.parent.resolve() if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) import os as _os from src.core.constants import DECOUPLER_DISCLAIMER_PER_SAMPLE from src.workflows.activity_scoring import ( plot_activity_landscape, score_bulk_samples_with_decoupler, ) from src.workflows.activity_stats import compare_activity_by_group from src.workflows.signatures import score_samples_against_signature def _resolve_output_dir() -> Path: """Return a writable output directory: env var > project default > /tmp fallback.""" env_val = _os.environ.get("RNA_OUTPUT_DIR") if env_val: p = Path(env_val) p.mkdir(parents=True, exist_ok=True) return p project_default = _PROJECT_ROOT / "tmp" / "outputs" try: project_default.mkdir(parents=True, exist_ok=True) return project_default except PermissionError: fallback = Path("/tmp/decoupleRpy/outputs") fallback.mkdir(parents=True, exist_ok=True) return fallback OUTPUT_DIR = _resolve_output_dir() _timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") bulk_dataset_mcp = FastMCP(name="bulk_dataset") # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _resolve_to_local_path(path_or_url: str) -> tuple[str, bool]: """ Resolve a local path or remote URL to a readable local file. Returns ``(local_path, is_temp)``. When ``is_temp`` is True the caller must delete ``local_path`` after reading. http(s)/ftp URLs (e.g. an h5ad hosted on the HuggingFace dataset repo) are streamed to a temp file, since scanpy's h5ad reader requires a local path. Local paths are returned unchanged — the caller is responsible for its own existence check. """ from src.core.data_io import resolve_to_local_path return resolve_to_local_path(path_or_url) def _load_expression_df(expression_path: str) -> pd.DataFrame: """ Load a samples × genes expression matrix from .h5ad, .csv, or .tsv. Accepts a local path or an http(s)/ftp URL (remote files are downloaded to a temp file first). For .h5ad files the dense matrix is relabelled to gene symbols when ``var.index`` holds Ensembl/probe IDs and a symbol column is available (see ``resolve_symbols_from_var``). Returns ``(dataframe, symbol_info)`` where ``symbol_info`` is the resolution report for h5ad inputs and ``None`` for CSV/TSV. """ local_path, is_temp = _resolve_to_local_path(expression_path) if not is_temp and not Path(local_path).exists(): raise FileNotFoundError(f"Expression file not found: {expression_path}") suffix = Path(expression_path.split("?", 1)[0]).suffix.lower() try: if suffix == ".h5ad": import scanpy as sc from src.workflows.microarray import resolve_symbols_from_var adata = sc.read_h5ad(local_path) df = adata.to_df() # An h5ad may be indexed by Ensembl IDs with the symbol parked in a # var column (gse205154_sears: var.index is versioned Ensembl, # var['SYMBOL'] holds the symbol). Activity scoring matches network # genes by NAME, so without this the tool dies with "zero gene # overlap" and the agent has to hand-roll the remap — which is # exactly what happened on the live prod run of this query. resolved = resolve_symbols_from_var(df, adata.var) resolved["obs"] = adata.obs.copy() return resolved["dataframe"], resolved elif suffix in (".tsv", ".txt"): return pd.read_csv(local_path, sep="\t", index_col=0), None else: return pd.read_csv(local_path, index_col=0), None finally: if is_temp: Path(local_path).unlink(missing_ok=True) # --------------------------------------------------------------------------- # MCP tool # --------------------------------------------------------------------------- @bulk_dataset_mcp.tool def dataset_score_bulk_samples( expression_path: Annotated[ str, "Path to a pre-normalised samples × genes expression file " "(.h5ad, .csv, or .tsv). Samples must be rows, gene symbols must " "be columns. Expression must already be normalised (log-CPM, " "log-intensity, log2-TPM, etc.). Do NOT pass raw integer counts — " "decoupleR activity estimation requires continuous normalised input.", ], resource: Annotated[ Literal["progeny", "collectri", "hallmark"], "Prior knowledge resource to score against: " "'progeny' — 14 cancer signalling pathway signatures (PROGENy), " "'collectri' — TF-target regulons from CollecTRI (~1 000 TFs), " "'hallmark' — 50 MSigDB Hallmark gene sets.", ] = "progeny", organism: Annotated[ Literal["human", "mouse"], "Organism for the prior knowledge resource.", ] = "human", method: Annotated[ Literal["ulm", "mlm", "zscore"], "Activity scoring method: " "'ulm' (Univariate Linear Model — recommended, fastest), " "'mlm' (Multivariate Linear Model — accounts for co-linearity), " "'wsum' (Weighted Sum — no distributional assumptions).", ] = "ulm", input_scale: Annotated[ Literal["auto", "log", "linear"], "Scale of the input expression matrix. " "'auto' (recommended) detects the scale and log2(x+1)-transforms the " "matrix when it looks linear (e.g. raw TPM), since ULM assumes roughly " "symmetric log-scale input; 'log' trusts the matrix as-is; 'linear' " "always log2(x+1)-transforms. The transform actually applied is " "reported as 'applied_transform' in the return value.", ] = "auto", exclude_samples_column: Annotated[ str | None, "Metadata column to match `exclude_samples_matching` against. Default " "None matches the expression matrix row index (sample IDs). Set this " "when the marker lives in a metadata column instead — e.g. in " "gse205154_sears the row index is GSM accessions and the fibroblast " "'-F' suffix is only in obs['sample_title'], so excluding them needs " "exclude_samples_column='sample_title'. Only available for .h5ad input.", ] = None, exclude_samples_matching: Annotated[ str | None, "Optional regular expression matched against sample IDs (the expression " "matrix row index); matching samples are dropped BEFORE scoring. Use to " "segment out sample types that a dataset's metadata does not separate — " "e.g. '-F$' drops the 7 fibroblast samples in gse205154_sears, which GEO " "annotates as 'Primary' and which can inflate stromal pathway scores " "(TGFb, NFkB). Excluded sample IDs are reported in 'excluded_samples'.", ] = None, out_prefix: Annotated[ str | None, "Output file prefix for saved CSVs. Defaults to '__'.", ] = None, ) -> dict: """ Score bulk expression samples for pathway, TF, or hallmark activity. Call this when: - You have a pre-normalised bulk expression matrix (samples × genes) and want per-sample activity scores across a cohort. - You want PROGENy pathway scores, CollecTRI TF regulon scores, or Hallmark gene set scores — without first computing DE statistics. - You are preparing activity matrices for downstream analysis such as subtype comparison, survival correlation, or visualisation. This tool complements the DE-based enrichment tools (decoupler_tf_enrichment_collectri, decoupler_pathway_enrichment_progeny, decoupler_hallmark_enrichment) which score from DE statistics. Use this tool when you want sample-level scores rather than contrast-level scores. Input expression must be pre-normalised. Raw read counts will produce incorrect activity scores — run normalisation first. Returns ------- Activity scores (samples × activities) and p-values are saved as CSV files in tmp/outputs/, together with a cohort-wide **landscape figure** (per-sample heatmap + mean±SD bar) at 'landscape_path'. Use that figure when the user asks to "see the landscape" — do NOT hand-roll a plot; a figure written outside tmp/outputs/ is never surfaced in the UI. The return dict includes artifact paths, gene coverage statistics, the input-scale transform applied, the pinned PROGENy footprint, and warnings. Next step: use the returned activity_path with downstream tools, or call dataset_validate_contrast() to define groups before comparing scores between conditions. """ # ── Load expression ─────────────────────────────────────────────────── try: expr_df, symbol_info = _load_expression_df(expression_path) except FileNotFoundError as exc: return { "valid": False, "error": str(exc), "expression_path": expression_path, } except Exception as exc: return { "valid": False, "error": f"Failed to load expression from '{expression_path}': {exc}", "expression_path": expression_path, } # ── Optional sample exclusion ───────────────────────────────────────── # Some cohorts label sample types only in the sample ID (gse205154_sears # annotates its 7 '-F' fibroblast samples as 'Primary' in GEO), so there is # no metadata column to filter on. This hook makes the exclusion explicit # and auditable rather than silent. excluded_samples: list[str] = [] exclusion_warnings: list[str] = [] if exclude_samples_matching: obs = (symbol_info or {}).get("obs") if exclude_samples_column: if obs is None: return { "valid": False, "error": ( f"exclude_samples_column='{exclude_samples_column}' requires .h5ad " "input (CSV/TSV carry no metadata table)." ), } if exclude_samples_column not in obs.columns: return { "valid": False, "error": ( f"exclude_samples_column='{exclude_samples_column}' not found. " f"Available metadata columns: {list(obs.columns)}" ), } match_against = obs[exclude_samples_column].reindex(expr_df.index).astype(str) else: match_against = expr_df.index.to_series().astype(str) try: mask = match_against.str.contains(exclude_samples_matching, regex=True).fillna(False) except Exception as exc: return { "valid": False, "error": f"Invalid exclude_samples_matching regex '{exclude_samples_matching}': {exc}", } mask = mask.to_numpy() # A filter that matches nothing must never pass silently: the caller # believes those samples are gone. Observed live — the agent passed # '-F$' against a GSM-accession index while the '-F' suffix only exists # in obs['sample_title'], so zero fibroblasts were excluded and nothing # said so. if not mask.any(): where = ( f"metadata column '{exclude_samples_column}'" if exclude_samples_column else "the sample-ID index" ) sample_vals = list(match_against[:3]) exclusion_warnings.append( f"exclude_samples_matching='{exclude_samples_matching}' matched NO samples " f"against {where} — nothing was excluded. Values look like {sample_vals}. " "If the marker lives in a metadata column, pass exclude_samples_column." ) excluded_samples = expr_df.index[mask].astype(str).tolist() expr_df = expr_df.loc[~mask] if expr_df.empty: return { "valid": False, "error": ( f"exclude_samples_matching='{exclude_samples_matching}' removed " f"all {len(excluded_samples)} samples — nothing left to score." ), "excluded_samples": excluded_samples, } # ── Score ───────────────────────────────────────────────────────────── try: result = score_bulk_samples_with_decoupler( expression_df=expr_df, resource=resource, organism=organism, method=method, input_scale=input_scale, ) except ValueError as exc: return { "valid": False, "error": str(exc), "expression_path": expression_path, } except Exception as exc: return { "valid": False, "error": f"Scoring failed: {exc}", "expression_path": expression_path, } # ── Save outputs ────────────────────────────────────────────────────── prefix = out_prefix or f"{resource}_{method}_{_timestamp}" activity_path = OUTPUT_DIR / f"{prefix}_activities.csv" pvalue_path = OUTPUT_DIR / f"{prefix}_pvalues.csv" result["activity_df"].to_csv(activity_path) result["pvalue_df"].to_csv(pvalue_path) # ── Landscape figure ────────────────────────────────────────────────── # Emitted by the tool (not left to agent-authored matplotlib) so it always # lands in OUTPUT_DIR and is picked up by the end-of-run inline-plot sweep. landscape_path = OUTPUT_DIR / f"{prefix}_landscape.png" warnings_out = list(result["warnings"]) if symbol_info: warnings_out.extend(symbol_info["warnings"]) landscape_info: dict | None = None try: landscape_info = plot_activity_landscape( result["activity_df"], landscape_path, title=f"{resource} activity landscape ({method}, n={result['n_samples']} samples)", ) if landscape_info["truncated"]: warnings_out.append( f"Landscape figure shows the {landscape_info['n_features_plotted']} " f"most variable of {landscape_info['n_features_total']} activities; " "all activities are in the CSV." ) except Exception as exc: # plotting must never fail the analysis warnings_out.append(f"Landscape figure could not be rendered: {exc}") warnings_out.extend(exclusion_warnings) if excluded_samples: warnings_out.append( f"Excluded {len(excluded_samples)} sample(s) matching " f"'{exclude_samples_matching}' before scoring: {excluded_samples}." ) return { "valid": True, "expression_path": expression_path, "n_samples": result["n_samples"], # samples in expression matrix "n_obs": result["n_samples"], # alias kept for backward compatibility "shape": (result["n_samples"], result["n_activities"]), "n_activities": result["n_activities"], "n_network_genes": result["n_network_genes"], "n_matched_genes": result["n_matched_genes"], "coverage_pct": result["coverage_pct"], "resource": result["resource"], "organism": result["organism"], "method": result["method"], "input_scale_requested": result["input_scale_requested"], "input_scale_detected": result["input_scale_detected"], "applied_transform": result["applied_transform"], "expression_max": result["expression_max"], "network_top": result["network_top"], "excluded_samples": excluded_samples, "n_excluded_samples": len(excluded_samples), "exclude_samples_column": exclude_samples_column, "csv_orientation": "rows = samples, columns = activity names", "gene_symbols_resolved": bool(symbol_info and symbol_info["renamed"]), "gene_symbol_column": (symbol_info or {}).get("symbol_column"), "method_caveat": DECOUPLER_DISCLAIMER_PER_SAMPLE, "warnings": warnings_out, "activity_path": str(activity_path.resolve()), "pvalue_path": str(pvalue_path.resolve()), "landscape_path": (str(landscape_path.resolve()) if landscape_info else None), "artifacts": [ { "description": ( f"{resource} activity scores ({method}) — " "samples × activities, save for downstream comparison" ), "path": str(activity_path.resolve()), }, { "description": f"{resource} activity p-values ({method})", "path": str(pvalue_path.resolve()), }, ] + ( [ { "description": ( f"{resource} cohort-wide activity landscape ({method}) — " "per-sample heatmap + mean±SD bar" ), "path": landscape_info["path"], } ] if landscape_info else [] ), "next_step": ( f"Activity scores for {result['n_activities']} {resource} features " f"saved to activity_path, and the cohort-wide landscape figure to " "landscape_path (show that figure rather than plotting your own). " "To compare scores between sample groups, call " "dataset_compare_activity_by_group(activity_path, metadata_path, ...)." ), } # --------------------------------------------------------------------------- # Custom-signature scoring (ADR-0006 Role 2) # --------------------------------------------------------------------------- @bulk_dataset_mcp.tool def dataset_score_signature( expression_path: Annotated[ str, "Path to a pre-normalised samples × genes expression file " "(.h5ad, .csv, or .tsv). Samples are rows, gene symbols are columns. " "Must be normalised (log-CPM, log-intensity, log2-TPM, etc.) — not raw " "integer counts.", ], signature_path: Annotated[ str, "Path to a signature table (.csv or .tsv) with columns " "'source, target[, weight]' — one row per gene in a signature. " "'source' is the signature name (e.g. a Loveless-derived cell-state " "program), 'target' is an HGNC gene symbol, and 'weight' (optional) is " "the gene's signed membership/loading. This is a precomputed, " "offline-derived signature artifact — NOT the single-cell atlas itself.", ], method: Annotated[ Literal["ulm", "mlm", "zscore"], "Activity scoring method (same engine as dataset_score_bulk_samples).", ] = "ulm", organism: Annotated[ Literal["human", "mouse"], "Organism label, carried through to diagnostics.", ] = "human", label: Annotated[ str, "Free-form name for this signature set, used in output filenames (e.g. 'loveless_basal').", ] = "signature", out_prefix: Annotated[ str | None, "Output file prefix for saved CSVs. Defaults to '