procurement-emission
Calculate Scope 3.1 emissions from procurement spending by analyzing spending amount, category, currency, and country. Powered by Climatiq MCP Server for accurate carbon impact insights.
Instructions
Calculate Scope 3.1 emissions from procurement spending using economic input-output life cycle assessment methods.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| amount | Yes | Amount of money spent | |
| category | Yes | Procurement category (e.g., electronics, food, construction) | |
| country | No | Country code where purchases were made | US |
| currency | No | Currency code (e.g., USD, EUR, GBP) | USD |
Implementation Reference
- src/climatiq_mcp_server/server.py:260-261 (registration)Registration and dispatch for the MCP tool 'procurement-emission'. Calls the specific handler function procurement_emission_tool with config, arguments, server, and climatiq_request helper.elif name == "procurement-emission": result_text, result, cache_id = await procurement_emission_tool(config, arguments, server, climatiq_request)
- Handler execution point for 'procurement-emission' tool. Delegates to dedicated procurement_emission_tool async function.elif name == "procurement-emission": result_text, result, cache_id = await procurement_emission_tool(config, arguments, server, climatiq_request)
- MCP @server.list_tools() handler that returns tool definitions (including input schemas) via get_tool_definitions(), covering 'procurement-emission'.async def handle_list_tools() -> list[types.Tool]: """ List available tools for interacting with the Climatiq API. """ return get_tool_definitions()
- Import of the procurement_emission_tool handler and get_tool_definitions for schemas from climatiq_mcp_server.tools module.procurement_emission_tool, hotel_emission_tool, travel_spend_tool, get_tool_definitions )
- Core helper function climatiq_request used by all emission tools, including procurement-emission, to query 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)}")