Update biopython/mcp_output/mcp_plugin/mcp_service.py

biopython/mcp_output/mcp_plugin/mcp_service.py

@@ -1,7 +1,6 @@
 import os
 import sys
-import ast
-import importlib
+import re
 from io import StringIO
 
 # Ensure Biopython source is importable when running the MCP service
@@ -11,485 +10,592 @@ if source_path not in sys.path:
 
 from fastmcp import FastMCP
 
-# Lightweight stand-ins for setup.py utilities to avoid running setup()
-def biopy_can_import(module_name):
-    try:
-        return importlib.import_module(module_name)
-    except ImportError:
-        return None
-
-
-def biopy_get_version():
-    init_path = os.path.join(source_path, "Bio", "__init__.py")
-    try:
-        with open(init_path, "r", encoding="utf-8") as handle:
-            for line in handle:
-                if line.startswith("__version__ = "):
-                    return ast.literal_eval(line.split("=", 1)[1].strip())
-    except FileNotFoundError:
-        return "Unknown"
-    return "Unknown"
-
-# ✅ 延迟导入：只在需要时才导入 Scripts 模块
-# 这样可以避免在服务启动时触发 setuptools
-_scripts_cache = {}
-
-def lazy_import_script(module_path, attr_name):
-    """延迟导入 Scripts 模块中的函数或类"""
-    cache_key = f"{module_path}.{attr_name}"
-    if cache_key not in _scripts_cache:
-        try:
-            module = importlib.import_module(module_path)
-            _scripts_cache[cache_key] = getattr(module, attr_name)
-        except Exception as e:
-            print(f"⚠️ Warning: Could not import {cache_key}: {e}")
-            _scripts_cache[cache_key] = None
-    return _scripts_cache[cache_key]
-
-# Biopython top-level imports (这些是安全的)
+# Biopython imports
 from Bio import __version__ as bio_version
-from Bio import Entrez, SeqIO, Phylo, AlignIO
-from Bio.PDB import PDBParser
-from Bio.Align import MultipleSeqAlignment, PairwiseAligner
+from Bio import Entrez, SeqIO
 from Bio.Seq import Seq
+from Bio.Blast import NCBIWWW, NCBIXML
 
-# GC function location varies by Biopython version
 try:
     from Bio.SeqUtils import GC
 except ImportError:
     try:
         from Bio.SeqUtils import gc_fraction as GC
     except ImportError:
-        # Fallback implementation
         def GC(seq):
             """Calculate GC content percentage"""
             seq = str(seq).upper()
             gc = seq.count('G') + seq.count('C')
             return (gc / len(seq)) * 100 if len(seq) > 0 else 0
 
-# pairwise2 is deprecated, but keep for compatibility
-try:
-    from Bio import pairwise2
-    HAS_PAIRWISE2 = True
-except (ImportError, AttributeError):
-    HAS_PAIRWISE2 = False
-
-mcp = FastMCP("biopython")
+mcp = FastMCP("biopython_gene_species_identification")
 
-# Configure Entrez email (required by NCBI); allow user override via env
+# Configure Entrez email (required by NCBI)
 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)}
+# ==============================================
+# 基础工具 - Basic Utilities
+# ==============================================
 
-@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="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):
+@mcp.tool(name="set_entrez_email", description="Set NCBI Entrez email (required for all NCBI operations)")
+def set_entrez_email(payload: dict):
+    """
+    设置 NCBI Entrez 邮箱
+    Required fields: email
+    """
     try:
-        func = lazy_import_script("Scripts.query_pubmed", "print_usage")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func()
-        return {"success": True, "result": result, "error": None}
+        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="line_wrap", description="Wrap long strings (from update_ncbi_codon_table.py)")
-def line_wrap_tool(payload: dict):
-    try:
-        func = lazy_import_script("Scripts.update_ncbi_codon_table", "line_wrap")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func(**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):
+@mcp.tool(name="validate_sequence", description="Validate if input is a valid DNA/RNA sequence")
+def validate_sequence(payload: dict):
+    """
+    验证序列是否为有效的 DNA/RNA 序列
+    Required fields: sequence
+    Optional fields: sequence_type (dna/rna/auto)
+    """
     try:
-        func = lazy_import_script("Scripts.scop_pdb", "main")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func()
-        return {"success": True, "result": result, "error": None}
+        sequence = payload.get("sequence", "").strip().upper()
+        seq_type = payload.get("sequence_type", "auto").lower()
+        
+        if not sequence:
+            return {"success": False, "result": None, "error": "sequence is required"}
+        
+        # Remove whitespace and newlines
+        sequence = "".join(sequence.split())
+        
+        # Check for valid DNA/RNA characters (including ambiguity codes)
+        dna_chars = set("ATCGN")
+        rna_chars = set("AUCGN")
+        seq_chars = set(sequence)
+        
+        is_dna = seq_chars.issubset(dna_chars)
+        is_rna = seq_chars.issubset(rna_chars)
+        
+        if seq_type == "auto":
+            if is_dna:
+                detected_type = "dna"
+            elif is_rna:
+                detected_type = "rna"
+            else:
+                return {"success": False, "result": None, 
+                       "error": f"Invalid sequence characters found: {seq_chars - dna_chars - rna_chars}"}
+        elif seq_type == "dna":
+            if not is_dna:
+                return {"success": False, "result": None, "error": "Not a valid DNA sequence"}
+            detected_type = "dna"
+        elif seq_type == "rna":
+            if not is_rna:
+                return {"success": False, "result": None, "error": "Not a valid RNA sequence"}
+            detected_type = "rna"
+        else:
+            return {"success": False, "result": None, "error": "sequence_type must be dna, rna, or auto"}
+        
+        return {
+            "success": True,
+            "result": {
+                "valid": True,
+                "sequence": sequence,
+                "length": len(sequence),
+                "type": detected_type
+            },
+            "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:
-        func = lazy_import_script("Scripts.scop_pdb", "open_pdb")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func(**payload)
-        return {"success": True, "result": str(result), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="scop_usage", description="Print usage for scop_pdb.py")
-def usage_tool(payload: dict | None = None):
+@mcp.tool(name="calculate_gc_content", description="Calculate GC content percentage of a sequence")
+def calculate_gc_content(payload: dict):
+    """
+    计算序列的 GC 含量
+    Required fields: sequence
+    """
     try:
-        func = lazy_import_script("Scripts.scop_pdb", "usage")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func()
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="notepad", description="Instantiate Scripts/xbbtools/xbb_utils.NotePad")
-def notepad_tool(payload: dict):
-    """NotePad class - pass constructor arguments in payload"""
-    try:
-        NotePad = lazy_import_script("Scripts.xbbtools.xbb_utils", "NotePad")
-        if NotePad is None:
-            return {"success": False, "result": None, "error": "Class NotePad is not available"}
+        sequence = payload.get("sequence", "")
+        if not sequence:
+            return {"success": False, "result": None, "error": "sequence is required"}
         
-        # Extract args and kwargs from payload
-        args = payload.get("args", [])
-        kwargs = payload.get("kwargs", {})
+        gc_pct = GC(sequence)
         
-        instance = NotePad(*args, **kwargs)
-        return {"success": True, "result": str(instance), "error": None}
+        return {
+            "success": True,
+            "result": {
+                "gc_content": round(gc_pct, 2),
+                "sequence_length": len(sequence)
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="nextorf_help", description="Help text for NextOrf (Scripts/xbbtools/nextorf.py)")
-def help_tool(payload: dict | None = None):
-    try:
-        func = lazy_import_script("Scripts.xbbtools.nextorf", "help")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func()
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="makeTableX", description="Generate NextOrf lookup table (Scripts/xbbtools/nextorf.py)")
-def makeTableX_tool(payload: dict):
-    try:
-        func = lazy_import_script("Scripts.xbbtools.nextorf", "makeTableX")
-        if func is None:
-            return {"success": False, "result": None, "error": "Function not available"}
-        result = func(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+# ==============================================
+# BLAST 搜索工具 - BLAST Search Tools
+# ==============================================
 
-@mcp.tool(name="missingtable", description="Instantiate MissingTable (Scripts/xbbtools/nextorf.py)")
-def missingtable_tool(payload: dict):
-    """MissingTable class - pass constructor arguments in payload"""
+@mcp.tool(name="blast_search", description="Search NCBI database using BLAST to find similar sequences")
+def blast_search(payload: dict):
+    """
+    使用 BLAST 搜索相似序列
+    Required fields: sequence
+    Optional fields: database (nt/nr), program (blastn/blastp), hitlist_size (default 20), expect (default 1e-10)
+    """
     try:
-        MissingTable = lazy_import_script("Scripts.xbbtools.nextorf", "MissingTable")
-        if MissingTable is None:
-            return {"success": False, "result": None, "error": "Class MissingTable is not available"}
+        sequence = str(payload.get("sequence", "")).strip()
+        if not sequence:
+            return {"success": False, "result": None, "error": "sequence is required"}
         
-        # Extract args and kwargs from payload
-        args = payload.get("args", [])
-        kwargs = payload.get("kwargs", {})
+        # Clean sequence
+        sequence = "".join(sequence.split()).upper()
         
-        instance = MissingTable(*args, **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(payload: dict):
-    """NextOrf class - pass constructor arguments in payload"""
-    try:
-        NextOrf = lazy_import_script("Scripts.xbbtools.nextorf", "NextOrf")
-        if NextOrf is None:
-            return {"success": False, "result": None, "error": "Class NextOrf is not available"}
+        database = payload.get("database", "nt")
+        program = payload.get("program", "blastn")
+        hitlist_size = int(payload.get("hitlist_size", 20))
+        expect = float(payload.get("expect", 1e-10))
         
-        # Extract args and kwargs from payload
-        args = payload.get("args", [])
-        kwargs = payload.get("kwargs", {})
+        print(f"🔍 Submitting BLAST search: {len(sequence)} bp, database={database}, hits={hitlist_size}")
+        
+        # Submit BLAST request
+        handle = NCBIWWW.qblast(
+            program=program,
+            database=database,
+            sequence=sequence,
+            hitlist_size=hitlist_size,
+            expect=expect,
+            format_type="XML"
+        )
+        
+        blast_record = NCBIXML.read(handle)
+        handle.close()
+        
+        print(f"✅ BLAST search completed: {len(blast_record.alignments)} alignments found")
+        
+        hits = []
+        for idx, alignment in enumerate(blast_record.alignments[:hitlist_size]):
+            if not alignment.hsps:
+                continue
+            
+            hsp = alignment.hsps[0]
+            align_len = int(hsp.align_length)
+            identities = int(hsp.identities)
+            identity_pct = (identities / align_len * 100.0) if align_len else 0
+            
+            hits.append({
+                "rank": idx + 1,
+                "accession": alignment.accession,
+                "hit_id": alignment.hit_id,
+                "hit_def": alignment.hit_def,
+                "length": alignment.length,
+                "bit_score": float(hsp.bits),
+                "evalue": float(hsp.expect),
+                "identity_pct": round(identity_pct, 2),
+                "identities": identities,
+                "align_length": align_len
+            })
+        
+        result = {
+            "query_length": len(sequence),
+            "database": database,
+            "program": program,
+            "total_hits": len(hits),
+            "hits": hits
+        }
         
-        instance = NextOrf(*args, **kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-# -----------------------
-# Core Biopython utilities
-# -----------------------
-
-
-@mcp.tool(name="seq_reverse_complement", description="Reverse-complement a DNA/RNA sequence string")
-def seq_reverse_complement(payload: dict):
-    try:
-        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="seq_transcribe", description="Transcribe DNA to RNA (replace T->U)")
-def seq_transcribe(payload: dict):
+@mcp.tool(name="filter_blast_hits", description="Filter BLAST hits by identity percentage threshold")
+def filter_blast_hits(payload: dict):
+    """
+    根据相似度阈值过滤 BLAST 结果
+    Required fields: hits (from blast_search result)
+    Optional fields: min_identity (default 70), max_hits (default 10)
+    """
     try:
-        seq = payload.get("sequence", "")
-        result = str(Seq(seq).transcribe())
-        return {"success": True, "result": result, "error": None}
+        hits = payload.get("hits", [])
+        min_identity = float(payload.get("min_identity", 70.0))
+        max_hits = int(payload.get("max_hits", 10))
+        
+        if not hits:
+            return {"success": False, "result": None, "error": "hits list is required"}
+        
+        # Filter by identity percentage
+        filtered = [h for h in hits if h.get("identity_pct", 0) >= min_identity]
+        
+        # Limit number of hits
+        filtered = filtered[:max_hits]
+        
+        return {
+            "success": True,
+            "result": {
+                "original_count": len(hits),
+                "filtered_count": len(filtered),
+                "min_identity": min_identity,
+                "hits": filtered
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
 
-@mcp.tool(name="seq_translate", description="Translate DNA/RNA to protein; accepts optional table, to_stop")
-def seq_translate(payload: dict):
-    try:
-        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)}
-
+# ==============================================
+# 物种提取工具 - Species Extraction Tools
+# ==============================================
 
-@mcp.tool(name="gc_content", description="Compute GC percentage of a sequence")
-def gc_content(payload: dict):
+@mcp.tool(name="extract_species_from_blast", description="Extract species names from BLAST hit definitions")
+def extract_species_from_blast(payload: dict):
+    """
+    从 BLAST 结果的描述中提取物种名称
+    Required fields: hits (from blast_search or filter_blast_hits)
+    """
     try:
-        seq = payload.get("sequence", "")
-        result = GC(seq)
-        return {"success": True, "result": result, "error": None}
+        hits = payload.get("hits", [])
+        if not hits:
+            return {"success": False, "result": None, "error": "hits list is required"}
+        
+        species_data = []
+        
+        for hit in hits:
+            hit_def = hit.get("hit_def", "")
+            identity_pct = hit.get("identity_pct", 0)
+            
+            # Extract species name from definition
+            # Common patterns: "[Species name]" or "Species name isolate/strain"
+            species = None
+            
+            # Try to find text in square brackets first
+            bracket_match = re.search(r'\[([^\]]+)\]', hit_def)
+            if bracket_match:
+                species = bracket_match.group(1)
+            else:
+                # Try to extract first two words (genus + species)
+                words = hit_def.split()
+                if len(words) >= 2:
+                    species = f"{words[0]} {words[1]}"
+            
+            if species:
+                species_data.append({
+                    "species": species,
+                    "identity_pct": identity_pct,
+                    "accession": hit.get("accession"),
+                    "hit_def": hit_def
+                })
+        
+        return {
+            "success": True,
+            "result": {
+                "total_species_found": len(species_data),
+                "species_data": species_data
+            },
+            "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):
+@mcp.tool(name="aggregate_species_scores", description="Aggregate species identifications by weighted scores")
+def aggregate_species_scores(payload: dict):
+    """
+    根据相似度对物种进行加权聚合
+    Required fields: species_data (from extract_species_from_blast)
+    """
     try:
-        seq1 = payload.get("seq1", "")
-        seq2 = payload.get("seq2", "")
-        limit = int(payload.get("limit", 3))
-        
-        if HAS_PAIRWISE2:
-            # Use deprecated pairwise2 if available
-            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
-            ]
-        else:
-            # Use modern PairwiseAligner
-            aligner = PairwiseAligner()
-            aligner.mode = 'global'
-            aligner.match_score = 1
-            aligner.mismatch_score = 0
-            alignments = list(aligner.align(seq1, seq2))[:limit]
-            result = [
-                {
-                    "seqA": str(alignment).split('\n')[0],
-                    "seqB": str(alignment).split('\n')[2] if len(str(alignment).split('\n')) > 2 else "",
-                    "score": alignment.score,
-                    "start": 0,
-                    "end": len(seq1),
+        species_data = payload.get("species_data", [])
+        if not species_data:
+            return {"success": False, "result": None, "error": "species_data is required"}
+        
+        # Aggregate by species
+        species_scores = {}
+        
+        for item in species_data:
+            species = item.get("species", "Unknown")
+            identity = item.get("identity_pct", 0)
+            
+            if species not in species_scores:
+                species_scores[species] = {
+                    "total_score": 0,
+                    "count": 0,
+                    "max_identity": 0,
+                    "accessions": []
                 }
-                for alignment in alignments
-            ]
+            
+            # Weight by identity percentage
+            species_scores[species]["total_score"] += identity
+            species_scores[species]["count"] += 1
+            species_scores[species]["max_identity"] = max(
+                species_scores[species]["max_identity"],
+                identity
+            )
+            species_scores[species]["accessions"].append(item.get("accession"))
+        
+        # Calculate average scores
+        species_list = []
+        for species, data in species_scores.items():
+            avg_identity = data["total_score"] / data["count"]
+            species_list.append({
+                "species": species,
+                "average_identity": round(avg_identity, 2),
+                "max_identity": round(data["max_identity"], 2),
+                "hit_count": data["count"],
+                "confidence_score": round(avg_identity * data["count"] / 10, 2),  # Weighted confidence
+                "accessions": data["accessions"][:3]  # Top 3 accessions
+            })
+        
+        # Sort by confidence score
+        species_list.sort(key=lambda x: x["confidence_score"], reverse=True)
         
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@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,
+            "result": {
+                "total_unique_species": len(species_list),
+                "species_rankings": species_list
+            },
+            "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:
-        fmt = payload.get("format", "fasta")
-        data = payload.get("data", "")
-        path = payload.get("path", "")
-        if data:
-            alignment = AlignIO.read(StringIO(data), fmt)
-        else:
-            alignment = AlignIO.read(path, fmt)
-        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):
+@mcp.tool(name="predict_species", description="Predict the most likely species for the input gene")
+def predict_species(payload: dict):
+    """
+    预测输入基因最可能的物种
+    Required fields: species_rankings (from aggregate_species_scores)
+    Optional fields: min_confidence (default 5.0)
+    """
     try:
-        fmt = payload.get("format", "fasta")
-        data = payload.get("data", "")
-        path = payload.get("path", "")
-        if data:
-            alignment = AlignIO.read(StringIO(data), fmt)
+        species_rankings = payload.get("species_rankings", [])
+        min_confidence = float(payload.get("min_confidence", 5.0))
+        
+        if not species_rankings:
+            return {"success": False, "result": None, "error": "species_rankings is required"}
+        
+        # Filter by minimum confidence
+        confident_predictions = [
+            sp for sp in species_rankings 
+            if sp.get("confidence_score", 0) >= min_confidence
+        ]
+        
+        if not confident_predictions:
+            return {
+                "success": True,
+                "result": {
+                    "prediction": "Unknown",
+                    "confidence": "Low",
+                    "reason": "No species met minimum confidence threshold",
+                    "top_candidates": species_rankings[:3]
+                },
+                "error": None
+            }
+        
+        # Get top prediction
+        top_prediction = confident_predictions[0]
+        
+        # Determine confidence level
+        confidence_score = top_prediction.get("confidence_score", 0)
+        if confidence_score >= 50:
+            confidence_level = "Very High"
+        elif confidence_score >= 20:
+            confidence_level = "High"
+        elif confidence_score >= 10:
+            confidence_level = "Medium"
         else:
-            alignment = AlignIO.read(path, fmt)
-        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}
+            confidence_level = "Low"
+        
+        return {
+            "success": True,
+            "result": {
+                "predicted_species": top_prediction.get("species"),
+                "confidence_level": confidence_level,
+                "confidence_score": confidence_score,
+                "average_identity": top_prediction.get("average_identity"),
+                "max_identity": top_prediction.get("max_identity"),
+                "supporting_hits": top_prediction.get("hit_count"),
+                "alternative_species": [
+                    {
+                        "species": sp.get("species"),
+                        "confidence_score": sp.get("confidence_score"),
+                        "average_identity": sp.get("average_identity")
+                    }
+                    for sp in confident_predictions[1:4]  # Next 3 candidates
+                ]
+            },
+            "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):
+# ==============================================
+# 完整流程工具 - Complete Workflow Tool
+# ==============================================
+
+@mcp.tool(name="identify_gene_species_complete", 
+          description="Complete workflow: input unknown gene sequence, identify species through BLAST search and analysis")
+def identify_gene_species_complete(payload: dict):
+    """
+    完整的基因物种鉴定流程
+    Required fields: sequence
+    Optional fields: min_identity (default 70), max_hits (default 20), database (default nt)
+    
+    This tool orchestrates multiple steps:
+    1. Validate sequence
+    2. Calculate GC content
+    3. BLAST search
+    4. Filter hits
+    5. Extract species
+    6. Aggregate scores
+    7. Predict species
+    """
     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}
+        sequence = payload.get("sequence", "")
+        min_identity = float(payload.get("min_identity", 70.0))
+        max_hits = int(payload.get("max_hits", 20))
+        database = payload.get("database", "nt")
+        
+        workflow_results = {
+            "steps": [],
+            "errors": []
+        }
+        
+        # Step 1: Validate sequence
+        print("📋 Step 1: Validating sequence...")
+        val_result = validate_sequence({"sequence": sequence})
+        workflow_results["steps"].append({
+            "step": 1,
+            "name": "validate_sequence",
+            "result": val_result
+        })
+        
+        if not val_result["success"]:
+            return {
+                "success": False,
+                "result": workflow_results,
+                "error": "Sequence validation failed"
+            }
+        
+        validated_seq = val_result["result"]["sequence"]
+        seq_length = val_result["result"]["length"]
+        
+        # Step 2: Calculate GC content
+        print("🧬 Step 2: Calculating GC content...")
+        gc_result = calculate_gc_content({"sequence": validated_seq})
+        workflow_results["steps"].append({
+            "step": 2,
+            "name": "calculate_gc_content",
+            "result": gc_result
+        })
+        
+        # Step 3: BLAST search
+        print("🔍 Step 3: Running BLAST search...")
+        blast_result = blast_search({
+            "sequence": validated_seq,
+            "database": database,
+            "hitlist_size": max_hits
+        })
+        workflow_results["steps"].append({
+            "step": 3,
+            "name": "blast_search",
+            "result": blast_result
+        })
+        
+        if not blast_result["success"]:
+            return {
+                "success": False,
+                "result": workflow_results,
+                "error": "BLAST search failed"
+            }
+        
+        hits = blast_result["result"]["hits"]
+        
+        # Step 4: Filter hits
+        print("🔬 Step 4: Filtering BLAST hits...")
+        filter_result = filter_blast_hits({
+            "hits": hits,
+            "min_identity": min_identity,
+            "max_hits": 10
+        })
+        workflow_results["steps"].append({
+            "step": 4,
+            "name": "filter_blast_hits",
+            "result": filter_result
+        })
+        
+        filtered_hits = filter_result["result"]["hits"]
+        
+        # Step 5: Extract species
+        print("🌍 Step 5: Extracting species information...")
+        species_result = extract_species_from_blast({"hits": filtered_hits})
+        workflow_results["steps"].append({
+            "step": 5,
+            "name": "extract_species_from_blast",
+            "result": species_result
+        })
+        
+        species_data = species_result["result"]["species_data"]
+        
+        # Step 6: Aggregate scores
+        print("📊 Step 6: Aggregating species scores...")
+        agg_result = aggregate_species_scores({"species_data": species_data})
+        workflow_results["steps"].append({
+            "step": 6,
+            "name": "aggregate_species_scores",
+            "result": agg_result
+        })
+        
+        species_rankings = agg_result["result"]["species_rankings"]
+        
+        # Step 7: Predict species
+        print("🎯 Step 7: Predicting species...")
+        pred_result = predict_species({"species_rankings": species_rankings})
+        workflow_results["steps"].append({
+            "step": 7,
+            "name": "predict_species",
+            "result": pred_result
+        })
+        
+        # Compile final result
+        final_result = {
+            "input_sequence_length": seq_length,
+            "gc_content": gc_result["result"]["gc_content"] if gc_result["success"] else None,
+            "total_blast_hits": len(hits),
+            "filtered_hits": len(filtered_hits),
+            "unique_species_found": len(species_rankings),
+            "prediction": pred_result["result"] if pred_result["success"] else None,
+            "workflow": workflow_results
+        }
+        
+        print("✅ Complete workflow finished successfully!")
+        
+        return {
+            "success": True,
+            "result": final_result,
+            "error": None
+        }
+        
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {
+            "success": False,
+            "result": None,
+            "error": f"Workflow 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)