wait_for_variant_analysis

Instructions

Implementation Reference

• server.py:448-521 (handler)

  The tool handler function that polls EVEE's on-demand interpretation endpoint until completion or timeout. Registered via @mcp.tool() decorator. Accepts variant_id, timeout_seconds (1-60), and poll_interval_seconds (0.5-10). Fetches variant data, repeatedly calls _fetch_analysis, then curates the result with _curate_variant_summary and _interpretation_from_analysis, adding a wait_status entry.

  @mcp.tool()
  def wait_for_variant_analysis(
      variant_id: str,
      timeout_seconds: float = 20.0,
      poll_interval_seconds: float = 2.0,
  ) -> dict:
      """Poll EVEE's on-demand interpretation until it completes or times out.
  
      Use this when `get_variant` reports `interpretation.status` as queued or
      processing. Returns the same curated variant summary as `get_variant`, plus
      a `wait_status` entry with attempts / elapsed_seconds. If the deadline
      hits before completion, call this tool again to keep polling.
  
      Args:
          variant_id: Variant identifier in chr:pos:ref:alt format.
          timeout_seconds: Maximum wall-clock time to wait (clamped to [1, 60]).
          poll_interval_seconds: Delay between polls (clamped to [0.5, 10]).
      """
      timeout_seconds = min(max(timeout_seconds, 1.0), 60.0)
      poll_interval_seconds = min(max(poll_interval_seconds, 0.5), 10.0)
  
      started = time.monotonic()
      deadline = started + timeout_seconds
      attempts = 0
      analysis: dict = {"status": "unavailable"}
  
      def _remaining() -> float:
          return max(0.0, deadline - time.monotonic())
  
      with _get_client() as client:
          resp = client.get(f"/variants/{variant_id}", timeout=max(1.0, _remaining()))
          if resp.status_code == 404:
              return {"error": "Variant not found", "variant_id": variant_id}
          resp.raise_for_status()
          data = resp.json()
  
          while True:
              remaining = _remaining()
              if remaining <= 0:
                  break
              try:
                  analysis = _fetch_analysis(client, variant_id, timeout=max(1.0, remaining))
              except httpx.TimeoutException:
                  analysis = {"status": "queued"}
                  break
              attempts += 1
              if analysis.get("status") in ("complete", "not_found"):
                  break
              if _remaining() <= 0:
                  break
              time.sleep(min(poll_interval_seconds, _remaining()))
  
      if analysis.get("status") == "not_found":
          return {"error": "Variant not found", "variant_id": variant_id}
  
      summary = _curate_variant_summary(data)
      interp = _interpretation_from_analysis(analysis)
      if interp:
          summary["interpretation"] = interp
          wait_status = "complete"
      else:
          summary["interpretation"] = {
              "status": analysis.get("status", "unavailable"),
              "detail": "Still generating after timeout. Call wait_for_variant_analysis again to continue polling.",
          }
          wait_status = "timeout"
  
      summary["wait_status"] = {
          "status": wait_status,
          "attempts": attempts,
          "elapsed_seconds": round(time.monotonic() - started, 2),
      }
      return summary

• server.py:127-141 (helper)

  Helper function that makes the actual HTTP call to the /variants/{id}/analysis endpoint. Called inside the polling loop of wait_for_variant_analysis with a timeout parameter.

  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:143-153 (helper)

  Helper that transforms a completed analysis response into a curated interpretation dict (summary, mechanism, key_evidence, confidence). Returns None if status is not 'complete'.

  def _interpretation_from_analysis(analysis: dict) -> dict | None:
      """Build the curated `interpretation` dict from an /analysis response."""
      if analysis.get("status") != "complete":
          return None
      r = analysis.get("result") or {}
      return {
          "summary": r.get("summary"),
          "mechanism": r.get("mechanism"),
          "key_evidence": r.get("key_evidence"),
          "confidence": r.get("confidence"),
      }

• server.py:224-349 (helper)

  Helper that curates the full variant API response into a structured summary dict (identity, gene, consequence, clinical, scores, domains, interpretation, similar variants). Used by both get_variant and wait_for_variant_analysis.

  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:448-448 (registration)

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

  @mcp.tool()