Repository navigation
Expand file tree
/
Copy pathsplicecraft_cli.py
More file actions
860 lines (740 loc) · 35.3 KB
/
Copy pathsplicecraft_cli.py
File metadata and controls
860 lines (740 loc) · 35.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
#!/usr/bin/env python3
"""splicecraft-cli — drive a running SpliceCraft GUI session from any
external CLI agent (Claude Code, Cursor, aider, hand-rolled scripts).
Connects to the localhost JSON API exposed by `splicecraft --agent`
(alias of the original `--agent-api`). Reads connection details
(port + bearer token) from
``~/.local/share/splicecraft/agent_token`` (or
``$SPLICECRAFT_DATA_DIR/agent_token`` when overridden), so the running
GUI is always the destination — no flag-fiddling required.
Stdlib-only by design: imports complete in ~50 ms (vs ~1.5 s for the
GUI module), so an AI agent firing dozens of commands per session
doesn't pay startup cost on every call.
Examples::
splicecraft-cli status
splicecraft-cli features
splicecraft-cli fetch L09137
splicecraft-cli load-entry pUC19
splicecraft-cli add-feature 100 200 --label lacZ --type CDS --strand 1
splicecraft-cli save
Use ``splicecraft-cli tools`` to list every endpoint the running
session exposes (handy when wiring a new tool surface for the agent).
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6701
TOKEN_FILENAME = "agent_token"
# Published by `splicecraft --agent --read-only` (an attach to a data dir
# another instance holds). Kept separate from TOKEN_FILENAME so a guest can
# never repoint — or, on quit, delete — the host's side-door.
READONLY_TOKEN_FILENAME = "agent_token.readonly"
def _data_dir() -> Path:
"""Resolve the SpliceCraft user-data directory the same way the
GUI does — env override first, then the platform default. We keep
a hand-rolled fallback so the CLI doesn't pull in `platformdirs`
just for one path lookup (keeps imports stdlib-only)."""
override = os.environ.get("SPLICECRAFT_DATA_DIR")
# Sweep #9 (2026-05-19): strip + truthiness check so a stray
# `SPLICECRAFT_DATA_DIR=" "` (shell scripting bug) doesn't land
# us on a literal whitespace-named relative dir.
if override and override.strip():
return Path(override.strip()).expanduser()
try:
from platformdirs import user_data_dir # noqa: WPS433 (lazy)
return Path(user_data_dir("splicecraft", appauthor=False,
roaming=False))
except ImportError:
return Path.home() / ".local" / "share" / "splicecraft"
def _token_file() -> Path:
"""The token file to talk to.
Normally `agent_token`, written by whichever instance holds the data dir.
A `splicecraft --agent --read-only` attach is a GUEST on a data dir some
other process owns, so it publishes to `agent_token.readonly` instead and
leaves the host's file alone. Prefer the host; fall back to the guest when
no host has published, so `splicecraft-cli status` finds a read-only
session with no extra flags. `SPLICECRAFT_READ_ONLY=1` pins the guest file
explicitly for when both are present and you want the read-only one.
"""
d = _data_dir()
primary = d / TOKEN_FILENAME
guest = d / READONLY_TOKEN_FILENAME
if os.environ.get("SPLICECRAFT_READ_ONLY", "").strip().lower() in (
"1", "true", "yes"):
return guest
if not primary.exists() and guest.exists():
return guest
return primary
_CLI_TOKEN_FILE_MAX_BYTES = 1024
def _read_session() -> tuple[str, int, str]:
"""Return `(host, port, token)` from the running session's token
file. Exits with a helpful message if no session is up.
Capped at 1 KB defensively. A local-attacker-with-write or a
misbehaving co-resident process can fill the token file; reading
unbounded would let a hostile fill DoS the CLI."""
f = _token_file()
if not f.exists():
sys.exit(
f"No SpliceCraft session found.\n"
f" Expected token file: {f}\n"
f" Start the GUI with: splicecraft --agent\n"
f" Already have one open? Attach read-only alongside it:\n"
f" splicecraft --agent --read-only"
)
# Sweep #25 (2026-05-23): `lstat` not `stat` — `f.stat()`
# follows symlinks. A pre-placed symlink at the token path
# pointing to (e.g.) `/etc/passwd` would be silently followed,
# read up to 1 KB, and stderr-leak via the malformed-port
# error message ("Malformed port in <token-file>: 'root:x:0...'").
# Niche but trivially fixable.
import stat as _stat
try:
st = f.lstat()
if _stat.S_ISLNK(st.st_mode):
sys.exit(
f"Refusing to read symlinked token file {f}. "
f"Restart the GUI to regenerate."
)
size = st.st_size
except OSError as exc:
sys.exit(f"Could not stat token file {f}: {exc}")
if size > _CLI_TOKEN_FILE_MAX_BYTES:
sys.exit(
f"Refusing to read oversized token file {f} "
f"({size:,} bytes > {_CLI_TOKEN_FILE_MAX_BYTES}-byte cap). "
f"Restart the GUI to regenerate."
)
lines = f.read_text(encoding="utf-8").strip().splitlines()
if len(lines) < 2:
sys.exit(
f"Malformed token file at {f} "
f"(expected `port\\ntoken`)."
)
try:
port = int(lines[0].strip())
except ValueError:
sys.exit(f"Malformed port in {f}: {lines[0]!r}")
return DEFAULT_HOST, port, lines[1].strip()
_CLI_RESPONSE_MAX_BYTES = 50 * 1024 * 1024
class _AgentCliError(Exception):
"""Raised by `_request` on an unrecoverable call failure (HTTP 4xx /
5xx, transport error, oversize response). CATCHABLE (snag #6) — a
4xx no longer hard-kills a batch that imports `_request`. `main()`
turns an uncaught one into a clean `sys.exit` (preserving the CLI's
message-and-non-zero-exit UX); `call` catches it to emit the
structured server error as JSON. Carries the HTTP `code` (None for
transport failures) and the parsed error `payload` (None if the body
wasn't a JSON object)."""
def __init__(self, message, *, code=None, payload=None):
super().__init__(message)
self.code = code
self.payload = payload
def _request(endpoint: str, method: str = "GET",
payload: "dict | None" = None,
timeout: float = 30.0,
idempotency_key: "str | None" = None) -> dict:
host, port, token = _read_session()
url = f"http://{host}:{port}/{endpoint}"
data = (json.dumps(payload or {}).encode("utf-8")
if method == "POST" else None)
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Authorization", f"Bearer {token}")
if data is not None:
req.add_header("Content-Type", "application/json")
if idempotency_key:
req.add_header("X-Idempotency-Key", idempotency_key)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
# Cap the read size symmetric with the server's cap; a
# compromised / buggy server returning an unbounded stream
# would otherwise OOM the CLI. Read 1 byte over the cap so
# we can detect oversize and abort with a useful message
# rather than silently truncating.
raw = resp.read(_CLI_RESPONSE_MAX_BYTES + 1)
if len(raw) > _CLI_RESPONSE_MAX_BYTES:
raise _AgentCliError(
f"Error: response from SpliceCraft exceeds "
f"{_CLI_RESPONSE_MAX_BYTES:,}-byte cap "
f"(endpoint={endpoint!r}). Refusing to read."
)
body = raw.decode("utf-8")
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8") if exc.fp else ""
err_payload: "dict | None" = None
try:
parsed = json.loads(body) if body else {}
if isinstance(parsed, dict):
err_payload = parsed
msg = parsed.get("error", body or exc.reason)
else:
msg = body or exc.reason
except json.JSONDecodeError:
msg = body or exc.reason
# Snag #6: raise a CATCHABLE error instead of a hard `sys.exit`,
# so a 4xx no longer kills an importing batch mid-run. `main()`
# turns an uncaught one into a clean exit; `call` catches it to
# surface the structured error as JSON.
raise _AgentCliError(f"Error: {msg} (HTTP {exc.code})",
code=exc.code, payload=err_payload)
except urllib.error.URLError as exc:
raise _AgentCliError(
f"Could not reach SpliceCraft at {host}:{port} ({exc.reason}).\n"
f" Is the GUI still running with --agent-api?"
)
try:
return json.loads(body) if body else {}
except json.JSONDecodeError:
return {"raw": body}
# ── Subcommand handlers ────────────────────────────────────────────────────────
# Each subcommand maps 1:1 to a `/<endpoint>` on the server. Keep these
# thin — the server is the source of truth for validation / error messages.
def _emit_json(obj) -> None:
print(json.dumps(obj, indent=2, default=str))
_IDEMPOTENCY_KEY_MAX = 128
_IDEMPOTENCY_KEY_RE = re.compile(r"^[A-Za-z0-9_-]+$")
def _validate_idempotency_key(key: str) -> str:
"""Reject a key the server would reject, before the request goes out.
The server constrains the key to ``[A-Za-z0-9_-]`` (so nothing that
could reach a log as an escape or a path separator) and caps it at 128
chars. Checking here turns a wasted round-trip into an argparse-time
error, and — more usefully — a caller who mistakenly passes something
like a whole JSON body or a path finds out immediately instead of
believing their batch was retry-safe when the header never applied.
"""
if len(key) > _IDEMPOTENCY_KEY_MAX:
raise argparse.ArgumentTypeError(
f"idempotency key is {len(key)} chars; the server caps it at "
f"{_IDEMPOTENCY_KEY_MAX}"
)
if not _IDEMPOTENCY_KEY_RE.match(key):
raise argparse.ArgumentTypeError(
"idempotency key may contain only letters, digits, hyphen and "
"underscore"
)
return key
def _call_pick_method(endpoint: str, method: str, payload: "dict | None",
*, allow_upgrade: bool,
idempotency_key: "str | None" = None) -> dict:
"""Issue the request, retrying once with the method the server names.
`call` has to guess a method: with no `--json` there is no body to
infer POST from, so a body-less WRITE endpoint (`add-current-to-library`,
`save`) guessed GET and came back 405 "requires POST" — leaving the
caller to re-run by hand. The server's 405 body carries the authoritative
`allow`, so honour it instead of guessing twice. Reads pay nothing: the
retry only fires on the 405 that would otherwise have been a hard failure.
An explicit `--method` is never overridden — the caller asked for that
method and gets the real 405 back.
"""
try:
return _request(endpoint, method=method, payload=payload,
idempotency_key=idempotency_key)
except _AgentCliError as exc:
allow = (exc.payload or {}).get("allow") if isinstance(
exc.payload, dict) else None
if (allow_upgrade and exc.code == 405 and method == "GET"
and str(allow or "").upper() == "POST"):
# Carry the key onto the retry: the whole point of the header is
# that the RETRY is the risky call, so dropping it here would
# leave exactly the request that needs protecting unprotected.
return _request(endpoint, method="POST", payload=payload or {},
idempotency_key=idempotency_key)
raise
def cmd_call(args) -> None:
"""Generic passthrough: call ANY agent endpoint by name (snag #1).
Reuses the same auth / host / port plumbing as the blessed
subcommands, so an agent integrating SpliceCraft no longer has to
hand-roll HTTP against the private `_request`. `tools` lists every
endpoint and its `doc_full` body schema."""
payload = None
if args.json:
try:
payload = json.loads(args.json)
except json.JSONDecodeError as exc:
sys.exit(f"Error: --json is not valid JSON: {exc}")
if not isinstance(payload, dict):
sys.exit("Error: --json must be a JSON object")
# Default to POST when a body is supplied (writes require POST), else
# GET — matching the server's read/write method split.
explicit_method = args.method is not None
method = (args.method or ("POST" if payload is not None else "GET")).upper()
try:
result = _call_pick_method(args.endpoint, method, payload,
allow_upgrade=not explicit_method,
idempotency_key=args.idempotency_key)
except _AgentCliError as exc:
# Surface the structured server error as JSON (then exit
# non-zero) rather than a bare message, so a calling script can
# parse the failure instead of scraping stderr.
out: dict = (dict(exc.payload) if isinstance(exc.payload, dict)
else {"error": str(exc)})
out["http_code"] = exc.code
_emit_json(out)
raise SystemExit(1)
_emit_json(result)
def cmd_status(args) -> None:
_emit_json(_request("status"))
def cmd_tools(args) -> None:
flt = getattr(args, "filter", None)
# `tools` is a GET, and the server reads a GET's payload from the QUERY
# STRING (`_parse_query_params`), not from a body — so the filter rides
# the URL rather than `_request`'s `payload`, which only ships on POST.
ep = ("tools?filter=" + urllib.parse.quote(flt, safe="")) if flt \
else "tools"
result = _request(ep)
if args.json:
_emit_json(result)
return
eps = result.get("endpoints", [])
# 34, not 14: `analyse-read-heterogeneity` is 26 characters and the old
# width pushed every long name's summary out of alignment, which is
# exactly the column a reader scans.
for ep in eps:
flag = "WRITE" if ep.get("write") else "READ "
print(f" {flag} {ep['name']:34} {ep.get('doc', '')}")
total = result.get("total_matches")
if flt and isinstance(total, int):
# Say how many matched, not just how many printed — a capped page
# read as the whole answer is the mistake this count exists to stop.
shown = len(eps)
more = f" (showing the best {shown})" if total > shown else ""
print(f"\n {total} endpoint(s) match {flt!r}{more}")
def cmd_features(args) -> None:
result = _request("features")
feats = result.get("features", [])
if args.json:
_emit_json(feats)
return
if not feats:
print("(no features)")
return
for f in feats:
strand = ("+" if f["strand"] == 1
else "-" if f["strand"] == -1 else ".")
label = f.get("label") or ""
print(
f" [{f['idx']:3}] {f.get('type','?'):14} "
f"{f['start'] + 1:>7,}..{f['end']:<7,} {strand} {label}"
)
def cmd_fetch(args) -> None:
payload = {"accession": args.accession}
if args.force:
payload["force"] = True
_emit_json(_request("fetch", "POST", payload, timeout=60))
def cmd_load_entry(args) -> None:
payload = {"name": args.name}
if args.force:
payload["force"] = True
_emit_json(_request("load-entry", "POST", payload))
def cmd_load_file(args) -> None:
payload = {"path": args.path}
if args.force:
payload["force"] = True
# Server-side parse on a chromosome can take 10s+; bump timeout.
_emit_json(_request("load-file", "POST", payload, timeout=120))
def cmd_add_feature(args) -> None:
payload = {
"start": args.start,
"end": args.end,
"label": args.label,
"type": args.type,
"strand": args.strand,
}
if args.force:
payload["force"] = True
_emit_json(_request("add-feature", "POST", payload))
def cmd_save(args) -> None:
_emit_json(_request("save", "POST"))
# ── Tier 1: sequence + feature CRUD ────────────────────────────────────────────
def cmd_get_sequence(args) -> None:
payload = {"start": args.start, "end": args.end, "bottom": args.bottom}
result = _request("get-sequence", "POST", payload)
if args.json:
_emit_json(result)
return
seq = result.get("seq", "")
print(seq)
def cmd_replace_sequence(args) -> None:
payload = {"start": args.start, "end": args.end, "bases": args.bases}
if args.force:
payload["force"] = True
_emit_json(_request("replace-sequence", "POST", payload))
def cmd_delete_feature(args) -> None:
payload = {"idx": args.idx}
if args.force:
payload["force"] = True
_emit_json(_request("delete-feature", "POST", payload))
def cmd_update_feature(args) -> None:
payload = {"idx": args.idx}
if args.label is not None: payload["label"] = args.label
if args.type is not None: payload["type"] = args.type
if args.strand is not None: payload["strand"] = args.strand
if args.force:
payload["force"] = True
_emit_json(_request("update-feature", "POST", payload))
def cmd_get_feature(args) -> None:
_emit_json(_request("get-feature", "POST", {"idx": args.idx}))
def cmd_export_genbank(args) -> None:
payload = {"path": args.path}
if args.force:
payload["force"] = True
_emit_json(_request("export-genbank", "POST", payload))
def cmd_export_fasta(args) -> None:
payload = {"path": args.path}
if args.force:
payload["force"] = True
_emit_json(_request("export-fasta", "POST", payload))
# ── Tier 2: library + collections ──────────────────────────────────────────────
def cmd_list_library(args) -> None:
result = _request("list-library")
if args.json:
_emit_json(result)
return
entries = result.get("library", [])
if not entries:
print("(empty library)")
return
for e in entries:
print(
f" {e.get('name','?'):24} {e.get('length',0):>7,} bp "
f"{e.get('n_features',0):>3} feat {e.get('topology','')}"
)
def cmd_list_collections(args) -> None:
result = _request("list-collections")
if args.json:
_emit_json(result)
return
active = result.get("active") or "(none)"
print(f" active: {active}")
for c in result.get("collections", []):
marker = "*" if c.get("name") == active else " "
print(f" {marker} {c.get('name','?'):24} "
f"{c.get('n_plasmids',0):>3} plasmids")
def cmd_delete_from_library(args) -> None:
payload = {"name": args.name}
if args.force:
payload["force"] = True
_emit_json(_request("delete-from-library", "POST", payload))
# ── Tier 3: cloning / design helpers ───────────────────────────────────────────
def cmd_list_restriction_sites(args) -> None:
payload: dict = {}
if args.enzymes:
payload["enzymes"] = args.enzymes
if args.min_length is not None:
payload["min_length"] = args.min_length
if args.unique_only:
payload["unique_only"] = True
result = _request("list-restriction-sites", "POST", payload)
if args.json:
_emit_json(result)
return
sites = result.get("sites", [])
if not sites:
print("(no sites)")
return
for s in sites:
strand = "+" if s.get("strand", 1) == 1 else "-"
print(f" {s.get('enzyme','?'):14} "
f"{(s.get('start') or 0)+1:>7,}..{s.get('end','?'):<7} "
f"{strand} cut@{s.get('cut_bp', '?')}")
def cmd_list_codon_tables(args) -> None:
result = _request("list-codon-tables")
if args.json:
_emit_json(result)
return
for t in result.get("tables", []):
print(f" {t.get('taxid',''):>6} {t.get('source','?'):8} "
f"{t.get('name','?')}")
def cmd_optimize_protein(args) -> None:
payload: "dict" = {"protein": args.protein}
# Straight scalar pass-throughs; only set when given so the server's
# defaults keep applying (and the response stays minimal).
for src, key in (("table", "table"), ("mode", "mode"),
("stops", "stops"), ("transl_table", "transl_table"),
("min_gc", "min_gc"), ("max_gc", "max_gc"),
("gc_window", "gc_window"), ("max_repeat", "max_repeat")):
val = getattr(args, src, None)
if val is not None:
payload[key] = val
# List params: argparse gives a list (nargs) or None.
if args.hazard_hosts:
payload["hazard_hosts"] = args.hazard_hosts
if args.forbidden_motifs:
payload["forbidden_motifs"] = args.forbidden_motifs
if args.avoid_repeats_with:
payload["avoid_repeats_with"] = args.avoid_repeats_with
result = _request("optimize-protein", "POST", payload)
if args.json:
_emit_json(result)
return
# DNA to stdout (pipeable); the audit trail + warnings to stderr so a
# `... | tee gene.txt` still captures just the sequence.
print(result.get("dna", ""))
for w in result.get("warnings") or []:
print(f"warning: {w}", file=sys.stderr)
fixes = result.get("fixes") or []
if fixes:
print(f"# {len(fixes)} synonymous fix(es) applied; "
f"CAI {result.get('cai')}, GC {result.get('gc')}%, "
f"GC3 {result.get('gc3')}%", file=sys.stderr)
# ── Argparse wiring ────────────────────────────────────────────────────────────
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="splicecraft-cli",
description=(__doc__ or "").splitlines()[0] if __doc__ else "",
)
sub = parser.add_subparsers(dest="cmd", required=True,
metavar="COMMAND")
p_status = sub.add_parser("status",
help="Show what's loaded + dirty flag.")
p_status.set_defaults(fn=cmd_status)
p_tools = sub.add_parser("tools",
help="List all endpoints the session exposes.")
p_tools.add_argument("--json", action="store_true",
help="Emit JSON instead of a table")
p_tools.add_argument("--filter", metavar="TERMS", default=None,
help="Rank by name + docs and show the best "
"matches (e.g. --filter primer) — search the "
"VERB you want instead of guessing a name.")
p_tools.set_defaults(fn=cmd_tools)
p_feat = sub.add_parser("features",
help="List features on the loaded record.")
p_feat.add_argument("--json", action="store_true",
help="Emit JSON instead of a table")
p_feat.set_defaults(fn=cmd_features)
p_fetch = sub.add_parser("fetch",
help="Fetch a GenBank record from NCBI.")
p_fetch.add_argument("accession",
help="GenBank accession (e.g. L09137).")
p_fetch.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_fetch.set_defaults(fn=cmd_fetch)
p_load = sub.add_parser("load-entry",
help="Load a plasmid library entry by name.")
p_load.add_argument("name", help="Library entry name or id.")
p_load.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_load.set_defaults(fn=cmd_load_entry)
p_loadf = sub.add_parser(
"load-file",
help="Load a .gb / .gbk / .dna file from a server-side path "
"(bypasses the JSON-body size cap; works for chromosome-"
"scale files).",
)
p_loadf.add_argument("path", help="Server-side file path.")
p_loadf.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_loadf.set_defaults(fn=cmd_load_file)
p_add = sub.add_parser(
"add-feature",
help="Add a feature to the loaded record.",
)
p_add.add_argument("start", type=int, help="0-based start bp.")
p_add.add_argument("end", type=int,
help="0-based end bp (exclusive). For wrap "
"features, pass end < start.")
p_add.add_argument("--label", default="",
help="Feature label (qualifier).")
p_add.add_argument("--type", default="misc_feature",
help="GenBank feature type (CDS, promoter, …).")
p_add.add_argument("--strand", type=int, default=1,
choices=[-1, 0, 1, 2],
help="1=forward (default), -1=reverse, 0=both.")
p_add.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_add.set_defaults(fn=cmd_add_feature)
p_save = sub.add_parser("save",
help="Save the loaded record (file + library).")
p_save.set_defaults(fn=cmd_save)
# ── Tier 1 ────────────────────────────────────────────────────────────
p_getseq = sub.add_parser(
"get-sequence",
help="Extract DNA from a bp range (forward or --bottom strand).",
)
p_getseq.add_argument("start", type=int, help="0-based start bp.")
p_getseq.add_argument("end", type=int, help="0-based end bp (exclusive).")
p_getseq.add_argument("--bottom", action="store_true",
help="Return reverse-complement (bottom strand 5'→3').")
p_getseq.add_argument("--json", action="store_true",
help="Emit JSON instead of plain seq.")
p_getseq.set_defaults(fn=cmd_get_sequence)
p_repseq = sub.add_parser(
"replace-sequence",
help="Replace bp range with new bases (mutagenesis).",
)
p_repseq.add_argument("start", type=int)
p_repseq.add_argument("end", type=int)
p_repseq.add_argument("bases",
help="New bases (IUPAC ACGTNRYWSMKBDHV; will be uppercased).")
p_repseq.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_repseq.set_defaults(fn=cmd_replace_sequence)
p_delfeat = sub.add_parser(
"delete-feature", help="Delete the feature at the given index.",
)
p_delfeat.add_argument("idx", type=int)
p_delfeat.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_delfeat.set_defaults(fn=cmd_delete_feature)
p_updfeat = sub.add_parser(
"update-feature",
help="Update label / type / strand of the feature at idx.",
)
p_updfeat.add_argument("idx", type=int)
p_updfeat.add_argument("--label", default=None)
p_updfeat.add_argument("--type", default=None,
help="GenBank feature type (CDS, promoter, …).")
p_updfeat.add_argument("--strand", type=int, default=None,
choices=[-1, 0, 1, 2])
p_updfeat.add_argument("--force", action="store_true")
p_updfeat.set_defaults(fn=cmd_update_feature)
p_getfeat = sub.add_parser(
"get-feature", help="Detail of one feature (idx, qualifiers, …).",
)
p_getfeat.add_argument("idx", type=int)
p_getfeat.set_defaults(fn=cmd_get_feature)
p_expgb = sub.add_parser(
"export-genbank", help="Write the loaded record to PATH as GenBank.",
)
p_expgb.add_argument("path", help="Output path (.gb / .gbk).")
p_expgb.add_argument("--force", action="store_true",
help="Override unsaved-changes guard.")
p_expgb.set_defaults(fn=cmd_export_genbank)
p_expfa = sub.add_parser(
"export-fasta", help="Write the loaded record's seq to PATH as FASTA.",
)
p_expfa.add_argument("path", help="Output path (.fa / .fasta / .fna).")
p_expfa.add_argument("--force", action="store_true")
p_expfa.set_defaults(fn=cmd_export_fasta)
# ── Tier 2 ────────────────────────────────────────────────────────────
p_lib = sub.add_parser(
"list-library", help="List saved plasmid library entries.",
)
p_lib.add_argument("--json", action="store_true")
p_lib.set_defaults(fn=cmd_list_library)
p_col = sub.add_parser(
"list-collections", help="List collections + active one.",
)
p_col.add_argument("--json", action="store_true")
p_col.set_defaults(fn=cmd_list_collections)
p_dellib = sub.add_parser(
"delete-from-library", help="Remove a plasmid library entry by name.",
)
p_dellib.add_argument("name")
p_dellib.add_argument("--force", action="store_true")
p_dellib.set_defaults(fn=cmd_delete_from_library)
# ── Tier 3 ────────────────────────────────────────────────────────────
p_re = sub.add_parser(
"list-restriction-sites",
help="Scan record for restriction sites (default: NEB catalog, "
"min recognition 4 bp).",
)
p_re.add_argument("--enzymes", nargs="*",
help="Limit to these enzyme names.")
p_re.add_argument("--min-length", type=int, default=None,
dest="min_length",
help="Minimum recognition site length (default 4).")
p_re.add_argument("--unique-only", action="store_true",
dest="unique_only",
help="Only return enzymes that cut once.")
p_re.add_argument("--json", action="store_true")
p_re.set_defaults(fn=cmd_list_restriction_sites)
p_codons = sub.add_parser(
"list-codon-tables", help="List available codon usage tables.",
)
p_codons.add_argument("--json", action="store_true")
p_codons.set_defaults(fn=cmd_list_codon_tables)
p_harm = sub.add_parser(
"optimize-protein",
help="Codon-optimize an AA sequence to DNA "
"(default table: E. coli K12, taxid 83333).",
)
p_harm.add_argument("protein", help="1-letter AA sequence.")
p_harm.add_argument("--table", default=None,
help="Codon-table taxid (see list-codon-tables; add "
"hosts with `add-codon-table`).")
p_harm.add_argument("--mode", choices=["frequency", "max_cai"],
default=None,
help="frequency (default; matches the host's codon "
"distribution) or max_cai (every position its "
"best synonym — watch the GC3 warning on AT-rich "
"hosts).")
p_harm.add_argument("--stops", type=int, default=None,
help="Stop codons to append (0-3); a trailing '*' run "
"in the protein overrides this.")
p_harm.add_argument("--transl-table", dest="transl_table", type=int,
default=None,
help="NCBI genetic-code id (default 1; e.g. 4, 6, 11).")
p_harm.add_argument("--hazard-hosts", dest="hazard_hosts", nargs="+",
metavar="HOST", default=None,
help="Scrub built-in expression hazards for these "
"hosts: plant, mammalian, bacterial.")
p_harm.add_argument("--forbidden-motifs", dest="forbidden_motifs",
nargs="+", metavar="MOTIF", default=None,
help="Also scrub these IUPAC motifs (both strands).")
p_harm.add_argument("--min-gc", dest="min_gc", type=float, default=None,
metavar="PCT",
help="Raise every gc-window below this percent.")
p_harm.add_argument("--max-gc", dest="max_gc", type=float, default=None,
metavar="PCT",
help="Lower every gc-window above this percent.")
p_harm.add_argument("--gc-window", dest="gc_window", type=int,
default=None, metavar="BASES",
help="Window size for --min-gc/--max-gc (default 50).")
p_harm.add_argument("--avoid-repeats-with", dest="avoid_repeats_with",
nargs="+", metavar="DNA", default=None,
help="Break perfect repeats shared with these already-"
"built DNA sequences.")
p_harm.add_argument("--max-repeat", dest="max_repeat", type=int,
default=None, metavar="BASES",
help="Shared-run length to break for "
"--avoid-repeats-with (default 25).")
p_harm.add_argument("--json", action="store_true")
p_harm.set_defaults(fn=cmd_optimize_protein)
p_call = sub.add_parser(
"call",
help="Call ANY agent endpoint by name (generic passthrough).",
)
p_call.add_argument("endpoint",
help="Endpoint name (see `tools`), e.g. "
"rename-plasmid.")
p_call.add_argument("--method", choices=["GET", "POST", "get", "post"],
default=None,
help="HTTP method. Default: POST when --json is "
"given, else GET auto-upgraded to POST if the "
"endpoint is a mutation. Setting this "
"explicitly disables the auto-upgrade.")
p_call.add_argument("--json", default=None, metavar="JSON",
help="Request body as a JSON object, e.g. "
"'{\"old\": \"a\", \"new\": \"b\"}'.")
p_call.add_argument("--idempotency-key", default=None, metavar="KEY",
type=_validate_idempotency_key,
help="Make this write safe to retry: send "
"X-Idempotency-Key. If the same key + same "
"body arrives again within 60s the first "
"result is replayed instead of the write "
"running twice. Use a FRESH key per distinct "
"request; the same key with a different body "
"is refused (409) and does nothing. "
"[A-Za-z0-9_-], max 128 chars.")
p_call.set_defaults(fn=cmd_call)
return parser
def main(argv=None) -> None:
parser = _build_parser()
args = parser.parse_args(argv)
try:
args.fn(args)
except _AgentCliError as exc:
# A call failure a subcommand didn't handle itself becomes a
# clean message + non-zero exit (back-compat with the old
# `sys.exit`), not a traceback. `call` handles its own.
sys.exit(str(exc))
if __name__ == "__main__":
main()