import subprocess import tempfile from pathlib import Path from typing import List, Optional # MCP decorator is commented out as per instructions # import mcp from mcp.server.fastmcp import FastMCP SERVER_NAME = 'local_bedops' mcp = FastMCP(SERVER_NAME) @mcp.tool() def bedops_complement( files: List[Path], chop_to_limits: bool = False, chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Computes the complement of one or more BED files. This corresponds to the `bedops -c` or `bedops --complement` operation. It finds the regions within chromosome boundaries that are not covered by any intervals in the input file(s). Args: files: A list of one or more input BED/Starch files. Must be sorted. chop_to_limits: If True, chop complementary regions to chromosome limits defined by the first input file (-L flag). chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if not files: raise ValueError("At least one input file must be provided for the complement operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--complement") if chop_to_limits: cmd.append("-L") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops complement failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_difference( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Computes the difference between a reference BED file and one or more other BED files. This corresponds to the `bedops -d` or `bedops --difference` operation. It returns regions from the first (reference) file that do not overlap with any regions in the subsequent files. Args: files: A list of two or more input BED/Starch files. The first file is the reference. All files must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if len(files) < 2: raise ValueError("At least two input files (a reference and one other) must be provided for the difference operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--difference") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops difference failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_element_of( files: List[Path], overlap_criterion: str = "100%", chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Finds elements in the reference file that overlap other files by a specified amount. This corresponds to the `bedops -e` or `bedops --element-of` operation. It returns elements from the first (reference) file that overlap elements in any of the other files by at least the specified amount. Args: files: A list of two or more input BED/Starch files. The first file is the reference. All files must be sorted. overlap_criterion: The required overlap, as base pairs (e.g., '1') or percentage (e.g., '50%'). Defaults to '100%'. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' or 'S' format. The first (reference) file is NOT padded with this operation. output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if len(files) < 2: raise ValueError("At least two input files (a reference and one other) must be provided for the element-of operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.extend(["--element-of", overlap_criterion]) cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops element-of failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_intersect( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Computes the intersection of two or more BED files. This corresponds to the `bedops -i` or `bedops --intersect` operation. It returns regions that are common to all input files. Args: files: A list of two or more input BED/Starch files. All files must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if len(files) < 2: raise ValueError("At least two input files must be provided for the intersect operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--intersect") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops intersect failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_merge( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Merges overlapping regions from one or more BED files. This corresponds to the `bedops -m` or `bedops --merge` operation. It combines overlapping or adjacent intervals into a single, larger interval. Args: files: A list of one or more input BED/Starch files. Must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if not files: raise ValueError("At least one input file must be provided for the merge operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--merge") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops merge failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_not_element_of( files: List[Path], overlap_criterion: str = "100%", chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Finds elements in the reference file that DO NOT overlap other files by a specified amount. This corresponds to the `bedops -n` or `bedops --not-element-of` operation. It is the inverse of the `element-of` operation. Args: files: A list of two or more input BED/Starch files. The first file is the reference. All files must be sorted. overlap_criterion: The required overlap, as base pairs (e.g., '1') or percentage (e.g., '50%'). Defaults to '100%'. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' or 'S' format. The first (reference) file is NOT padded with this operation. output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if len(files) < 2: raise ValueError("At least two input files (a reference and one other) must be provided for the not-element-of operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.extend(["--not-element-of", overlap_criterion]) cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops not-element-of failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_partition( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Partitions the input BED file(s) into disjoint segments. This corresponds to the `bedops -p` or `bedops --partition` operation. It breaks the input regions into non-overlapping segments, reporting each new segment and which input files it came from. Args: files: A list of one or more input BED/Starch files. Must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if not files: raise ValueError("At least one input file must be provided for the partition operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--partition") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops partition failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_symmdiff( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Computes the symmetric difference of two or more BED files. This corresponds to the `bedops -s` or `bedops --symmdiff` operation. It returns regions that are unique to any of the input files (i.e., not present in their intersection). Args: files: A list of two or more input BED/Starch files. All files must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if len(files) < 2: raise ValueError("At least two input files must be provided for the symmetric difference operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--symmdiff") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops symmdiff failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_everything( files: List[Path], chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Returns the union of all elements from input files without merging. This corresponds to the `bedops -u` or `bedops --everything` operation. It effectively concatenates the input files while maintaining sort order and preserving all original columns. Args: files: A list of one or more input BED/Starch files. Must be sorted. chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if not files: raise ValueError("At least one input file must be provided for the everything operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.append("--everything") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops everything failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e @mcp.tool() def bedops_chop( files: List[Path], bp: int = 1, stagger: Optional[int] = None, exclusive_chop: bool = False, chrom: Optional[str] = None, ec: bool = False, header: bool = False, range_str: Optional[str] = None, output_file: Optional[Path] = None, ) -> dict: """ Chops elements into fixed-size, potentially staggered sub-elements. This corresponds to the `bedops -w` or `bedops --chop` operation. Args: files: A list of one or more input BED/Starch files. Must be sorted. bp: The size in base pairs of each chopped element. Defaults to 1. stagger: The stagger distance in nucleotides. If not set, no staggering is done. exclusive_chop: If True, removes single-base elements that can result from chopping (-x flag). chrom: Process data for the given chromosome only. ec: Error check input files (slower). header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a single value 'S' for symmetric padding (e.g., '100'). output_file: Optional path to save the output. If not provided, output is returned as a string in the result dictionary. Returns: A dictionary containing the command executed, stdout, stderr, and a list of output files generated. """ if not files: raise ValueError("At least one input file must be provided for the chop operation.") for file_path in files: if not file_path.exists(): raise FileNotFoundError(f"Input file not found: {file_path}") cmd = ["bedops"] if chrom: cmd.extend(["--chrom", chrom]) if ec: cmd.append("--ec") if header: cmd.append("--header") if range_str: cmd.extend(["--range", range_str]) cmd.extend(["--chop", str(bp)]) if stagger is not None: cmd.extend(["--stagger", str(stagger)]) if exclusive_chop: cmd.append("-x") cmd.extend([str(p) for p in files]) try: if output_file: with open(output_file, "w") as f: result = subprocess.run( cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE ) stdout_capture = "" output_files_list = [str(output_file)] else: result = subprocess.run( cmd, check=True, text=True, capture_output=True ) stdout_capture = result.stdout output_files_list = [] return { "command_executed": " ".join(cmd), "stdout": stdout_capture, "stderr": result.stderr, "output_files": output_files_list, } except subprocess.CalledProcessError as e: raise RuntimeError( f"bedops chop failed with exit code {e.returncode}\n" f"Stderr: {e.stderr}\n" f"Stdout: {e.stdout}\n" f"Command: {' '.join(cmd)}" ) from e if __name__ == "__main__": mcp.run(transport="stdio")