Upload 14 files

Dockerfile

@@ -1,17 +1,22 @@
-FROM python:3.10
+FROM python:3.11-slim
 
-RUN useradd -m -u 1000 user && python -m pip install --upgrade pip
-USER user
-ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV MCP_TRANSPORT=http
+ENV MCP_PORT=7860
 
 WORKDIR /app
 
-COPY --chown=user ./requirements.txt requirements.txt
-RUN pip install --no-cache-dir --upgrade -r requirements.txt
+RUN useradd -m -u 1000 appuser
 
-COPY --chown=user . /app
-ENV MCP_TRANSPORT=http
-ENV MCP_PORT=7860
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY scanpy /app/scanpy
+COPY app.py /app/app.py
+
+RUN chown -R appuser:appuser /app
+USER appuser
 
 EXPOSE 7860
 

README.md

@@ -1,10 +1,60 @@
 ---
-title: Scanpy
-emoji: 👁
+title: scanpy MCP Service
+emoji: 🔧
 colorFrom: blue
-colorTo: yellow
+colorTo: indigo
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# scanpy MCP Service
+
+这是一个面向 `scanpy` 的 MCP 服务部署包，支持：
+
+- 本地 `stdio`（Claude Desktop / CLI）
+- Docker / HuggingFace Spaces 的 `http` 传输
+
+## 可用工具
+
+- `scanpy_health`
+- `create_synthetic_adata`
+- `load_h5ad`
+- `preprocess_basic`
+- `run_pca_neighbors_umap`
+- `run_leiden`
+- `rank_marker_genes`
+- `adapter_call_function`
+
+详见 `scanpy/mcp_output/README_MCP.md`。
+
+## 本地 stdio 连接
+
+```bash
+cd scanpy/mcp_output
+python mcp_plugin/main.py
+```
+
+或：
+
+```bash
+MCP_TRANSPORT=stdio python scanpy/mcp_output/start_mcp.py
+```
+
+## HTTP 客户端连接
+
+```bash
+MCP_TRANSPORT=http MCP_PORT=8000 python scanpy/mcp_output/start_mcp.py
+```
+
+客户端连接地址：`http://localhost:8000/mcp`
+
+## Docker / HF Spaces
+
+该部署默认使用 `7860` 端口并直接启动 FastMCP HTTP 服务：
+
+```bash
+python scanpy/mcp_output/start_mcp.py
+```
+
+在 HuggingFace Spaces 上，客户端连接：`https://<your-space-host>/mcp`。

app.py

@@ -1,45 +1,46 @@
-from fastapi import FastAPI
+from __future__ import annotations
+
 import os
 import sys
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI
 
-mcp_plugin_path = os.path.join(os.path.dirname(__file__), "scanpy", "mcp_output", "mcp_plugin")
-sys.path.insert(0, mcp_plugin_path)
+PLUGIN_DIR = Path(__file__).resolve().parent / "scanpy" / "mcp_output" / "mcp_plugin"
+plugin_path = str(PLUGIN_DIR)
+if plugin_path not in sys.path:
+    sys.path.insert(0, plugin_path)
+
+app = FastAPI(title="scanpy-mcp-info", version="1.0.0")
 
-app = FastAPI(
-    title="Scanpy MCP Service",
-    description="Auto-generated MCP service for scanpy",
-    version="1.0.0"
-)
 
 @app.get("/")
-def root():
+def root() -> dict[str, Any]:
     return {
-        "service": "Scanpy MCP Service",
-        "version": "1.0.0",
-        "status": "running",
-        "transport": os.environ.get("MCP_TRANSPORT", "http")
+        "name": "scanpy MCP service",
+        "description": "Supplementary info API for local development",
+        "mcp_entrypoint": "scanpy/mcp_output/start_mcp.py",
+        "default_port": int(os.getenv("PORT", "7860")),
     }
 
+
 @app.get("/health")
-def health_check():
-    return {"status": "healthy", "service": "scanpy MCP"}
+def health() -> dict[str, str]:
+    return {"status": "healthy"}
+
 
 @app.get("/tools")
-def list_tools():
-    try:
-        from mcp_service import create_app
-        mcp_app = create_app()
-        tools = []
-        for tool_name, tool_func in mcp_app.tools.items():
-            tools.append({
-                "name": tool_name,
-                "description": tool_func.__doc__ or "No description available"
-            })
-        return {"tools": tools}
-    except Exception as e:
-        return {"error": f"Failed to load tools: {str(e)}"}
-
-if __name__ == "__main__":
-    import uvicorn
-    port = int(os.environ.get("PORT", 7860))
-    uvicorn.run(app, host="0.0.0.0", port=port)
+def tools() -> dict[str, Any]:
+    from mcp_service import create_app
+
+    mcp = create_app()
+    tool_items = []
+    for tool in getattr(mcp, "tools", []):
+        tool_items.append(
+            {
+                "name": getattr(tool, "name", "unknown"),
+                "description": getattr(tool, "description", ""),
+            }
+        )
+    return {"tools": tool_items, "count": len(tool_items)}

port.json

@@ -1,5 +1 @@
-{
-  "repo": "scanpy",
-  "port": 7918,
-  "timestamp": 1773467942
-}
+{"port": 7860}

requirements.txt

@@ -1,25 +1,8 @@
 fastmcp
+scanpy
+anndata
+numpy
+pandas
+scipy
 fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-anndata>=0.10.8
-fast-array-utils[accel,sparse]>=1.2.1
-h5py>=3.11
-joblib
-matplotlib>=3.9
-natsort
-networkx>=2.8.8
-numba>=0.60
-numpy>=2
-packaging>=25
-pandas>=2.2.2
-patsy
-pynndescent>=0.5.13
-scikit-learn>=1.4.2
-scipy>=1.13
-seaborn>=0.13.2
-session-info2
-statsmodels>=0.14.5
-tqdm
-typing-extensions; python_version<'3.13'
-umap-learn>=0.5.7
+uvicorn

run_docker.ps1

@@ -1,26 +1,7 @@
-cd $PSScriptRoot
 $ErrorActionPreference = "Stop"
-$entryName = if ($env:MCP_ENTRY_NAME) { $env:MCP_ENTRY_NAME } else { "scanpy" }
-$entryUrl  = if ($env:MCP_ENTRY_URL)  { $env:MCP_ENTRY_URL  } else { "http://localhost:7918/mcp" }
-$imageName = if ($env:MCP_IMAGE_NAME) { $env:MCP_IMAGE_NAME } else { "scanpy-mcp" }
-$mcpDir = Join-Path $env:USERPROFILE ".cursor"
-$mcpPath = Join-Path $mcpDir "mcp.json"
-if (!(Test-Path $mcpDir)) { New-Item -ItemType Directory -Path $mcpDir | Out-Null }
-$config = @{}
-if (Test-Path $mcpPath) {
-  try { $config = Get-Content $mcpPath -Raw | ConvertFrom-Json } catch { $config = @{} }
-}
-$serversOrdered = [ordered]@{}
-if ($config -and ($config.PSObject.Properties.Name -contains "mcpServers") -and $config.mcpServers) {
-  $existing = $config.mcpServers
-  if ($existing -is [pscustomobject]) {
-    foreach ($p in $existing.PSObject.Properties) { if ($p.Name -ne $entryName) { $serversOrdered[$p.Name] = $p.Value } }
-  } elseif ($existing -is [System.Collections.IDictionary]) {
-    foreach ($k in $existing.Keys) { if ($k -ne $entryName) { $serversOrdered[$k] = $existing[$k] } }
-  }
-}
-$serversOrdered[$entryName] = @{ url = $entryUrl }
-$config = @{ mcpServers = $serversOrdered }
-$config | ConvertTo-Json -Depth 10 | Set-Content -Path $mcpPath -Encoding UTF8
+
+$port = (Get-Content -Raw -Path "port.json" | ConvertFrom-Json).port
+$imageName = "scanpy-mcp"
+
 docker build -t $imageName .
-docker run --rm -p 7918:7860 $imageName
+docker run --rm -it -p "${port}:${port}" $imageName

run_docker.sh

@@ -1,75 +1,8 @@
 #!/usr/bin/env bash
 set -euo pipefail
-cd "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-mcp_entry_name="${MCP_ENTRY_NAME:-scanpy}"
-mcp_entry_url="${MCP_ENTRY_URL:-http://localhost:7918/mcp}"
-mcp_dir="${HOME}/.cursor"
-mcp_path="${mcp_dir}/mcp.json"
-mkdir -p "${mcp_dir}"
-if command -v python3 >/dev/null 2>&1; then
-python3 - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v python >/dev/null 2>&1; then
-python - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v jq >/dev/null 2>&1; then
-  name="${mcp_entry_name}"; url="${mcp_entry_url}"
-  if [ -f "${mcp_path}" ]; then
-    tmp="$(mktemp)"
-    jq --arg name "$name" --arg url "$url" '
-      .mcpServers = (.mcpServers // {})
-      | .mcpServers as $s
-      | ($s | with_entries(select(.key != $name))) as $base
-      | .mcpServers = ($base + {($name): {"url": $url}})
-    ' "${mcp_path}" > "${tmp}" && mv "${tmp}" "${mcp_path}"
-  else
-    printf '{ "mcpServers": { "%s": { "url": "%s" } } }
-' "$name" "$url" > "${mcp_path}"
-  fi
-fi
-docker build -t scanpy-mcp .
-docker run --rm -p 7918:7860 scanpy-mcp
+
+PORT=$(python3 -c 'import json;print(json.load(open("port.json"))["port"])')
+IMAGE_NAME="scanpy-mcp"
+
+docker build -t "${IMAGE_NAME}" .
+docker run --rm -it -p "${PORT}:${PORT}" "${IMAGE_NAME}"

scanpy/mcp_output/README_MCP.md

@@ -1,130 +1,73 @@
-# Scanpy MCP (Model Context Protocol) Service README
+# Scanpy MCP 插件说明
 
-## 1) Project Introduction
+该目录提供基于 FastMCP 的 Scanpy 工具封装，支持本地 `stdio` 与 HTTP 传输。
 
-This service exposes core **Scanpy** single-cell analysis capabilities through an MCP (Model Context Protocol) interface.  
-It is designed for developer workflows that need programmatic access to:
+## 已暴露工具
 
-- Data I/O for AnnData and 10x formats
-- Preprocessing (QC, normalization, HVG selection, PCA, neighbor graph)
-- Analysis tools (Leiden/Louvain clustering, UMAP/t-SNE, PAGA, marker ranking)
-- Plotting helpers (UMAP/scatter, dotplot, matrixplot, violin, etc.)
-- Dataset loading and tabular extraction utilities
+1. `scanpy_health()`
+   - 参数：无
+   - 作用：返回服务状态、已加载模块、已缓存数据集
+   - 示例：`scanpy_health()`
 
-Repository: https://github.com/scverse/scanpy
+2. `create_synthetic_adata(dataset_key="demo", n_obs=200, n_vars=50, seed=0)`
+   - 作用：创建内存中的合成 AnnData
+   - 示例：`create_synthetic_adata(dataset_key="toy", n_obs=300, n_vars=100, seed=42)`
 
----
+3. `load_h5ad(file_path, dataset_key="adata")`
+   - 作用：加载本地 `.h5ad` 到内存注册表
+   - 示例：`load_h5ad(file_path="/data/sample.h5ad", dataset_key="pbmc")`
 
-## 2) Installation Method
+4. `preprocess_basic(dataset_key, target_sum=10000.0, apply_log1p=True, n_top_genes=2000, scale_data=False, max_value=10.0)`
+   - 作用：执行标准预处理流程
+   - 示例：`preprocess_basic(dataset_key="pbmc", n_top_genes=1500, scale_data=True)`
 
-### Requirements
-Core runtime dependencies typically include:
+5. `run_pca_neighbors_umap(dataset_key, n_pcs=50, n_neighbors=15, min_dist=0.5, random_state=0)`
+   - 作用：执行 PCA + 邻居图 + UMAP
+   - 示例：`run_pca_neighbors_umap(dataset_key="pbmc", n_pcs=30, n_neighbors=10)`
 
-- anndata, numpy, scipy, pandas
-- matplotlib, scikit-learn, h5py
-- numba, networkx, packaging
+6. `run_leiden(dataset_key, resolution=1.0, key_added="leiden", random_state=0)`
+   - 作用：执行 Leiden 聚类
+   - 示例：`run_leiden(dataset_key="pbmc", resolution=0.8)`
 
-Common optional dependencies (feature-dependent):
+7. `rank_marker_genes(dataset_key, groupby="leiden", method="wilcoxon", n_genes=10)`
+   - 作用：按分组计算 marker genes
+   - 示例：`rank_marker_genes(dataset_key="pbmc", n_genes=20)`
 
-- umap-learn
-- igraph / python-igraph, leidenalg, louvain
-- dask, zarr
-- statsmodels, seaborn
-- harmonypy, bbknn, scanorama, magic-impute, palantir, phate, phenograph
+8. `adapter_call_function(module_name, function_name, args_json="[]", kwargs_json="{}")`
+   - 作用：通过适配器调用任意可导入函数
+   - 示例：`adapter_call_function(module_name="scanpy.pp", function_name="log1p", args_json='[null]', kwargs_json='{}')`
 
-### Install
-- Install Scanpy:
-  - `pip install scanpy`
-- For richer workflows, also install optional ecosystem packages as needed.
-- If deploying as an MCP (Model Context Protocol) service, include Scanpy and your MCP runtime in the same environment.
+## 返回格式
 
----
+所有 MCP 工具统一返回：
 
-## 3) Quick Start
+```json
+{
+  "success": true,
+  "result": {},
+  "error": null
+}
+```
 
-Typical service flow:
+失败时 `success=false`，并在 `error` 字段返回错误信息。
 
-1. Load data (`read_h5ad`, `read_10x_h5`, `read_10x_mtx`, or dataset loaders such as `pbmc3k`)
-2. Run preprocessing (`filter_cells`, `filter_genes`, `calculate_qc_metrics`, `normalize_total`, `log1p`, `highly_variable_genes`, `pca`, `neighbors`)
-3. Run analysis (`leiden`/`louvain`, `umap`, `rank_genes_groups`, `paga`)
-4. Return tabular results (`obs_df`, `var_df`, `rank_genes_groups_df`) and/or plotting artifacts
+## 本地 stdio 运行
 
-Example pipeline sequence:
-- read → qc/filter → normalize/log1p → hvg → pca → neighbors → leiden → umap → marker ranking
+```bash
+cd scanpy/mcp_output
+python mcp_plugin/main.py
+```
 
----
+或：
 
-## 4) Available Tools and Endpoints List
+```bash
+MCP_TRANSPORT=stdio python start_mcp.py
+```
 
-Recommended MCP (Model Context Protocol) service endpoints (grouped by module):
+## HTTP 运行
 
-### I/O
-- `read`, `read_h5ad`, `read_csv`, `read_loom` — Load AnnData-compatible files
-- `read_10x_h5`, `read_10x_mtx` — Read 10x Genomics data
-- `write` — Persist processed AnnData objects
+```bash
+MCP_TRANSPORT=http MCP_PORT=8000 python start_mcp.py
+```
 
-### Datasets
-- `pbmc3k`, `pbmc68k_reduced`, `blobs`, `krumsiek11`, `toggleswitch`, `ebi_expression_atlas` — Quick demo/test datasets
-
-### Preprocessing (`pp`)
-- `calculate_qc_metrics` — Per-cell/per-gene QC metrics
-- `filter_cells`, `filter_genes` — Basic filtering
-- `normalize_total`, `log1p` — Library-size normalization and transform
-- `highly_variable_genes` — Feature selection
-- `scale`, `regress_out` — Feature scaling and regression
-- `pca`, `neighbors` — Dimensionality reduction and graph construction
-- `scrublet`, `subsample` — Doublet detection and sampling helpers
-
-### Tools (`tl`)
-- `leiden`, `louvain` — Graph clustering
-- `umap`, `tsne`, `diffmap`, `draw_graph` — Embedding
-- `paga`, `dendrogram`, `embedding_density`, `ingest` — Graph/trajectory and transfer utilities
-- `rank_genes_groups`, `score_genes`, `marker_gene_overlap` — Marker and scoring analysis
-
-### Plotting (`pl`)
-- `umap`, `scatter`, `spatial` — Embedding/spatial visualization
-- `dotplot`, `matrixplot`, `stacked_violin`, `violin`, `heatmap` — Expression summary plots
-- `paga`, `rank_genes_groups` — Result-oriented visual outputs
-
-### Get/Tabular Accessors
-- `obs_df`, `var_df`, `rank_genes_groups_df` — Dataframe outputs for downstream systems
-
-### Metrics
-- `morans_i`, `gearys_c` — Spatial/autocorrelation metrics
-
-### External Integrations (`external`)
-- `bbknn`, `harmony_integrate`, `scanorama_integrate`, `magic`, `mnn_correct` — Optional integrations requiring extra packages
-
-### CLI
-- `scanpy`
-- `python -m scanpy`
-
----
-
-## 5) Common Issues and Notes
-
-- **Missing optional dependencies**: many advanced endpoints require extra installs (e.g., `leidenalg`, `igraph`, `umap-learn`).
-- **Version compatibility**: keep `scanpy`, `anndata`, and numeric stack versions aligned.
-- **Memory/performance**: large datasets can be expensive for PCA/neighbors/UMAP; consider subsampling, sparse matrices, and staged processing.
-- **Headless environments**: plotting may require non-interactive matplotlib backend.
-- **External methods**: wrappers in `scanpy.external` fail gracefully only if dependency checks are handled in service code.
-- **Data format assumptions**: most endpoints expect valid AnnData structure (`.obs`, `.var`, `.X`, optional `.raw`, `.obsm`, `.uns`).
-
----
-
-## 6) Reference Links and Documentation
-
-- Scanpy repository: https://github.com/scverse/scanpy
-- Scanpy docs index: `docs/index.md` in repository
-- API docs sections:
-  - `docs/api/preprocessing.md`
-  - `docs/api/tools.md`
-  - `docs/api/plotting.md`
-  - `docs/api/io.md`
-  - `docs/api/datasets.md`
-  - `docs/api/get.md`
-  - `docs/api/metrics.md`
-- Developer docs:
-  - `docs/dev/getting-set-up.md`
-  - `docs/dev/testing.md`
-  - `docs/dev/documentation.md`
+HTTP 模式下，MCP 端点由 FastMCP 直接提供，客户端连接 `http://localhost:8000/mcp`。

scanpy/mcp_output/mcp_plugin/adapter.py

@@ -1,222 +1,189 @@
-import os
+from __future__ import annotations
+
+import importlib
+import inspect
+import pkgutil
 import sys
-from typing import Any, Callable, Dict, Optional
+from pathlib import Path
+from typing import Any
 
-source_path = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-    "source",
-)
-sys.path.insert(0, source_path)
+SOURCE_DIR = Path(__file__).resolve().parents[2] / "source"
+if SOURCE_DIR.exists():
+    source_path = str(SOURCE_DIR)
+    if source_path not in sys.path:
+        sys.path.insert(0, source_path)
 
 
 class Adapter:
-    """
-    MCP import-mode adapter for the Scanpy repository.
-
-    This adapter prioritizes direct imports from repository source code and falls back
-    to a lightweight CLI-compatible mode when imports are unavailable.
-    """
-
-    def __init__(self) -> None:
-        self.mode = "import"
-        self._modules: Dict[str, Any] = {}
-        self._import_error: Optional[str] = None
-        self._initialize_modules()
-
-    # ---------------------------------------------------------------------
-    # Internal helpers
-    # ---------------------------------------------------------------------
-    def _ok(self, data: Optional[Dict[str, Any]] = None, message: str = "success") -> Dict[str, Any]:
-        payload = {"status": "success", "mode": self.mode, "message": message}
-        if data:
-            payload.update(data)
-        return payload
-
-    def _error(self, message: str, guidance: Optional[str] = None, exc: Optional[Exception] = None) -> Dict[str, Any]:
-        payload = {"status": "error", "mode": self.mode, "message": message}
-        if guidance:
-            payload["guidance"] = guidance
-        if exc is not None:
-            payload["error"] = str(exc)
-        return payload
-
-    def _initialize_modules(self) -> None:
+    def __init__(self, package_name: str = "scanpy") -> None:
+        self.package_name = package_name
+        self.loaded_modules: dict[str, Any] = {}
+        self.failed_modules: dict[str, str] = {}
+        self.mode = "normal"
+        self._instances: dict[str, Any] = {}
+        self._instance_counter = 0
+
+    def _discover_submodules(self) -> list[str]:
+        discovered = [self.package_name]
         try:
-            import source.src.scanpy as scanpy_pkg
-            import source.src.scanpy.cli as scanpy_cli
-            import source.src.scanpy.__main__ as scanpy_main
+            root_module = importlib.import_module(self.package_name)
+        except Exception:
+            return discovered
+
+        if hasattr(root_module, "__path__"):
+            for module_info in pkgutil.walk_packages(
+                root_module.__path__, prefix=f"{self.package_name}."
+            ):
+                discovered.append(module_info.name)
+        return discovered
+
+    def load_modules(self) -> dict[str, Any]:
+        self.loaded_modules.clear()
+        self.failed_modules.clear()
+
+        for module_name in self._discover_submodules():
+            try:
+                self.loaded_modules[module_name] = importlib.import_module(module_name)
+            except Exception as exc:
+                self.failed_modules[module_name] = str(exc)
+
+        if not self.loaded_modules:
+            self.mode = "blackbox"
+            return {
+                "status": "fallback",
+                "mode": self.mode,
+                "message": "No modules loaded; operating in blackbox mode",
+                "loaded_count": 0,
+                "failed_count": len(self.failed_modules),
+            }
+
+        self.mode = "normal"
+        return {
+            "status": "ok",
+            "mode": self.mode,
+            "loaded_count": len(self.loaded_modules),
+            "failed_count": len(self.failed_modules),
+        }
+
+    def health(self) -> dict[str, Any]:
+        if not self.loaded_modules and not self.failed_modules:
+            return self.load_modules()
+
+        status = "ok" if self.loaded_modules else "fallback"
+        return {
+            "status": status,
+            "mode": self.mode,
+            "loaded_count": len(self.loaded_modules),
+            "failed_count": len(self.failed_modules),
+            "package": self.package_name,
+        }
+
+    def list_modules(self) -> dict[str, Any]:
+        if not self.loaded_modules and not self.failed_modules:
+            self.load_modules()
+
+        status = "ok" if self.loaded_modules else "fallback"
+        return {
+            "status": status,
+            "mode": self.mode,
+            "loaded_modules": sorted(self.loaded_modules.keys()),
+            "failed_modules": self.failed_modules,
+        }
+
+    def list_symbols(self, module_name: str, limit: int = 200) -> dict[str, Any]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            try:
+                module = importlib.import_module(module_name)
+                self.loaded_modules[module_name] = module
+            except Exception as exc:
+                return {
+                    "status": "error",
+                    "module": module_name,
+                    "error": str(exc),
+                }
+
+        symbols = sorted(name for name, _ in inspect.getmembers(module))
+        return {
+            "status": "ok",
+            "module": module_name,
+            "count": len(symbols),
+            "symbols": symbols[:limit],
+            "truncated": len(symbols) > limit,
+        }
+
+    def call_function(
+        self,
+        module_name: str,
+        function_name: str,
+        args: list[Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        args = args or []
+        kwargs = kwargs or {}
 
-            self._modules["scanpy"] = scanpy_pkg
-            self._modules["cli"] = scanpy_cli
-            self._modules["main"] = scanpy_main
-        except Exception as exc:
-            self.mode = "cli"
-            self._import_error = str(exc)
+        try:
+            module = self.loaded_modules.get(module_name)
+            if module is None:
+                module = importlib.import_module(module_name)
+                self.loaded_modules[module_name] = module
+
+            func = getattr(module, function_name)
+            if not callable(func):
+                return {
+                    "status": "error",
+                    "module": module_name,
+                    "function": function_name,
+                    "error": f"{function_name} is not callable",
+                }
 
-    def _get_module(self, key: str) -> Optional[Any]:
-        return self._modules.get(key)
+            result = func(*args, **kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "function": function_name,
+                "result": result,
+            }
+        except Exception as exc:
+            return {
+                "status": "error",
+                "module": module_name,
+                "function": function_name,
+                "error": str(exc),
+            }
+
+    def create_instance(
+        self,
+        module_name: str,
+        class_name: str,
+        init_args: list[Any] | None = None,
+        init_kwargs: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        init_args = init_args or []
+        init_kwargs = init_kwargs or {}
 
-    def _safe_call(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Dict[str, Any]:
         try:
-            result = func(*args, **kwargs)
-            return self._ok({"result": result})
+            module = self.loaded_modules.get(module_name)
+            if module is None:
+                module = importlib.import_module(module_name)
+                self.loaded_modules[module_name] = module
+
+            klass = getattr(module, class_name)
+            instance = klass(*init_args, **init_kwargs)
+            self._instance_counter += 1
+            instance_id = f"instance_{self._instance_counter}"
+            self._instances[instance_id] = instance
+
+            return {
+                "status": "ok",
+                "module": module_name,
+                "class": class_name,
+                "instance_id": instance_id,
+            }
         except Exception as exc:
-            return self._error(
-                "Function execution failed.",
-                guidance="Check arguments and data schema, then retry.",
-                exc=exc,
-            )
-
-    # ---------------------------------------------------------------------
-    # Status and environment
-    # ---------------------------------------------------------------------
-    def health_check(self) -> Dict[str, Any]:
-        """
-        Report adapter readiness and import state.
-
-        Returns:
-            dict: Unified response with status, mode, and module availability.
-        """
-        available = {k: bool(v) for k, v in self._modules.items()}
-        if self.mode == "import":
-            return self._ok({"modules": available}, "Adapter is ready in import mode.")
-        return self._error(
-            "Adapter is running in fallback mode due to import failure.",
-            guidance=f"Verify repository path and dependencies. Import error: {self._import_error}",
-        )
-
-    # ---------------------------------------------------------------------
-    # Core scanpy package access
-    # ---------------------------------------------------------------------
-    def instance_scanpy(self) -> Dict[str, Any]:
-        """
-        Return the imported scanpy module instance.
-
-        Returns:
-            dict: Unified response containing module object when available.
-        """
-        mod = self._get_module("scanpy")
-        if mod is None:
-            return self._error(
-                "scanpy module is unavailable in current mode.",
-                guidance="Install required dependencies and ensure source/src is importable.",
-            )
-        return self._ok({"module": mod}, "scanpy module loaded.")
-
-    def call_scanpy_attr(self, attr_name: str, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call a top-level attribute from source.src.scanpy dynamically.
-
-        Parameters:
-            attr_name: Name of function/attribute on the scanpy package.
-            *args: Positional arguments for callable attributes.
-            **kwargs: Keyword arguments for callable attributes.
-
-        Returns:
-            dict: Unified response with call result or attribute value.
-        """
-        mod = self._get_module("scanpy")
-        if mod is None:
-            return self._error(
-                "scanpy module is not imported.",
-                guidance="Use health_check() and resolve import issues first.",
-            )
-        if not hasattr(mod, attr_name):
-            return self._error(
-                f"Attribute '{attr_name}' not found in scanpy module.",
-                guidance="Check the attribute name against source.src.scanpy exports.",
-            )
-        target = getattr(mod, attr_name)
-        if callable(target):
-            return self._safe_call(target, *args, **kwargs)
-        return self._ok({"result": target}, f"Attribute '{attr_name}' retrieved.")
-
-    # ---------------------------------------------------------------------
-    # CLI module wrappers (source.src.scanpy.cli)
-    # ---------------------------------------------------------------------
-    def instance_cli(self) -> Dict[str, Any]:
-        """
-        Return the imported CLI module instance.
-
-        Returns:
-            dict: Unified response containing CLI module object.
-        """
-        mod = self._get_module("cli")
-        if mod is None:
-            return self._error(
-                "CLI module is unavailable.",
-                guidance="Ensure source.src.scanpy.cli can be imported.",
-            )
-        return self._ok({"module": mod}, "CLI module loaded.")
-
-    def call_cli_main(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Invoke CLI entry behavior when available.
-
-        Parameters:
-            args: Optional argument list to emulate command-line input.
-
-        Returns:
-            dict: Unified response with CLI execution result.
-        """
-        mod = self._get_module("cli")
-        if mod is None:
-            return self._error(
-                "CLI module is not available in fallback context.",
-                guidance="Run 'python -m scanpy --help' in environment with dependencies.",
-            )
-        for candidate in ("main", "cli", "run"):
-            if hasattr(mod, candidate) and callable(getattr(mod, candidate)):
-                fn = getattr(mod, candidate)
-                if args is None:
-                    return self._safe_call(fn)
-                return self._safe_call(fn, args)
-        return self._error(
-            "No callable CLI entry function found in source.src.scanpy.cli.",
-            guidance="Inspect module for available public entry points and update adapter mapping.",
-        )
-
-    # ---------------------------------------------------------------------
-    # __main__ module wrappers (source.src.scanpy.__main__)
-    # ---------------------------------------------------------------------
-    def instance_main(self) -> Dict[str, Any]:
-        """
-        Return the imported __main__ module instance.
-
-        Returns:
-            dict: Unified response containing __main__ module object.
-        """
-        mod = self._get_module("main")
-        if mod is None:
-            return self._error(
-                "__main__ module is unavailable.",
-                guidance="Ensure source.src.scanpy.__main__ is importable.",
-            )
-        return self._ok({"module": mod}, "__main__ module loaded.")
-
-    def call_module_main(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Invoke python -m scanpy style entry point behavior.
-
-        Parameters:
-            args: Optional list of arguments for module main function.
-
-        Returns:
-            dict: Unified response with execution outcome.
-        """
-        mod = self._get_module("main")
-        if mod is None:
-            return self._error(
-                "__main__ module is unavailable in current mode.",
-                guidance="Use environment with full dependencies or run CLI directly.",
-            )
-        for candidate in ("main",):
-            if hasattr(mod, candidate) and callable(getattr(mod, candidate)):
-                fn = getattr(mod, candidate)
-                if args is None:
-                    return self._safe_call(fn)
-                return self._safe_call(fn, args)
-        return self._error(
-            "No callable main() found in source.src.scanpy.__main__.",
-            guidance="Verify the repository version and update adapter expectations.",
-        )
+            return {
+                "status": "error",
+                "module": module_name,
+                "class": class_name,
+                "error": str(exc),
+            }

scanpy/mcp_output/mcp_plugin/main.py

@@ -1,13 +1,8 @@
-"""
-MCP Service Auto-Wrapper - Auto-generated
-"""
-from mcp_service import create_app
+from __future__ import annotations
 
-def main():
-    """Main entry point"""
-    app = create_app()
-    return app
+from mcp_service import create_app
 
+# For local stdio use only (Claude Desktop / CLI), not for web or Docker deployment.
 if __name__ == "__main__":
-    app = main()
-    app.run()
+    app = create_app()
+    app.run(transport="stdio")

scanpy/mcp_output/mcp_plugin/mcp_service.py

@@ -1,273 +1,355 @@
-import os
-import sys
-from typing import Any, Dict, List, Optional
+from __future__ import annotations
 
-source_path = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-    "source",
-)
-if source_path not in sys.path:
-    sys.path.insert(0, source_path)
+import json
+import sys
+from pathlib import Path
+from typing import Any
 
+import numpy as np
+from anndata import AnnData
 from fastmcp import FastMCP
 
+SOURCE_DIR = Path(__file__).resolve().parents[2] / "source"
+if SOURCE_DIR.exists():
+    source_path = str(SOURCE_DIR)
+    if source_path not in sys.path:
+        sys.path.insert(0, source_path)
+
+import scanpy as sc
+
 try:
-    import scanpy as sc
-except Exception:
-    from src.scanpy import __init__ as sc  # type: ignore
+    from .adapter import Adapter
+except ImportError:
+    from adapter import Adapter
 
-mcp = FastMCP("scanpy_mcp_service")
+mcp = FastMCP("scanpy-mcp-service")
+adapter = Adapter("scanpy")
+DATASETS: dict[str, AnnData] = {}
 
 
-def _ok(result: Any) -> Dict[str, Any]:
+def _ok(result: Any) -> dict[str, Any]:
     return {"success": True, "result": result, "error": None}
 
 
-def _err(exc: Exception) -> Dict[str, Any]:
-    return {"success": False, "result": None, "error": str(exc)}
+def _error(message: str) -> dict[str, Any]:
+    return {"success": False, "result": None, "error": message}
 
 
-@mcp.tool(name="scanpy_version", description="Get installed Scanpy version information.")
-def scanpy_version() -> Dict[str, Any]:
-    """
-    Return Scanpy version.
+def _get_dataset(dataset_key: str) -> AnnData:
+    if dataset_key not in DATASETS:
+        raise KeyError(f"Dataset '{dataset_key}' not found. Create or load it first.")
+    return DATASETS[dataset_key]
 
-    Returns:
-        Dict with success/result/error. result is the version string.
-    """
+
+@mcp.tool(name="scanpy_health", description="Return Scanpy MCP service and adapter health.")
+def scanpy_health() -> dict[str, Any]:
+    """Get current service health and module loading status."""
     try:
-        version = getattr(sc, "__version__", "unknown")
-        return _ok(version)
-    except Exception as e:
-        return _err(e)
+        health_info = adapter.health()
+        result = {
+            "service": "scanpy-mcp-service",
+            "scanpy_version": getattr(sc, "__version__", "unknown"),
+            "adapter": health_info,
+            "datasets_loaded": sorted(DATASETS.keys()),
+        }
+        return _ok(result)
+    except Exception as exc:
+        return _error(str(exc))
 
 
 @mcp.tool(
-    name="load_builtin_dataset",
-    description="Load a built-in Scanpy dataset by name.",
+    name="create_synthetic_adata",
+    description="Create an in-memory synthetic AnnData object for analysis workflows.",
 )
-def load_builtin_dataset(dataset_name: str) -> Dict[str, Any]:
-    """
-    Load a Scanpy built-in dataset.
-
-    Parameters:
-        dataset_name: Name of dataset function under sc.datasets (e.g., 'pbmc3k').
-
-    Returns:
-        Dict with success/result/error. result includes shape and metadata keys.
+def create_synthetic_adata(
+    dataset_key: str = "demo",
+    n_obs: int = 200,
+    n_vars: int = 50,
+    seed: int = 0,
+) -> dict[str, Any]:
+    """Create synthetic dataset.
+
+    Parameters
+    ----------
+    dataset_key
+        Key used to store the dataset in memory.
+    n_obs
+        Number of cells (observations).
+    n_vars
+        Number of genes (variables).
+    seed
+        Random seed for reproducibility.
     """
     try:
-        if not hasattr(sc.datasets, dataset_name):
-            raise ValueError(f"Unknown dataset: {dataset_name}")
-        ds_fn = getattr(sc.datasets, dataset_name)
-        adata = ds_fn()
+        if n_obs <= 0 or n_vars <= 0:
+            return _error("n_obs and n_vars must be positive integers")
+
+        rng = np.random.default_rng(seed)
+        matrix = rng.poisson(lam=1.2, size=(n_obs, n_vars)).astype(np.float32)
+        adata = AnnData(X=matrix)
+        adata.obs_names = [f"cell_{i}" for i in range(n_obs)]
+        adata.var_names = [f"gene_{i}" for i in range(n_vars)]
+        DATASETS[dataset_key] = adata
         return _ok(
             {
-                "dataset": dataset_name,
+                "dataset_key": dataset_key,
                 "n_obs": int(adata.n_obs),
                 "n_vars": int(adata.n_vars),
-                "obs_columns": list(map(str, adata.obs.columns.tolist())),
-                "var_columns": list(map(str, adata.var.columns.tolist())),
             }
         )
-    except Exception as e:
-        return _err(e)
+    except Exception as exc:
+        return _error(str(exc))
 
 
-@mcp.tool(
-    name="read_h5ad_summary",
-    description="Read an .h5ad file and return high-level summary.",
-)
-def read_h5ad_summary(path: str) -> Dict[str, Any]:
-    """
-    Read AnnData from disk and return summary stats.
-
-    Parameters:
-        path: Path to .h5ad file.
+@mcp.tool(name="load_h5ad", description="Load a local .h5ad file into in-memory dataset registry.")
+def load_h5ad(file_path: str, dataset_key: str = "adata") -> dict[str, Any]:
+    """Load an h5ad dataset from local filesystem.
 
-    Returns:
-        Dict with success/result/error. result includes dimensions and annotation keys.
+    Parameters
+    ----------
+    file_path
+        Path to a local .h5ad file.
+    dataset_key
+        Key used for storing the dataset in memory.
     """
     try:
-        adata = sc.read_h5ad(path)
+        adata = sc.read_h5ad(file_path)
+        DATASETS[dataset_key] = adata
         return _ok(
             {
-                "path": path,
+                "dataset_key": dataset_key,
                 "n_obs": int(adata.n_obs),
                 "n_vars": int(adata.n_vars),
-                "obs_keys": list(map(str, adata.obs_keys())),
-                "var_keys": list(map(str, adata.var_keys())),
-                "obsm_keys": list(map(str, adata.obsm_keys())),
-                "uns_keys": list(map(str, adata.uns_keys())),
+                "obs_columns": list(adata.obs.columns),
+                "var_columns": list(adata.var.columns),
             }
         )
-    except Exception as e:
-        return _err(e)
+    except Exception as exc:
+        return _error(str(exc))
 
 
 @mcp.tool(
     name="preprocess_basic",
-    description="Run standard Scanpy preprocessing pipeline on a dataset.",
+    description="Run basic preprocessing: normalize_total, optional log1p, HVG selection, optional scale.",
 )
 def preprocess_basic(
-    dataset_name: str = "pbmc3k",
-    min_genes: int = 200,
-    min_cells: int = 3,
+    dataset_key: str,
     target_sum: float = 10000.0,
+    apply_log1p: bool = True,
     n_top_genes: int = 2000,
-    max_n_genes_by_counts: Optional[int] = None,
-    max_pct_counts_mt: Optional[float] = None,
-) -> Dict[str, Any]:
-    """
-    Execute a lightweight preprocessing workflow.
-
-    Parameters:
-        dataset_name: Built-in Scanpy dataset loader name.
-        min_genes: Minimum genes per cell filter.
-        min_cells: Minimum cells per gene filter.
-        target_sum: Total-count normalization target.
-        n_top_genes: Number of HVGs to keep.
-        max_n_genes_by_counts: Optional cell QC upper bound.
-        max_pct_counts_mt: Optional mitochondrial percentage upper bound.
-
-    Returns:
-        Dict with success/result/error. result includes filtered dimensions and HVG count.
+    scale_data: bool = False,
+    max_value: float = 10.0,
+) -> dict[str, Any]:
+    """Preprocess a dataset in place.
+
+    Parameters
+    ----------
+    dataset_key
+        Dataset key in memory registry.
+    target_sum
+        Target total counts after normalization.
+    apply_log1p
+        Whether to apply log1p transform.
+    n_top_genes
+        Number of highly variable genes to keep.
+    scale_data
+        Whether to z-score scale genes.
+    max_value
+        Clipping value for scaling.
     """
     try:
-        if not hasattr(sc.datasets, dataset_name):
-            raise ValueError(f"Unknown dataset: {dataset_name}")
-        adata = getattr(sc.datasets, dataset_name)()
-
-        sc.pp.filter_cells(adata, min_genes=min_genes)
-        sc.pp.filter_genes(adata, min_cells=min_cells)
-
-        adata.var["mt"] = adata.var_names.str.upper().str.startswith("MT-")
-        sc.pp.calculate_qc_metrics(adata, qc_vars=["mt"], inplace=True)
-
-        if max_n_genes_by_counts is not None:
-            adata = adata[adata.obs["n_genes_by_counts"] < max_n_genes_by_counts].copy()
-        if max_pct_counts_mt is not None and "pct_counts_mt" in adata.obs:
-            adata = adata[adata.obs["pct_counts_mt"] < max_pct_counts_mt].copy()
-
+        adata = _get_dataset(dataset_key)
         sc.pp.normalize_total(adata, target_sum=target_sum)
-        sc.pp.log1p(adata)
+        if apply_log1p:
+            sc.pp.log1p(adata)
         sc.pp.highly_variable_genes(adata, n_top_genes=n_top_genes, inplace=True)
-
-        hvg_count = int(adata.var["highly_variable"].sum()) if "highly_variable" in adata.var else 0
+        if scale_data:
+            sc.pp.scale(adata, max_value=max_value)
         return _ok(
             {
-                "dataset": dataset_name,
+                "dataset_key": dataset_key,
                 "n_obs": int(adata.n_obs),
                 "n_vars": int(adata.n_vars),
-                "highly_variable_genes": hvg_count,
+                "hvg_selected": int(adata.var.get("highly_variable", []).sum())
+                if "highly_variable" in adata.var
+                else 0,
             }
         )
-    except Exception as e:
-        return _err(e)
+    except Exception as exc:
+        return _error(str(exc))
 
 
 @mcp.tool(
-    name="compute_neighbors_and_umap",
-    description="Compute PCA, neighborhood graph, and UMAP embedding.",
+    name="run_pca_neighbors_umap",
+    description="Compute PCA, neighborhood graph, and UMAP embedding for a dataset.",
 )
-def compute_neighbors_and_umap(
-    dataset_name: str = "pbmc3k",
-    n_pcs: int = 30,
-    n_neighbors: int = 10,
+def run_pca_neighbors_umap(
+    dataset_key: str,
+    n_pcs: int = 50,
+    n_neighbors: int = 15,
+    min_dist: float = 0.5,
     random_state: int = 0,
-) -> Dict[str, Any]:
-    """
-    Run dimensionality reduction and graph embedding workflow.
-
-    Parameters:
-        dataset_name: Built-in Scanpy dataset loader name.
-        n_pcs: Number of principal components for neighbors graph.
-        n_neighbors: Number of neighbors in graph construction.
-        random_state: Random seed for PCA/UMAP reproducibility.
-
-    Returns:
-        Dict with success/result/error. result includes embedding presence and shape.
+) -> dict[str, Any]:
+    """Run core embedding workflow.
+
+    Parameters
+    ----------
+    dataset_key
+        Dataset key in memory registry.
+    n_pcs
+        Number of principal components.
+    n_neighbors
+        Number of neighbors for graph construction.
+    min_dist
+        UMAP min_dist parameter.
+    random_state
+        Random state for reproducibility.
     """
     try:
-        if not hasattr(sc.datasets, dataset_name):
-            raise ValueError(f"Unknown dataset: {dataset_name}")
-        adata = getattr(sc.datasets, dataset_name)()
-
-        sc.pp.normalize_total(adata, target_sum=1e4)
-        sc.pp.log1p(adata)
-        sc.pp.highly_variable_genes(adata, n_top_genes=2000, inplace=True)
-        if "highly_variable" in adata.var:
-            adata = adata[:, adata.var["highly_variable"]].copy()
-
-        sc.pp.scale(adata, max_value=10)
-        sc.tl.pca(adata, svd_solver="arpack", random_state=random_state)
+        adata = _get_dataset(dataset_key)
+        sc.tl.pca(adata, n_comps=n_pcs)
         sc.pp.neighbors(adata, n_neighbors=n_neighbors, n_pcs=n_pcs)
-        sc.tl.umap(adata, random_state=random_state)
-
-        umap = adata.obsm.get("X_umap")
-        umap_shape: List[int] = list(umap.shape) if umap is not None else [0, 0]
-
+        sc.tl.umap(adata, min_dist=min_dist, random_state=random_state)
         return _ok(
             {
-                "dataset": dataset_name,
-                "n_obs": int(adata.n_obs),
-                "n_vars": int(adata.n_vars),
-                "umap_shape": umap_shape,
-                "obsm_keys": list(map(str, adata.obsm_keys())),
+                "dataset_key": dataset_key,
+                "obsm_keys": list(adata.obsm.keys()),
+                "uns_keys": list(adata.uns.keys()),
             }
         )
-    except Exception as e:
-        return _err(e)
+    except Exception as exc:
+        return _error(str(exc))
 
 
 @mcp.tool(
-    name="cluster_leiden",
-    description="Run Leiden clustering after neighbors graph construction.",
+    name="run_leiden",
+    description="Run Leiden clustering and write labels to adata.obs.",
 )
-def cluster_leiden(
-    dataset_name: str = "pbmc3k",
+def run_leiden(
+    dataset_key: str,
     resolution: float = 1.0,
-    n_neighbors: int = 10,
-    n_pcs: int = 30,
-) -> Dict[str, Any]:
+    key_added: str = "leiden",
+    random_state: int = 0,
+) -> dict[str, Any]:
+    """Run Leiden clustering.
+
+    Parameters
+    ----------
+    dataset_key
+        Dataset key in memory registry.
+    resolution
+        Clustering granularity.
+    key_added
+        Column name in adata.obs storing cluster labels.
+    random_state
+        Random state.
     """
-    Perform Leiden clustering on a built-in dataset.
+    try:
+        adata = _get_dataset(dataset_key)
+        sc.tl.leiden(
+            adata,
+            resolution=resolution,
+            key_added=key_added,
+            random_state=random_state,
+        )
+        counts = adata.obs[key_added].value_counts().to_dict()
+        return _ok(
+            {
+                "dataset_key": dataset_key,
+                "key_added": key_added,
+                "cluster_counts": {str(k): int(v) for k, v in counts.items()},
+            }
+        )
+    except Exception as exc:
+        return _error(str(exc))
 
-    Parameters:
-        dataset_name: Built-in Scanpy dataset loader name.
-        resolution: Leiden resolution parameter.
-        n_neighbors: Graph neighbor count.
-        n_pcs: Number of PCs for graph construction.
 
-    Returns:
-        Dict with success/result/error. result includes cluster labels and counts.
+@mcp.tool(
+    name="rank_marker_genes",
+    description="Rank marker genes by group using tl.rank_genes_groups.",
+)
+def rank_marker_genes(
+    dataset_key: str,
+    groupby: str = "leiden",
+    method: str = "wilcoxon",
+    n_genes: int = 10,
+) -> dict[str, Any]:
+    """Rank marker genes.
+
+    Parameters
+    ----------
+    dataset_key
+        Dataset key in memory registry.
+    groupby
+        adata.obs column used for group comparison.
+    method
+        Statistical method for ranking.
+    n_genes
+        Number of genes per group to return.
     """
     try:
-        if not hasattr(sc.datasets, dataset_name):
-            raise ValueError(f"Unknown dataset: {dataset_name}")
-        adata = getattr(sc.datasets, dataset_name)()
-
-        sc.pp.normalize_total(adata, target_sum=1e4)
-        sc.pp.log1p(adata)
-        sc.pp.highly_variable_genes(adata, n_top_genes=2000, inplace=True)
-        if "highly_variable" in adata.var:
-            adata = adata[:, adata.var["highly_variable"]].copy()
-
-        sc.pp.pca(adata)
-        sc.pp.neighbors(adata, n_neighbors=n_neighbors, n_pcs=n_pcs)
-        sc.tl.leiden(adata, resolution=resolution)
-
-        counts = adata.obs["leiden"].value_counts().to_dict()
+        adata = _get_dataset(dataset_key)
+        sc.tl.rank_genes_groups(adata, groupby=groupby, method=method, n_genes=n_genes)
+        ranked = adata.uns.get("rank_genes_groups", {})
+        names = ranked.get("names")
+        output: dict[str, list[str]] = {}
+        if names is not None and hasattr(names, "dtype") and names.dtype.names:
+            for group_name in names.dtype.names:
+                output[group_name] = [str(x) for x in names[group_name][:n_genes]]
         return _ok(
             {
-                "dataset": dataset_name,
-                "resolution": resolution,
-                "n_clusters": int(len(counts)),
-                "cluster_counts": {str(k): int(v) for k, v in counts.items()},
+                "dataset_key": dataset_key,
+                "groupby": groupby,
+                "method": method,
+                "top_genes": output,
             }
         )
-    except Exception as e:
-        return _err(e)
+    except Exception as exc:
+        return _error(str(exc))
+
+
+@mcp.tool(
+    name="adapter_call_function",
+    description="Call any importable function via adapter using JSON-encoded args/kwargs.",
+)
+def adapter_call_function(
+    module_name: str,
+    function_name: str,
+    args_json: str = "[]",
+    kwargs_json: str = "{}",
+) -> dict[str, Any]:
+    """Call arbitrary function through adapter.
+
+    Parameters
+    ----------
+    module_name
+        Python module path.
+    function_name
+        Callable name inside module.
+    args_json
+        JSON array string for positional arguments.
+    kwargs_json
+        JSON object string for keyword arguments.
+    """
+    try:
+        args = json.loads(args_json)
+        kwargs = json.loads(kwargs_json)
+        if not isinstance(args, list):
+            return _error("args_json must decode to a JSON list")
+        if not isinstance(kwargs, dict):
+            return _error("kwargs_json must decode to a JSON object")
+
+        result = adapter.call_function(
+            module_name=module_name,
+            function_name=function_name,
+            args=args,
+            kwargs=kwargs,
+        )
+        if result.get("status") == "error":
+            return _error(result.get("error", "adapter call failed"))
+        return _ok(result)
+    except Exception as exc:
+        return _error(str(exc))
 
 
 def create_app() -> FastMCP:
@@ -275,4 +357,4 @@ def create_app() -> FastMCP:
 
 
 if __name__ == "__main__":
-    mcp.run()
+    mcp.run(transport="stdio")

scanpy/mcp_output/requirements.txt

@@ -1,25 +1,6 @@
 fastmcp
-fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-anndata>=0.10.8
-fast-array-utils[accel,sparse]>=1.2.1
-h5py>=3.11
-joblib
-matplotlib>=3.9
-natsort
-networkx>=2.8.8
-numba>=0.60
-numpy>=2
-packaging>=25
-pandas>=2.2.2
-patsy
-pynndescent>=0.5.13
-scikit-learn>=1.4.2
-scipy>=1.13
-seaborn>=0.13.2
-session-info2
-statsmodels>=0.14.5
-tqdm
-typing-extensions; python_version<'3.13'
-umap-learn>=0.5.7
+scanpy
+anndata
+numpy
+pandas
+scipy

scanpy/mcp_output/start_mcp.py

@@ -1,30 +1,38 @@
+from __future__ import annotations
 
-"""
-MCP Service Startup Entry
-"""
-import sys
 import os
+import sys
+from pathlib import Path
 
-project_root = os.path.dirname(os.path.abspath(__file__))
-mcp_plugin_dir = os.path.join(project_root, "mcp_plugin")
-if mcp_plugin_dir not in sys.path:
-    sys.path.insert(0, mcp_plugin_dir)
+PLUGIN_DIR = Path(__file__).resolve().parent / "mcp_plugin"
+plugin_path = str(PLUGIN_DIR)
+if plugin_path not in sys.path:
+    sys.path.insert(0, plugin_path)
 
 from mcp_service import create_app
 
-def main():
-    """Start FastMCP service"""
+
+def main() -> None:
+    transport = os.getenv("MCP_TRANSPORT", "stdio").strip().lower()
+    port_value = os.getenv("MCP_PORT", "8000").strip()
+
+    try:
+        port = int(port_value)
+    except ValueError:
+        raise ValueError(f"Invalid MCP_PORT: {port_value}")
+
     app = create_app()
-    # Use environment variable to configure port, default 8000
-    port = int(os.environ.get("MCP_PORT", "8000"))
-    
-    # Choose transport mode based on environment variable
-    transport = os.environ.get("MCP_TRANSPORT", "stdio")
+
+    if transport == "stdio":
+        app.run(transport="stdio")
+        return
+
     if transport == "http":
         app.run(transport="http", host="0.0.0.0", port=port)
-    else:
-        # Default to STDIO mode
-        app.run()
+        return
+
+    raise ValueError(f"Unsupported MCP_TRANSPORT: {transport}")
+
 
 if __name__ == "__main__":
     main()