get_vision_history

Instructions

Implementation Reference

• src/pythia_oracle_mcp/server.py:1033-1099 (handler)

  The 'get_vision_history' tool handler. It is defined as an async MCP tool function that takes a token symbol (default 'BTC'), fetches live data from feed-status.json, filters recent visions for the given token, and returns formatted history with pattern breakdown and stats.

  @mcp.tool()
  async def get_vision_history(token: str = "BTC") -> str:
      """Get recent Pythia Visions fired for a token with pattern breakdown and stats.
  
      Args:
          token: Token symbol to check (default: BTC). Case-insensitive.
                 Currently live: BTC, ETH.
      """
      data = await _fetch_data()
      visions = data.get("visions", {})
  
      if not visions:
          return ("Pythia Visions data not available yet. "
                  "Use get_visions_info() for pattern details and contract address.")
  
      recent = visions.get("recent", [])
      stats = visions.get("stats", {})
      registry = visions.get("registry", "")
      token_upper = token.upper()
  
      # Filter by token
      filtered = [v for v in recent if v.get("token", "").upper() == token_upper]
  
      lines = [f"Pythia Visions — {token_upper} History\n"]
      lines.append(f"Registry: {registry}")
      lines.append(f"Subscription: FREE\n")
  
      if not filtered:
          lines.append(f"No Visions have fired for {token_upper} yet.")
          lines.append(f"\nAvailable tokens: {', '.join(visions.get('tokens', []))}")
          lines.append("\nUse get_visions_info() for pattern details.")
          return "\n".join(lines)
  
      lines.append(f"Recent Visions ({len(filtered)} shown):\n")
      for v in filtered:
          lines.append(f"  {v.get('fired_at', '?')}")
          lines.append(f"    Pattern:    {v.get('pattern_name', '?')}")
          lines.append(f"    Confidence: {v.get('confidence', '?')}")
          lines.append(f"    Direction:  {v.get('direction', '?')}")
          lines.append(f"    Price:      ${v.get('price_usd', 0):,.2f}")
          lines.append("")
  
      # Pattern breakdown
      pattern_counts: dict[str, list[int]] = {}
      for v in filtered:
          name = v.get("pattern_name", "?")
          conf = v.get("confidence", 0)
          if name not in pattern_counts:
              pattern_counts[name] = []
          pattern_counts[name].append(conf)
  
      lines.append("Pattern Breakdown:\n")
      lines.append(f"  {'Pattern':<28} {'Fires':<7} {'Avg Confidence'}")
      lines.append(f"  {'-'*28} {'-'*7} {'-'*14}")
      for name, confs in sorted(pattern_counts.items(), key=lambda x: -len(x[1])):
          avg = sum(confs) / len(confs) if confs else 0
          lines.append(f"  {name:<28} {len(confs):<7} {avg:.1f}")
      lines.append("")
  
      if stats.get("total_fired"):
          lines.append(f"Overall: {stats['total_fired']} total fired, "
                       f"avg confidence {stats.get('avg_confidence', 'N/A')}")
  
      lines.append("\nUse get_vision_payload(vision_id) for the full enriched object")
      lines.append("(failure profile, cooldown context, concurrent fires).")
      lines.append("Use get_visions_guide() for Solidity integration code.")
      return "\n".join(lines)

• src/pythia_oracle_mcp/server.py:1033-1034 (registration)

  The tool is registered via the @mcp.tool() decorator on the get_vision_history function at line 1033.

  @mcp.tool()
  async def get_vision_history(token: str = "BTC") -> str:

• src/pythia_oracle_mcp/server.py:134-161 (helper)

  The _fetch_data() helper function fetches and caches the feed-status.json data that get_vision_history relies on.

  async def _fetch_data() -> dict:
      """Fetch feed-status.json from the live Pythia data engine.
  
      Cached for CACHE_TTL_SECONDS to keep tool responses fast. Raises RuntimeError
      with a clear message if the live URL is unreachable — there is no baked-in
      fallback. AI consumers should retry shortly or check status; serving stale
      data silently would be worse than a clear failure.
      """
      now = datetime.now(timezone.utc)
      cached = _cache.get("data")
      if cached and (now - cached["at"]).total_seconds() < CACHE_TTL_SECONDS:
          return cached["data"]
  
      try:
          async with httpx.AsyncClient(timeout=15) as client:
              resp = await client.get(DATA_URL)
              resp.raise_for_status()
              data = resp.json()
      except httpx.HTTPError as e:
          raise RuntimeError(
              f"Pythia data unreachable: GET {DATA_URL} failed with "
              f"{type(e).__name__}: {e}. "
              "MCP cannot serve token, pattern, pricing, or contract data without the "
              "live JSON. Retry shortly, or check https://pythia.c3x-solutions.com/status."
          ) from e
  
      _cache["data"] = {"data": data, "at": now}
      return data

• src/pythia_oracle_mcp/server.py:1038-1039 (schema)

  The input schema for get_vision_history is defined as a single optional string parameter 'token' with a default of 'BTC', documented in the docstring.

  token: Token symbol to check (default: BTC). Case-insensitive.
         Currently live: BTC, ETH.