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

Update pybedtools/mcp_output/mcp_plugin/mcp_service.py

Browse files
pybedtools/mcp_output/mcp_plugin/mcp_service.py CHANGED
@@ -1,140 +1,670 @@
 
 
 
 
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
 
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