Alpha ESS MCP Server

Instructions

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": "%" } )