Skills › Scientific Agent Skills › protocolsio-integration › protocols_read.py
protocols_read.py
502 lines · skills/protocolsio-integration/scripts/protocols_read.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded, read-only protocols.io REST client with an explicit execute gate."""
4from __future__ import annotations
6import argparse
7import os
8import re
9from typing import Any, Mapping, Sequence
11try:
12 from ._common import (
13 ACCESS_TOKEN_ENV,
14 DEFAULT_ORIGIN,
15 MAX_JSON_RESPONSE_BYTES,
16 MAX_PDF_RESPONSE_BYTES,
17 ApiError,
18 SafetyError,
19 build_url,
20 emit_error,
21 emit_json,
22 encode_protocol_identifier,
23 parse_json_bytes,
24 request_bytes,
25 require_api_success,
26 safe_output_path,
27 sanitize_untrusted,
28 validate_origin,
29 validate_protocol_identifier,
30 write_private_bytes,
31 )
32 from .pagination_helper import inspect_pagination
33except ImportError:
34 from _common import ( # type: ignore
35 ACCESS_TOKEN_ENV,
36 DEFAULT_ORIGIN,
37 MAX_JSON_RESPONSE_BYTES,
38 MAX_PDF_RESPONSE_BYTES,
39 ApiError,
40 SafetyError,
41 build_url,
42 emit_error,
43 emit_json,
44 encode_protocol_identifier,
45 parse_json_bytes,
46 request_bytes,
47 require_api_success,
48 safe_output_path,
49 sanitize_untrusted,
50 validate_origin,
51 validate_protocol_identifier,
52 write_private_bytes,
53 )
54 from pagination_helper import inspect_pagination # type: ignore
57MAX_LIST_PAGES = 20
58MAX_LIST_ITEMS = 2_000
59_EXPORT_GUID_RE = re.compile(r"^[A-Fa-f0-9]{32}$")
60_ORG_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
63def build_parser() -> argparse.ArgumentParser:
64 parser = argparse.ArgumentParser(
65 description=(
66 "Plan or execute bounded read-only protocols.io requests. Network "
67 "access occurs only with --execute."
68 )
69 )
70 parser.add_argument(
71 "--execute",
72 action="store_true",
73 help="Perform the planned read-only network request.",
74 )
75 parser.add_argument(
76 "--origin",
77 default=DEFAULT_ORIGIN,
78 help="Core API origin (default: %(default)s).",
79 )
80 parser.add_argument(
81 "--timeout",
82 type=float,
83 default=15.0,
84 help="Per-request timeout in seconds, 1..60 (default: %(default)s).",
85 )
86 parser.add_argument(
87 "--retries",
88 type=int,
89 default=1,
90 help="Additional retries for 429/5xx, 0..2 (default: %(default)s).",
91 )
92 parser.add_argument(
93 "--max-response-bytes",
94 type=int,
95 default=MAX_JSON_RESPONSE_BYTES,
96 help=(
97 "Per-page JSON response cap, up to "
98 f"{MAX_JSON_RESPONSE_BYTES} (default: %(default)s)."
99 ),
100 )
101 subparsers = parser.add_subparsers(dest="command", required=True)
103 list_parser = subparsers.add_parser(
104 "list",
105 help="Search public protocols through the documented v3 list endpoint.",
106 )
107 list_parser.add_argument("--query", required=True, help="Non-empty search key.")
108 list_parser.add_argument(
109 "--page-size",
110 type=int,
111 default=10,
112 help="Documented page size 1..100 (default: %(default)s).",
113 )
114 list_parser.add_argument(
115 "--max-pages",
116 type=int,
117 default=3,
118 help=f"Local page cap 1..{MAX_LIST_PAGES} (default: %(default)s).",
119 )
120 list_parser.add_argument(
121 "--max-items",
122 type=int,
123 default=100,
124 help=f"Local item cap 1..{MAX_LIST_ITEMS} (default: %(default)s).",
125 )
127 get_parser = subparsers.add_parser(
128 "get",
129 help="Get one protocol through the documented v4 endpoint.",
130 )
131 _add_protocol_arguments(get_parser)
133 steps_parser = subparsers.add_parser(
134 "steps",
135 help="Get protocol steps through the documented v4 endpoint.",
136 )
137 _add_protocol_arguments(steps_parser)
139 pdf_parser = subparsers.add_parser(
140 "export-pdf",
141 help="Download the documented read-only PDF representation.",
142 )
143 pdf_parser.add_argument(
144 "--id", required=True, help="Protocol ID or URI; resolve a DOI with get first."
145 )
146 pdf_parser.add_argument(
147 "--output",
148 required=True,
149 help="New .pdf path below the current working directory.",
150 )
151 pdf_parser.add_argument(
152 "--anonymous",
153 action="store_true",
154 help="Intentionally use the signed-out PDF path without a bearer token.",
155 )
156 pdf_parser.add_argument(
157 "--compact-view",
158 action="store_true",
159 help="Request compact_view=1.",
160 )
161 pdf_parser.add_argument(
162 "--only",
163 choices=("materials", "commands", "steps"),
164 help="Request one documented PDF subset.",
165 )
166 pdf_parser.add_argument(
167 "--max-pdf-bytes",
168 type=int,
169 default=MAX_PDF_RESPONSE_BYTES,
170 help=f"Local PDF cap up to {MAX_PDF_RESPONSE_BYTES} bytes.",
171 )
173 export_parser = subparsers.add_parser(
174 "export-status",
175 help="Read an existing v4 organization content-export status.",
176 )
177 export_parser.add_argument(
178 "--tenant-origin",
179 required=True,
180 help="Organization/VPC origin, for example https://tenant.protocols.io.",
181 )
182 export_parser.add_argument("--organization", required=True)
183 export_parser.add_argument("--export-guid", required=True)
184 return parser
187def _add_protocol_arguments(parser: argparse.ArgumentParser) -> None:
188 parser.add_argument("--id", required=True, help="Protocol ID, URI, or DOI.")
189 parser.add_argument(
190 "--content-format",
191 choices=("json", "html", "markdown"),
192 default="json",
193 help="Documented content representation (default: %(default)s).",
194 )
195 parser.add_argument(
196 "--last-version",
197 action="store_true",
198 help="Request last_version=1. Prefer an explicit /vN identifier for archives.",
199 )
202def _require_token(environ: Mapping[str, str]) -> str:
203 token = environ.get(ACCESS_TOKEN_ENV)
204 if not token:
205 raise SafetyError(
206 f"{ACCESS_TOKEN_ENV} is required for this documented REST endpoint"
207 )
208 if token != token.strip() or "\r" in token or "\n" in token:
209 raise SafetyError(f"{ACCESS_TOKEN_ENV} contains invalid whitespace")
210 return token
213def _get_items(payload: Mapping[str, Any]) -> list[Any]:
214 items = payload.get("items")
215 if items is None and isinstance(payload.get("payload"), Mapping):
216 items = payload["payload"].get("items")
217 if not isinstance(items, list):
218 raise ApiError("list response does not contain an items array")
219 return items
222def _has_pagination(payload: Mapping[str, Any]) -> bool:
223 if isinstance(payload.get("pagination"), Mapping):
224 return True
225 nested = payload.get("payload")
226 return isinstance(nested, Mapping) and isinstance(
227 nested.get("pagination"),
228 Mapping,
229 )
232def _request_json(
233 url: str,
234 *,
235 token: str,
236 args: argparse.Namespace,
237 opener: Any | None,
238 sleep: Any,
239) -> tuple[Mapping[str, Any], int]:
240 result = request_bytes(
241 url,
242 token=token,
243 accept="application/json",
244 timeout=args.timeout,
245 retries=args.retries,
246 max_bytes=args.max_response_bytes,
247 opener=opener,
248 sleep=sleep,
249 )
250 payload = require_api_success(
251 parse_json_bytes(result.body, source="protocols.io response")
252 )
253 return payload, result.attempts
256def _plan(args: argparse.Namespace) -> dict[str, Any]:
257 origin = validate_origin(args.origin)
258 params: dict[str, Any] = {}
259 authentication = "bearer"
260 if args.command == "list":
261 query = args.query.strip()
262 if not query or len(query) > 1_000:
263 raise SafetyError("query must contain 1 to 1000 non-whitespace characters")
264 if not 1 <= args.page_size <= 100:
265 raise SafetyError("page-size must be between 1 and 100")
266 if not 1 <= args.max_pages <= MAX_LIST_PAGES:
267 raise SafetyError(f"max-pages must be between 1 and {MAX_LIST_PAGES}")
268 if not 1 <= args.max_items <= MAX_LIST_ITEMS:
269 raise SafetyError(f"max-items must be between 1 and {MAX_LIST_ITEMS}")
270 params = {
271 "filter": "public",
272 "key": query,
273 "page_size": args.page_size,
274 "page_id": 1,
275 }
276 url = build_url(origin, "/api/v3/protocols", params)
277 elif args.command in {"get", "steps"}:
278 identifier = validate_protocol_identifier(args.id)
279 encoded = encode_protocol_identifier(identifier)
280 suffix = "/steps" if args.command == "steps" else ""
281 params = {"content_format": args.content_format}
282 if args.last_version:
283 params["last_version"] = 1
284 url = build_url(origin, f"/api/v4/protocols/{encoded}{suffix}", params)
285 elif args.command == "export-pdf":
286 identifier = encode_protocol_identifier(args.id)
287 if args.id.startswith(("10.17504/", "protocols.io.")):
288 raise SafetyError(
289 "PDF identifiers are documented as IDs or URIs; resolve the DOI with get first"
290 )
291 params = {}
292 if args.compact_view:
293 params["compact_view"] = 1
294 if args.only:
295 params[f"only_{args.only}"] = 1
296 if not 1 <= args.max_pdf_bytes <= MAX_PDF_RESPONSE_BYTES:
297 raise SafetyError(
298 f"max-pdf-bytes must be between 1 and {MAX_PDF_RESPONSE_BYTES}"
299 )
300 safe_output_path(args.output, suffix=".pdf")
301 url = build_url(origin, f"/view/{identifier}.pdf", params)
302 authentication = "anonymous" if args.anonymous else "bearer"
303 elif args.command == "export-status":
304 tenant = validate_origin(args.tenant_origin, allow_tenant=True)
305 if tenant in {"https://protocols.io", "https://www.protocols.io"}:
306 raise SafetyError("export-status requires an explicit tenant subdomain")
307 if _ORG_RE.fullmatch(args.organization) is None:
308 raise SafetyError("organization must be a bounded URI identifier")
309 if _EXPORT_GUID_RE.fullmatch(args.export_guid) is None:
310 raise SafetyError("export-guid must be a 32-character hex GUID")
311 url = build_url(
312 tenant,
313 "/api/v4/organizations/"
314 f"{args.organization}/content/exports/{args.export_guid.upper()}",
315 )
316 else:
317 raise SafetyError("unsupported read command")
319 return {
320 "ok": True,
321 "plan_kind": "read_only",
322 "command": args.command,
323 "method": "GET",
324 "url": url,
325 "authentication": authentication,
326 "credential_source": (
327 None if authentication == "anonymous" else ACCESS_TOKEN_ENV
328 ),
329 "network_requires_execute": True,
330 "network_accessed": False,
331 "redirects_followed": False,
332 "untrusted_remote_data": True,
333 }
336def _execute_list(
337 args: argparse.Namespace,
338 plan: Mapping[str, Any],
339 *,
340 token: str,
341 opener: Any | None,
342 sleep: Any,
343) -> dict[str, Any]:
344 current_url = str(plan["url"])
345 collected: list[Any] = []
346 pages = 0
347 attempts = 0
348 final_pagination: dict[str, Any] | None = None
349 while pages < args.max_pages and len(collected) < args.max_items:
350 payload, request_attempts = _request_json(
351 current_url,
352 token=token,
353 args=args,
354 opener=opener,
355 sleep=sleep,
356 )
357 attempts += request_attempts
358 pages += 1
359 remaining = args.max_items - len(collected)
360 collected.extend(_get_items(payload)[:remaining])
361 if not _has_pagination(payload):
362 break
363 final_pagination = inspect_pagination(
364 payload,
365 current_url=current_url,
366 pages_seen=pages,
367 items_seen=len(collected),
368 max_pages=args.max_pages,
369 max_items=args.max_items,
370 )
371 next_url = final_pagination["next_page_url"]
372 if not final_pagination["can_continue"] or not isinstance(next_url, str):
373 break
374 current_url = next_url
376 return {
377 "ok": True,
378 "command": "list",
379 "network_accessed": True,
380 "request_count": pages,
381 "attempt_count": attempts,
382 "returned_items": len(collected),
383 "local_caps": {
384 "max_pages": args.max_pages,
385 "max_items": args.max_items,
386 },
387 "pagination": sanitize_untrusted(final_pagination),
388 "untrusted_remote_data": True,
389 "embedded_instructions_followed": False,
390 "items": sanitize_untrusted(collected),
391 }
394def _execute_pdf(
395 args: argparse.Namespace,
396 plan: Mapping[str, Any],
397 *,
398 environ: Mapping[str, str],
399 opener: Any | None,
400 sleep: Any,
401) -> dict[str, Any]:
402 token = None if args.anonymous else _require_token(environ)
403 result = request_bytes(
404 str(plan["url"]),
405 token=token,
406 accept="application/pdf",
407 timeout=args.timeout,
408 retries=args.retries,
409 max_bytes=args.max_pdf_bytes,
410 opener=opener,
411 sleep=sleep,
412 )
413 content_type = result.headers.get("Content-Type", "").lower()
414 if "application/pdf" not in content_type or not result.body.startswith(b"%PDF-"):
415 raise ApiError("export response is not a validated PDF")
416 output = safe_output_path(args.output, suffix=".pdf")
417 write_private_bytes(output, result.body)
418 return {
419 "ok": True,
420 "command": "export-pdf",
421 "network_accessed": True,
422 "authentication": "anonymous" if args.anonymous else "bearer",
423 "attempt_count": result.attempts,
424 "bytes_written": len(result.body),
425 "source_url": str(plan["url"]),
426 "partial_export": args.only is not None,
427 "pdf_filter": args.only,
428 "output": str(output.relative_to(os.getcwd())),
429 "redirects_followed": False,
430 "remote_file_content_executed": False,
431 }
434def execute(
435 args: argparse.Namespace,
436 plan: Mapping[str, Any],
437 *,
438 environ: Mapping[str, str],
439 opener: Any | None = None,
440 sleep: Any = None,
441) -> dict[str, Any]:
442 if sleep is None:
443 import time
445 sleep = time.sleep
446 if args.command == "export-pdf":
447 return _execute_pdf(
448 args,
449 plan,
450 environ=environ,
451 opener=opener,
452 sleep=sleep,
453 )
455 token = _require_token(environ)
456 if args.command == "list":
457 return _execute_list(
458 args,
459 plan,
460 token=token,
461 opener=opener,
462 sleep=sleep,
463 )
464 payload, attempts = _request_json(
465 str(plan["url"]),
466 token=token,
467 args=args,
468 opener=opener,
469 sleep=sleep,
470 )
471 return {
472 "ok": True,
473 "command": args.command,
474 "network_accessed": True,
475 "attempt_count": attempts,
476 "untrusted_remote_data": True,
477 "embedded_instructions_followed": False,
478 "data": sanitize_untrusted(payload),
479 "version_preservation": (
480 "Retain the response's DOI, version_uri, authors, source URL, and "
481 "license metadata; do not silently replace /vN with /latest."
482 ),
483 }
486def main(argv: Sequence[str] | None = None) -> int:
487 args = build_parser().parse_args(argv)
488 try:
489 if not 1 <= args.max_response_bytes <= MAX_JSON_RESPONSE_BYTES:
490 raise SafetyError(
491 f"max-response-bytes must be between 1 and {MAX_JSON_RESPONSE_BYTES}"
492 )
493 plan = _plan(args)
494 report = execute(args, plan, environ=os.environ) if args.execute else plan
495 emit_json(report)
496 return 0
497 except (ApiError, SafetyError, ValueError) as exc:
498 return emit_error(exc)
501if __name__ == "__main__":
502 raise SystemExit(main())