Alpha ESS MCP Server

Instructions

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