computeEmissions
Computes greenhouse gas emissions from structured activity data across Scope 1, 2, and 3. Use with generateEmissionsReport to save results.
Instructions
Computes greenhouse gas emissions from structured activity data.
IMPORTANT: After calling this tool, you MUST call generateEmissionsReport with the
full output of this tool. Do not present results to the user without first saving
the report files. The _required_next_step field in the response will remind you.
Args:
inputs_json: A JSON string containing categorized activity data.
Required fields vary by scope:
- Scope 1: stationary_combustion, mobile_combustion, refrigerants
- Scope 2: electricity_kwh, egrid_subregion, steam_mmbtu
- Scope 3: business_travel, employee_commuting, transportation, waste
Optional: annual_revenue, headcount (for scoring), period, source
Returns:
JSON string containing computed emissions by scope, totals, breakdown,
carbon intensity scores, and a _required_next_step instruction.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| inputs_json | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The MCP tool handler function 'computeEmissions' — decorated with @mcp.tool(). It parses the JSON input string, calls compute_all() from compute_emissions.py, adds a _required_next_step instruction, and returns the result as a JSON string.
@mcp.tool() def computeEmissions(inputs_json: str) -> str: """ Computes greenhouse gas emissions from structured activity data. IMPORTANT: After calling this tool, you MUST call generateEmissionsReport with the full output of this tool. Do not present results to the user without first saving the report files. The _required_next_step field in the response will remind you. Args: inputs_json: A JSON string containing categorized activity data. Required fields vary by scope: - Scope 1: stationary_combustion, mobile_combustion, refrigerants - Scope 2: electricity_kwh, egrid_subregion, steam_mmbtu - Scope 3: business_travel, employee_commuting, transportation, waste Optional: annual_revenue, headcount (for scoring), period, source Returns: JSON string containing computed emissions by scope, totals, breakdown, carbon intensity scores, and a _required_next_step instruction. """ try: inputs = json.loads(inputs_json) except json.JSONDecodeError as e: logger.error(f"Failed to parse inputs_json: {e}") return json.dumps({"error": f"Invalid JSON input: {e}"}) result = compute_all(inputs) result["_required_next_step"] = ( "MANDATORY: You must now call generateEmissionsReport with this full JSON output " "and an explicit output_dir (user's Desktop or a path they specified). " "Do not present results to the user until the report files are saved. " "Pass the entire contents of this response as the emissions_json argument." ) return json.dumps(result, indent=2) - src/carbon_footprint_mcp/server.py:1-6 (registration)Import of 'compute_all' from compute_emissions module and registration via @mcp.tool() decorator on line 217.
from mcp.server.fastmcp import FastMCP import json import logging from pathlib import Path from .compute_emissions import compute_all - The compute_all() function — the main computation engine invoked by computeEmissions. Accepts categorized activity data, delegates to compute_scope1, compute_scope2, compute_scope3, sums totals, computes scores, and returns structured JSON output with scope breakdowns, source distribution, and top sources.
def compute_all(raw_inputs): """ Main entry point. Accepts a dict of categorized activity data and returns a full emissions breakdown by scope, totals, and scoring. """ inputs = dict(raw_inputs) scope1 = compute_scope1(inputs) scope2 = compute_scope2(inputs) scope3 = compute_scope3(inputs) def _sum_scope(scope_results): total = 0.0 for key, item in scope_results.items(): val = item.get("value") if val is not None and item.get("label") == "computed": total += val return total scope1_total = _sum_scope(scope1) scope2_total = _sum_scope(scope2) scope3_total = _sum_scope(scope3) grand_total = scope1_total + scope2_total + scope3_total # Scope breakdown percentages breakdown = {} if grand_total > 0: breakdown = { "scope_1_pct": _safe_round(scope1_total / grand_total * 100, 1), "scope_2_pct": _safe_round(scope2_total / grand_total * 100, 1), "scope_3_pct": _safe_round(scope3_total / grand_total * 100, 1), } period_months, period_label = _infer_period_months(inputs) scores = compute_score(grand_total, inputs, period_months, period_label) # ── Source distribution (for pie chart — by bank transaction category) ── CATEGORY_DISPLAY = { "stationary_combustion": ("Natural Gas / Heating", "#f97316", "1"), "mobile_combustion": ("Fleet Fuel", "#ef4444", "1"), "refrigerants": ("Refrigerants", "#f59e0b", "1"), "electricity": ("Electricity", "#3b82f6", "2"), "steam_heat": ("Steam & Heat", "#0ea5e9", "2"), "transportation": ("Shipping & Freight", "#06b6d4", "3"), "waste": ("Waste Disposal", "#84cc16", "3"), "business_travel": ("Business Travel", "#8b5cf6", "3"), "employee_commuting": ("Employee Commuting", "#ec4899", "3"), } source_distribution = [] all_scopes = {**scope1, **scope2, **scope3} for cat_key, cat_data in all_scopes.items(): val = cat_data.get("value") if val is not None and val > 0 and cat_data.get("label") == "computed": display_name, color, scope_num = CATEGORY_DISPLAY.get( cat_key, (cat_key.replace("_", " ").title(), "#a3a3a3", "?") ) pct = _safe_round(val / grand_total * 100, 1) if grand_total > 0 else 0 source_distribution.append({ "category": cat_key, "display_name": display_name, "scope": f"Scope {scope_num}", "kg_co2e": _safe_round(val), "metric_tons_co2e": _safe_round(val / 1000, 2), "pct_of_total": pct, "color": color, }) # Sort by emissions descending source_distribution.sort(key=lambda x: x["kg_co2e"], reverse=True) # Top sources (ranked table data) top_sources = [] for rank, src in enumerate(source_distribution, 1): top_sources.append({ "rank": rank, "category": src["display_name"], "scope": src["scope"], "metric_tons_co2e": src["metric_tons_co2e"], "pct_of_total": src["pct_of_total"], "color": src["color"], }) return { "source": inputs.get("source", "manual"), "period": inputs.get("period", "Not specified"), "egrid_subregion": inputs.get("egrid_subregion", "US Average"), "scope_1": { "total_kg_co2e": _safe_round(scope1_total), "total_metric_tons_co2e": _safe_round(scope1_total / 1000, 2), "categories": scope1, }, "scope_2": { "total_kg_co2e": _safe_round(scope2_total), "total_metric_tons_co2e": _safe_round(scope2_total / 1000, 2), "categories": scope2, }, "scope_3": { "total_kg_co2e": _safe_round(scope3_total), "total_metric_tons_co2e": _safe_round(scope3_total / 1000, 2), "categories": scope3, }, "totals": { "total_kg_co2e": _safe_round(grand_total), "total_metric_tons_co2e": _safe_round(grand_total / 1000, 2), "breakdown": breakdown, }, "scores": scores, "source_distribution": source_distribution, "top_sources": top_sources, } - compute_scope1 (direct emissions: stationary combustion, mobile combustion, refrigerant leakage) — helper called by compute_all.
def compute_scope1(inputs): """Compute Scope 1 (direct) emissions.""" results = {} # --- Stationary Combustion --- stationary_items = inputs.get("stationary_combustion", []) if isinstance(stationary_items, list) and stationary_items: total_co2e = 0.0 details = [] for item in stationary_items: fuel = item.get("fuel_type") quantity = _to_number(item.get("quantity")) quantity_unit = item.get("quantity_unit", "mmbtu") # mmbtu or native unit if not fuel or quantity is None: continue factors = lookup_stationary(fuel) if not factors: details.append({"fuel": fuel, "error": f"Unknown fuel type: {fuel}"}) continue # Calculate based on unit if quantity_unit.lower() in ("mmbtu", "mm_btu"): co2_kg = quantity * factors["co2_per_mmbtu"] ch4_g = quantity * factors["ch4_per_mmbtu"] n2o_g = quantity * factors["n2o_per_mmbtu"] else: # Use per-unit factors (gallon, scf, short_ton) co2_kg = quantity * (factors["co2_per_unit"] or 0) ch4_g = quantity * (factors["ch4_per_unit"] or 0) n2o_g = quantity * (factors["n2o_per_unit"] or 0) co2e = _to_co2e(co2_kg, ch4_g, n2o_g) total_co2e += co2e details.append({ "fuel": fuel, "quantity": quantity, "unit": quantity_unit, "co2_kg": _safe_round(co2_kg), "ch4_g": _safe_round(ch4_g), "n2o_g": _safe_round(n2o_g), "co2e_kg": _safe_round(co2e), }) results["stationary_combustion"] = { "value": _safe_round(total_co2e), "unit": "kg CO2e", "label": "computed", "details": details, "missing_inputs": [], } else: results["stationary_combustion"] = _metric( "stationary_combustion", None, "kg CO2e", ["stationary_combustion (list of {fuel_type, quantity, quantity_unit})"] ) # --- Mobile Combustion --- mobile_items = inputs.get("mobile_combustion", []) if isinstance(mobile_items, list) and mobile_items: total_co2e = 0.0 details = [] for item in mobile_items: fuel = item.get("fuel_type") gallons = _to_number(item.get("gallons")) if not fuel or gallons is None: continue factors = lookup_mobile_co2(fuel) if not factors: details.append({"fuel": fuel, "error": f"Unknown fuel type: {fuel}"}) continue co2_kg = gallons * factors["kg_co2_per_unit"] # CH4/N2O from mobile are small; use default vehicle factors ch4_g = gallons * 0.0 # Needs vehicle-specific data n2o_g = gallons * 0.0 co2e = _to_co2e(co2_kg, ch4_g, n2o_g) total_co2e += co2e details.append({ "fuel": fuel, "gallons": gallons, "co2_kg": _safe_round(co2_kg), "co2e_kg": _safe_round(co2e), }) results["mobile_combustion"] = { "value": _safe_round(total_co2e), "unit": "kg CO2e", "label": "computed", "details": details, "missing_inputs": [], } else: results["mobile_combustion"] = _metric( "mobile_combustion", None, "kg CO2e", ["mobile_combustion (list of {fuel_type, gallons})"] ) # --- Refrigerant Leakage (if provided) --- refrigerant_items = inputs.get("refrigerants", []) if isinstance(refrigerant_items, list) and refrigerant_items: total_co2e = 0.0 details = [] for item in refrigerant_items: gas = item.get("gas") quantity_kg = _to_number(item.get("quantity_kg")) gwp = _to_number(item.get("gwp")) if gas and quantity_kg is not None and gwp is not None: co2e = quantity_kg * gwp total_co2e += co2e details.append({ "gas": gas, "quantity_kg": quantity_kg, "gwp": gwp, "co2e_kg": _safe_round(co2e), }) if details: results["refrigerants"] = { "value": _safe_round(total_co2e), "unit": "kg CO2e", "label": "computed", "details": details, "missing_inputs": [], } return results - compute_scope2 (purchased energy: electricity and steam/heat) — helper called by compute_all.
def compute_scope2(inputs): """Compute Scope 2 (purchased energy) emissions.""" results = {} # --- Purchased Electricity --- kwh = _to_number(inputs.get("electricity_kwh")) egrid_subregion = inputs.get("egrid_subregion", "US Average") if kwh is not None: factors = lookup_egrid(egrid_subregion) if factors is None: factors = lookup_egrid("US Average") egrid_subregion = "US Average (fallback)" # Convert lb/MWh to kg/kWh co2_kg = kwh * (factors["co2_lb_per_mwh"] / 2204.62) / 1000 * 1000 # lb/MWh -> kg/kWh # Correct: lb/MWh * kWh / 1000 (to MWh) / 2.20462 (lb to kg) mwh = kwh / 1000.0 co2_kg = mwh * factors["co2_lb_per_mwh"] / 2.20462 ch4_g = mwh * factors["ch4_lb_per_mwh"] / 2.20462 * 1000 # lb -> g n2o_g = mwh * factors["n2o_lb_per_mwh"] / 2.20462 * 1000 co2e = _to_co2e(co2_kg, ch4_g, n2o_g) results["electricity"] = { "value": _safe_round(co2e), "unit": "kg CO2e", "label": "computed", "details": { "kwh": kwh, "egrid_subregion": egrid_subregion, "co2_kg": _safe_round(co2_kg), "ch4_g": _safe_round(ch4_g), "n2o_g": _safe_round(n2o_g), }, "missing_inputs": [], } else: results["electricity"] = _metric( "electricity", None, "kg CO2e", ["electricity_kwh"] ) # --- Purchased Steam/Heat --- steam_mmbtu = _to_number(inputs.get("steam_mmbtu")) if steam_mmbtu is not None: co2_kg = steam_mmbtu * STEAM_HEAT["co2_kg_per_mmbtu"] ch4_g = steam_mmbtu * STEAM_HEAT["ch4_g_per_mmbtu"] n2o_g = steam_mmbtu * STEAM_HEAT["n2o_g_per_mmbtu"] co2e = _to_co2e(co2_kg, ch4_g, n2o_g) results["steam_heat"] = { "value": _safe_round(co2e), "unit": "kg CO2e", "label": "computed", "details": { "mmbtu": steam_mmbtu, "co2_kg": _safe_round(co2_kg), "ch4_g": _safe_round(ch4_g), "n2o_g": _safe_round(n2o_g), }, "missing_inputs": [], } return results