travel-emission
Calculate carbon emissions for car, plane, or train travel, considering vehicle type, passenger count, and distance. Use origin and destination for automatic distance calculations to assess climate impact accurately.
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 |
|---|---|---|---|
| air_details | No | Detailed flight specifications for more accurate calculations | |
| car_details | No | Detailed car specifications for more accurate calculations | |
| destination | No | Destination location for travel API (allows automatic distance calculation) | |
| distance | No | Distance traveled (optional if origin and destination are provided) | |
| distance_unit | No | Distance unit (km, mi) | km |
| fuel_type | No | For car: regular/electric/hybrid | regular |
| mode | Yes | Travel mode (car, plane, train) | |
| origin | No | Origin location for travel API (allows automatic distance calculation) | |
| 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 |
| year | No | Year of travel for more accurate emission factors |
Implementation Reference
- The entry point for executing the 'travel-emission' tool. It calls the dedicated travel_emission_tool function passing configuration, arguments, server context, and the climatiq_request helper.elif name == "travel-emission": result_text, result, cache_id = await travel_emission_tool(config, arguments, server, climatiq_request)
- src/climatiq_mcp_server/server.py:224-230 (registration)Registers the 'travel-emission' tool (among others) by returning its schema and description 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()
- Shared helper function used by all emission calculation tools, including 'travel-emission', to make authenticated requests to the Climatiq API.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:86-145 (helper)CLI implementation of travel emission calculation with identical logic likely used or mirrored in the MCP tool.async def calculate_travel_emission(mode, distance, distance_unit, region="US"): """Calculate emissions from travel""" logger.info(f"Calculating emissions for {distance} {distance_unit} of {mode} travel in {region}") # Map mode to activity ID mode_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" } if mode.lower() not in mode_mapping: print(f"Unsupported travel mode: {mode}. Supported modes are: car, plane, train") return None # Construct request data request_data = { "emission_factor": { "activity_id": mode_mapping[mode.lower()], "data_version": config["data_version"], "region": region }, "parameters": { "distance": distance, "distance_unit": distance_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"\n{mode.capitalize()} travel of {distance} {distance_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