get_current_power_price

Instructions

Implementation Reference

• mcp_server.py:497-632 (handler)

  The main handler function for the get_current_power_price tool. It fetches hourly day-ahead electricity spot prices for Nordic bidding zones. Supports Norwegian zones via hvakosterstrommen.no API and other zones via ENTSO-E API. Returns current hour price, today's summary (min/max/avg + hourly), and optionally tomorrow's prices.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True))
  async def get_current_power_price(
      zone: Annotated[str, "Bidding zone: NO1–NO5, SE1–SE4, DK1, DK2, or FI"] = "NO1",
      include_tomorrow: Annotated[bool, "Also fetch tomorrow's prices if available (published after 13:00 CET)"] = False,
  ) -> dict:
      """Fetch today's hourly day-ahead electricity spot prices for a Nordic bidding zone.
  
      Use this for current and near-term (today/tomorrow) price queries. Do not use
      for historical price analysis — use search_filings with report_type='macro_summary'
      and a date reference in the query for that purpose.
      Tomorrow's prices are published by NordPool around 13:00 CET; requests before
      that time will return "not yet available" for the tomorrow field.
  
      All zones return prices in EUR/kWh (NordPool day-ahead, native currency).
      Norwegian zones (NO1–NO5) use hvakosterstrommen.no; all other zones use ENTSO-E.
  
      Args:
          zone:             Bidding zone code. Options:
                            NO1 (East/Oslo), NO2 (Southwest), NO3 (Central/Trondheim),
                            NO4 (North), NO5 (West/Bergen),
                            SE1–SE4, DK1, DK2, FI.
          include_tomorrow: Set to True to also fetch tomorrow's hourly prices if
                            already published (default False).
  
      Returns:
          Dict containing zone, date, current_hour_utc, current price, and a 'today'
          summary with min/max/avg and the full hourly list. Includes a 'tomorrow'
          key if include_tomorrow=True. Returns {'error': '<message>'} if price data
          is unavailable for the requested zone or date.
      """
      import xml.etree.ElementTree as ET
      from datetime import date, datetime, timedelta, timezone
  
      zone = zone.upper()
      if zone not in _ZONE_EIC:
          return {"error": f"Unknown zone '{zone}'. Valid zones: {sorted(_ZONE_EIC)}"}
  
      today = date.today()
  
      async def fetch_no(d: date) -> list[dict] | None:
          url = (f"https://www.hvakosterstrommen.no/api/v1/prices"
                 f"/{d.year}/{d.month:02d}-{d.day:02d}_{zone}.json")
          async with httpx.AsyncClient(timeout=10) as client:
              resp = await client.get(url)
              if resp.status_code == 404:
                  return None
              resp.raise_for_status()
              raw = resp.json()
          return [
              {"time": h["time_start"][11:16],
               "EUR_per_kWh": round(h["EUR_per_kWh"], 5)}
              for h in raw
          ]
  
      async def fetch_entsoe(d: date) -> list[dict] | None:
          if not _ENTSOE_KEY:
              return None
          eic = _ZONE_EIC[zone]
          period_start = d.strftime("%Y%m%d") + "0000"
          period_end   = (d + timedelta(days=1)).strftime("%Y%m%d") + "0000"
          params = {
              "securityToken": _ENTSOE_KEY,
              "documentType": "A44",
              "in_Domain": eic, "out_Domain": eic,
              "periodStart": period_start, "periodEnd": period_end,
          }
          async with httpx.AsyncClient(timeout=15) as client:
              resp = await client.get("https://web-api.tp.entsoe.eu/api", params=params)
              if resp.status_code != 200 or "<Acknowledgement" in resp.text:
                  return None
          ns = {"ns": "urn:iec62325.351:tc57wg16:451-3:publicationdocument:7:3"}
          root = ET.fromstring(resp.text)
          # Collect raw points keyed by UTC datetime
          raw: dict[datetime, list[float]] = {}
          for ts in root.findall(".//ns:TimeSeries", ns):
              resolution = ts.findtext(".//ns:resolution", namespaces=ns)
              if resolution == "PT60M":
                  step = timedelta(hours=1)
              elif resolution == "PT15M":
                  step = timedelta(minutes=15)
              else:
                  continue
              start_str = ts.findtext(".//ns:timeInterval/ns:start", namespaces=ns)
              if not start_str:
                  continue
              start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
              for pt in ts.findall(".//ns:Point", ns):
                  pos   = int(pt.findtext("ns:position", namespaces=ns))
                  price = float(pt.findtext("ns:price.amount", namespaces=ns))
                  dt    = start_dt + (pos - 1) * step
                  # Round down to hour for aggregation
                  hour_dt = dt.replace(minute=0, second=0, microsecond=0)
                  raw.setdefault(hour_dt, []).append(price)
          if not raw:
              return None
          return [
              {"time": dt.strftime("%H:%M"),
               "EUR_per_kWh": round(sum(prices) / len(prices) / 1000, 5)}
              for dt, prices in sorted(raw.items())
          ]
  
      def summarize(hours: list[dict]) -> dict:
          prices = [h["EUR_per_kWh"] for h in hours]
          result = {
              "min_EUR_per_kWh": round(min(prices), 5),
              "max_EUR_per_kWh": round(max(prices), 5),
              "avg_EUR_per_kWh": round(sum(prices) / len(prices), 5),
              "hours": hours,
          }
          return result
  
      fetch = fetch_no if zone in _NO_ZONES else fetch_entsoe
      today_data = await fetch(today)
      if not today_data:
          return {"error": f"No price data available for zone {zone} on {today}"}
  
      now_h = datetime.now(timezone.utc).hour
      current = next(
          (h for h in today_data if int(h["time"][:2]) == now_h),
          today_data[0],
      )
  
      result = {
          "zone": zone,
          "date": today.isoformat(),
          "current_hour_utc": f"{now_h:02d}:00",
          "current": current,
          "today": summarize(today_data),
      }
  
      if include_tomorrow:
          tomorrow_data = await fetch(today + timedelta(days=1))
          result["tomorrow"] = summarize(tomorrow_data) if tomorrow_data else "not yet available"
  
      _log.info(f'get_current_power_price zone="{zone}" current={current["EUR_per_kWh"]} EUR/kWh')
      return result

• mcp_server.py:498-500 (schema)

  Input parameter definitions for the tool: 'zone' (str, default NO1) with valid bidding zone codes, and 'include_tomorrow' (bool, default False) to optionally fetch tomorrow's prices.

  async def get_current_power_price(
      zone: Annotated[str, "Bidding zone: NO1–NO5, SE1–SE4, DK1, DK2, or FI"] = "NO1",
      include_tomorrow: Annotated[bool, "Also fetch tomorrow's prices if available (published after 13:00 CET)"] = False,

• mcp_server.py:497-497 (registration)

  Tool registration via @mcp.tool decorator with ToolAnnotations(readOnlyHint=True, openWorldHint=True).

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True))

• mcp_server.py:482-496 (helper)

  Supporting data structures for the handler: _ENTSOE_KEY environment variable, _ZONE_EIC mapping of zone codes to ENTSO-E EIC codes, and _NO_ZONES set identifying Norwegian zones that use hvakosterstrommen.no.

  _ENTSOE_KEY = os.getenv("ENTSOE_API_KEY")
  
  _ZONE_EIC = {
      "NO1": "10YNO-1--------2", "NO2": "10YNO-2--------T",
      "NO3": "10YNO-3--------J", "NO4": "10YNO-4--------9",
      "NO5": "10Y1001A1001A48H",
      "SE1": "10Y1001A1001A44P", "SE2": "10Y1001A1001A45N",
      "SE3": "10Y1001A1001A46L", "SE4": "10Y1001A1001A47J",
      "FI":  "10YFI-1--------U",
      "DK1": "10YDK-1--------W", "DK2": "10YDK-2--------M",
  }
  
  _NO_ZONES = {"NO1", "NO2", "NO3", "NO4", "NO5"}

• mcp_server.py:596-598 (helper)

  Internal summarize() helper that computes min/max/avg price from hourly data.

      ]
  
  def summarize(hours: list[dict]) -> dict: