scanpy_mcp.py

"""
Model Context Protocol (MCP) for scanpy

Scanpy is a scalable toolkit for analyzing single-cell gene expression data built jointly with anndata. 
It provides preprocessing, visualization, clustering, pseudotime and trajectory inference, differential expression testing, and integration of heterogeneous datasets.
This codebase focuses on fundamental single-cell RNA sequencing analysis workflows including quality control, normalization, dimensionality reduction, and clustering.

This MCP Server contains the tools extracted from the following tutorials:
1. clustering
    - quality_control: Calculate and visualize QC metrics, filter cells and genes, detect doublets
    - normalize_data: Normalize count data with median total counts and log transformation
    - select_features: Identify highly variable genes for feature selection
    - reduce_dimensionality: Perform PCA analysis and variance visualization
    - build_neighborhood_graph: Construct nearest neighbor graph and UMAP embedding
    - cluster_cells: Perform Leiden clustering with visualization
    - annotate_cell_types: Multi-resolution clustering, marker gene analysis, and differential expression
"""

import sys
from pathlib import Path
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import PlainTextResponse, JSONResponse
import os
from fastapi.staticfiles import StaticFiles
import uuid
import os


# Import the MCP tools from the tools folder
from tools.clustering import clustering_mcp

# Define the MCP server
mcp = FastMCP(name = "scanpy")

# Mount the tools
mcp.mount(clustering_mcp)

# Use absolute directory for uploads
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
UPLOAD_DIR = os.path.join(BASE_DIR, "/data/upload")
os.makedirs(UPLOAD_DIR, exist_ok=True)

@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> PlainTextResponse:
    return PlainTextResponse("OK")


@mcp.custom_route("/", methods=["GET"])
async def index(request: Request) -> PlainTextResponse:
    return PlainTextResponse("MCP is on https://Paper2Agent-scanpy-mcp.hf.space/mcp")

# Upload route
@mcp.custom_route("/upload", methods=["POST"])
async def upload(request: Request):
    form = await request.form()
    up = form.get("file")
    if up is None:
        return JSONResponse({"error": "missing form field 'file'"}, status_code=400)

    # Generate a safe filename
    orig = getattr(up, "filename", "") or ""
    ext = os.path.splitext(orig)[1]
    name = f"{uuid.uuid4().hex}{ext}"
    dst = os.path.join(UPLOAD_DIR, name)
    
    # up is a Starlette UploadFile-like object
    with open(dst, "wb") as out:
        out.write(await up.read())

    # Return only the absolute local path
    abs_path = os.path.abspath(dst)
    return JSONResponse({"path": abs_path})

app = mcp.http_app(path="/mcp")
# Saved uploaded input files
app.mount("/files", StaticFiles(directory=UPLOAD_DIR), name="files")
# Saved output files
app.mount("/outputs", StaticFiles(directory="/data/tmp_outputs"), name="outputs")

# Run the MCP server
if __name__ == "__main__":
  mcp.run(transport="http", host="127.0.0.1", port=8003)