get_contracts

Instructions

Implementation Reference

• src/pythia_oracle_mcp/server.py:400-440 (handler)

  The main tool handler for 'get_contracts'. It fetches live data, extracts contract addresses using _get_contracts() helper, tier fees, and formats a human-readable response with all supported chains, consumer contracts, and event registry addresses.

  @mcp.tool()
  async def get_contracts() -> str:
      """Get Pythia contract addresses for on-chain integration. Shows all supported chains."""
      data = await _fetch_data()
      all_contracts = _get_contracts(data)
      fees = _get_tier_fees(data)
      events = data.get("events", {}) if data else {}
  
      lines = ["Pythia Oracle — Contract Addresses\n"]
  
      for chain_key, chain in sorted(all_contracts.items()):
          chain_id = chain.get("chain_id", "?")
          lines.append(f"  {chain['display_name']} (Chain ID {chain_id})")
          lines.append(f"    Operator:            {chain['operator']}")
          lines.append(f"    LINK Token (ERC-677): {chain['link_token']}")
          lines.append("")
          lines.append("    Consumer Contracts (by tier):")
          for tier in ("discovery", "analysis", "speed", "complete"):
              addr = chain["consumers"].get(tier)
              if not addr:
                  continue
              fee_val = fees.get(tier, "?")
              lines.append(f"      {tier.upper()} — {fee_val} LINK")
              lines.append(f"        Address: {addr}")
              lines.append(f"        Returns: {_TIER_RETURNS.get(tier, '?')}")
              lines.append(f"        Job ID:  {_JOB_IDS.get(tier, 'see website')}")
          lines.append("")
  
      # Events registries
      registries = events.get("registries", [])
      if registries:
          lines.append("  Event Registry (indicator alerts):")
          for reg in registries:
              lines.append(f"    {reg['chain']}: {reg['address']}")
          lines.append("")
  
      lines.append(f"  Faucet (free trial): {FAUCET_ADDRESS}")
      lines.append("\nIMPORTANT: Use ERC-677 LINK only (0xb08976...).")
      lines.append("Bridged ERC-20 LINK (0x53e0bc...) does NOT work with Chainlink.")
      lines.append("Use PegSwap (pegswap.chain.link) to convert if needed.")
      return "\n".join(lines)

• src/pythia_oracle_mcp/server.py:400-401 (registration)

  Registration of 'get_contracts' as a FastMCP tool via the @mcp.tool() decorator on line 400.

  @mcp.tool()
  async def get_contracts() -> str:

• src/pythia_oracle_mcp/server.py:75-99 (helper)

  The _get_contracts() helper function that extracts normalized contract data (display_name, chain_id, explorer, operator, link_token, consumers) from the live feed-status.json data.

  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:65-72 (helper)

  The _parse_consumers() helper function that converts raw consumer contract data from display-name format ('Discovery (0.01 LINK)': '0x...') to normalized tier keys ('discovery': '0x...').

  def _parse_consumers(raw: dict) -> dict[str, str]:
      """Convert {"Discovery (0.01 LINK)": "0x..."} → {"discovery": "0x..."}."""
      parsed = {}
      for display_name, address in raw.items():
          tier = display_name.split()[0].lower() if display_name else ""
          if tier and address:
              parsed[tier] = address
      return parsed

• src/pythia_oracle_mcp/server.py:108-119 (helper)

  The _get_tier_fees() helper used by get_contracts to extract tier fee amounts from live data.

  def _get_tier_fees(data: dict) -> dict[str, float]:
      """Extract tier fees from live feed-status.json data.
  
      Raises RuntimeError if tiers section is missing.
      """
      tiers = data.get("tiers")
      if not tiers:
          raise RuntimeError(
              "feed-status.json is missing the tiers section. "
              "This is a structural error in the live data — check the data engine."
          )
      return {t["id"]: t["fee"] for t in tiers if "id" in t and "fee" in t}