get_variant

Instructions

Implementation Reference

• server.py:402-445 (handler)

  The `get_variant` tool handler function. Registered via @mcp.tool() decorator, it fetches variant data from the EVEE API, curates it via _curate_variant_summary, and optionally triggers on-demand analysis if no stored interpretation is available.

  @mcp.tool()
  def get_variant(variant_id: str) -> dict:
      """Get comprehensive information about a specific genetic variant.
  
      Returns clinical significance, model-derived scores from EVEE's heads
      (aligned to AlphaMissense, CADD, REVEL, SIFT, etc.), reference predictor
      scores from external databases when present, gene constraint (LOEUF), HGVS
      notation, disease associations, protein domains, and the AI-generated
      mechanistic interpretation.
  
      If the stored interpretation isn't ready, this tool hits EVEE's on-demand
      /analysis endpoint once: if generation has already completed, the fresh
      interpretation is returned inline; otherwise the response carries an
      `interpretation = {status: queued/processing, detail: ...}` entry and you
      should call `wait_for_variant_analysis` to poll until it finishes.
  
      Args:
          variant_id: Variant identifier in chr:pos:ref:alt format
                      (e.g. "chr17:43092918:G:A" for BRCA1 ClinVar ID 41812).
                      NOTE: EVEE uses 0-based positions; ClinVar/VCF/HGVS are
                      1-based. Subtract 1 from ClinVar pos for SNVs; for indels
                      the offset varies — prefer search_variants.
      """
      with _get_client() as client:
          resp = client.get(f"/variants/{variant_id}")
          if resp.status_code == 404:
              return {"error": "Variant not found", "variant_id": variant_id}
          resp.raise_for_status()
          data = resp.json()
  
          summary = _curate_variant_summary(data)
  
          if summary["interpretation"] is None:
              analysis = _fetch_analysis(client, variant_id)
              interp = _interpretation_from_analysis(analysis)
              if interp:
                  summary["interpretation"] = interp
              else:
                  summary["interpretation"] = {
                      "status": analysis.get("status", "unavailable"),
                      "detail": "Interpretation is being generated on-demand. Call wait_for_variant_analysis to poll for completion.",
                  }
  
      return summary

• server.py:224-349 (helper)

  Helper function `_curate_variant_summary` that transforms the raw API response into a structured dict with identity, gene, consequence, clinical, model-derived scores, reference scores, domains, interpretation, and similar variants.

  def _curate_variant_summary(data: dict) -> dict:
      """Curate the massive variant response into a structured summary for LLM consumption."""
      summary = {}
  
      # --- Identity ---
      variant_id = data.get("variant_id")
      summary["variant_id"] = variant_id
      summary["evee_url"] = _evee_url(variant_id)
      summary["rs_id"] = data.get("rs_id")
      summary["chrom"] = data.get("chrom")
      summary["pos"] = data.get("pos")
      summary["ref"] = data.get("ref")
      summary["alt"] = data.get("alt")
      summary["variation_id"] = data.get("variation_id")
  
      # --- Gene ---
      summary["gene"] = data.get("gene_name")
      summary["gene_id"] = data.get("gene_id")
      summary["gene_strand"] = data.get("gene_strand")
      summary["loeuf"] = data.get("loeuf")
      summary["loeuf_label"] = data.get("loeuf_label")
  
      # --- Consequence & HGVS ---
      summary["consequence"] = data.get("consequence_display") or data.get("consequence")
      summary["hgvs_coding"] = data.get("hgvsc")
      summary["hgvs_protein"] = data.get("hgvsp")
      summary["hgvs_coding_short"] = data.get("hgvsc_short")
      summary["hgvs_protein_short"] = data.get("hgvsp_short")
      summary["vep_transcript_id"] = data.get("vep_transcript_id")
      summary["vep_protein_id"] = data.get("vep_protein_id")
      summary["exon"] = data.get("exon")
      summary["vep_impact"] = data.get("vep_impact")
  
      # --- Clinical ---
      summary["clinical_label"] = data.get("label_display") or data.get("label")
      summary["pathogenicity_score"] = data.get("pathogenicity") or data.get("score")
      summary["disease"] = data.get("disease")
      summary["clinical_features"] = data.get("clinical_features")
      summary["significance"] = data.get("significance")
      summary["review_status"] = data.get("review_status")
      summary["stars"] = data.get("stars")
      summary["n_submissions"] = data.get("n_submissions")
      summary["last_evaluated"] = data.get("last_evaluated")
      summary["origin"] = data.get("origin")
      summary["acmg"] = data.get("acmg")
  
      # --- Model-derived scores (EVEE heads / probes aligned to external predictors) ---
      scores = {}
      score_keys = {
          "evee_pathogenic": "eff_pathogenic",
          "evee_splice_disrupting": "eff_splice_disrupting",
          "alphamissense": "eff_alphamissense_c",
          "cadd": "eff_cadd_c",
          "revel": "eff_revel_c",
          "sift": "eff_sift_c",
          "polyphen": "eff_polyphen_c",
          "spliceai_max": "eff_spliceai_max_c",
          "clinpred": "eff_clinpred_c",
          "bayesdel": "eff_bayesdel_c",
          "vest4": "eff_vest4_c",
          "blosum62": "eff_blosum62_c",
          "grantham": "eff_grantham_c",
          "charge_altering": "eff_charge_altering",
          "hydrophobicity": "eff_hydrophobicity_c",
          "mpc": "eff_mpc_c",
          "mcap": "eff_mcap_c",
          "metalr": "eff_metalr_c",
          "mvp": "eff_mvp_c",
          "primateai": "eff_primateai_c",
          "deogen2": "eff_deogen2_c",
          "mutpred": "eff_mutpred_c",
          "cadd_wg": "eff_cadd_wg_c",
      }
      for name, key in score_keys.items():
          val = data.get(key)
          if val is not None:
              scores[name] = val
      summary["model_derived_scores"] = scores
  
      # --- Reference predictor scores (raw values from source databases, when present) ---
      gt_scores = {}
      gt_keys = {
          "alphamissense": "gt_alphamissense_c",
          "cadd": "gt_cadd_c",
          "revel": "gt_revel_c",
          "sift": "gt_sift_c",
          "spliceai_max": "gt_spliceai_max_c",
      }
      for name, key in gt_keys.items():
          val = data.get(key)
          if val is not None:
              gt_scores[name] = val
      if gt_scores:
          summary["reference_predictor_scores"] = gt_scores
  
      # --- Protein domains ---
      summary["domains"] = data.get("domains")
  
      # --- AI interpretation (from stored processed_result) ---
      pr = data.get("processed_result")
      if pr and isinstance(pr, dict) and pr.get("status") == "ok":
          summary["interpretation"] = {
              "summary": pr.get("summary"),
              "mechanism": pr.get("mechanism"),
              "key_evidence": pr.get("key_evidence"),
              "confidence": pr.get("confidence"),
          }
      else:
          summary["interpretation"] = None
  
      # --- Similar variants ---
      neighbors = data.get("neighbors", [])
      if neighbors:
          summary["similar_variants"] = [
              {
                  "variant_id": n.get("id"),
                  "gene": n.get("gene"),
                  "consequence": n.get("consequence_display"),
                  "label": n.get("label_display") or n.get("label"),
                  "score": n.get("score"),
                  "similarity": n.get("similarity"),
              }
              for n in neighbors
          ]
  
      return summary

• server.py:123-124 (helper)

  Helper function `_evee_url` that builds the EVEE web URL for a variant ID.

  def _evee_url(variant_id: str | None) -> str | None:
      return f"https://evee.goodfire.ai/#/variant/{variant_id}" if variant_id else None

• server.py:127-140 (helper)

  Helper function `_fetch_analysis` that hits the /variants/{id}/analysis endpoint to trigger/retrieve on-demand AI interpretation.

  def _fetch_analysis(client: httpx.Client, variant_id: str, timeout: float | None = None) -> dict:
      """Hit /variants/{id}/analysis once.
  
      Returns one of:
        {"status": "complete", "result": {...}}         — interpretation ready
        {"status": "queued", "retry_after": N}          — generation in progress
        {"status": "not_found"}                         — variant missing
      """
      kwargs = {"timeout": timeout} if timeout is not None else {}
      resp = client.get(f"/variants/{variant_id}/analysis", **kwargs)
      if resp.status_code == 404:
          return {"status": "not_found"}
      resp.raise_for_status()
      return resp.json()

• server.py:402-403 (registration)

  Registration of `get_variant` as an MCP tool via the @mcp.tool() decorator on the FastMCP instance.

  @mcp.tool()
  def get_variant(variant_id: str) -> dict: