travel-emission
Calculate carbon emissions for travel by car, plane, or train using vehicle types, passenger counts, and distance data to assess environmental impact.
Instructions
Calculate emissions from passenger travel via car, plane, or train, with options for vehicle types and passenger count.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| mode | Yes | Travel mode (car, plane, train) | |
| distance | No | Distance traveled (optional if origin and destination are provided) | |
| distance_unit | No | Distance unit (km, mi) | km |
| passengers | No | Number of passengers (for car travel) | |
| vehicle_type | No | For car: small/medium/large; For plane: short/medium/long-haul/domestic/international; For train: electric/diesel | medium |
| fuel_type | No | For car: regular/electric/hybrid | regular |
| origin | No | Origin location for travel API (allows automatic distance calculation) | |
| destination | No | Destination location for travel API (allows automatic distance calculation) | |
| year | No | Year of travel for more accurate emission factors | |
| car_details | No | Detailed car specifications for more accurate calculations | |
| air_details | No | Detailed flight specifications for more accurate calculations |
Implementation Reference
- src/climatiq_mcp_server/tools.py:108-136 (handler)Main handler function for the 'travel-emission' MCP tool that dispatches to appropriate helper based on input (locations or direct distance).async def travel_emission_tool(config, arguments, server, climatiq_request): """ Calculate carbon emissions from passenger travel. This tool estimates the greenhouse gas emissions for various modes of passenger transportation: - Car travel (with options for vehicle size, fuel type, and passenger count) - Air travel (with options for flight length categories) - Train travel (with options for train types) It supports both direct distance input and origin-destination location lookup for automatic distance calculation. The emissions calculation accounts for the specific transportation mode characteristics and regional factors. """ mode = arguments.get("mode") if not mode: raise ValueError("Missing travel mode (car, plane, train)") # Check if we need to use the distance API or the direct calculation if (arguments.get("origin") and arguments.get("destination")) or mode.lower() == "flight": try: return await _travel_distance_api(config, arguments, climatiq_request) except Exception as e: raise ValueError(f"Error calculating travel emissions with locations: {str(e)}") else: try: return await _travel_estimate_api(config, arguments, climatiq_request) except Exception as e: raise ValueError(f"Error calculating travel emissions with distance: {str(e)}")
- Helper function handling travel emissions calculation using Climatiq's travel distance API for origin/destination inputs, supporting car, train, and flight.async def _travel_distance_api(config, arguments, climatiq_request): """Helper function to calculate travel emissions using the distance API with origin/destination.""" mode = arguments.get("mode").lower() origin = arguments.get("origin") destination = arguments.get("destination") # Check for flight-specific parameters if mode == "flight" or mode == "plane": # Get flight-specific parameters flight_type = arguments.get("vehicle_type", "international") passengers = arguments.get("passengers", 1) cabin_class = arguments.get("cabin_class", "economy") # For flights with specific origin/destination if origin and destination: request_data = { "legs": [ { "from": origin, "to": destination, "passengers": passengers, "cabin_class": cabin_class } ] } result = await climatiq_request("/travel/flights", request_data) # Format the result co2e = result.get("co2e", 0) co2e_unit = result.get("co2e_unit", "kg") distance = result.get("distance_value", 0) distance_unit = result.get("distance_unit", "km") origin_text = origin.get("query") if isinstance(origin, dict) else origin destination_text = destination.get("query") if isinstance(destination, dict) else destination result_text = f"Flight from {origin_text} to {destination_text} " result_text += f"({distance} {distance_unit}) with {passengers} passenger(s) in {cabin_class} class " result_text += f"results in {co2e} {co2e_unit} of CO2e emissions." # Store in cache cache_id = f"flight_{origin_text}_{destination_text}_{cabin_class}_{id(result)}" return result_text, result, cache_id # For other modes (car, train) with origin/destination if not origin or not destination: raise ValueError("Origin and destination are required for travel distance calculation") # Prepare car/train specific details vehicle_details = {} if mode == "car": car_details = arguments.get("car_details", {}) vehicle_type = car_details.get("car_type", arguments.get("vehicle_type", "medium")) fuel_type = car_details.get("fuel_type", arguments.get("fuel_type", "regular")) engine_size = car_details.get("engine_size") vehicle_details = { "vehicle_type": vehicle_type, "fuel_type": fuel_type } if engine_size: vehicle_details["engine_size"] = engine_size elif mode == "train": train_type = arguments.get("vehicle_type", "national_rail") vehicle_details = {"train_type": train_type} # Construct the request request_data = { "origin": origin, "destination": destination, "transport_method": mode, } # Add vehicle specific details if any if vehicle_details: request_data.update(vehicle_details) result = await climatiq_request("/travel/distance", request_data) # Format the result co2e = result.get("co2e", 0) co2e_unit = result.get("co2e_unit", "kg") distance = result.get("distance", {}).get("value", 0) distance_unit = result.get("distance", {}).get("unit", "km") origin_text = origin.get("query") if isinstance(origin, dict) else origin destination_text = destination.get("query") if isinstance(destination, dict) else destination result_text = f"{mode.capitalize()} travel from {origin_text} to {destination_text} " result_text += f"({distance} {distance_unit}) results in {co2e} {co2e_unit} of CO2e emissions." # Store in cache cache_id = f"{mode}_{origin_text}_{destination_text}_{id(result)}" return result_text, result, cache_id
- Helper function for direct distance-based travel emission estimation using specific emission factors for different modes.async def _travel_estimate_api(config, arguments, climatiq_request): """Helper function to calculate travel emissions using direct distance input.""" mode = arguments.get("mode").lower() distance = arguments.get("distance") distance_unit = arguments.get("distance_unit", "km") if not distance: raise ValueError(f"Missing distance for {mode} travel") # Construct the request based on mode activity_mapping = { "car": "passenger_vehicle-vehicle_type_car-fuel_source_na-engine_size_na-vehicle_age_na-vehicle_weight_na", "plane": "passenger_flight-route_type_international-aircraft_type_na-distance_na-class_na-rf_included", "train": "passenger_train-route_type_commuter_rail-fuel_source_na" } # Simpler implementation that matches the example in the notebook # and ensures compatibility with the API request_data = { "emission_factor": { "activity_id": activity_mapping.get(mode, activity_mapping["car"]), "data_version": config["data_version"] }, "parameters": { "distance": distance, "distance_unit": distance_unit } } # Add region if provided region = arguments.get("region") if region: request_data["emission_factor"]["region"] = region # We no longer add these parameters to avoid API errors # Instead of trying to use complex parameters, keep it simple try: result = await climatiq_request("/data/v1/estimate", request_data) # Store in cache cache_id = f"travel_{mode}_{distance}_{distance_unit}_{id(result)}" co2e = result.get("co2e", 0) co2e_unit = result.get("co2e_unit", "kg") result_text = f"{mode.capitalize()} travel of {distance} {distance_unit} " if region: result_text += f"in {region} " result_text += f"results in {co2e} {co2e_unit} of CO2e emissions." if "emission_factor" in result and result["emission_factor"].get("name"): ef_name = result["emission_factor"]["name"] result_text += f"\nEmission factor used: {ef_name}" result_text += f"\n\nDetailed results are available as a resource with ID: {cache_id}" return result_text, result, cache_id except Exception as e: logger.error(f"Error in travel estimation: {str(e)}") raise ValueError(f"Error calculating travel emissions: {str(e)}")
- src/climatiq_mcp_server/tools.py:769-789 (registration)Tool registration in get_tool_definitions(), defining the name 'travel-emission' and its input schema for MCP server.types.Tool( name="travel-emission", description="Calculate emissions from passenger travel via car, plane, or train, with options for vehicle types and passenger count.", inputSchema={ "type": "object", "properties": { "mode": {"type": "string", "description": "Travel mode (car, plane, train)"}, "distance": {"type": "number", "description": "Distance traveled (optional if origin and destination are provided)"}, "distance_unit": {"type": "string", "description": "Distance unit (km, mi)", "default": "km"}, "passengers": {"type": "integer", "description": "Number of passengers (for car travel)", "default": 1}, "vehicle_type": {"type": "string", "description": "For car: small/medium/large; For plane: short/medium/long-haul/domestic/international; For train: electric/diesel", "default": "medium"}, "fuel_type": {"type": "string", "description": "For car: regular/electric/hybrid", "default": "regular"}, "origin": {"type": "object", "description": "Origin location for travel API (allows automatic distance calculation)"}, "destination": {"type": "object", "description": "Destination location for travel API (allows automatic distance calculation)"}, "year": {"type": "integer", "description": "Year of travel for more accurate emission factors", "default": 2022}, "car_details": {"type": "object", "description": "Detailed car specifications for more accurate calculations"}, "air_details": {"type": "object", "description": "Detailed flight specifications for more accurate calculations"} }, "required": ["mode"], }, ),
- src/climatiq_mcp_server/server.py:250-251 (registration)Dispatch in the MCP server's call_tool handler that routes 'travel-emission' calls to the implementation.elif name == "travel-emission": result_text, result, cache_id = await travel_emission_tool(config, arguments, server, climatiq_request)