|
5 | 5 | import json |
6 | 6 | import uuid |
7 | 7 | from Bio import Entrez, SeqIO |
8 | | -from Bio.Align import PairwiseAligner |
9 | 8 | from Bio.Seq import Seq |
10 | 9 | from pydantic import BaseModel, Field, ConfigDict |
11 | 10 | from pathlib import Path |
@@ -37,36 +36,6 @@ class EntrezDownloadRequest(BaseModel): |
37 | 36 | } |
38 | 37 | } |
39 | 38 |
|
40 | | -class PairwiseAlignmentRequest(BaseModel): |
41 | | - sequence1: str = Field(..., description="First sequence for alignment.") |
42 | | - sequence2: str = Field(..., description="Second sequence for alignment.") |
43 | | - match_score: float = Field(1.0, description="Score for a match.") |
44 | | - mismatch_penalty: float = Field(-1.0, description="Penalty for a mismatch. Should be negative or zero.") |
45 | | - open_gap_penalty: float = Field(-0.5, description="Penalty for opening a gap. Should be negative or zero.") |
46 | | - extend_gap_penalty: float = Field(-0.1, description="Penalty for extending a gap. Should be negative or zero.") |
47 | | - mode: Literal["global", "local"] = Field("global", description="Alignment mode: 'global' or 'local'.") |
48 | | - |
49 | | - model_config = ConfigDict( |
50 | | - json_schema_extra = { |
51 | | - "example": { |
52 | | - "sequence1": "GATTACA", |
53 | | - "sequence2": "GCATGCU", |
54 | | - "match_score": 2.0, |
55 | | - "mismatch_penalty": -1.0, |
56 | | - "open_gap_penalty": -1.0, |
57 | | - "extend_gap_penalty": -0.5, |
58 | | - "mode": "global" |
59 | | - } |
60 | | - } |
61 | | - ) |
62 | | - |
63 | | -class PairwiseAlignmentResponse(BaseModel): |
64 | | - score: float |
65 | | - aligned_sequence1: str |
66 | | - aligned_sequence2: str |
67 | | - full_alignment_str: str |
68 | | - parameters_used: Dict |
69 | | - |
70 | 39 | # Type hint for local file results |
71 | 40 | LocalFileResult = Dict[Literal["path", "format", "success", "error"], Any] |
72 | 41 |
|
@@ -286,215 +255,7 @@ def download_entrez_data_local( |
286 | 255 | "success": False, |
287 | 256 | "error": f"An unexpected error occurred during Entrez download: {str(e)}" |
288 | 257 | } |
289 | | - |
290 | | - @self.mcp_server.tool( |
291 | | - name=f"{self.prefix}perform_pairwise_alignment", |
292 | | - description="""Perform pairwise sequence alignment using Biopython's PairwiseAligner. |
293 | | - |
294 | | - Performs a pairwise sequence alignment (global or local) using Biopython's PairwiseAligner. |
295 | | - You can specify sequences and alignment scoring parameters. |
296 | | - |
297 | | - **Parameters:** |
298 | | - - `sequence1` (str): First sequence for alignment |
299 | | - - `sequence2` (str): Second sequence for alignment |
300 | | - - `match_score` (float): Score for a match (default: 1.0) |
301 | | - - `mismatch_penalty` (float): Penalty for a mismatch (default: -1.0, should be negative or zero) |
302 | | - - `open_gap_penalty` (float): Penalty for opening a gap (default: -0.5, should be negative or zero) |
303 | | - - `extend_gap_penalty` (float): Penalty for extending a gap (default: -0.1, should be negative or zero) |
304 | | - - `mode` (str): Alignment mode - 'global' or 'local' (default: 'global') |
305 | | - |
306 | | - **Returns:** |
307 | | - PairwiseAlignmentResponse containing: |
308 | | - - `score`: Alignment score |
309 | | - - `aligned_sequence1`: First sequence with gaps |
310 | | - - `aligned_sequence2`: Second sequence with gaps |
311 | | - - `full_alignment_str`: Complete alignment visualization |
312 | | - - `parameters_used`: Parameters used for the alignment |
313 | | - |
314 | | - **Example Usage:** |
315 | | - ``` |
316 | | - perform_pairwise_alignment( |
317 | | - sequence1="GATTACA", |
318 | | - sequence2="GCATGCU", |
319 | | - match_score=2.0, |
320 | | - mismatch_penalty=-1.0, |
321 | | - mode="global" |
322 | | - ) |
323 | | - ``` |
324 | | - """) |
325 | | - def perform_pairwise_alignment( |
326 | | - sequence1: str, |
327 | | - sequence2: str, |
328 | | - match_score: float = 1.0, |
329 | | - mismatch_penalty: float = -1.0, |
330 | | - open_gap_penalty: float = -0.5, |
331 | | - extend_gap_penalty: float = -0.1, |
332 | | - mode: Literal["global", "local"] = "global" |
333 | | - ) -> PairwiseAlignmentResponse: |
334 | | - with start_action(action_type="perform_pairwise_alignment", |
335 | | - sequence1_length=len(sequence1), |
336 | | - sequence2_length=len(sequence2), |
337 | | - mode=mode) as action: |
338 | | - try: |
339 | | - request = PairwiseAlignmentRequest( |
340 | | - sequence1=sequence1, |
341 | | - sequence2=sequence2, |
342 | | - match_score=match_score, |
343 | | - mismatch_penalty=mismatch_penalty, |
344 | | - open_gap_penalty=open_gap_penalty, |
345 | | - extend_gap_penalty=extend_gap_penalty, |
346 | | - mode=mode |
347 | | - ) |
348 | | - response = run_pairwise_alignment(request) |
349 | | - action.add_success_fields(alignment_score=response.score) |
350 | | - return response |
351 | | - except ValueError as ve: |
352 | | - action.add_error_fields(error=str(ve)) |
353 | | - raise ValueError(str(ve)) from ve |
354 | | - except Exception as e: |
355 | | - action.add_error_fields(error=f"Unexpected error during alignment: {str(e)}") |
356 | | - raise ValueError(f"An unexpected error occurred during alignment: {str(e)}") from e |
357 | | - |
358 | | - @self.mcp_server.tool( |
359 | | - name=f"{self.prefix}perform_pairwise_alignment_local", |
360 | | - description="""Perform pairwise sequence alignment and save results to local file. |
361 | | - |
362 | | - Same as perform_pairwise_alignment but saves the alignment result to a local file instead of returning it. |
363 | | - This is useful for preserving alignment results for further analysis. |
364 | | - |
365 | | - **Parameters:** |
366 | | - - `sequence1` (str): First sequence for alignment |
367 | | - - `sequence2` (str): Second sequence for alignment |
368 | | - - `match_score` (float): Score for a match (default: 1.0) |
369 | | - - `mismatch_penalty` (float): Penalty for a mismatch (default: -1.0, should be negative or zero) |
370 | | - - `open_gap_penalty` (float): Penalty for opening a gap (default: -0.5, should be negative or zero) |
371 | | - - `extend_gap_penalty` (float): Penalty for extending a gap (default: -0.1, should be negative or zero) |
372 | | - - `mode` (str): Alignment mode - 'global' or 'local' (default: 'global') |
373 | | - - `output_path` (Optional[str]): Custom output path. If None, generates unique filename |
374 | | - |
375 | | - **Returns:** |
376 | | - LocalFileResult containing: |
377 | | - - `path`: Path to the saved alignment file |
378 | | - - `format`: File format used ('alignment') |
379 | | - - `success`: Whether the operation succeeded |
380 | | - - `error`: Error message if failed |
381 | | - |
382 | | - **Example Usage:** |
383 | | - ``` |
384 | | - perform_pairwise_alignment_local( |
385 | | - sequence1="GATTACA", |
386 | | - sequence2="GCATGCU", |
387 | | - match_score=2.0, |
388 | | - mismatch_penalty=-1.0, |
389 | | - mode="global", |
390 | | - output_path="alignment_result.aln" |
391 | | - ) |
392 | | - ``` |
393 | | - """) |
394 | | - def perform_pairwise_alignment_local( |
395 | | - sequence1: str, |
396 | | - sequence2: str, |
397 | | - match_score: float = 1.0, |
398 | | - mismatch_penalty: float = -1.0, |
399 | | - open_gap_penalty: float = -0.5, |
400 | | - extend_gap_penalty: float = -0.1, |
401 | | - mode: Literal["global", "local"] = "global", |
402 | | - output_path: Optional[str] = None |
403 | | - ) -> LocalFileResult: |
404 | | - with start_action(action_type="perform_pairwise_alignment_local", |
405 | | - sequence1_length=len(sequence1), |
406 | | - sequence2_length=len(sequence2), |
407 | | - mode=mode) as action: |
408 | | - try: |
409 | | - request = PairwiseAlignmentRequest( |
410 | | - sequence1=sequence1, |
411 | | - sequence2=sequence2, |
412 | | - match_score=match_score, |
413 | | - mismatch_penalty=mismatch_penalty, |
414 | | - open_gap_penalty=open_gap_penalty, |
415 | | - extend_gap_penalty=extend_gap_penalty, |
416 | | - mode=mode |
417 | | - ) |
418 | | - response = run_pairwise_alignment(request) |
419 | | - |
420 | | - # Create detailed alignment output |
421 | | - alignment_content = f"""Pairwise Alignment Results |
422 | | -============================= |
423 | | -
|
424 | | -Parameters: |
425 | | -- Match Score: {response.parameters_used['match_score']} |
426 | | -- Mismatch Penalty: {response.parameters_used['mismatch_penalty']} |
427 | | -- Open Gap Penalty: {response.parameters_used['open_gap_penalty']} |
428 | | -- Extend Gap Penalty: {response.parameters_used['extend_gap_penalty']} |
429 | | -- Mode: {response.parameters_used['mode']} |
430 | | -
|
431 | | -Alignment Score: {response.score} |
432 | | -
|
433 | | -Aligned Sequences: |
434 | | -{response.full_alignment_str} |
435 | | -
|
436 | | -Sequence 1 (aligned): {response.aligned_sequence1} |
437 | | -Sequence 2 (aligned): {response.aligned_sequence2} |
438 | | -""" |
439 | | - |
440 | | - result = self._save_to_local_file( |
441 | | - data=alignment_content, |
442 | | - format_type="alignment", |
443 | | - output_path=output_path, |
444 | | - default_prefix="pairwise_alignment" |
445 | | - ) |
446 | | - action.add_success_fields( |
447 | | - alignment_score=response.score, |
448 | | - saved_to=result.get("path"), |
449 | | - success=result.get("success") |
450 | | - ) |
451 | | - return result |
452 | | - except ValueError as ve: |
453 | | - action.add_error_fields(error=str(ve)) |
454 | | - return { |
455 | | - "path": None, |
456 | | - "format": "alignment", |
457 | | - "success": False, |
458 | | - "error": str(ve) |
459 | | - } |
460 | | - except Exception as e: |
461 | | - action.add_error_fields(error=f"Unexpected error during alignment: {str(e)}") |
462 | | - return { |
463 | | - "path": None, |
464 | | - "format": "alignment", |
465 | | - "success": False, |
466 | | - "error": f"An unexpected error occurred during alignment: {str(e)}" |
467 | | - } |
468 | | - |
469 | | -def run_pairwise_alignment(request: PairwiseAlignmentRequest) -> PairwiseAlignmentResponse: |
470 | | - """Run pairwise alignment using Biopython.""" |
471 | | - aligner = PairwiseAligner() |
472 | | - aligner.match_score = request.match_score |
473 | | - aligner.mismatch_score = request.mismatch_penalty |
474 | | - aligner.open_gap_score = request.open_gap_penalty |
475 | | - aligner.extend_gap_score = request.extend_gap_penalty |
476 | | - aligner.mode = request.mode |
477 | | - |
478 | | - seq1 = Seq(request.sequence1) |
479 | | - seq2 = Seq(request.sequence2) |
480 | | - |
481 | | - alignments = list(aligner.align(seq1, seq2)) |
482 | | - |
483 | | - if not alignments: |
484 | | - raise ValueError("No alignment could be produced with the given sequences and parameters.") |
485 | | - |
486 | | - best_alignment = alignments[0] |
487 | | - |
488 | | - aligned_seq1_str = str(best_alignment[0]) |
489 | | - aligned_seq2_str = str(best_alignment[1]) |
490 | | - |
491 | | - return PairwiseAlignmentResponse( |
492 | | - score=best_alignment.score, |
493 | | - aligned_sequence1=aligned_seq1_str, |
494 | | - aligned_sequence2=aligned_seq2_str, |
495 | | - full_alignment_str=str(best_alignment), |
496 | | - parameters_used=request.model_dump() |
497 | | - ) |
| 258 | + |
498 | 259 |
|
499 | 260 | def get_entrez(ids: List[str], db: DB_LITERAL, reftype: Literal["fasta", "gb"]) -> str: |
500 | 261 | """ |
|
0 commit comments