Alpha ESS MCP Server

Instructions

Get energy data for a specific date and Alpha ESS system.
If no serial provided, auto-selects if only one system exists.

Args:
    query_date: Date in YYYY-MM-DD format
    serial: The serial number of the Alpha ESS system (optional)
    
Returns:
    dict: Energy data for the specified date with success status

Implementation Reference

• main.py:592-663 (handler)

  Primary handler for the MCP tool 'get_one_date_energy_data'. Fetches daily power data from Alpha ESS API, structures it into TimeSeries summary, maps to energy metrics (epv, eCharge, etc.), handles serial auto-detection and errors.

  @mcp.tool()
  async def get_one_date_energy_data(query_date: str, serial: Optional[str] = None) -> dict[str, Any]:
      """
      Get energy data for a specific date and Alpha ESS system.
      If no serial provided, auto-selects if only one system exists.
      
      Args:
          query_date: Date in YYYY-MM-DD format
          serial: The serial number of the Alpha ESS system (optional)
          
      Returns:
          dict: Energy data for the specified date with success status
      """
      client = None
      try:
          # Auto-discover serial if not provided
          if not serial:
              serial_info = await get_default_serial()
              if not serial_info['success'] or not serial_info['serial']:
                  return {
                      "success": False,
                      "message": f"Serial auto-discovery failed: {serial_info['message']}",
                      "data": None,
                      "available_systems": serial_info.get('systems', [])
                  }
              serial = serial_info['serial']
  
          app_id, app_secret = get_alpha_credentials()
          client = alphaess(app_id, app_secret)
  
          # Get one day power data
          power_data = await client.getOneDayPowerBySn(serial, query_date)
  
          # Structure the timeseries data
          structured = structure_timeseries_data(power_data, serial)
          
          summary = structured.summary
          
          # Map the summary to the expected output format
          energy_data = {
              "eCharge": summary.battery.get('total_charge_kwh', 0),
              "epv": summary.solar.get('total_generation_kwh', 0),
              "eOutput": summary.grid.get('total_feedin_kwh', 0),
              "eInput": summary.grid.get('total_import_kwh', 0),
              "eGridCharge": summary.grid.get('total_charge_from_grid_kwh', 0),
              "eDischarge": summary.battery.get('total_discharge_kwh', 0),
              "eChargingPile": 0 # This data is not available in the summary
          }
  
          return {
              "success": True,
              "message": f"Successfully retrieved energy data for {serial} on {query_date}",
              "data": energy_data,
              "serial_used": serial
          }
  
      except ValueError as e:
          return {
              "success": False,
              "message": f"Configuration or parameter error: {str(e)}",
              "data": None
          }
      except Exception as e:
          return {
              "success": False,
              "message": f"Error retrieving energy data: {str(e)}",
              "data": None
          }
      finally:
          if client:
              await client.close()

• main.py:55-133 (helper)

  Key helper function that aggregates raw API timeseries data into hourly averages and computes summary statistics (e.g., total_generation_kwh for solar), directly used by the handler to process power data into energy metrics.

  def structure_timeseries_data(raw_data: List[Dict], serial: str) -> TimeSeries:
      """Convert inefficient timeseries to structured format with hourly aggregation"""
      if not raw_data:
          return TimeSeries(series=[], summary=TimeSeriesSummary(total_records=0, interval="1 hour", time_span_hours=0, solar={}, battery={}, grid={}, load={}))
  
      # Group data by hour
      hourly_data = {}
      for record in raw_data:
          timestamp = record.get('uploadTime', '')
          if not timestamp:
              continue
  
          # Extract hour from timestamp (assumes format like "2024-03-21 14:30:00")
          hour = timestamp[:13] + ":00:00"  # Truncate to hour
  
          if hour not in hourly_data:
              hourly_data[hour] = {
                  "solar_power": [],
                  "load_power": [],
                  "battery_soc": [],
                  "grid_feedin": [],
                  "grid_import": [],
                  "ev_charging": []
              }
  
          # Collect all values for this hour
          hourly_data[hour]["solar_power"].append(record.get('ppv', 0))
          hourly_data[hour]["load_power"].append(record.get('load', 0))
          hourly_data[hour]["battery_soc"].append(record.get('cbat', 0))
          hourly_data[hour]["grid_feedin"].append(record.get('feedIn', 0))
          hourly_data[hour]["grid_import"].append(record.get('gridCharge', 0))
          hourly_data[hour]["ev_charging"].append(record.get('pchargingPile', 0))
  
      # Convert hourly data to averages
      series_entries = []
      for hour, data in sorted(hourly_data.items()):
          series_entries.append(TimeSeriesEntry(
              timestamp=hour,
              solar_power=round(sum(data["solar_power"]) / len(data["solar_power"])) if data["solar_power"] else 0,
              load_power=round(sum(data["load_power"]) / len(data["load_power"])) if data["load_power"] else 0,
              battery_soc=round(sum(data["battery_soc"]) / len(data["battery_soc"]), 1) if data["battery_soc"] else 0,
              grid_feedin=round(sum(data["grid_feedin"]) / len(data["grid_feedin"])) if data["grid_feedin"] else 0,
              grid_import=round(sum(data["grid_import"]) / len(data["grid_import"])) if data["grid_import"] else 0,
              ev_charging=round(sum(data["ev_charging"]) / len(data["ev_charging"])) if data["ev_charging"] else 0
          ))
  
      # Calculate summary statistics using hourly averages
      solar_values = [r.solar_power for r in series_entries]
      load_values = [r.load_power for r in series_entries]
      battery_values = [r.battery_soc for r in series_entries]
      feedin_values = [r.grid_feedin for r in series_entries]
  
      summary = TimeSeriesSummary(
          total_records=len(series_entries),
          interval="1 hour",
          time_span_hours=len(series_entries),
          solar={
              "peak_power": max(solar_values) if solar_values else 0,
              "avg_power": round(sum(solar_values) / len(solar_values)) if solar_values else 0,
              "total_generation_kwh": round(sum(solar_values) / 1000, 2)  # Convert W to kWh
          },
          battery={
              "max_soc": max(battery_values) if battery_values else 0,
              "min_soc": min(battery_values) if battery_values else 0,
              "avg_soc": round(sum(battery_values) / len(battery_values), 1) if battery_values else 0
          },
          grid={
              "total_feedin_kwh": round(sum(feedin_values) / 1000, 2),
              "peak_feedin": max(feedin_values) if feedin_values else 0
          },
          load={
              "peak_power": max(load_values) if load_values else 0,
              "avg_power": round(sum(load_values) / len(load_values)) if load_values else 0,
              "total_consumption_kwh": round(sum(load_values) / 1000, 2)
          }
      )
  
      return TimeSeries(series=series_entries, summary=summary)

• main.py:240-308 (helper)

  Helper function for auto-detecting the default system serial number by querying the ESS list, used when serial parameter is not provided in the tool call.

  async def get_default_serial() -> dict[str, Any]:
      """
      Get the default serial number to use. If only one system is registered,
      returns that serial. If multiple systems, returns list for user to choose.
      
      Returns:
          dict: Result with serial info
      """
      client = None
      try:
          app_id, app_secret = get_alpha_credentials()
          client = alphaess(app_id, app_secret)
  
          # Get ESS list
          ess_list = await client.getESSList()
  
          if not ess_list or len(ess_list) == 0:
              return {
                  "success": False,
                  "message": "No Alpha ESS systems found in your account",
                  "serial": None,
                  "systems": []
              }
  
          if len(ess_list) == 1:
              # Auto-select the only system
              system = ess_list[0]
              serial = system.get('sysSn') if isinstance(system, dict) else getattr(system, 'sysSn', None)
              return {
                  "success": True,
                  "message": f"Auto-selected single system: {serial}",
                  "serial": serial,
                  "systems": ess_list
              }
          else:
              # Multiple systems - return list for user choice
              systems_info = []
              for system in ess_list:
                  if isinstance(system, dict):
                      systems_info.append({
                          "serial": system.get('sysSn'),
                          "name": system.get('sysName', 'Unknown'),
                          "status": system.get('sysStatus', 'Unknown')
                      })
                  else:
                      systems_info.append({
                          "serial": getattr(system, 'sysSn', 'Unknown'),
                          "name": getattr(system, 'sysName', 'Unknown'),
                          "status": getattr(system, 'sysStatus', 'Unknown')
                      })
  
              return {
                  "success": True,
                  "message": f"Found {len(ess_list)} systems. Please specify which serial to use.",
                  "serial": None,
                  "systems": systems_info
              }
  
      except Exception as e:
          return {
              "success": False,
              "message": f"Error getting system list: {str(e)}",
              "serial": None,
              "systems": []
          }
      finally:
          if client:
              await client.close()

• models.py:27-31 (schema)

  Dataclass defining the structured TimeSeries format used internally by the handler's helper to represent processed timeseries data with series entries and summary.

  @dataclass
  class TimeSeries:
      series: List[TimeSeriesEntry]
      summary: TimeSeriesSummary