electricity-emission
Calculate carbon emissions from electricity consumption by inputting energy amount, unit, and region-specific grid mix data using Climatiq MCP Server.
Instructions
Calculate carbon emissions from electricity consumption based on energy amount and regional grid mix.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| energy | Yes | Amount of energy consumed | |
| energy_unit | Yes | Energy unit (kWh, MWh, etc.) | |
| region | No | Region code (e.g., US, GB, FR) representing the electricity grid location | US |
Implementation Reference
- src/climatiq_mcp_server/server.py:248-249 (registration)Registration and dispatch to the electricity-emission tool handler function.elif name == "electricity-emission": result_text, result, cache_id = await electricity_emission_tool(config, arguments, server, climatiq_request)
- The main tool handler function that routes 'electricity-emission' calls to its specific implementation and handles response formatting and caching.async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ Handle tool execution requests for Climatiq API operations. """ if not arguments: raise ValueError("Missing arguments") result_text = "" result = None cache_id = None # Route to the appropriate tool handler if name == "set-api-key": result_text = await set_api_key_tool(config, arguments, server, climatiq_request) elif name == "electricity-emission": result_text, result, cache_id = await electricity_emission_tool(config, arguments, server, climatiq_request) elif name == "travel-emission": result_text, result, cache_id = await travel_emission_tool(config, arguments, server, climatiq_request) elif name == "search-emission-factors": result_text, result, cache_id = await search_emission_factors_tool(config, arguments, server, climatiq_request) elif name == "custom-emission-calculation": result_text, result, cache_id = await custom_emission_calculation_tool(config, arguments, server, climatiq_request) elif name == "cloud-computing-emission": result_text, result, cache_id = await cloud_computing_emission_tool(config, arguments, server, climatiq_request) elif name == "freight-emission": result_text, result, cache_id = await freight_emission_tool(config, arguments, server, climatiq_request) elif name == "procurement-emission": result_text, result, cache_id = await procurement_emission_tool(config, arguments, server, climatiq_request) elif name == "hotel-emission": result_text, result, cache_id = await hotel_emission_tool(config, arguments, server, climatiq_request) elif name == "travel-spend": result_text, result, cache_id = await travel_spend_tool(config, arguments, server, climatiq_request) else: raise ValueError(f"Unknown tool: {name}") # Store result in cache if available if result and cache_id: cached_results[cache_id] = result # Don't send notification to avoid potential issues during testing # await server.request_context.session.send_resource_list_changed() return [ types.TextContent( type="text", text=result_text, ) ]
- src/climatiq_mcp_server/server.py:224-230 (registration)Tool listing handler that returns tool definitions (including schema for electricity-emission) via get_tool_definitions().@server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools for interacting with the Climatiq API. """ return get_tool_definitions()
- Core helper function used by all emission calculation tools to make API requests to Climatiq, including electricity-emission.async def climatiq_request(endpoint: str, json_data: dict, method: str = "POST") -> dict: """Make a request to the Climatiq API.""" if not config["api_key"]: raise ValueError("Climatiq API key not set. Please configure it using the set-api-key tool.") url = f"{config['base_url']}{endpoint}" headers = { "Authorization": f"Bearer {config['api_key']}", "Content-Type": "application/json" } logger.debug(f"Request URL: {url}") logger.debug(f"Request method: {method}") logger.debug(f"Request headers: {headers}") logger.debug(f"Request data: {json.dumps(json_data, indent=2)}") try: async with httpx.AsyncClient(timeout=60.0) as client: if method.upper() == "POST": response = await client.post(url, headers=headers, json=json_data) else: response = await client.get(url, headers=headers, params=json_data) logger.debug(f"Response status code: {response.status_code}") if response.status_code != 200: error_detail = response.text try: error_json = response.json() if "error" in error_json: error_detail = error_json["error"] elif "message" in error_json: error_detail = error_json["message"] except: pass logger.error(f"API request failed with status {response.status_code}: {error_detail}") raise ValueError(f"API request failed with status {response.status_code}: {error_detail}") result = response.json() logger.debug(f"Response data: {json.dumps(result, indent=2)}") return result except httpx.TimeoutException: logger.error("Request to Climatiq API timed out") raise ValueError("Request to Climatiq API timed out. Please try again later.") except httpx.RequestError as e: logger.error(f"Request error: {str(e)}") raise ValueError(f"Failed to connect to Climatiq API: {str(e)}") except Exception as e: logger.error(f"Unexpected error during API request: {str(e)}", exc_info=True) raise ValueError(f"Error during Climatiq API request: {str(e)}")
- utils/climatiq_cli.py:36-84 (helper)CLI implementation of electricity emission calculation, likely mirrors the tool logic with specific activity_id and parameters.async def calculate_electricity_emission(energy, energy_unit, region): """Calculate emissions from electricity consumption""" logger.info(f"Calculating emissions for {energy} {energy_unit} of electricity in {region}") # Construct request data request_data = { "emission_factor": { "activity_id": "electricity-supply_grid-source_residual_mix", "data_version": config["data_version"], "region": region }, "parameters": { "energy": energy, "energy_unit": energy_unit } } # Make the API request result = await climatiq_request("/data/v1/estimate", request_data) # Format the result co2e = result.get("co2e", 0) co2e_unit = result.get("co2e_unit", "kg") # Create a user-friendly output output = f"\nElectricity consumption of {energy} {energy_unit} in {region} results in {co2e} {co2e_unit} of CO2e emissions.\n" if "emission_factor" in result and result["emission_factor"].get("name"): ef_name = result["emission_factor"]["name"] output += f"Emission factor used: {ef_name}\n" # Add source information if "emission_factor" in result: ef = result["emission_factor"] if "source" in ef: output += f"Source: {ef.get('source')}" if "source_dataset" in ef: output += f" - {ef.get('source_dataset')}" output += "\n" if "year" in ef: output += f"Year of data: {ef.get('year')}\n" print(output) # Ask if user wants to see the full JSON response if input("Would you like to see the full API response? (y/n): ").lower() == 'y': print(json.dumps(result, indent=2)) return result