get_carbon_intensity
Calculate CO2 emissions per unit of electricity generated from Spain's grid data. Retrieve carbon intensity time series for specified periods to analyze energy mix cleanliness.
Instructions
Get carbon intensity over time (gCO2/kWh).
Calculates CO2 emissions per unit of electricity generated. Lower values indicate cleaner energy mix.
Args: start_date: Start datetime in ISO format (YYYY-MM-DDTHH:MM) end_date: End datetime in ISO format (YYYY-MM-DDTHH:MM) time_granularity: Time aggregation (raw, hour, day, fifteen_minutes)
Returns: JSON string with carbon intensity time series and statistics.
Examples: Get hourly carbon intensity for a day: >>> await get_carbon_intensity("2025-10-08T00:00", "2025-10-08T23:59", "hour")
Get daily carbon intensity for a week:
>>> await get_carbon_intensity("2025-10-01T00:00", "2025-10-07T23:59", "day")Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| start_date | Yes | ||
| end_date | Yes | ||
| time_granularity | No | hour |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- Core handler implementing get_carbon_intensity tool. Fetches CO2 emissions and generation data using specific REE indicators, calculates gCO2/kWh intensity, computes statistics (min/max/avg), and returns formatted JSON response with interpretation guidelines.
@mcp.tool() async def get_carbon_intensity( start_date: str, end_date: str, time_granularity: str = "hour" ) -> str: """Get carbon intensity over time (gCO2/kWh). Calculates CO2 emissions per unit of electricity generated. Lower values indicate cleaner energy mix. Args: start_date: Start datetime in ISO format (YYYY-MM-DDTHH:MM) end_date: End datetime in ISO format (YYYY-MM-DDTHH:MM) time_granularity: Time aggregation (raw, hour, day, fifteen_minutes) Returns: JSON string with carbon intensity time series and statistics. Examples: Get hourly carbon intensity for a day: >>> await get_carbon_intensity("2025-10-08T00:00", "2025-10-08T23:59", "hour") Get daily carbon intensity for a week: >>> await get_carbon_intensity("2025-10-01T00:00", "2025-10-07T23:59", "day") """ try: async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() # Get CO2 emissions co2_request = GetIndicatorDataRequest( indicator_id=IndicatorIDs.CO2_EMISSIONS.id, start_date=start_date, end_date=end_date, time_granularity=time_granularity, ) co2_response = await use_case.execute(co2_request) co2_data = co2_response.model_dump() # Get generation/demand demand_request = GetIndicatorDataRequest( indicator_id=IndicatorIDs.REAL_DEMAND_SUM_GENERATION.id, start_date=start_date, end_date=end_date, time_granularity=time_granularity, ) demand_response = await use_case.execute(demand_request) demand_data = demand_response.model_dump() # Calculate carbon intensity (gCO2/kWh) co2_values = co2_data.get("values", []) demand_values = demand_data.get("values", []) intensity_values = [] for co2_val, demand_val in zip(co2_values, demand_values, strict=False): if demand_val["value"] > 0: # Convert tCO2 to gCO2, MW to MWh (for hourly data they're equivalent) intensity_g_per_kwh = (co2_val["value"] * 1_000_000) / (demand_val["value"] * 1_000) intensity_values.append( { "datetime": co2_val["datetime"], "carbon_intensity_g_per_kwh": round(intensity_g_per_kwh, 2), "co2_tonnes": co2_val["value"], "generation_mw": demand_val["value"], } ) # Calculate statistics stats = {} if intensity_values: intensities = [v["carbon_intensity_g_per_kwh"] for v in intensity_values] stats = { "min_g_per_kwh": min(intensities), "max_g_per_kwh": max(intensities), "avg_g_per_kwh": round(sum(intensities) / len(intensities), 2), "count": len(intensities), } result = { "period": {"start": start_date, "end": end_date, "granularity": time_granularity}, "values": intensity_values, "statistics": stats, "interpretation": { "excellent": "< 50 g/kWh", "good": "50-150 g/kWh", "moderate": "150-300 g/kWh", "poor": "> 300 g/kWh", }, } return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error calculating carbon intensity") - Defines the CO2_EMISSIONS indicator ID (10355) used as data source for carbon calculations in get_carbon_intensity.
CO2_EMISSIONS = IndicatorMetadata( id=10355, name="CO₂ Emissions", category=IndicatorCategory.EMISSION, description="CO₂ emissions associated with real-time generation (tCO₂eq)", ) - Defines the REAL_DEMAND_SUM_GENERATION indicator ID (10004) used as denominator (generation proxy) for carbon intensity calculation.
REAL_DEMAND_SUM_GENERATION = IndicatorMetadata( id=10004, name="Real Demand Sum of Generation", category=IndicatorCategory.DEMAND, description="Real demand calculated from generation sum", ) - src/ree_mcp/interface/mcp_server.py:28-29 (registration)FastMCP server initialization where all @mcp.tool() decorators register the get_carbon_intensity tool.
# Initialize MCP server mcp = FastMCP("REE MCP Server", dependencies=["httpx", "pydantic", "pydantic-settings"]) - Docstring providing input/output schema description, parameters, return type, and usage examples for MCP tool schema generation.
"""Get carbon intensity over time (gCO2/kWh). Calculates CO2 emissions per unit of electricity generated. Lower values indicate cleaner energy mix. Args: start_date: Start datetime in ISO format (YYYY-MM-DDTHH:MM) end_date: End datetime in ISO format (YYYY-MM-DDTHH:MM) time_granularity: Time aggregation (raw, hour, day, fifteen_minutes) Returns: JSON string with carbon intensity time series and statistics. Examples: Get hourly carbon intensity for a day: >>> await get_carbon_intensity("2025-10-08T00:00", "2025-10-08T23:59", "hour") Get daily carbon intensity for a week: >>> await get_carbon_intensity("2025-10-01T00:00", "2025-10-07T23:59", "day") """