Source code for tooluniverse.giab_tool

# giab_tool.py
"""
Genome in a Bottle (GIAB) benchmark file browser for ToolUniverse.

GIAB (NIST) publishes the reference-standard high-confidence variant call
sets (VCF) and callable regions (BED) used to validate variant-calling
pipelines, for samples HG001-HG007 across multiple reference builds and
release versions (NISTv3.2 through v5.0q, plus CMRG, structural-variant,
and tandem-repeat benchmark sets). There is no REST/JSON API for this data
-- only a plain Apache directory listing at
https://ftp-trace.ncbi.nlm.nih.gov/ReferenceSamples/giab/release/ -- so
this tool parses that listing to let an agent navigate the tree (trio ->
sample -> version -> reference build -> files) and get direct download
URLs for the benchmark VCF/BED files it needs.

No authentication required.
"""

import re
from typing import Any, Dict, List
from urllib.parse import urljoin

import requests

from .base_tool import BaseTool
from .tool_registry import register_tool

GIAB_HOST = "https://ftp-trace.ncbi.nlm.nih.gov"
GIAB_RELEASE_ROOT = "/ReferenceSamples/giab/release/"

_ENTRY_PATTERN = re.compile(
    r'<a href="([^"]+)">([^<]+)</a>\s*(\d{4}-\d{2}-\d{2} \d{2}:\d{2})?\s*(\S*)'
)


def _parse_listing(html: str, listing_url: str) -> List[Dict[str, Any]]:
    """Parse an Apache-style directory listing into structured entries."""
    entries = []
    for href, name, modified, size in _ENTRY_PATTERN.findall(html):
        name = name.strip()
        if name == "Parent Directory" or href.startswith("http"):
            continue
        is_dir = href.endswith("/")
        entries.append(
            {
                "name": name.rstrip("/") if is_dir else name,
                "type": "directory" if is_dir else "file",
                "size": None if size == "-" else size,
                "last_modified": modified or None,
                "url": urljoin(listing_url, href),
            }
        )
    return entries


def _normalize_path(path: str):
    """Validate and normalize a path relative to the GIAB release root.

    Returns (normalized_path, error_message). Exactly one is None.
    """
    path = (path or "").strip().strip("/")
    if ".." in path.split("/"):
        return None, "path must not contain '..' path traversal segments."
    return path, None


[docs] @register_tool("GIABTool") class GIABTool(BaseTool): """Browse the GIAB benchmark file release tree (NIST FTP mirror). No authentication required. """
[docs] def __init__(self, tool_config: Dict[str, Any]): super().__init__(tool_config) self.timeout = tool_config.get("timeout", 30)
[docs] def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]: path, err = _normalize_path(arguments.get("path", "")) if err is not None: return {"status": "error", "error": err} listing_url = urljoin(GIAB_HOST + GIAB_RELEASE_ROOT, path + "/" if path else "") try: resp = requests.get(listing_url, timeout=self.timeout) except requests.exceptions.Timeout: return { "status": "error", "error": f"GIAB request timed out after {self.timeout}s", } except requests.exceptions.RequestException as e: return {"status": "error", "error": f"GIAB request failed: {e}"} if resp.status_code == 404: return { "status": "error", "error": f"No such GIAB path: '{path or '/'}'. Start with an " "empty path to list the top-level trios/samples.", } resp.raise_for_status() entries = _parse_listing(resp.text, listing_url) if not entries: return { "status": "error", "error": f"'{path or '/'}' is empty or not a directory listing.", } return { "status": "success", "data": entries, "metadata": { "path": path or "/", "listing_url": listing_url, "returned": len(entries), "source": "NIST Genome in a Bottle (ftp-trace.ncbi.nlm.nih.gov)", }, }