hotel-emission
Measure carbon emissions from hotel stays by inputting the number of nights and hotel location. Utilize Climatiq MCP Server for accurate climate impact insights.
Instructions
Calculate carbon emissions from hotel stays based on the number of nights stayed and the location of the hotel.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hotel_nights | Yes | Number of nights stayed | |
| location | Yes | Location of the hotel | |
| year | No | Year of the hotel stay |
Implementation Reference
- src/climatiq_mcp_server/server.py:262-263 (registration)Registration and dispatch point for the 'hotel-emission' tool. Calls the hotel_emission_tool function with config, arguments, server context, and the climatiq_request helper.elif name == "hotel-emission": result_text, result, cache_id = await hotel_emission_tool(config, arguments, server, climatiq_request)
- Main MCP tool handler that routes tool calls by name to specific tool functions. Includes the dispatch for 'hotel-emission'.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, ) ]
- Helper function used by tool handlers to make authenticated requests to the Climatiq API for emission calculations.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)}")
- src/climatiq_mcp_server/server.py:225-229 (registration)Tool listing handler that returns definitions (including schema) for all tools via get_tool_definitions(), which includes 'hotel-emission'.async def handle_list_tools() -> list[types.Tool]: """ List available tools for interacting with the Climatiq API. """ return get_tool_definitions()