list_subscriptions

Instructions

Implementation Reference

• src/pythia_oracle_mcp/server.py:1275-1332 (handler)

  The main handler function for the 'list_subscriptions' tool. It takes an owner_address parameter, fetches live data from feed-status.json, filters subscriptions by owner, and returns a formatted report of active Pythia Event subscriptions.

  async def list_subscriptions(owner_address: str) -> str:
      """Enumerate active Pythia Event subscriptions owned by an address.
  
      Returns every subscription where active=true (not yet fired, expired, or
      cancelled). Without this tool, dApps and dashboards have to replay every
      SubscriptionCreated log from the registry deploy block to discover what
      an owner is currently subscribed to.
  
      Args:
          owner_address: subscriber wallet address ('0x...', case-insensitive).
  
      Returns:
          Multi-section report listing each active subscription with feed name,
          condition + threshold, expiry, registry address, and creation tx.
      """
      addr = owner_address.strip().lower()
      if not addr.startswith("0x"):
          addr = "0x" + addr
  
      data = await _fetch_data()
      events = data.get("events", {}) if data else {}
      subs = events.get("subscriptions", [])
  
      matched = [s for s in subs if (s.get("owner") or "").lower() == addr]
  
      if not matched:
          return (
              f"No active subscriptions for {addr}.\n"
              "This means: never subscribed on a tracked registry, all "
              "subscriptions already fired / expired / cancelled, or the\n"
              "off-chain sync hasn't caught up yet (event_sync.py runs every "
              "2 minutes)."
          )
  
      out = [
          f"Active Pythia Event subscriptions for {addr}",
          f"Count: {len(matched)}",
          "",
      ]
      for i, s in enumerate(matched, 1):
          out.append(f"[{i}] sub_id={s.get('sub_id')} on {s.get('source_chain')}")
          out.append(f"    Registry:   {s.get('registry')}")
          feed_id = s.get("feed_id") or ""
          feed_id_short = (feed_id[:12] + "…") if len(feed_id) > 14 else feed_id
          out.append(f"    Feed:       {s.get('feed_name', '?')} ({feed_id_short})")
          out.append(f"    Condition:  {s.get('condition')} {s.get('threshold')}")
          out.append(f"    Feed chain: {s.get('feed_chain')}")
          out.append(f"    Expires:    {s.get('expires_at')}")
          out.append(f"    Created:    {s.get('created_at')}")
          if s.get("tx_hash"):
              out.append(f"    Tx:         {s.get('tx_hash')}")
          out.append("")
  
      out.append(
          "Cancel early via cancelSubscription(sub_id) on the registry — "
          "refunds remaining whole-day LINK."
      )
      return "\n".join(out)

• src/pythia_oracle_mcp/server.py:1274-1276 (registration)

  Registration of 'list_subscriptions' as an MCP tool via the @mcp.tool() decorator on line 1274.

  @mcp.tool()
  async def list_subscriptions(owner_address: str) -> str:
      """Enumerate active Pythia Event subscriptions owned by an address.