get_visions_guide

Instructions

Implementation Reference

• src/pythia_oracle_mcp/server.py:929-1030 (handler)

  The main handler function for the 'get_visions_guide' tool. It fetches live data from feed-status.json, retrieves the Vision Registry address, and returns a complete Solidity contract for subscribing to Pythia Visions and listening for VisionFired events.

  @mcp.tool()
  async def get_visions_guide() -> str:
      """Get Solidity code to subscribe to Pythia Visions and listen for VisionFired events.
  
      Returns a complete contract that subscribes to the PythiaVisionRegistry,
      receives VisionFired events with pattern type, confidence, direction, price,
      and full analysis payload. Subscription is FREE (no LINK required).
      """
      data = await _fetch_data()
      visions = data.get("visions", {})
      registry = visions.get("registry", "")
      if not registry:
          raise RuntimeError(
              "Pythia Visions registry address is missing from live data. "
              "Visions may not be deployed on this environment yet. "
              "Check https://pythia.c3x-solutions.com/feed-status.json visions.registry."
          )
      mainnet = _get_mainnet(data)
      link_token = mainnet["link_token"]
  
      return f"""Pythia Visions Integration — Walk-Forward Validated Market Intelligence On-Chain
  
  Vision Registry (Mainnet): {registry}
  LINK Token: {link_token}
  
  ```solidity
  // SPDX-License-Identifier: MIT
  pragma solidity ^0.8.19;
  
  interface IPythiaVisionRegistry {{
      function subscribe(bytes32 tokenId) external;
      function unsubscribe(bytes32 tokenId) external;
      function isSubscribed(address subscriber, bytes32 tokenId) external view returns (bool);
      function getSubscriberCount(bytes32 tokenId) external view returns (uint256);
      function getVisionCount(bytes32 tokenId) external view returns (uint256);
      function getLastVisionAt(bytes32 tokenId) external view returns (uint256);
  }}
  
  contract MyVisionSubscriber {{
      IPythiaVisionRegistry public immutable registry;
  
      // Token IDs are keccak256 hashes of the token name
      bytes32 public constant BTC = keccak256("BTC");
      bytes32 public constant ETH = keccak256("ETH");
  
      // Fired by PythiaVisionRegistry when a pattern is detected
      event VisionFired(
          bytes32 indexed tokenId,   // keccak256("BTC") or keccak256("ETH")
          uint8 patternType,          // per-token high nibble: BTC=0x1_, ETH=0x2_
          uint8 confidence,           // Confidence score within pattern's historical range
          uint8 direction,            // 1 = BULLISH
          uint256 price,              // 18 decimals
          bytes payload               // Full JSON (indicators + analysis)
      );
  
      constructor(address _registry) {{
          registry = IPythiaVisionRegistry(_registry);
      }}
  
      /// @notice Subscribe to Visions for a token. FREE — no LINK required.
      function subscribeToken(bytes32 tokenId) external {{
          registry.subscribe(tokenId);
      }}
  
      /// @notice Unsubscribe.
      function unsubscribeToken(bytes32 tokenId) external {{
          registry.unsubscribe(tokenId);
      }}
  
      /// @notice Subscribe to BTC Visions.
      function subscribeBTC() external {{
          registry.subscribe(BTC);
      }}
  
      /// @notice Subscribe to ETH Visions.
      function subscribeETH() external {{
          registry.subscribe(ETH);
      }}
  }}
  ```
  
  Steps:
  1. Deploy with (_registry) = {registry}
  2. Call subscribeBTC() / subscribeETH() — no LINK needed, subscription is FREE
  3. Listen for VisionFired events on the registry contract via RPC/WebSocket
  4. Decode the payload bytes to get the full analysis JSON
  
  Pattern Types (walk-forward validated):
    Per-token high nibble: BTC=0x1_, ETH=0x2_, next token=0x3_, etc.
    Live pattern catalog (codes, accuracy ranges, fold validation, fires/yr,
    failure profile) — call get_visions_info() for the current set.
  
  Payload JSON includes: indicators (RSI, EMA, Bollinger, VWAP, ATR),
  pattern details, confidence score, analysis narrative,
  and feeds-to-watch for confirmation.
  
  Token IDs: keccak256 of the token name.
    BTC = keccak256("BTC") = 0xe98e2830be1a7e4156d656a7505e65d08c67660dc618072422e9c78053c261e9
    ETH = keccak256("ETH") = 0xaaaebeba3810b1e6b70781f14b2d72c1cb89c0b2b320c43bb67ff79f562f5ff4
  
  Deployment:
    Mainnet: _registry={registry}"""

• src/pythia_oracle_mcp/server.py:929-931 (registration)

  Registered as an MCP tool via the @mcp.tool() decorator on line 929. The FastMCP instance ('mcp') is defined on line 17.

  @mcp.tool()
  async def get_visions_guide() -> str:
      """Get Solidity code to subscribe to Pythia Visions and listen for VisionFired events.

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

  The _fetch_data() helper used by get_visions_guide to retrieve live data from the Pythia data engine with caching.

  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:75-99 (helper)

  The _get_contracts() helper used indirectly through _get_mainnet() to resolve contract addresses for the integration guide.

  def _get_contracts(data: dict) -> dict:
      """Extract normalized contracts from live feed-status.json data.
  
      Raises RuntimeError if the data is missing the developer.contracts section
      (would only happen if generate_site_data.py is broken or schema changed).
      """
      contracts = data.get("developer", {}).get("contracts")
      if not contracts:
          raise RuntimeError(
              "feed-status.json is missing developer.contracts. "
              "This is a structural error in the live data — check the data engine."
          )
  
      result = {}
      for chain_key, chain_data in contracts.items():
          consumers_raw = chain_data.get("consumers", {})
          result[chain_key] = {
              "display_name": chain_data.get("display_name", chain_key),
              "chain_id": chain_data.get("chain_id"),
              "explorer": chain_data.get("explorer", ""),
              "operator": chain_data.get("operator", ""),
              "link_token": chain_data.get("link_token", ""),
              "consumers": _parse_consumers(consumers_raw),
          }
      return result

• src/pythia_oracle_mcp/server.py:863-926 (handler)

  The related get_visions_info() tool which provides the overview of Pythia Visions and references get_visions_guide() for Solidity code.

  @mcp.tool()
  async def get_visions_info() -> str:
      """Get overview of Pythia Visions — walk-forward validated market intelligence on-chain.
  
      Returns the walk-forward validated patterns with accuracy stats, the Vision Registry
      contract address, subscription info (FREE), evaluation frequency, and
      supported tokens. Visions are pattern detections that passed walk-forward
      validation across multiple years of history. Live token + pattern set is
      returned in the response (canonical source: feed-status.json visions section).
      """
      data = await _fetch_data()
      visions = data.get("visions", {})
  
      registry = visions.get("registry", "")
      patterns = visions.get("patterns", [])
      tokens = visions.get("tokens", [])
      stats = visions.get("stats", {})
  
      if not patterns:
          return (
              "Pythia Visions catalog is empty in the live data. "
              "This may mean Visions are not yet deployed on this environment, "
              "or feed-status.json is missing the visions.patterns section. "
              "Check https://pythia.c3x-solutions.com/feed-status.json directly."
          )
  
      lines = ["Pythia Visions — Walk-Forward Validated Market Intelligence On-Chain\n"]
      lines.append(
          "Walk-forward validated pattern detections delivered on-chain via "
          "Chainlink. FREE to subscribe."
      )
      lines.append("")
  
      lines.append("Detected Patterns (validated 2017-2026, 75K+ candles):\n")
      lines.append(f"  {'Token':<5} {'Pattern':<22} {'Code':<6} {'Accuracy':<10} {'Avg Return':<14} {'Frequency':<14} {'Folds'}")
      lines.append(f"  {'-'*5} {'-'*22} {'-'*6} {'-'*10} {'-'*14} {'-'*14} {'-'*5}")
      for p in patterns:
          lines.append(
              f"  {p.get('token', '?'):<5} {p['name']:<22} {p['code']:<6} {p['accuracy']:<10} "
              f"{p['avg_return']:<14} {p['frequency']:<14} {p.get('fold_validation', '?')}"
          )
      lines.append("")
  
      lines.append("How to Use:")
      lines.append("  1. Subscribe — one call, free, permissionless: subscribe(keccak256(\"BTC\"))")
      lines.append("  2. Receive — your contract gets VisionFired events with pattern, confidence, indicators, analysis, and feeds-to-watch")
      lines.append("  3. React — read the payload on-chain, auto-subscribe to confirmation Events, trigger your strategy")
      lines.append("")
  
      lines.append(f"Vision Registry: {registry}")
      lines.append(f"Chain: Polygon PoS (mainnet)")
      lines.append(f"Subscription fee: FREE")
      lines.append(f"Tokens: {', '.join(tokens)}")
      lines.append(f"Signal frequency: ~107 Visions/year for BTC (100 OVERSOLD + 7 CAPITULATION), ~13/year for ETH")
      lines.append("")
  
      if stats.get("total_fired"):
          lines.append(f"Stats: {stats['total_fired']} total fired, "
                       f"avg confidence {stats.get('avg_confidence', 'N/A')}")
          lines.append("")
  
      lines.append("Use get_visions_guide() for Solidity integration code.")
      lines.append("Use get_vision_history() for recent Visions fired.")
      return "\n".join(lines)