comprehensive-variant-annotation

name	comprehensive-variant-annotation
description	Given an rsID, query multiple databases (dbSNP, FAVOR, GWAS Catalog, ClinVar, gnomAD, PharmGKB, ClinGen) for comprehensive annotation. Use when user asks a general question about a variant without specifying which aspect.
license	MIT license
metadata	{"skill-author":"PJLab"}

Comprehensive Variant Annotation

Usage

1. Tool Descriptions

This skill chains 7 public genomics database APIs sequentially to build a comprehensive annotation for a given variant. Use this when the user asks a general/vague question like "帮我查一下 rs7412" or "tell me about rs7412".

Tool 1: dbSNP — Variant Basic Info & ClinVar RCV Records

Query NCBI dbSNP REST API to get SNP basic information and ClinVar clinical records.
API: GET https://api.ncbi.nlm.nih.gov/variation/v0/refsnp/{rsid_number}
Args:
    rs_id (str): dbSNP rsID (e.g. "rs7412")
Return:
    primary_snapshot_data: allele_annotations (gene associations, functional impact,
        ClinVar RCV clinical records), placements_with_allele (GRCh38 coordinates).

Tool 2: FAVOR — Functional Annotation & Scores

Query FAVOR (GenoHub) API to get functional annotation and conservation scores.
API: GET https://api.genohub.org/v1/rsids/{rsid}
Return:
    Functional prediction scores (CADD, REVEL, etc.), conservation scores,
    gene annotations, variant effect predictions, and variant_id for gnomAD.

Tool 3: gnomAD — Population Allele Frequency

Query gnomAD GraphQL API for population allele frequencies.
API: POST https://gnomad.broadinstitute.org/api
Note: Requires variant_id (chr-pos-ref-alt format) from FAVOR Step 2.
Return:
    Allele frequencies across populations (global, AFR, AMR, ASJ, EAS, FIN, NFE, SAS, etc.)

Tool 4: GWAS Catalog — Trait Associations

Query EBI GWAS Catalog REST API for GWAS statistical associations.
API: GET https://www.ebi.ac.uk/gwas/rest/api/associations/search/findByRsId?rsId={rsid}
Return:
    associations: pvalue, risk allele, associated trait/disease, source study.

Tool 5: ClinVar — Clinical Pathogenicity (extracted from dbSNP Step 1)

Extract ClinVar RCV records from dbSNP response (already obtained in Step 1).
Return:
    Clinical significance (Pathogenic/Benign/VUS/drug-response),
    review status, associated diseases, RCV accession numbers.

Tool 6: PharmGKB — Pharmacogenomic Annotations

Query PharmGKB clinPGx API for drug-gene-variant interactions.
API: GET https://api.clinpgx.org/v1/data/clinicalAnnotation?location.fingerprint={rsid}&view=base
Return:
    Related drugs, evidence level (1A-4), related diseases, annotation types.

Tool 7: ClinGen — Cross-Database ID Mapping

Query ClinGen Allele Registry for cross-database identifiers.
API: GET https://reg.genome.network/alleles?dbSNP.rs={rs_id}
Headers: Accept: application/json
Return:
    CA ID, ClinVar IDs, COSMIC ID, gnomAD IDs, external cross-references.

2. Comprehensive Variant Annotation

Query 7 databases for a given rsID, then save all results into a single JSON file {rsID}_annotation.json.

import requests
import json
from datetime import datetime

rs_id = "rs7412"
results = {"query_rsid": rs_id, "timestamp": datetime.now().isoformat()}

def safe_request(name, func, fallback=None):
    """统一的容错请求包装器。任一数据库超时/报错不会中断整个流程。"""
    try:
        return func()
    except requests.exceptions.Timeout:
        print(f"[{name}] ⚠ 连接超时，跳过")
        results.setdefault("errors", {})[name] = "timeout"
        return fallback
    except Exception as e:
        print(f"[{name}] ⚠ 请求失败: {e}，跳过")
        results.setdefault("errors", {})[name] = str(e)
        return fallback

# ── Step 1: dbSNP — 变异基本信息 + ClinVar RCV ──
rsid_num = rs_id.replace("rs", "")
dbsnp_url = f"https://api.ncbi.nlm.nih.gov/variation/v0/refsnp/{rsid_num}"
dbsnp = safe_request("dbSNP",
    lambda: requests.get(dbsnp_url, timeout=30).json(), fallback={})
results["dbsnp"] = dbsnp
print(f"[dbSNP] {rs_id} 查询{'成功' if dbsnp else '失败'}")

# 从 dbSNP 提取 ClinVar RCV 记录（Step 5）
snapshot = dbsnp.get(, {})
clinvar_records = []
 ann  snapshot.get(, []):
     clin  ann.get(, []):
        clinvar_records.append({
            : clin.get(, ),
            : clin.get(, []),
            : clin.get(, []),
            : clin.get(, )
        })
results[] = clinvar_records
()


favor_url = 
favor = safe_request(,
    : requests.get(favor_url, timeout=).json(), fallback={})
results[] = favor
()


variant_id = 
favor_results = favor  (favor, )  favor.get(, [favor])  favor  []
 favor_results  (favor_results, ):
    first = favor_results[]  favor_results  {}
    chrom = (first.get(, ))
    pos = (first.get(, ))
    ref = first.get(, )
    alt = first.get(, )
     chrom  pos  ref  alt:
        variant_id = 


 variant_id:
    gnomad_query = 
    gnomad_data = safe_request(,
        : requests.post(
            ,
            json={: gnomad_query, : {: variant_id}},
            timeout=
        ).json(), fallback={})
    results[] = (gnomad_data  {}).get(, {}).get(, {})
    genome = results[].get(, {})  results[]  {}
    ()
:
    results[] = {: }
    ()


gwas_url = 
gwas = safe_request(,
    : requests.get(gwas_url, headers={: }, timeout=).json(),
    fallback={})
associations = (gwas  {}).get(, {}).get(, [])
results[] = {
    : (associations),
    : associations
}
()


pgx_url = 
pgx_resp = safe_request(,
    : requests.get(pgx_url, timeout=).json(), fallback=[])
pgx_annotations = pgx_resp  (pgx_resp, )  (pgx_resp  {}).get(, [])
results[] = pgx_annotations
()



clingen_url = 
clingen_resp = 
 attempt  ():  
    clingen_resp = safe_request(,
        : requests.get(clingen_url,
            headers={: }, timeout=).json(),
        fallback=)
     clingen_resp   :
        
    ()

 clingen_resp   :
      (clingen_resp, ):
        clingen_resp = [clingen_resp]
    
    clingen_alleles = []
     allele  clingen_resp:
        titles = allele.get(, [])
         titles  (  t  t  titles):
            
        clingen_alleles.append({
            : allele.get(, ).split()[-],
            : titles,
            : allele.get(, {})
        })
    results[] = clingen_alleles
    ()
:
    results[] = {: }
    ()


output_file = 
 (output_file, , encoding=)  f:
    json.dump(results, f, indent=, ensure_ascii=)


errors = results.get(, {})
 errors:
    ()
()