"""
ClinVar REST API Tool
This tool provides access to the ClinVar database for clinical variant information,
disease associations, and clinical significance data.
"""
import re
import requests
import time
from typing import Dict, Any, Optional
from .base_tool import BaseTool
from .tool_registry import register_tool
[docs]
@register_tool("ClinVarSearchVariants")
class ClinVarSearchVariants(ClinVarRESTTool):
"""Search for variants in ClinVar by gene or condition."""
[docs]
def __init__(self, tool_config):
super().__init__(tool_config)
self.endpoint = "/esearch.fcgi"
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search variants by gene or condition."""
# Normalize aliases before dispatch
if not arguments.get("gene") and arguments.get("gene_symbol"):
arguments = dict(arguments, gene=arguments["gene_symbol"])
if not arguments.get("clinical_significance") and arguments.get("significance"):
arguments = dict(arguments, clinical_significance=arguments["significance"])
if not arguments.get("condition") and arguments.get("query"):
arguments = dict(arguments, condition=arguments["query"])
params = {
"db": "clinvar",
"retmode": "json",
"retmax": arguments.get("max_results") or arguments.get("limit", 20),
}
# Build search query
query_parts = []
gene_hyphen_variant = None
if "gene" in arguments:
# Fix-R10D-1: ClinVar's [gene] index only matches a bare HGNC
# symbol, but a natural clinical phrasing like "NF2 gene" (used
# verbatim in real research questions -- e.g. "look up the NF2
# gene") silently returns 0 results, since the literal trailing
# word "gene" becomes part of the queried string (confirmed
# live and via raw NCBI E-utils curl: "Nf2 gene[gene]" -> 0
# hits, "NF2[gene]" -> 2612 hits). Strip a trailing/leading
# "gene"/"protein" qualifier word before querying.
gene = re.sub(
r"^\s*(?:gene|protein)\s+|\s+(?:gene|protein)\s*$",
"",
arguments["gene"],
flags=re.IGNORECASE,
).strip()
# Fix-R8B-1 (revised, Fix-R49A-1): some genes are informally
# hyphenated even though their current HGNC symbol has none (e.g.
# "BRCA-2" for "BRCA2", "HER-2" for "HER2" -- confirmed live via
# raw NCBI E-utils: "BRCA-2[gene]" -> 0 hits, "BRCA2[gene]" ->
# 21879 hits). R8B-1 originally handled this by unconditionally
# stripping every hyphen, but that silently broke every gene
# whose CURRENT official HGNC symbol genuinely contains one --
# confirmed live for the entire HLA family (HLA-B[gene] -> 130
# hits, HLAB[gene] -> 0), the NKX homeobox family (NKX2-5[gene]
# -> real hits, NKX25[gene] -> 0), and MT- mitochondrial genes
# (MT-ND1[gene] -> 194 hits, MTND1[gene] -> 0) -- all clinically
# significant gene families (HLA pharmacogenomics/transplant,
# NKX cardiac/pancreatic development, MT- mitochondrial disease).
# Query with the hyphen preserved first; only fall back to the
# stripped form if that returns zero results, so both the old
# informally-hyphenated-name case and genuinely-hyphenated
# official symbols work.
if "-" in gene:
gene_hyphen_variant = gene.replace("-", "")
query_parts.append(f"{gene}[gene]")
if "condition" in arguments:
# Feature-70B-005: [disease/phenotype] is not a valid ClinVar eSearch field.
# Use [dis] (disease/phenotype) field tag for condition searches.
# Quote multi-word conditions so ClinVar treats them as a phrase.
condition = arguments["condition"].strip()
if " " in condition and not condition.startswith('"'):
condition = f'"{condition}"'
query_parts.append(f"{condition}[dis]")
if "variant_id" in arguments:
# Feature-70B-004: [variant_id] is not recognized by ClinVar eSearch.
# Use [uid] to look up by numeric variation ID.
query_parts.append(f"{arguments['variant_id']}[uid]")
if arguments.get("variant_name"):
# Fix-R78B-1: the Fix-R5D-1/R8C-1 comment above assumed there was
# "no such parameter" -- but ClinVar eSearch DOES support a real
# [Variant name] field tag (confirmed live: "HBB[gene] AND
# Glu7Val[Variant name]" -> 9 hits including VCV 15333, the
# canonical sickle-cell NM_000518.5(HBB):c.20A>T (p.Glu7Val)
# Pathogenic record). Without this, an exact-match lookup for a
# specific protein change/HGVS notation could only be done by
# fetching gene-level rows (capped at 100) and filtering
# client-side -- which silently misses old/low-ID records in any
# variant-dense gene, since ClinVar's default gene-level order
# isn't clinical-significance- or fame-ranked (confirmed live:
# HBB's own sickle-cell record, one of the best-known variants in
# human genetics, isn't within the first 100 or even the first
# 425-Pathogenic-filtered gene-level rows). Accepts either a
# single name or a list (a multi-allelic site yields one protein
# change per allele; only the queried allele's name will hit, so
# every candidate must be tried) -- combined with OR so one
# request covers all candidates.
names = arguments["variant_name"]
names = names if isinstance(names, list) else [names]
names = [str(n).strip() for n in names if str(n).strip()]
if names:
terms = " OR ".join(f"{n}[Variant name]" for n in names)
query_parts.append(f"({terms})" if len(names) > 1 else terms)
if "clinical_significance" in arguments:
# Feature-82A-002: NCBI silently translates [clnsig] to [All Fields],
# returning unrelated variants. The correct syntax is the [Filter] field:
# "clinsig pathogenic"[Filter] which properly restricts to the clinsig index.
#
# Fix-R6C-1: that [Filter] form only indexes single-word clinsig
# values -- confirmed live that "clinsig risk factor"[Filter] and
# "clinsig likely pathogenic"[Filter] both silently return 0
# results even though matching variants exist (e.g. HFE C282Y,
# whose own classification literally contains "risk factor").
# ClinVar separately indexes compound values via clinsig_<value
# with underscores>[prop] (confirmed live for risk_factor and
# likely_pathogenic). OR both forms so single-word values keep
# using the already-verified [Filter] path while compound values
# fall through to the [prop] path -- this can only add matches
# the old query missed, never drop ones it already found.
clnsig = arguments["clinical_significance"].lower().replace("_", " ")
clnsig_prop = clnsig.replace(" ", "_")
query_parts.append(
f'("clinsig {clnsig}"[Filter] OR clinsig_{clnsig_prop}[prop])'
)
# Fix-R5D-1/R8C-1: a caller-supplied param that doesn't match any
# recognized name/alias (e.g. "gene_name" instead of "gene") was
# silently dropped. (Fix-R78B-1: "variant_name" -- originally called
# out here as "no such parameter" -- is now a real, supported param;
# see above.) When it's the ONLY param supplied, this
# produced a generic "at least one search parameter is required"
# error with no hint the parameter itself was misnamed. When OTHER
# valid params are also supplied, the query still runs but silently
# ignores the unrecognized one -- confirmed live that
# {"gene": "GJB2", "made_up_param": "35delG"} returns all 739 GJB2
# variants with no indication the made_up_param filter had zero
# effect. Name unrecognized parameter(s) in both cases.
recognized = {
"gene",
"gene_symbol",
"condition",
"query",
"variant_id",
"variant_name",
"clinical_significance",
"significance",
"max_results",
"limit",
}
unrecognized = sorted(set(arguments) - recognized)
if not query_parts:
if unrecognized:
return {
"status": "error",
"error": (
f"Unrecognized parameter(s): {', '.join(unrecognized)}. "
"Valid search parameters: gene (or gene_symbol), "
"condition (or query), variant_id, variant_name, "
"clinical_significance (or significance)."
),
}
return {
"status": "error",
"error": "At least one search parameter is required",
}
params["term"] = " AND ".join(query_parts)
result = self._make_request(self.endpoint, params)
if result.get("status") != "success":
return result
data = result.get("data", {})
if "esearchresult" not in data:
return result
# Fix-R49A-1: the hyphen-preserved gene query above found nothing --
# retry once with the hyphen stripped, for the informally-hyphenated
# BRCA-2/HER-2-style names ClinVar's index doesn't recognize.
if gene_hyphen_variant and int(data["esearchresult"].get("count", 0)) == 0:
# "gene" is always the first query part appended when present.
fallback_query_parts = [f"{gene_hyphen_variant}[gene]"] + query_parts[1:]
fallback_params = dict(params, term=" AND ".join(fallback_query_parts))
fallback_result = self._make_request(self.endpoint, fallback_params)
if (
fallback_result.get("status") == "success"
and "esearchresult" in fallback_result.get("data", {})
and int(fallback_result["data"]["esearchresult"].get("count", 0)) > 0
):
result = fallback_result
data = result["data"]
# Fix-R62A-1: some gene symbols are HGNC-deprecated/renamed entirely,
# not just informally hyphenated -- e.g. "GBA" (glucocerebrosidase,
# Gaucher disease) was renamed to "GBA1" in 2023 to disambiguate from
# the GBA2/GBA3 gene family, and ClinVar's own [gene] index has fully
# absorbed the rename with no backward-compat alias (confirmed live
# and via raw NCBI E-utils: "GBA[gene]" -> 0 hits, not even recognized
# as a search phrase; "GBA1[gene]" -> 786 hits). Unlike the hyphen
# case above, no string transform recovers the current symbol --
# resolve it via NCBI's own gene database (which tracks "GBA" as an
# alias of gene ID 2629, current official symbol "GBA1") instead of
# hardcoding a growing alias list.
if "gene" in arguments and int(data["esearchresult"].get("count", 0)) == 0:
resolved_gene = self._resolve_deprecated_gene_symbol(gene)
if resolved_gene and resolved_gene.upper() != gene.upper():
resolved_query_parts = [f"{resolved_gene}[gene]"] + query_parts[1:]
resolved_params = dict(params, term=" AND ".join(resolved_query_parts))
resolved_result = self._make_request(self.endpoint, resolved_params)
if (
resolved_result.get("status") == "success"
and "esearchresult" in resolved_result.get("data", {})
and int(resolved_result["data"]["esearchresult"].get("count", 0))
> 0
):
result = resolved_result
data = result["data"]
esearch = data["esearchresult"]
ids = esearch.get("idlist", [])
count = int(esearch.get("count", 0))
variants = []
if ids:
summary_result = self._make_request(
"/esummary.fcgi",
{"db": "clinvar", "id": ",".join(ids[:200]), "retmode": "json"},
)
if summary_result.get("status") == "success":
result_map = summary_result.get("data", {}).get("result", {})
for vid in ids:
vdata = result_map.get(vid)
if vdata and not vdata.get("error"):
variants.append(
{"variant_id": vid, **self._parse_variant_summary(vdata)}
)
search_params = {
k: v
for k, v in {
"gene": arguments.get("gene"),
"condition": arguments.get("condition"),
"variant_id": arguments.get("variant_id"),
"clinical_significance": arguments.get("clinical_significance"),
}.items()
if v is not None
}
response_data = {
"total_count": count,
"variant_ids": ids,
"variants": variants,
"query_translation": esearch.get("querytranslation", ""),
"search_params": search_params,
}
if unrecognized:
response_data["ignored_parameters"] = unrecognized
return {"status": "success", "data": response_data}
[docs]
def _resolve_deprecated_gene_symbol(self, gene: str) -> Optional[str]:
"""Look up `gene` in NCBI's own gene database (which tracks
deprecated/previous symbols as aliases, e.g. "GBA" -> current
official symbol "GBA1") and return its current official symbol.
Best-effort: returns None on any failure or if no gene is found,
never raises -- this is a fallback, not a hard requirement."""
try:
search_result = self._make_request(
"/esearch.fcgi",
{
"db": "gene",
"term": f"{gene}[sym] AND human[orgn]",
"retmode": "json",
},
)
if search_result.get("status") != "success":
return None
ids = (
search_result.get("data", {}).get("esearchresult", {}).get("idlist", [])
)
if not ids:
return None
summary_result = self._make_request(
"/esummary.fcgi", {"db": "gene", "id": ids[0], "retmode": "json"}
)
if summary_result.get("status") != "success":
return None
gene_data = summary_result.get("data", {}).get("result", {}).get(ids[0], {})
return gene_data.get("name") or None
except Exception:
return None
[docs]
@register_tool("ClinVarGetVariantDetails")
class ClinVarGetVariantDetails(ClinVarRESTTool):
"""Get detailed variant information by ClinVar ID."""
[docs]
def __init__(self, tool_config):
super().__init__(tool_config)
self.endpoint = "/esummary.fcgi"
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get variant details by ClinVar ID."""
variant_id = arguments.get("variant_id", "")
if not variant_id:
return {"status": "error", "error": "variant_id is required"}
fetch = self._fetch_variant(variant_id)
if "error_result" in fetch:
return fetch["error_result"]
variant_data = fetch["variant_data"]
result = fetch["result"]
result["variant_id"] = variant_id
result["formatted_data"] = {
"variant_id": variant_id,
"accession": variant_data.get("accession", ""),
"obj_type": variant_data.get("obj_type", ""),
"chromosome": variant_data.get("chr_sort", ""),
"location": variant_data.get("variation_set", [{}])[0]
.get("variation_loc", [{}])[0]
.get("band", ""),
"variation_name": variant_data.get("variation_set", [{}])[0].get(
"variation_name", ""
),
**self._parse_variant_summary(variant_data),
"raw_data": variant_data,
}
# Fix-R8E-1/R6C-2: `result["data"]` (the full, unprocessed esummary
# envelope from _make_request) duplicates the same content already
# exposed at formatted_data["raw_data"] -- keeping both roughly
# tripled payload size for no informational gain and made this
# tool's output nearly indistinguishable from
# ClinVarGetClinicalSignificance's, which has the identical
# duplication. Drop the redundant top-level copy; raw access is
# still available via formatted_data.raw_data.
result.pop("data", None)
return result
[docs]
@register_tool("ClinVarGetClinicalSignificance")
class ClinVarGetClinicalSignificance(ClinVarRESTTool):
"""Get clinical significance information for variants."""
[docs]
def __init__(self, tool_config):
super().__init__(tool_config)
self.endpoint = "/esummary.fcgi"
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get clinical significance by variant ID."""
variant_id = arguments.get("variant_id", "")
if not variant_id:
return {"status": "error", "error": "variant_id is required"}
fetch = self._fetch_variant(variant_id)
if "error_result" in fetch:
return fetch["error_result"]
variant_data = fetch["variant_data"]
result = fetch["result"]
result["variant_id"] = variant_id
# Extract clinical significance information
germline_class = variant_data.get("germline_classification", {})
clinical_impact = variant_data.get("clinical_impact_classification", {})
oncogenicity = variant_data.get("oncogenicity_classification", {})
result["formatted_data"] = {
"variant_id": variant_id,
"germline_classification": {
"description": germline_class.get("description", ""),
"review_status": germline_class.get("review_status", ""),
"last_evaluated": germline_class.get("last_evaluated", ""),
"fda_recognized": germline_class.get("fda_recognized_database", ""),
"traits": [
trait.get("trait_name", "")
for trait in germline_class.get("trait_set", [])
],
},
"clinical_impact": {
"description": clinical_impact.get("description", ""),
"review_status": clinical_impact.get("review_status", ""),
"last_evaluated": clinical_impact.get("last_evaluated", ""),
},
"oncogenicity": {
"description": oncogenicity.get("description", ""),
"review_status": oncogenicity.get("review_status", ""),
"last_evaluated": oncogenicity.get("last_evaluated", ""),
},
"raw_data": variant_data,
}
# Fix-R8E-1/R6C-2: see the matching fix in ClinVarGetVariantDetails
# -- result["data"] duplicates formatted_data["raw_data"].
result.pop("data", None)
return result
