Alpha ESS MCP Server

Instructions

Get the latest real-time power data for a specific Alpha ESS system.
Returns structured snapshot with clear field names and units.
If no serial provided, auto-selects if only one system exists.

Args:
    serial: The serial number of the Alpha ESS system (optional)
    
Returns:
    dict: Enhanced response with structured real-time power data

Implementation Reference

• main.py:451-518 (handler)

  The primary handler function for the MCP tool 'get_last_power_data'. It handles serial auto-discovery, API call to client.getLastPowerData(serial), data structuring, error handling, and response formatting using @mcp.tool() decorator for registration.

  @mcp.tool()
  async def get_last_power_data(serial: Optional[str] = None) -> dict[str, Any]:
      """
      Get the latest real-time power data for a specific Alpha ESS system.
      Returns structured snapshot with clear field names and units.
      If no serial provided, auto-selects if only one system exists.
      
      Args:
          serial: The serial number of the Alpha ESS system (optional)
          
      Returns:
          dict: Enhanced response with structured real-time power data
      """
      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 create_enhanced_response(
                      success=False,
                      message=f"Serial auto-discovery failed: {serial_info['message']}",
                      raw_data=None,
                      data_type="snapshot",
                      metadata={"available_systems": serial_info.get('systems', [])}
                  )
              serial = serial_info['serial']
  
          app_id, app_secret = get_alpha_credentials()
          client = alphaess(app_id, app_secret)
  
          # Get last power data
          power_data = await client.getLastPowerData(serial)
  
          # Structure the snapshot data
          structured = structure_snapshot_data(power_data)
  
          return create_enhanced_response(
              success=True,
              message=f"Successfully retrieved last power data for {serial}",
              raw_data=power_data,
              data_type="snapshot",
              serial_used=serial,
              metadata={
                  "snapshot_type": "real_time_power",
                  "units": {"power": "W", "soc": "%"}
              },
              structured_data=structured
          )
  
      except ValueError as e:
          return create_enhanced_response(
              success=False,
              message=f"Configuration error: {str(e)}",
              raw_data=None,
              data_type="snapshot"
          )
      except Exception as e:
          return create_enhanced_response(
              success=False,
              message=f"Error retrieving power data: {str(e)}",
              raw_data=None,
              data_type="snapshot"
          )
      finally:
          if client:
              await client.close()

• models.py:58-64 (schema)

  Dataclass defining the structured Snapshot output schema used by get_last_power_data tool for real-time power data.

  class Snapshot:
      solar: Dict[str, Any]
      battery: Dict[str, Any]
      grid: Dict[str, Any]
      load: Dict[str, Any]
      ev_charging: Dict[str, Any]
      units: Dict[str, str]

• main.py:180-226 (helper)

  Helper function that transforms raw API power data into the structured Snapshot dataclass instance used by the tool.

  def structure_snapshot_data(raw_data: Dict[str, Any]) -> Snapshot:
      """Structure real-time snapshot data with clear field names"""
      return Snapshot(
          solar={
              "total_power": raw_data.get('ppv', 0),
              "panels": {
                  "panel_1": raw_data.get('ppvDetail', {}).get('ppv1', 0),
                  "panel_2": raw_data.get('ppvDetail', {}).get('ppv2', 0),
                  "panel_3": raw_data.get('ppvDetail', {}).get('ppv3', 0),
                  "panel_4": raw_data.get('ppvDetail', {}).get('ppv4', 0)
              }
          },
          battery={
              "state_of_charge": raw_data.get('soc', 0),
              "power": raw_data.get('pbat', 0)  # Positive = charging, Negative = discharging
          },
          grid={
              "total_power": raw_data.get('pgrid', 0),  # Positive = importing, Negative = exporting
              "phases": {
                  "l1_power": raw_data.get('pgridDetail', {}).get('pmeterL1', 0),
                  "l2_power": raw_data.get('pgridDetail', {}).get('pmeterL2', 0),
                  "l3_power": raw_data.get('pgridDetail', {}).get('pmeterL3', 0)
              }
          },
          load={
              "total_power": raw_data.get('pload', 0),
              "phases": {
                  "l1_real": raw_data.get('prealL1', 0),
                  "l2_real": raw_data.get('prealL2', 0),
                  "l3_real": raw_data.get('prealL3', 0)
              }
          },
          ev_charging={
              "total_power": raw_data.get('pev', 0),
              "stations": {
                  "ev1": raw_data.get('pevDetail', {}).get('ev1Power', 0),
                  "ev2": raw_data.get('pevDetail', {}).get('ev2Power', 0),
                  "ev3": raw_data.get('pevDetail', {}).get('ev3Power', 0),
                  "ev4": raw_data.get('pevDetail', {}).get('ev4Power', 0)
              }
          },
          units={
              "power": "W",
              "soc": "%"
          }
      )