Update biopython/mcp_output/mcp_plugin/mcp_service.py

biopython/mcp_output/mcp_plugin/mcp_service.py

@@ -1,213 +1,474 @@
 import os
 import sys
+from io import StringIO
 
+# Ensure Biopython source is importable when running the MCP service
 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)
+if source_path not in sys.path:
+    sys.path.insert(0, source_path)
 
 from fastmcp import FastMCP
-from Bio.SeqIO import parse, read, write
+
+# Core Biopython utilities from setup.py
+from setup import can_import as biopy_can_import
+from setup import get_version as biopy_get_version
+
+# Script entrypoints (aliased to avoid name shadowing)
+from Scripts.query_pubmed import print_usage as query_pubmed_print_usage
+from Scripts.update_ncbi_codon_table import line_wrap as codon_line_wrap
+from Scripts.scop_pdb import main as scop_main
+from Scripts.scop_pdb import open_pdb as scop_open_pdb
+from Scripts.scop_pdb import usage as scop_usage
+from Scripts.xbbtools.nextorf import MissingTable
+from Scripts.xbbtools.nextorf import NextOrf
+from Scripts.xbbtools.nextorf import help as nextorf_help
+from Scripts.xbbtools.nextorf import makeTableX
+from Scripts.xbbtools.xbb_utils import NotePad
+
+# Biopython top-level version
+from Bio import __version__ as bio_version
+from Bio import Entrez, SeqIO, pairwise2, Phylo, AlignIO
+from Bio.PDB import PDBParser
+from Bio.Align import MultipleSeqAlignment
 from Bio.Seq import Seq
-from Bio.SeqRecord import SeqRecord
-from Bio.Blast.NCBIWWW import qblast
-from Bio.Entrez import efetch, esearch, email
-import Bio.Entrez
-
-# 设置 NCBI Entrez email（必需）
-Bio.Entrez.email = "biopython-mcp@huggingface.co"
-
-mcp = FastMCP("biopython_service")
-
-def seqrecord_to_dict(record):
-    """Convert a SeqRecord object to a serializable dictionary."""
-    return {
-        "id": str(record.id),
-        "name": str(record.name),
-        "description": str(record.description),
-        "sequence": str(record.seq),
-        "length": len(record.seq),
-        "annotations": dict(record.annotations) if record.annotations else {},
-        "features": [
-            {
-                "type": f.type,
-                "location": str(f.location),
-                "qualifiers": dict(f.qualifiers)
-            }
-            for f in record.features
-        ] if record.features else []
-    }
-
-def dict_to_seqrecord(seq_dict):
-    """Convert a dictionary back to a SeqRecord object."""
-    if isinstance(seq_dict, SeqRecord):
-        return seq_dict
-    
-    # 从字典创建 SeqRecord
-    record = SeqRecord(
-        Seq(seq_dict.get("sequence", "")),
-        id=seq_dict.get("id", ""),
-        name=seq_dict.get("name", ""),
-        description=seq_dict.get("description", "")
-    )
-    
-    # 添加注释
-    if "annotations" in seq_dict:
-        record.annotations.update(seq_dict["annotations"])
-    
-    return record
-
-@mcp.tool(name="seqio_parse", description="Parse sequence data from a file.")
-def seqio_parse(file_path: str, format: str) -> dict:
-    """
-    Parses sequence data from a file.
-
-    Parameters:
-    - file_path: Path to the sequence file.
-    - format: Format of the sequence file (e.g., 'fasta').
-
-    Returns:
-    - A dictionary with success status and parsed sequences or error message.
-    """
-    try:
-        sequences = list(parse(file_path, format))
-        # Convert SeqRecord objects to serializable dictionaries
-        result = [seqrecord_to_dict(seq) for seq in sequences]
+from Bio.SeqUtils import GC
+
+mcp = FastMCP("biopython")
+
+# Configure Entrez email (required by NCBI); allow user override via env
+Entrez.email = os.getenv("BIOPYTHON_ENTREZ_EMAIL", "biopython-mcp@huggingface.co")
+
+
+@mcp.tool(name="can_import", description="Check if a module can be imported (from setup.py)")
+def can_import_tool(payload: dict):
+    try:
+        result = biopy_can_import(**payload)
+        return {"success": True, "result": str(result), "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="get_version", description="Biopython package version from setup.py")
+def get_version_tool(payload: dict | None = None):
+    try:
+        result = biopy_get_version()
         return {"success": True, "result": result, "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="seqio_read", description="Read a single sequence from a file.")
-def seqio_read(file_path: str, format: str) -> dict:
-    """
-    Reads a single sequence from a file.
 
-    Parameters:
-    - file_path: Path to the sequence file.
-    - format: Format of the sequence file (e.g., 'fasta').
+@mcp.tool(name="bio_version", description="Return Bio.__version__ from the Biopython package")
+def bio_version_tool(payload: dict | None = None):
+    try:
+        return {"success": True, "result": bio_version, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="query_pubmed_print_usage", description="Print usage for query_pubmed script")
+def print_usage_tool(payload: dict | None = None):
+    try:
+        result = query_pubmed_print_usage()
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="line_wrap", description="Wrap long strings (from update_ncbi_codon_table.py)")
+def line_wrap_tool(payload: dict):
+    try:
+        result = codon_line_wrap(**payload)
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="scop_main", description="Extract SCOP domain PDB records (Scripts/scop_pdb.py main)")
+def scop_main_tool(payload: dict | None = None):
+    try:
+        result = scop_main()
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="scop_open_pdb", description="Download/open a PDB file (Scripts/scop_pdb.py open_pdb)")
+def open_pdb_tool(payload: dict):
+    try:
+        result = scop_open_pdb(**payload)
+        return {"success": True, "result": str(result), "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
 
-    Returns:
-    - A dictionary with success status and the sequence or error message.
-    """
+@mcp.tool(name="scop_usage", description="Print usage for scop_pdb.py")
+def usage_tool(payload: dict | None = None):
     try:
-        sequence = read(file_path, format)
-        # Convert SeqRecord object to serializable dictionary
-        result = seqrecord_to_dict(sequence)
+        result = scop_usage()
         return {"success": True, "result": result, "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="seqio_write", description="Write sequences to a file.")
-def seqio_write(sequences, file_path: str, format: str) -> dict:
-    """
-    Writes sequences to a file.
-
-    Parameters:
-    - sequences: List of sequences to write (can be SeqRecord objects or dicts).
-    - file_path: Path to the output file.
-    - format: Format of the output file (e.g., 'fasta').
-
-    Returns:
-    - A dictionary with success status and number of records written or error message.
-    """
-    try:
-        # 如果 sequences 是列表，将每个元素转换为 SeqRecord
-        if isinstance(sequences, list):
-            records = []
-            for seq in sequences:
-                if isinstance(seq, dict):
-                    records.append(dict_to_seqrecord(seq))
-                else:
-                    records.append(seq)
-        else:
-            # 如果是单个对象
-            if isinstance(sequences, dict):
-                records = [dict_to_seqrecord(sequences)]
+@mcp.tool(name="notepad", description="Instantiate Scripts/xbbtools/xbb_utils.NotePad")
+def notepad_tool(*args, **kwargs):
+    """NotePad class"""
+    try:
+        if NotePad is None:
+            return {"success": False, "result": None, "error": "Class NotePad is not available, path may need adjustment"}
+        
+        # MCP parameter type conversion
+        converted_args = []
+        converted_kwargs = kwargs.copy()
+        
+        # Handle position argument type conversion
+        for arg in args:
+            if isinstance(arg, str):
+                # Try to convert to numeric type
+                try:
+                    if '.' in arg:
+                        converted_args.append(float(arg))
+                    else:
+                        converted_args.append(int(arg))
+                except ValueError:
+                    converted_args.append(arg)
             else:
-                records = [sequences]
+                converted_args.append(arg)
+        
+        # Handle keyword argument type conversion
+        for key, value in converted_kwargs.items():
+            if isinstance(value, str):
+                try:
+                    if '.' in value:
+                        converted_kwargs[key] = float(value)
+                    else:
+                        converted_kwargs[key] = int(value)
+                except ValueError:
+                    pass
         
-        count = write(records, file_path, format)
-        return {"success": True, "result": count, "error": None}
+        instance = NotePad(*converted_args, **converted_kwargs)
+        return {"success": True, "result": str(instance), "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="blast_qblast", description="Run a BLAST query using NCBI's BLAST service.")
-def blast_qblast(program: str, database: str, sequence: str) -> dict:
-    """
-    Runs a BLAST query using NCBI's BLAST service.
+@mcp.tool(name="nextorf_help", description="Help text for NextOrf (Scripts/xbbtools/nextorf.py)")
+def help_tool(payload: dict | None = None):
+    try:
+        result = nextorf_help()
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
 
-    Parameters:
-    - program: BLAST program to use (e.g., 'blastn').
-    - database: Database to search against (e.g., 'nt').
-    - sequence: Sequence to search.
+@mcp.tool(name="makeTableX", description="Generate NextOrf lookup table (Scripts/xbbtools/nextorf.py)")
+def makeTableX_tool(payload: dict):
+    try:
+        result = makeTableX(**payload)
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
 
-    Returns:
-    - A dictionary with success status and BLAST result or error message.
-    """
+@mcp.tool(name="missingtable", description="Instantiate MissingTable (Scripts/xbbtools/nextorf.py)")
+def missingtable_tool(*args, **kwargs):
+    """MissingTable class"""
     try:
-        result_handle = qblast(program, database, sequence)
-        return {"success": True, "result": result_handle.read(), "error": None}
+        if MissingTable is None:
+            return {"success": False, "result": None, "error": "Class MissingTable is not available, path may need adjustment"}
+        
+        # MCP parameter type conversion
+        converted_args = []
+        converted_kwargs = kwargs.copy()
+        
+        # Handle position argument type conversion
+        for arg in args:
+            if isinstance(arg, str):
+                # Try to convert to numeric type
+                try:
+                    if '.' in arg:
+                        converted_args.append(float(arg))
+                    else:
+                        converted_args.append(int(arg))
+                except ValueError:
+                    converted_args.append(arg)
+            else:
+                converted_args.append(arg)
+        
+        # Handle keyword argument type conversion
+        for key, value in converted_kwargs.items():
+            if isinstance(value, str):
+                try:
+                    if '.' in value:
+                        converted_kwargs[key] = float(value)
+                    else:
+                        converted_kwargs[key] = int(value)
+                except ValueError:
+                    pass
+        
+        instance = MissingTable(*converted_args, **converted_kwargs)
+        return {"success": True, "result": str(instance), "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+@mcp.tool(name="nextorf", description="Instantiate NextOrf (Scripts/xbbtools/nextorf.py)")
+def nextorf_tool(*args, **kwargs):
+    """NextOrf class"""
+    try:
+        if NextOrf is None:
+            return {"success": False, "result": None, "error": "Class NextOrf is not available, path may need adjustment"}
+        
+        # MCP parameter type conversion
+        converted_args = []
+        converted_kwargs = kwargs.copy()
+        
+        # Handle position argument type conversion
+        for arg in args:
+            if isinstance(arg, str):
+                # Try to convert to numeric type
+                try:
+                    if '.' in arg:
+                        converted_args.append(float(arg))
+                    else:
+                        converted_args.append(int(arg))
+                except ValueError:
+                    converted_args.append(arg)
+            else:
+                converted_args.append(arg)
+        
+        # Handle keyword argument type conversion
+        for key, value in converted_kwargs.items():
+            if isinstance(value, str):
+                try:
+                    if '.' in value:
+                        converted_kwargs[key] = float(value)
+                    else:
+                        converted_kwargs[key] = int(value)
+                except ValueError:
+                    pass
+        
+        instance = NextOrf(*converted_args, **converted_kwargs)
+        return {"success": True, "result": str(instance), "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="entrez_efetch", description="Fetch data from NCBI's Entrez databases.")
-def entrez_efetch(db: str, id: str, rettype: str, retmode: str) -> dict:
-    """
-    Fetches data from NCBI's Entrez databases.
 
-    Parameters:
-    - db: Database to fetch from (e.g., 'nucleotide').
-    - id: ID of the record to fetch.
-    - rettype: Return type (e.g., 'gb').
-    - retmode: Return mode (e.g., 'text').
+# -----------------------
+# Core Biopython utilities
+# -----------------------
 
-    Returns:
-    - A dictionary with success status and fetched data or error message.
-    """
+
+@mcp.tool(name="seq_reverse_complement", description="Reverse-complement a DNA/RNA sequence string")
+def seq_reverse_complement(payload: dict):
     try:
-        handle = efetch(db=db, id=id, rettype=rettype, retmode=retmode)
-        return {"success": True, "result": handle.read(), "error": None}
+        seq = payload.get("sequence", "")
+        result = str(Seq(seq).reverse_complement())
+        return {"success": True, "result": result, "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="entrez_esearch", description="Search NCBI's Entrez databases.")
-def entrez_esearch(db: str, term: str) -> dict:
-    """
-    Searches NCBI's Entrez databases.
 
-    Parameters:
-    - db: Database to search (e.g., 'nucleotide', 'protein', 'gene', 'pubmed').
-    - term: Search term. Use proper NCBI query syntax:
-        - gene_name[GENE] - Search by gene name
-        - organism[ORGN] - Search by organism
-        - "Homo sapiens"[ORGN] - Species filter
-        - "RefSeq"[Filter] - Only RefSeq records
+@mcp.tool(name="seq_transcribe", description="Transcribe DNA to RNA (replace T->U)")
+def seq_transcribe(payload: dict):
+    try:
+        seq = payload.get("sequence", "")
+        result = str(Seq(seq).transcribe())
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
 
-    Returns:
-    - A dictionary with success status and search results or error message.
-    
-    Examples:
-    - "TP53[gene] AND human[organism]"
-    - "BRCA1[gene]"
-    - "insulin[protein name]"
-    """
+@mcp.tool(name="seq_translate", description="Translate DNA/RNA to protein; accepts optional table, to_stop")
+def seq_translate(payload: dict):
     try:
-        # 确保 email 已设置
-        if not Bio.Entrez.email:
-            Bio.Entrez.email = "biopython-mcp@huggingface.co"
-        
-        # 使用 retmax 参数获取更多结果
-        handle = esearch(db=db, term=term, retmax=100)
-        result = handle.read()
-        handle.close()
-        
+        seq = payload.get("sequence", "")
+        table = payload.get("table", 1)
+        to_stop = bool(payload.get("to_stop", False))
+        cds = bool(payload.get("cds", False))
+        result = str(Seq(seq).translate(table=table, to_stop=to_stop, cds=cds))
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="gc_content", description="Compute GC percentage of a sequence")
+def gc_content(payload: dict):
+    try:
+        seq = payload.get("sequence", "")
+        result = GC(seq)
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="pairwise_align_globalxx", description="Global alignment (identity scoring) between two sequences")
+def pairwise_align_globalxx(payload: dict):
+    try:
+        seq1 = payload.get("seq1", "")
+        seq2 = payload.get("seq2", "")
+        limit = int(payload.get("limit", 3))
+        aligns = pairwise2.align.globalxx(seq1, seq2)[:limit]
+        result = [
+            {
+                "seqA": a.seqA,
+                "seqB": a.seqB,
+                "score": a.score,
+                "start": a.start,
+                "end": a.end,
+            }
+            for a in aligns
+        ]
         return {"success": True, "result": result, "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-def create_app() -> FastMCP:
-    """
-    Creates and returns the FastMCP application instance.
 
-    Returns:
-    - FastMCP instance.
-    """
-    return mcp
+@mcp.tool(name="seqio_parse_string", description="Parse FASTA/GenBank data from string; returns list of records")
+def seqio_parse_string(payload: dict):
+    try:
+        data = payload.get("data", "")
+        fmt = payload.get("format", "fasta")
+        handle = StringIO(data)
+        records = []
+        for rec in SeqIO.parse(handle, fmt):
+            records.append({"id": rec.id, "name": rec.name, "description": rec.description, "seq": str(rec.seq)})
+        return {"success": True, "result": records, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="seqio_convert_file", description="Convert sequence file between formats (e.g., fasta->genbank)")
+def seqio_convert_file(payload: dict):
+    try:
+        input_path = payload["input_path"]
+        input_format = payload.get("input_format", "fasta")
+        output_path = payload["output_path"]
+        output_format = payload.get("output_format", "genbank")
+        count = SeqIO.convert(input_path, input_format, output_path, output_format)
+        return {"success": True, "result": {"converted_records": count, "output_path": output_path}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="entrez_search", description="NCBI Entrez esearch; provide db, term; optional retmax")
+def entrez_search(payload: dict):
+    try:
+        db = payload.get("db", "pubmed")
+        term = payload.get("term", "")
+        retmax = int(payload.get("retmax", 20))
+        handle = Entrez.esearch(db=db, term=term, retmax=retmax, usehistory="y")
+        rec = Entrez.read(handle)
+        handle.close()
+        return {
+            "success": True,
+            "result": {"ids": rec.get("IdList", []), "count": int(rec.get("Count", 0)), "webenv": rec.get("WebEnv"), "query_key": rec.get("QueryKey")},
+            "error": None,
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="entrez_fetch", description="NCBI Entrez efetch; provide db, id(s), rettype, retmode")
+def entrez_fetch(payload: dict):
+    try:
+        db = payload.get("db", "pubmed")
+        ids = payload.get("ids", [])
+        if isinstance(ids, list):
+            ids = ",".join(ids)
+        rettype = payload.get("rettype", "medline")
+        retmode = payload.get("retmode", "text")
+        handle = Entrez.efetch(db=db, id=ids, rettype=rettype, retmode=retmode)
+        data = handle.read()
+        handle.close()
+        return {"success": True, "result": data, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="set_entrez_email", description="Override Entrez.email at runtime; required for NCBI")
+def set_entrez_email(payload: dict):
+    try:
+        email = payload.get("email", "")
+        if not email:
+            return {"success": False, "result": None, "error": "email is required"}
+        Entrez.email = email
+        return {"success": True, "result": {"email": Entrez.email}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="phylo_parse_newick", description="Parse Newick string; return leaf names and counts")
+def phylo_parse_newick(payload: dict):
+    try:
+        newick = payload.get("newick", "")
+        tree = Phylo.read(StringIO(newick), "newick")
+        leaves = [term.name for term in tree.get_terminals()]
+        return {"success": True, "result": {"leaf_count": len(leaves), "leaves": leaves}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="pdb_summary", description="Parse a PDB file; report chains, residues, atoms")
+def pdb_summary(payload: dict):
+    try:
+        path = payload.get("path", "")
+        structure_id = payload.get("structure_id", "structure")
+        parser = PDBParser(QUIET=True)
+        structure = parser.get_structure(structure_id, path)
+        chains = []
+        for model in structure:
+            for chain in model:
+                atom_count = sum(1 for _ in chain.get_atoms())
+                res_count = sum(1 for _ in chain.get_residues())
+                chains.append({"id": chain.id, "residues": res_count, "atoms": atom_count})
+        return {"success": True, "result": {"chains": chains}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="msa_read_fasta", description="Read a FASTA alignment file; return length and sequences")
+def msa_read_fasta(payload: dict):
+    try:
+        path = payload.get("path", "")
+        alignment = AlignIO.read(path, "fasta")
+        seqs = [{"id": rec.id, "seq": str(rec.seq)} for rec in alignment]
+        aln_len = alignment.get_alignment_length() if isinstance(alignment, MultipleSeqAlignment) else None
+        return {"success": True, "result": {"alignment_length": aln_len, "sequences": seqs}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="msa_consensus", description="Compute simple consensus from FASTA alignment file")
+def msa_consensus(payload: dict):
+    try:
+        path = payload.get("path", "")
+        alignment = AlignIO.read(path, "fasta")
+        consensus = alignment.column_annotations.get("consensus") if hasattr(alignment, "column_annotations") else None
+        if consensus is None:
+            # naive majority rule per column
+            aln_len = alignment.get_alignment_length()
+            cols = []
+            for i in range(aln_len):
+                counts = {}
+                for rec in alignment:
+                    c = rec.seq[i]
+                    counts[c] = counts.get(c, 0) + 1
+                cols.append(max(counts, key=counts.get))
+            consensus = "".join(cols)
+        return {"success": True, "result": {"consensus": str(consensus)}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="fasta_filter_length", description="Filter FASTA sequences by minimum length; returns kept sequences")
+def fasta_filter_length(payload: dict):
+    try:
+        data = payload.get("data", "")
+        min_len = int(payload.get("min_len", 1))
+        handle = StringIO(data)
+        kept = []
+        for rec in SeqIO.parse(handle, "fasta"):
+            if len(rec.seq) >= min_len:
+                kept.append({"id": rec.id, "seq": str(rec.seq), "len": len(rec.seq)})
+        return {"success": True, "result": {"kept": kept, "count": len(kept)}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+
+def create_app():
+    """Create and return FastMCP application instance"""
+    return mcp
+
+if __name__ == "__main__":
+    mcp.run(transport="http", host="0.0.0.0", port=8000)