Biomni/mcp_generated/mcp_bcftools/app/bcftools_server.py

import subprocess
import tempfile
from pathlib import Path
from typing import List, Optional

# Note: The @mcp.tool decorator is not defined here.
# It is assumed to be provided by the Model Context Protocol (MCP) framework.
# This code is designed to be used within that framework.

# Helper function to handle subprocess execution and error reporting
def _run_bcftools_command(cmd: List[str], output_file: Optional[Path] = None):
    """
    A helper function to execute a bcftools command, handle errors,
    and return a structured dictionary.
    """
    command_str = " ".join(cmd)
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True,
        )
        output_files = [str(output_file)] if output_file and output_file.exists() else []
        
        # For commands that create index files automatically
        if output_file and str(output_file).endswith((".bcf", ".vcf.gz")):
            index_extensions = [".csi", ".tbi"]
            for ext in index_extensions:
                index_file = Path(str(output_file) + ext)
                if index_file.exists():
                    output_files.append(str(index_file))

        return {
            "command_executed": command_str,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "output_files": output_files,
            "return_code": 0,
        }
    except FileNotFoundError:
        return {
            "command_executed": command_str,
            "stdout": "",
            "stderr": "Error: 'bcftools' command not found. Please ensure it is in your PATH.",
            "output_files": [],
            "return_code": 1,
            "error": "FileNotFoundError"
        }
    except subprocess.CalledProcessError as e:
        return {
            "command_executed": command_str,
            "stdout": e.stdout,
            "stderr": e.stderr,
            "output_files": [],
            "return_code": e.returncode,
            "error": "CalledProcessError"
        }

from mcp.server.fastmcp import FastMCP

SERVER_NAME = 'local_bcftools'
mcp = FastMCP(SERVER_NAME)

@mcp.tool()
def bcftools_annotate(
    input_file: Path,
    annotations: Optional[Path] = None,
    columns: Optional[str] = None,
    header_lines: Optional[Path] = None,
    output: Optional[Path] = None,
    output_type: str = "v",
    remove: Optional[str] = None,
    set_id: Optional[str] = None,
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    targets: Optional[str] = None,
    targets_file: Optional[Path] = None,
    include: Optional[str] = None,
    exclude: Optional[str] = None,
    samples: Optional[str] = None,
    samples_file: Optional[Path] = None,
    threads: int = 1,
    rename_chrs: Optional[Path] = None,
    no_version: bool = False,
    single_overlaps: bool = False,
    mark_sites: Optional[str] = None,
):
    """
    Annotate VCF/BCF files with information from other files.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if annotations and not annotations.exists():
        raise FileNotFoundError(f"Annotations file not found: {annotations}")
    if header_lines and not header_lines.exists():
        raise FileNotFoundError(f"Header lines file not found: {header_lines}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "annotate", str(input_file)]

    if annotations:
        cmd.extend(["-a", str(annotations)])
    if columns:
        cmd.extend(["-c", columns])
    if header_lines:
        cmd.extend(["-h", str(header_lines)])
    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if remove:
        cmd.extend(["-x", remove])
    if set_id:
        cmd.extend(["--set-id", set_id])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if targets:
        cmd.extend(["-t", targets])
    if targets_file:
        cmd.extend(["-T", str(targets_file)])
    if include:
        cmd.extend(["-i", include])
    if exclude:
        cmd.extend(["-e", exclude])
    if samples:
        cmd.extend(["-s", samples])
    if samples_file:
        cmd.extend(["-S", str(samples_file)])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])
    if rename_chrs:
        cmd.extend(["--rename-chrs", str(rename_chrs)])
    if no_version:
        cmd.append("--no-version")
    if single_overlaps:
        cmd.append("--single-overlaps")
    if mark_sites:
        cmd.extend(["--mark-sites", mark_sites])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_call(
    input_file: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    ploidy: Optional[str] = None,
    ploidy_file: Optional[Path] = None,
    call_method: str = "multiallelic",
    constrain_alleles: bool = False,
    group_samples: Optional[str] = None,
    samples: Optional[str] = None,
    samples_file: Optional[Path] = None,
    format_fields: Optional[str] = None,
    keep_alts: bool = False,
    keep_masked_ref: bool = False,
    skip_variants: bool = False,
    threads: int = 1,
):
    """
    Call genotypes from likelihoods.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")
    if call_method not in ["consensus", "multiallelic"]:
        raise ValueError("call_method must be 'consensus' or 'multiallelic'.")

    cmd = ["bcftools", "call", str(input_file)]

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if ploidy:
        cmd.extend(["-p", ploidy])
    if ploidy_file:
        cmd.extend(["-P", str(ploidy_file)])
    if call_method == "consensus":
        cmd.append("-c")
    else: # multiallelic is default, but -m is explicit
        cmd.append("-m")
    if constrain_alleles:
        cmd.append("-C alleles")
    if group_samples:
        cmd.extend(["-G", group_samples])
    if samples:
        cmd.extend(["-s", samples])
    if samples_file:
        cmd.extend(["-S", str(samples_file)])
    if format_fields:
        cmd.extend(["-f", format_fields])
    if keep_alts:
        cmd.append("-A")
    if keep_masked_ref:
        cmd.append("-M")
    if skip_variants:
        cmd.append("-V")
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_concat(
    input_files: List[Path],
    output: Optional[Path] = None,
    output_type: str = "v",
    allow_overlaps: bool = False,
    remove_duplicates: bool = False,
    naive: bool = False,
    threads: int = 1,
):
    """
    Concatenate VCF/BCF files from non-overlapping regions.
    """
    if not input_files:
        raise ValueError("At least one input file must be provided.")
    for f in input_files:
        if not f.exists():
            raise FileNotFoundError(f"Input file not found: {f}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "concat"]

    if allow_overlaps:
        cmd.append("-a")
    if remove_duplicates:
        cmd.append("-d")
    if naive:
        cmd.append("-n")
    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    cmd.extend([str(f) for f in input_files])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_consensus(
    input_file: Path,
    fasta_ref: Path,
    output: Optional[Path] = None,
    chain: Optional[Path] = None,
    haplotype: Optional[str] = None,
    iupac_codes: bool = False,
    mark_del: Optional[str] = None,
    mark_ins: Optional[str] = None,
    mark_snv: Optional[str] = None,
    missing: Optional[str] = None,
    sample: Optional[str] = None,
    include: Optional[str] = None,
    exclude: Optional[str] = None,
):
    """
    Create a consensus sequence by applying VCF variants to a reference FASTA file.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input VCF/BCF file not found: {input_file}")
    if not fasta_ref.exists():
        raise FileNotFoundError(f"Reference FASTA file not found: {fasta_ref}")
    if haplotype and haplotype not in ["1", "2", "R", "A", "IUPAC", "all"]:
        raise ValueError("haplotype must be one of '1', '2', 'R', 'A', 'IUPAC', 'all'.")

    cmd = ["bcftools", "consensus", str(input_file)]
    cmd.extend(["-f", str(fasta_ref)])

    if output:
        cmd.extend(["-o", str(output)])
    if chain:
        cmd.extend(["-c", str(chain)])
    if haplotype:
        cmd.extend(["-H", haplotype])
    if iupac_codes:
        cmd.append("-i")
    if mark_del:
        cmd.extend(["--mark-del", mark_del])
    if mark_ins:
        cmd.extend(["--mark-ins", mark_ins])
    if mark_snv:
        cmd.extend(["--mark-snv", mark_snv])
    if missing:
        cmd.extend(["-m", missing])
    if sample:
        cmd.extend(["-s", sample])
    if include:
        cmd.extend(["--include", include])
    if exclude:
        cmd.extend(["--exclude", exclude])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_filter(
    input_file: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    include: Optional[str] = None,
    exclude: Optional[str] = None,
    soft_filter: Optional[str] = None,
    set_gt: Optional[str] = None,
    mode: str = "+",
    threads: int = 1,
):
    """
    Apply filters to VCF/BCF files.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")
    if mode not in ["+", "x", "X"]:
        raise ValueError("mode must be one of '+', 'x', 'X'.")

    cmd = ["bcftools", "filter", str(input_file)]

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if include:
        cmd.extend(["-i", include])
    if exclude:
        cmd.extend(["-e", exclude])
    if soft_filter:
        cmd.extend(["-s", soft_filter])
    if set_gt:
        cmd.extend(["-S", set_gt])
    if mode != "+":
        cmd.extend(["-m", mode])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_index(
    input_file: Path,
    force: bool = False,
    stats: bool = False,
    threads: int = 1,
    csi: bool = False,
    tbi: bool = False,
    min_csi_shift: int = 14,
):
    """
    Index VCF/BCF files for fast random access.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if csi and tbi:
        raise ValueError("Cannot specify both --csi (-c) and --tbi (-t).")

    cmd = ["bcftools", "index"]

    if force:
        cmd.append("-f")
    if stats:
        cmd.append("-n")
    if threads > 1:
        cmd.extend(["--threads", str(threads)])
    if csi:
        cmd.append("-c")
    if tbi:
        cmd.append("-t")
    if csi and min_csi_shift != 14:
        cmd.extend(["-m", str(min_csi_shift)])

    cmd.append(str(input_file))
    
    # The output file is the index file itself
    output_idx_file = None
    if csi:
        output_idx_file = Path(str(input_file) + ".csi")
    else: # default is tbi for vcf.gz, csi for bcf
        if str(input_file).endswith(".bcf"):
            output_idx_file = Path(str(input_file) + ".csi")
        else:
            output_idx_file = Path(str(input_file) + ".tbi")

    return _run_bcftools_command(cmd, output_idx_file)

@mcp.tool()
def bcftools_isec(
    input_files: List[Path],
    output_dir: Path,
    output_type: str = "v",
    complement: bool = False,
    collapse: Optional[str] = None,
    exclude: Optional[str] = None,
    include: Optional[str] = None,
    n_sites: Optional[str] = None,
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    targets: Optional[str] = None,
    targets_file: Optional[Path] = None,
    threads: int = 1,
):
    """
    Create intersections, unions, and complements of VCF/BCF files.
    """
    if len(input_files) < 2:
        raise ValueError("At least two input files must be provided for isec.")
    for f in input_files:
        if not f.exists():
            raise FileNotFoundError(f"Input file not found: {f}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")
    if collapse and collapse not in ["none", "all", "snps", "indels", "both", "any"]:
        raise ValueError("Invalid value for collapse.")

    output_dir.mkdir(parents=True, exist_ok=True)

    cmd = ["bcftools", "isec"]
    cmd.extend(["-p", str(output_dir)])
    cmd.extend(["-O", output_type])

    if complement:
        cmd.append("-c")
    if collapse:
        cmd.extend(["--collapse", collapse])
    if exclude:
        cmd.extend(["-e", exclude])
    if include:
        cmd.extend(["-i", include])
    if n_sites:
        cmd.extend(["-n", n_sites])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if targets:
        cmd.extend(["-t", targets])
    if targets_file:
        cmd.extend(["-T", str(targets_file)])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    cmd.extend([str(f) for f in input_files])

    # isec doesn't have a single output file, it creates a directory
    # We will return the directory path in a structured way if needed,
    # but for now, we'll rely on the user knowing the output is a directory.
    # The _run_bcftools_command helper won't find a single output file.
    result = _run_bcftools_command(cmd)
    
    # Manually list created files
    if result["return_code"] == 0:
        created_files = [str(p) for p in output_dir.glob('*')]
        result["output_files"] = created_files

    return result

@mcp.tool()
def bcftools_merge(
    input_files: List[Path],
    output: Optional[Path] = None,
    output_type: str = "v",
    info_rules: Optional[str] = None,
    merge_logic: Optional[str] = None,
    threads: int = 1,
):
    """
    Merge multiple VCF/BCF files from different samples into a single file.
    """
    if not input_files:
        raise ValueError("At least one input file must be provided.")
    for f in input_files:
        if not f.exists():
            raise FileNotFoundError(f"Input file not found: {f}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "merge"]

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if info_rules:
        cmd.extend(["--info-rules", info_rules])
    if merge_logic:
        cmd.extend(["-m", merge_logic])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    cmd.extend([str(f) for f in input_files])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_mpileup(
    input_bams: List[Path],
    fasta_ref: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    annotate: Optional[str] = None,
    max_depth: int = 250,
    min_base_quality: int = 13,
    min_mapping_quality: int = 0,
    adjust_mq: int = 50,
    no_baq: bool = False,
    threads: int = 1,
):
    """
    Generate VCF/BCF from alignment files (BAM/CRAM).
    """
    if not input_bams:
        raise ValueError("At least one input BAM/CRAM file must be provided.")
    for f in input_bams:
        if not f.exists():
            raise FileNotFoundError(f"Input alignment file not found: {f}")
    if not fasta_ref.exists():
        raise FileNotFoundError(f"Reference FASTA file not found: {fasta_ref}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "mpileup"]
    cmd.extend(["-f", str(fasta_ref)])

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if annotate:
        cmd.extend(["-a", annotate])
    if max_depth != 250:
        cmd.extend(["-d", str(max_depth)])
    if min_base_quality != 13:
        cmd.extend(["-q", str(min_base_quality)])
    if min_mapping_quality != 0:
        cmd.extend(["-Q", str(min_mapping_quality)])
    if adjust_mq != 50:
        cmd.extend(["-C", str(adjust_mq)])
    if no_baq:
        cmd.append("-B")
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    cmd.extend([str(f) for f in input_bams])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_norm(
    input_file: Path,
    fasta_ref: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    check_ref: str = "w",
    multiallelics: Optional[str] = None,
    atomize: bool = False,
    rm_dup: Optional[str] = None,
    threads: int = 1,
):
    """
    Left-align and normalize indels, check REF, split multiallelic sites.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if not fasta_ref.exists():
        raise FileNotFoundError(f"Reference FASTA file not found: {fasta_ref}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")
    if check_ref not in ["w", "e", "x", "s"]:
        raise ValueError("check_ref must be one of 'w', 'e', 'x', 's'.")

    cmd = ["bcftools", "norm", str(input_file)]
    cmd.extend(["-f", str(fasta_ref)])

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if check_ref != "w":
        cmd.extend(["-c", check_ref])
    if multiallelics:
        cmd.extend(["-m", multiallelics])
    if atomize:
        cmd.append("-a")
    if rm_dup:
        cmd.extend(["-d", rm_dup])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_query(
    input_file: Path,
    format_string: str,
    output: Optional[Path] = None,
    include: Optional[str] = None,
    exclude: Optional[str] = None,
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    samples: Optional[str] = None,
    samples_file: Optional[Path] = None,
    list_samples: bool = False,
):
    """
    Extracts fields from VCF/BCF files and prints them in a user-defined format.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")

    cmd = ["bcftools", "query", str(input_file)]

    if list_samples:
        cmd.append("-l")
    else:
        cmd.extend(["-f", format_string])

    if output:
        cmd.extend(["-o", str(output)])
    if include:
        cmd.extend(["-i", include])
    if exclude:
        cmd.extend(["-e", exclude])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if samples:
        cmd.extend(["-s", samples])
    if samples_file:
        cmd.extend(["-S", str(samples_file)])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_reheader(
    input_file: Path,
    header: Optional[Path] = None,
    samples: Optional[Path] = None,
    output: Optional[Path] = None,
    output_type: str = "v",
):
    """
    Modify VCF/BCF header, e.g., rename samples.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if header and not header.exists():
        raise FileNotFoundError(f"Header file not found: {header}")
    if samples and not samples.exists():
        raise FileNotFoundError(f"Samples file not found: {samples}")
    if not header and not samples:
        raise ValueError("Either 'header' or 'samples' file must be provided.")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "reheader"]

    if header:
        cmd.extend(["-h", str(header)])
    if samples:
        cmd.extend(["-s", str(samples)])
    if output:
        cmd.extend(["-o", str(output)])
    
    cmd.append(str(input_file))

    # Reheader does not support -O, output type is inferred from extension
    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_sort(
    input_file: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    max_mem: str = "768M",
    temp_dir: Optional[Path] = None,
):
    """
    Sort VCF/BCF file by chromosome and position.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "sort", str(input_file)]

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if max_mem:
        cmd.extend(["-m", max_mem])
    if temp_dir:
        temp_dir.mkdir(exist_ok=True, parents=True)
        cmd.extend(["-T", str(temp_dir)])

    return _run_bcftools_command(cmd, output)

@mcp.tool()
def bcftools_stats(
    input_files: List[Path],
    fasta_ref: Optional[Path] = None,
    output_dir: Optional[Path] = None,
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    samples: Optional[str] = None,
    samples_file: Optional[Path] = None,
    threads: int = 1,
):
    """
    Produce VCF/BCF stats, create plots with plot-vcfstats.
    """
    if not input_files:
        raise ValueError("At least one input file must be provided.")
    for f in input_files:
        if not f.exists():
            raise FileNotFoundError(f"Input file not found: {f}")
    if fasta_ref and not fasta_ref.exists():
        raise FileNotFoundError(f"Reference FASTA file not found: {fasta_ref}")

    cmd = ["bcftools", "stats"]

    if fasta_ref:
        cmd.extend(["-f", str(fasta_ref)])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if samples:
        cmd.extend(["-s", samples])
    if samples_file:
        cmd.extend(["-S", str(samples_file)])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    cmd.extend([str(f) for f in input_files])

    # stats command outputs to stdout by default.
    # If output_dir is provided, we redirect stdout to a file.
    output_file = None
    if output_dir:
        output_dir.mkdir(exist_ok=True, parents=True)
        output_file = output_dir / "stats.txt"
        with open(output_file, "w") as f_out:
            command_str = " ".join(cmd)
            try:
                result = subprocess.run(
                    cmd,
                    stdout=f_out,
                    stderr=subprocess.PIPE,
                    text=True,
                    check=True,
                )
                return {
                    "command_executed": command_str,
                    "stdout": f"Stats written to {output_file}",
                    "stderr": result.stderr,
                    "output_files": [str(output_file)],
                    "return_code": 0,
                }
            except subprocess.CalledProcessError as e:
                return {
                    "command_executed": command_str,
                    "stdout": e.stdout or "",
                    "stderr": e.stderr,
                    "output_files": [],
                    "return_code": e.returncode,
                    "error": "CalledProcessError"
                }
    else:
        # No output file, capture stdout directly
        return _run_bcftools_command(cmd)

@mcp.tool()
def bcftools_view(
    input_file: Path,
    output: Optional[Path] = None,
    output_type: str = "v",
    header_only: bool = False,
    no_header: bool = False,
    drop_genotypes: bool = False,
    trim_alt_alleles: bool = False,
    min_alleles: Optional[int] = None,
    max_alleles: Optional[int] = None,
    min_ac: Optional[int] = None,
    max_ac: Optional[int] = None,
    min_af: Optional[float] = None,
    max_af: Optional[float] = None,
    types: Optional[str] = None,
    include: Optional[str] = None,
    exclude: Optional[str] = None,
    regions: Optional[str] = None,
    regions_file: Optional[Path] = None,
    samples: Optional[str] = None,
    samples_file: Optional[Path] = None,
    threads: int = 1,
):
    """
    Subset, filter, and convert VCF and BCF files.
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")
    if output_type not in ["b", "u", "z", "v"]:
        raise ValueError("output_type must be one of 'b', 'u', 'z', 'v'.")

    cmd = ["bcftools", "view", str(input_file)]

    if output:
        cmd.extend(["-o", str(output)])
    if output_type:
        cmd.extend(["-O", output_type])
    if header_only:
        cmd.append("-h")
    if no_header:
        cmd.append("-H")
    if drop_genotypes:
        cmd.append("-G")
    if trim_alt_alleles:
        cmd.append("-a")
    if min_alleles is not None:
        cmd.extend(["-m", str(min_alleles)])
    if max_alleles is not None:
        cmd.extend(["-M", str(max_alleles)])
    if min_ac is not None:
        cmd.extend(["-c", str(min_ac)])
    if max_ac is not None:
        cmd.extend(["-C", str(max_ac)])
    if min_af is not None:
        cmd.extend(["-q", str(min_af)])
    if max_af is not None:
        cmd.extend(["-Q", str(max_af)])
    if types:
        cmd.extend(["-v", types])
    if include:
        cmd.extend(["-i", include])
    if exclude:
        cmd.extend(["-e", exclude])
    if regions:
        cmd.extend(["-r", regions])
    if regions_file:
        cmd.extend(["-R", str(regions_file)])
    if samples:
        cmd.extend(["-s", samples])
    if samples_file:
        cmd.extend(["-S", str(samples_file)])
    if threads > 1:
        cmd.extend(["--threads", str(threads)])

    return _run_bcftools_command(cmd, output)

if __name__ == "__main__":
    mcp.run(transport="stdio")