travel-spend
Calculate carbon emissions from travel expenses by inputting spend type, amount, currency, location, and year. Use this tool to assess climate impact of travel-related spending.
Instructions
Calculate carbon emissions from travel-related spending based on spend type, amount, currency, location, and year.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| money | Yes | Amount of money spent | |
| money_unit | Yes | Currency unit for the spent money | |
| spend_location | Yes | Location of the travel spending | |
| spend_type | Yes | Type of travel spending (air, road, rail, sea, hotel) | |
| spend_year | No | Year of the travel spending |
Implementation Reference
- src/climatiq_mcp_server/server.py:264-265 (registration)Registration and dispatch for the 'travel-spend' tool, which delegates to the travel_spend_tool implementation function.elif name == "travel-spend": result_text, result, cache_id = await travel_spend_tool(config, arguments, server, climatiq_request)
- src/climatiq_mcp_server/server.py:32-44 (registration)Import statement bringing in the travel_spend_tool function and get_tool_definitions for schemas.from climatiq_mcp_server.tools import ( set_api_key_tool, electricity_emission_tool, travel_emission_tool, search_emission_factors_tool, custom_emission_calculation_tool, cloud_computing_emission_tool, freight_emission_tool, procurement_emission_tool, hotel_emission_tool, travel_spend_tool, get_tool_definitions )
- MCP tool handler function that routes 'travel-spend' tool calls to specific implementation.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, ) ]
- Shared helper function for making requests to Climatiq API, used by travel_spend_tool.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)}")