guohanghui commited on
Commit
ef76e4e
·
verified ·
1 Parent(s): eb1baf2

Update pybedtools/mcp_output/mcp_plugin/mcp_service.py

Browse files
pybedtools/mcp_output/mcp_plugin/mcp_service.py CHANGED
@@ -1,670 +1,140 @@
1
- import os
2
- import sys
3
- from typing import List, Optional
4
-
5
  from fastmcp import FastMCP
6
-
7
- # pybedtools requires compiled Cython extensions, so we must use the installed package
8
- # Do NOT import from source directory as cbedtools needs to be compiled
9
- import pybedtools
10
- from pybedtools import BedTool, cleanup
11
 
12
  # Create the FastMCP service application
13
  mcp = FastMCP("pybedtools_service")
14
 
15
-
16
- # =============================================================================
17
- # Basic BED Operations
18
- # =============================================================================
19
-
20
- @mcp.tool(name="intersect_bed", description="Intersect two BED files and return the result.")
21
- def intersect_bed(file1: str, file2: str) -> dict:
22
- """
23
- Intersects two BED files using BedTool and returns the result.
24
-
25
- :param file1: Path to the first BED file.
26
- :param file2: Path to the second BED file.
27
- :return: Dictionary containing success status and result or error message.
28
- """
29
- try:
30
- a = BedTool(file1)
31
- b = BedTool(file2)
32
- result = a.intersect(b)
33
- return {"success": True, "result": str(result), "error": None}
34
- except Exception as e:
35
- return {"success": False, "result": None, "error": str(e)}
36
-
37
-
38
- @mcp.tool(name="merge_bed", description="Merge overlapping intervals in a BED file.")
39
- def merge_bed(file: str) -> dict:
40
- """
41
- Merges overlapping intervals in a BED file using BedTool.
42
-
43
- :param file: Path to the BED file.
44
- :return: Dictionary containing success status and result or error message.
45
- """
46
- try:
47
- a = BedTool(file)
48
- result = a.merge()
49
- return {"success": True, "result": str(result), "error": None}
50
- except Exception as e:
51
- return {"success": False, "result": None, "error": str(e)}
52
-
53
-
54
- @mcp.tool(name="annotate_bed", description="Annotate a BED file with additional information.")
55
- def annotate_bed(file: str, annotation_file: str) -> dict:
56
- """
57
- Annotates a BED file with additional information from another file.
58
-
59
- :param file: Path to the BED file to be annotated.
60
- :param annotation_file: Path to the annotation file.
61
- :return: Dictionary containing success status and result or error message.
62
- """
63
- try:
64
- a = BedTool(file)
65
- result = a.annotate(files=annotation_file)
66
- return {"success": True, "result": str(result), "error": None}
67
- except Exception as e:
68
- return {"success": False, "result": None, "error": str(e)}
69
-
70
-
71
- @mcp.tool(name="subtract_bed", description="Subtract intervals in file2 from file1.")
72
- def subtract_bed(file1: str, file2: str) -> dict:
73
- """
74
- Removes portions of intervals in file1 that overlap with intervals in file2.
75
-
76
- :param file1: Path to the first BED file (features to subtract from).
77
- :param file2: Path to the second BED file (features to subtract).
78
- :return: Dictionary containing success status and result or error message.
79
- """
80
- try:
81
- a = BedTool(file1)
82
- b = BedTool(file2)
83
- result = a.subtract(b)
84
- return {"success": True, "result": str(result), "error": None}
85
- except Exception as e:
86
- return {"success": False, "result": None, "error": str(e)}
87
-
88
-
89
- @mcp.tool(name="sort_bed", description="Sort a BED file by chromosome and position.")
90
- def sort_bed(file: str) -> dict:
91
- """
92
- Sorts a BED file by chromosome and then by start position.
93
-
94
- :param file: Path to the BED file.
95
- :return: Dictionary containing success status and sorted result.
96
- """
97
- try:
98
- a = BedTool(file)
99
- result = a.sort()
100
- return {"success": True, "result": str(result), "error": None}
101
- except Exception as e:
102
- return {"success": False, "result": None, "error": str(e)}
103
-
104
-
105
- @mcp.tool(name="closest_bed", description="Find the closest features between two BED files.")
106
- def closest_bed(file1: str, file2: str, report_distance: bool = True) -> dict:
107
- """
108
- For each feature in file1, finds the closest feature in file2.
109
-
110
- :param file1: Path to the first BED file.
111
- :param file2: Path to the second BED file.
112
- :param report_distance: If True, report the distance to the closest feature.
113
- :return: Dictionary containing success status and result or error message.
114
- """
115
- try:
116
- a = BedTool(file1)
117
- b = BedTool(file2)
118
- result = a.closest(b, d=report_distance)
119
- return {"success": True, "result": str(result), "error": None}
120
- except Exception as e:
121
- return {"success": False, "result": None, "error": str(e)}
122
-
123
-
124
- @mcp.tool(name="window_bed", description="Find overlapping features within a window.")
125
- def window_bed(file1: str, file2: str, window_size: int = 1000) -> dict:
126
- """
127
- Reports features in file2 that are within a window of features in file1.
128
-
129
- :param file1: Path to the first BED file.
130
- :param file2: Path to the second BED file.
131
- :param window_size: Window size in base pairs (default: 1000).
132
- :return: Dictionary containing success status and result or error message.
133
- """
134
- try:
135
- a = BedTool(file1)
136
- b = BedTool(file2)
137
- result = a.window(b, w=window_size)
138
- return {"success": True, "result": str(result), "error": None}
139
- except Exception as e:
140
- return {"success": False, "result": None, "error": str(e)}
141
-
142
-
143
- @mcp.tool(name="complement_bed", description="Get regions not covered by the BED file.")
144
- def complement_bed(file: str, genome: str) -> dict:
145
- """
146
- Returns regions of the genome not covered by the BED file.
147
-
148
- :param file: Path to the BED file.
149
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
150
- :return: Dictionary containing success status and result or error message.
151
- """
152
- try:
153
- a = BedTool(file)
154
- result = a.complement(g=pybedtools.chromsizes_to_file(pybedtools.chromsizes(genome)))
155
- return {"success": True, "result": str(result), "error": None}
156
- except Exception as e:
157
- return {"success": False, "result": None, "error": str(e)}
158
-
159
-
160
- @mcp.tool(name="flank_bed", description="Create flanking regions for each feature.")
161
- def flank_bed(file: str, genome: str, left: int = 0, right: int = 0, both: int = 0) -> dict:
162
- """
163
- Creates flanking regions for each feature in the BED file.
164
-
165
- :param file: Path to the BED file.
166
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
167
- :param left: Number of bases to extend on the left (5' for + strand).
168
- :param right: Number of bases to extend on the right (3' for + strand).
169
- :param both: Number of bases to extend on both sides.
170
- :return: Dictionary containing success status and result or error message.
171
  """
172
- try:
173
- a = BedTool(file)
174
- if both > 0:
175
- result = a.flank(genome=genome, b=both)
176
- else:
177
- result = a.flank(genome=genome, l=left, r=right)
178
- return {"success": True, "result": str(result), "error": None}
179
- except Exception as e:
180
- return {"success": False, "result": None, "error": str(e)}
181
 
 
 
 
182
 
183
- @mcp.tool(name="slop_bed", description="Extend features by a specified number of bases.")
184
- def slop_bed(file: str, genome: str, left: int = 0, right: int = 0, both: int = 0) -> dict:
185
- """
186
- Extends features by a specified number of bases.
187
-
188
- :param file: Path to the BED file.
189
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
190
- :param left: Number of bases to extend on the left.
191
- :param right: Number of bases to extend on the right.
192
- :param both: Number of bases to extend on both sides.
193
- :return: Dictionary containing success status and result or error message.
194
  """
195
  try:
196
- a = BedTool(file)
197
- if both > 0:
198
- result = a.slop(genome=genome, b=both)
 
199
  else:
200
- result = a.slop(genome=genome, l=left, r=right)
201
- return {"success": True, "result": str(result), "error": None}
202
- except Exception as e:
203
- return {"success": False, "result": None, "error": str(e)}
204
-
205
-
206
- @mcp.tool(name="shift_bed", description="Shift features by a specified number of bases.")
207
- def shift_bed(file: str, genome: str, shift_amount: int) -> dict:
208
- """
209
- Shifts all features by a specified number of bases.
210
-
211
- :param file: Path to the BED file.
212
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
213
- :param shift_amount: Number of bases to shift (positive or negative).
214
- :return: Dictionary containing success status and result or error message.
215
- """
216
- try:
217
- a = BedTool(file)
218
- result = a.shift(genome=genome, s=shift_amount)
219
- return {"success": True, "result": str(result), "error": None}
220
- except Exception as e:
221
- return {"success": False, "result": None, "error": str(e)}
222
-
223
-
224
- @mcp.tool(name="shuffle_bed", description="Randomly shuffle features within a genome.")
225
- def shuffle_bed(file: str, genome: str, seed: int = None) -> dict:
226
- """
227
- Randomly repositions features within the genome.
228
-
229
- :param file: Path to the BED file.
230
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
231
- :param seed: Random seed for reproducibility (optional).
232
- :return: Dictionary containing success status and result or error message.
233
- """
234
- try:
235
- a = BedTool(file)
236
- kwargs = {"genome": genome}
237
- if seed is not None:
238
- kwargs["seed"] = seed
239
- result = a.shuffle(**kwargs)
240
- return {"success": True, "result": str(result), "error": None}
241
- except Exception as e:
242
- return {"success": False, "result": None, "error": str(e)}
243
-
244
-
245
- @mcp.tool(name="cluster_bed", description="Cluster overlapping or nearby features.")
246
- def cluster_bed(file: str, distance: int = 0) -> dict:
247
- """
248
- Clusters overlapping or nearby features together.
249
-
250
- :param file: Path to the BED file.
251
- :param distance: Maximum distance between features to cluster (default: 0).
252
- :return: Dictionary containing success status and result with cluster IDs.
253
- """
254
- try:
255
- a = BedTool(file)
256
- result = a.cluster(d=distance)
257
- return {"success": True, "result": str(result), "error": None}
258
- except Exception as e:
259
- return {"success": False, "result": None, "error": str(e)}
260
-
261
 
262
- # =============================================================================
263
- # Coverage and Statistics
264
- # =============================================================================
265
-
266
- @mcp.tool(name="coverage_bed", description="Calculate coverage of features.")
267
- def coverage_bed(file1: str, file2: str) -> dict:
268
- """
269
- Computes the depth and breadth of coverage of features in file1 by file2.
270
-
271
- :param file1: Path to the first BED file (features to check coverage on).
272
- :param file2: Path to the second BED file (features to use for coverage).
273
- :return: Dictionary containing success status and coverage result.
274
- """
275
- try:
276
- a = BedTool(file1)
277
- b = BedTool(file2)
278
- result = a.coverage(b)
279
- return {"success": True, "result": str(result), "error": None}
280
- except Exception as e:
281
- return {"success": False, "result": None, "error": str(e)}
282
-
283
-
284
- @mcp.tool(name="genome_coverage", description="Calculate genome-wide coverage histogram.")
285
- def genome_coverage(file: str, genome: str) -> dict:
286
- """
287
- Computes a histogram of genome-wide coverage.
288
-
289
- :param file: Path to the BED file.
290
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
291
- :return: Dictionary containing success status and coverage histogram.
292
- """
293
- try:
294
- a = BedTool(file)
295
- result = a.genome_coverage(g=pybedtools.chromsizes_to_file(pybedtools.chromsizes(genome)))
296
- return {"success": True, "result": str(result), "error": None}
297
- except Exception as e:
298
- return {"success": False, "result": None, "error": str(e)}
299
-
300
-
301
- @mcp.tool(name="jaccard_index", description="Calculate Jaccard index between two BED files.")
302
- def jaccard_index(file1: str, file2: str) -> dict:
303
- """
304
- Calculates the Jaccard index (intersection over union) between two BED files.
305
-
306
- :param file1: Path to the first BED file.
307
- :param file2: Path to the second BED file.
308
- :return: Dictionary containing Jaccard statistics.
309
- """
310
- try:
311
- a = BedTool(file1)
312
- b = BedTool(file2)
313
- result = a.jaccard(b)
314
  return {
315
  "success": True,
316
- "result": {
317
- "intersection": result.get("intersection", 0),
318
- "union": result.get("union-intersection", 0),
319
- "jaccard": result.get("jaccard", 0),
320
- "n_intersections": result.get("n_intersections", 0)
321
- },
322
- "error": None
323
  }
324
  except Exception as e:
325
- return {"success": False, "result": None, "error": str(e)}
326
-
327
 
328
- @mcp.tool(name="fisher_test", description="Perform Fisher's exact test on interval overlaps.")
329
- def fisher_test(file1: str, file2: str, genome: str) -> dict:
330
  """
331
- Performs Fisher's exact test on the overlap between two BED files.
332
 
333
- :param file1: Path to the first BED file.
334
- :param file2: Path to the second BED file.
335
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
336
- :return: Dictionary containing Fisher's test results.
 
 
337
  """
338
  try:
339
- a = BedTool(file1)
340
- b = BedTool(file2)
341
- result = a.fisher(b, genome=genome)
 
342
  return {
343
  "success": True,
344
- "result": str(result),
345
- "error": None
346
  }
347
  except Exception as e:
348
- return {"success": False, "result": None, "error": str(e)}
349
-
350
-
351
- @mcp.tool(name="count_features", description="Count the number of features in a BED file.")
352
- def count_features(file: str) -> dict:
353
- """
354
- Counts the number of features (lines) in a BED file.
355
-
356
- :param file: Path to the BED file.
357
- :return: Dictionary containing the feature count.
358
- """
359
- try:
360
- a = BedTool(file)
361
- count = len(a)
362
- return {"success": True, "result": {"count": count}, "error": None}
363
- except Exception as e:
364
- return {"success": False, "result": None, "error": str(e)}
365
-
366
-
367
- @mcp.tool(name="total_coverage_bp", description="Calculate total base pairs covered.")
368
- def total_coverage_bp(file: str) -> dict:
369
- """
370
- Calculates the total number of base pairs covered by the BED file.
371
- Merges overlapping regions first to avoid double-counting.
372
-
373
- :param file: Path to the BED file.
374
- :return: Dictionary containing total covered base pairs.
375
- """
376
- try:
377
- a = BedTool(file)
378
- total_bp = a.total_coverage()
379
- return {"success": True, "result": {"total_bp": total_bp}, "error": None}
380
- except Exception as e:
381
- return {"success": False, "result": None, "error": str(e)}
382
-
383
-
384
- # =============================================================================
385
- # Multi-file Operations
386
- # =============================================================================
387
-
388
- @mcp.tool(name="multi_intersect", description="Find intersection of multiple BED files.")
389
- def multi_intersect(files: List[str]) -> dict:
390
- """
391
- Identifies regions that are common to multiple BED files.
392
-
393
- :param files: List of paths to BED files.
394
- :return: Dictionary containing multi-intersection result.
395
- """
396
- try:
397
- x = BedTool()
398
- result = x.multi_intersect(i=files)
399
- return {"success": True, "result": str(result), "error": None}
400
- except Exception as e:
401
- return {"success": False, "result": None, "error": str(e)}
402
-
403
-
404
- @mcp.tool(name="union_bedgraphs", description="Combine multiple BedGraph files.")
405
- def union_bedgraphs(files: List[str]) -> dict:
406
- """
407
- Combines multiple BedGraph files into a single unified BedGraph.
408
-
409
- :param files: List of paths to BedGraph files.
410
- :return: Dictionary containing union result.
411
- """
412
- try:
413
- x = BedTool()
414
- result = x.union_bedgraphs(i=files)
415
- return {"success": True, "result": str(result), "error": None}
416
- except Exception as e:
417
- return {"success": False, "result": None, "error": str(e)}
418
-
419
 
420
- # =============================================================================
421
- # Feature Manipulation
422
- # =============================================================================
423
-
424
- @mcp.tool(name="get_fasta", description="Extract sequences from a FASTA file for BED regions.")
425
- def get_fasta(bed_file: str, fasta_file: str, strand: bool = False) -> dict:
426
  """
427
- Extracts sequences from a FASTA file based on BED coordinates.
428
 
429
- :param bed_file: Path to the BED file.
430
- :param fasta_file: Path to the FASTA file.
431
- :param strand: If True, sequences on - strand are reverse complemented.
432
- :return: Dictionary containing extracted sequences.
433
- """
434
- try:
435
- a = BedTool(bed_file)
436
- result = a.sequence(fi=fasta_file, s=strand)
437
- # Read the sequence file
438
- with open(result.seqfn, 'r') as f:
439
- sequences = f.read()
440
- return {"success": True, "result": sequences, "error": None}
441
- except Exception as e:
442
- return {"success": False, "result": None, "error": str(e)}
443
-
444
-
445
- @mcp.tool(name="map_features", description="Map values from one BED file to another.")
446
- def map_features(file1: str, file2: str, column: int = 5, operation: str = "mean") -> dict:
447
- """
448
- Maps values from file2 onto features in file1.
449
 
450
- :param file1: Path to the first BED file (features to map onto).
451
- :param file2: Path to the second BED file (source of values).
452
- :param column: Column number in file2 to extract values from (1-based).
453
- :param operation: Operation to apply (mean, sum, min, max, count, etc.).
454
- :return: Dictionary containing mapped result.
455
  """
456
  try:
457
- a = BedTool(file1)
458
- b = BedTool(file2)
459
- result = a.map(b, c=column, o=operation)
460
- return {"success": True, "result": str(result), "error": None}
461
- except Exception as e:
462
- return {"success": False, "result": None, "error": str(e)}
463
-
464
-
465
- @mcp.tool(name="groupby_bed", description="Group features and apply operations.")
466
- def groupby_bed(file: str, group_columns: List[int], operation_column: int, operation: str = "sum") -> dict:
467
- """
468
- Groups features by specified columns and applies an operation.
469
 
470
- :param file: Path to the BED file.
471
- :param group_columns: List of column numbers to group by (1-based).
472
- :param operation_column: Column number to apply operation on (1-based).
473
- :param operation: Operation to apply (sum, mean, count, min, max, etc.).
474
- :return: Dictionary containing grouped result.
475
- """
476
- try:
477
- a = BedTool(file)
478
- result = a.groupby(g=group_columns, c=operation_column, o=operation)
479
- return {"success": True, "result": str(result), "error": None}
480
- except Exception as e:
481
- return {"success": False, "result": None, "error": str(e)}
482
-
483
-
484
- @mcp.tool(name="sample_bed", description="Randomly sample features from a BED file.")
485
- def sample_bed(file: str, n: int = None, fraction: float = None, seed: int = None) -> dict:
486
- """
487
- Randomly samples features from a BED file.
488
-
489
- :param file: Path to the BED file.
490
- :param n: Number of features to sample.
491
- :param fraction: Fraction of features to sample (0.0 to 1.0).
492
- :param seed: Random seed for reproducibility.
493
- :return: Dictionary containing sampled features.
494
- """
495
- try:
496
- a = BedTool(file)
497
- result = a.random_subset(n=n, f=fraction, seed=seed)
498
- return {"success": True, "result": str(result), "error": None}
499
  except Exception as e:
500
- return {"success": False, "result": None, "error": str(e)}
501
-
502
-
503
- # =============================================================================
504
- # Window and Genome Operations
505
- # =============================================================================
506
-
507
- @mcp.tool(name="make_windows", description="Create windows across a genome or BED file.")
508
- def make_windows(genome: str = None, window_size: int = 1000, step_size: int = None) -> dict:
509
- """
510
- Creates fixed-size windows across a genome.
511
 
512
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
513
- :param window_size: Size of each window in base pairs.
514
- :param step_size: Step size for sliding windows (default: same as window_size).
515
- :return: Dictionary containing window coordinates.
516
  """
517
- try:
518
- x = BedTool()
519
- kwargs = {"genome": genome, "w": window_size}
520
- if step_size is not None:
521
- kwargs["s"] = step_size
522
- result = x.window_maker(**kwargs)
523
- return {"success": True, "result": str(result), "error": None}
524
- except Exception as e:
525
- return {"success": False, "result": None, "error": str(e)}
526
 
 
 
 
527
 
528
- @mcp.tool(name="random_intervals", description="Generate random intervals in a genome.")
529
- def random_intervals(genome: str, length: int = 100, num_intervals: int = 10, seed: int = None) -> dict:
530
- """
531
- Generates random intervals within a genome.
532
-
533
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
534
- :param length: Length of each interval in base pairs.
535
- :param num_intervals: Number of intervals to generate.
536
- :param seed: Random seed for reproducibility.
537
- :return: Dictionary containing random intervals.
538
  """
539
  try:
540
- x = BedTool()
541
- kwargs = {"genome": genome, "l": length, "n": num_intervals}
542
- if seed is not None:
543
- kwargs["seed"] = seed
544
- result = x.random(**kwargs)
545
- return {"success": True, "result": str(result), "error": None}
546
- except Exception as e:
547
- return {"success": False, "result": None, "error": str(e)}
548
-
549
-
550
- # =============================================================================
551
- # File Information
552
- # =============================================================================
553
 
554
- @mcp.tool(name="get_bed_info", description="Get information about a BED file.")
555
- def get_bed_info(file: str) -> dict:
556
- """
557
- Returns information about a BED file including feature count, field count, and file type.
558
-
559
- :param file: Path to the BED file.
560
- :return: Dictionary containing file information.
561
- """
562
- try:
563
- a = BedTool(file)
564
- info = {
565
- "feature_count": len(a),
566
- "field_count": a.field_count(),
567
- "file_type": a.file_type,
568
  }
569
- return {"success": True, "result": info, "error": None}
570
  except Exception as e:
571
- return {"success": False, "result": None, "error": str(e)}
572
-
573
 
574
- @mcp.tool(name="head_bed", description="Get the first N features from a BED file.")
575
- def head_bed(file: str, n: int = 10) -> dict:
576
  """
577
- Returns the first N features from a BED file.
578
 
579
- :param file: Path to the BED file.
580
- :param n: Number of features to return (default: 10).
581
- :return: Dictionary containing the first N features.
582
- """
583
- try:
584
- a = BedTool(file)
585
- result = a.head(n=n, as_string=True)
586
- return {"success": True, "result": result, "error": None}
587
- except Exception as e:
588
- return {"success": False, "result": None, "error": str(e)}
589
-
590
-
591
- @mcp.tool(name="get_chromsizes", description="Get chromosome sizes for a genome.")
592
- def get_chromsizes(genome: str) -> dict:
593
- """
594
- Returns chromosome sizes for a specified genome.
595
 
596
- :param genome: Genome name (e.g., 'hg19', 'hg38', 'mm10').
597
- :return: Dictionary containing chromosome sizes.
598
  """
599
  try:
600
- chromsizes = pybedtools.chromsizes(genome)
601
- # Convert to serializable format
602
- result = {chrom: {"start": coords[0], "end": coords[1]}
603
- for chrom, coords in chromsizes.items()}
604
- return {"success": True, "result": result, "error": None}
605
- except Exception as e:
606
- return {"success": False, "result": None, "error": str(e)}
607
-
608
-
609
- # =============================================================================
610
- # Utility Functions
611
- # =============================================================================
612
-
613
- @mcp.tool(name="create_bed_from_string", description="Create a BED file from a string.")
614
- def create_bed_from_string(bed_content: str, output_file: str = None) -> dict:
615
- """
616
- Creates a BED file from a string content.
617
-
618
- :param bed_content: BED content as a string (tab or space separated).
619
- :param output_file: Optional output file path. If not provided, returns temp file path.
620
- :return: Dictionary containing the file path.
621
- """
622
- try:
623
- a = BedTool(bed_content, from_string=True)
624
- if output_file:
625
- a.saveas(output_file)
626
- return {"success": True, "result": {"file_path": output_file}, "error": None}
627
- else:
628
- return {"success": True, "result": {"file_path": a.fn, "content": str(a)}, "error": None}
629
- except Exception as e:
630
- return {"success": False, "result": None, "error": str(e)}
631
-
632
-
633
- @mcp.tool(name="save_bed", description="Save BED operation result to a file.")
634
- def save_bed(file: str, output_file: str) -> dict:
635
- """
636
- Saves a BED file to a specified location.
637
 
638
- :param file: Path to the input BED file.
639
- :param output_file: Path to save the output file.
640
- :return: Dictionary containing save status.
641
- """
642
- try:
643
- a = BedTool(file)
644
- a.saveas(output_file)
645
- return {"success": True, "result": {"saved_to": output_file}, "error": None}
646
- except Exception as e:
647
- return {"success": False, "result": None, "error": str(e)}
648
-
649
-
650
- @mcp.tool(name="cleanup_temp_files", description="Clean up temporary files created by pybedtools.")
651
- def cleanup_temp_files() -> dict:
652
- """
653
- Cleans up all temporary files created by pybedtools.
654
-
655
- :return: Dictionary containing cleanup status.
656
- """
657
- try:
658
- cleanup()
659
- return {"success": True, "result": "Temporary files cleaned up", "error": None}
660
  except Exception as e:
661
- return {"success": False, "result": None, "error": str(e)}
662
-
663
 
 
664
  def create_app() -> FastMCP:
665
  """
666
- Creates and returns the FastMCP application instance.
667
 
668
- :return: FastMCP instance.
 
669
  """
670
  return mcp
 
 
 
 
 
1
  from fastmcp import FastMCP
2
+ from pybedtools import BedTool
 
 
 
 
3
 
4
  # Create the FastMCP service application
5
  mcp = FastMCP("pybedtools_service")
6
 
7
+ @mcp.tool(name="create_bedtool", description="Create a BedTool object from a file or string")
8
+ def create_bedtool(file_path: str = None, from_string: str = None) -> dict:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
9
  """
10
+ Create a BedTool object from a file or a string.
 
 
 
 
 
 
 
 
11
 
12
+ Parameters:
13
+ - file_path: Path to the BED file.
14
+ - from_string: String containing BED file content.
15
 
16
+ Returns:
17
+ - dict: Success status and BedTool object information.
 
 
 
 
 
 
 
 
 
18
  """
19
  try:
20
+ if file_path:
21
+ bedtool = BedTool(file_path)
22
+ elif from_string:
23
+ bedtool = BedTool(from_string, from_string=True)
24
  else:
25
+ return {"success": False, "error": "Either file_path or from_string must be provided."}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
26
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
27
  return {
28
  "success": True,
29
+ "message": "BedTool object created successfully.",
30
+ "bedtool": str(bedtool)
 
 
 
 
 
31
  }
32
  except Exception as e:
33
+ return {"success": False, "error": str(e)}
 
34
 
35
+ @mcp.tool(name="intersect_bedtools", description="Find the intersection of two BedTool objects")
36
+ def intersect_bedtools(file1: str, file2: str) -> dict:
37
  """
38
+ Find the intersection of two BedTool objects.
39
 
40
+ Parameters:
41
+ - file1: Path to the first BED file.
42
+ - file2: Path to the second BED file.
43
+
44
+ Returns:
45
+ - dict: Success status and intersection result.
46
  """
47
  try:
48
+ bed1 = BedTool(file1)
49
+ bed2 = BedTool(file2)
50
+ intersection = bed1.intersect(bed2)
51
+
52
  return {
53
  "success": True,
54
+ "message": "Intersection computed successfully.",
55
+ "result": str(intersection)
56
  }
57
  except Exception as e:
58
+ return {"success": False, "error": str(e)}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
59
 
60
+ @mcp.tool(name="union_bedtools", description="Find the union of two BedTool objects")
61
+ def union_bedtools(file1: str, file2: str) -> dict:
 
 
 
 
62
  """
63
+ Find the union of two BedTool objects.
64
 
65
+ Parameters:
66
+ - file1: Path to the first BED file.
67
+ - file2: Path to the second BED file.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
68
 
69
+ Returns:
70
+ - dict: Success status and union result.
 
 
 
71
  """
72
  try:
73
+ bed1 = BedTool(file1)
74
+ bed2 = BedTool(file2)
75
+ union = bed1.cat(bed2, postmerge=True)
 
 
 
 
 
 
 
 
 
76
 
77
+ return {
78
+ "success": True,
79
+ "message": "Union computed successfully.",
80
+ "result": str(union)
81
+ }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
82
  except Exception as e:
83
+ return {"success": False, "error": str(e)}
 
 
 
 
 
 
 
 
 
 
84
 
85
+ @mcp.tool(name="filter_bedtool", description="Filter a BedTool object based on a condition")
86
+ def filter_bedtool(file_path: str, condition: str) -> dict:
 
 
87
  """
88
+ Filter a BedTool object based on a condition.
 
 
 
 
 
 
 
 
89
 
90
+ Parameters:
91
+ - file_path: Path to the BED file.
92
+ - condition: A lambda function as a string to filter intervals.
93
 
94
+ Returns:
95
+ - dict: Success status and filtered result.
 
 
 
 
 
 
 
 
96
  """
97
  try:
98
+ bedtool = BedTool(file_path)
99
+ filtered = bedtool.filter(eval(condition))
 
 
 
 
 
 
 
 
 
 
 
100
 
101
+ return {
102
+ "success": True,
103
+ "message": "Filtering completed successfully.",
104
+ "result": str(filtered)
 
 
 
 
 
 
 
 
 
 
105
  }
 
106
  except Exception as e:
107
+ return {"success": False, "error": str(e)}
 
108
 
109
+ @mcp.tool(name="sort_bedtool", description="Sort a BedTool object")
110
+ def sort_bedtool(file_path: str) -> dict:
111
  """
112
+ Sort a BedTool object.
113
 
114
+ Parameters:
115
+ - file_path: Path to the BED file.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
116
 
117
+ Returns:
118
+ - dict: Success status and sorted result.
119
  """
120
  try:
121
+ bedtool = BedTool(file_path)
122
+ sorted_bed = bedtool.sort()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
123
 
124
+ return {
125
+ "success": True,
126
+ "message": "Sorting completed successfully.",
127
+ "result": str(sorted_bed)
128
+ }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
129
  except Exception as e:
130
+ return {"success": False, "error": str(e)}
 
131
 
132
+ @mcp.tool(name="create_app", description="Create and return the FastMCP application instance")
133
  def create_app() -> FastMCP:
134
  """
135
+ Create and return the FastMCP application instance.
136
 
137
+ Returns:
138
+ - FastMCP: The FastMCP application instance.
139
  """
140
  return mcp