alphagenome_mcp.py

"""
Model Context Protocol (MCP) for AlphaGenome

AlphaGenome provides state-of-the-art AI-driven tools for genomic sequence analysis, variant effect prediction, and functional genomics visualization. The platform enables researchers to predict regulatory elements, assess variant impacts, and explore molecular mechanisms across diverse cell types and tissues.

This MCP Server contains the tools extracted from the following tutorials with their tools:
1. batch_variant_scoring
    - score_variants_batch: Score multiple genetic variants in batch using configurable variant scorers
    - filter_variant_scores: Filter variant scores by ontology criteria (e.g., cell types)
2. essential_commands
    - create_genomic_interval: Create genomic intervals for DNA regions
    - create_genomic_variant: Create genomic variants for genetic changes  
    - create_track_data: Create TrackData objects from arrays and metadata
    - create_variant_scores: Create AnnData objects for variant scoring results
    - genomic_interval_operations: Perform operations on genomic intervals
    - variant_interval_operations: Check variant overlaps with intervals  
    - track_data_operations: Filter, resize, slice and transform TrackData
    - track_data_resolution_conversion: Convert between track data resolutions
3. quick_start
    - predict_dna_sequence: Predict genomic tracks from DNA sequence
    - predict_genome_interval: Predict genomic tracks for reference genome intervals  
    - ism_analysis: Perform in silico mutagenesis analysis with sequence logos
4. tissue_ontology_mapping
    - explore_output_metadata: Explore and filter output metadata for specific organisms and search terms
    - count_tracks_by_output_type: Count tracks by output type for human and mouse organisms
5. variant_scoring_ui
    - visualize_variant_effects: Generate comprehensive variant effect visualization across multiple modalities
6. visualization_modality_tour
    - visualize_gene_expression: Visualize RNA_SEQ and CAGE gene expression predictions
    - visualize_chromatin_accessibility: Visualize DNASE and ATAC chromatin accessibility
    - visualize_splicing_effects: Visualize splicing predictions with SPLICE_SITES, SPLICE_SITE_USAGE, SPLICE_JUNCTIONS
    - visualize_histone_modifications: Visualize CHIP_HISTONE predictions with custom colors
    - visualize_tf_binding: Visualize CHIP_TF transcription factor binding predictions
    - visualize_contact_maps: Visualize CONTACT_MAPS DNA-DNA contact predictions
"""
from fastmcp import FastMCP

# Import the MCP tools from the tools folder
from tools.batch_variant_scoring import batch_variant_scoring_mcp
from tools.essential_commands import essential_commands_mcp
from tools.quick_start import quick_start_mcp
from tools.tissue_ontology_mapping import tissue_ontology_mapping_mcp
from tools.variant_scoring_ui import variant_scoring_ui_mcp
from tools.visualization_modality_tour import visualization_modality_tour_mcp

from starlette.requests import Request
from starlette.responses import PlainTextResponse, JSONResponse
import os
from fastapi.staticfiles import StaticFiles
import uuid

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

# Mount the tools
mcp.mount(batch_variant_scoring_mcp)
mcp.mount(essential_commands_mcp)
mcp.mount(quick_start_mcp)
mcp.mount(tissue_ontology_mapping_mcp)
mcp.mount(variant_scoring_ui_mcp)
mcp.mount(visualization_modality_tour_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-alphagenome-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")

    
if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=int(os.getenv("PORT", 7860)), path="/mcp")