Climatiq MCP Server

Instructions

Implementation Reference

• src/climatiq_mcp_server/server.py:258-259 (registration)

  The tool 'freight-emission' is registered and dispatched here in the handle_call_tool function. It calls the freight_emission_tool function with config, arguments, server, and the climatiq_request helper.
  elif name == "freight-emission": result_text, result, cache_id = await freight_emission_tool(config, arguments, server, climatiq_request)
• src/climatiq_mcp_server/server.py:32-44 (registration)

  Import statement bringing in the freight_emission_tool and other tools, indicating where the implementation is imported from (though the module location could not be resolved).
  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 )
• src/climatiq_mcp_server/server.py:59-280 (helper)

  Helper function used by tool handlers to make requests to the Climatiq API, likely used by freight_emission_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)}") @server.list_resources() async def handle_list_resources() -> list[types.Resource]: """ List available resources including cached results. """ resources = [] # Add cached emission calculation results for cache_id, result in cached_results.items(): if "co2e" in result and "emission_factor" in result: name = result.get("emission_factor", {}).get("name", "Unknown emission factor") resources.append( types.Resource( uri=AnyUrl(f"climatiq://calculation/{cache_id}"), name=f"Calculation: {name}", description=f"CO2e: {result.get('co2e')} {result.get('co2e_unit', 'kg')}", mimeType="application/json", ) ) return resources @server.read_resource() async def handle_read_resource(uri: AnyUrl) -> str: """ Read a specific calculation result by its URI. """ if uri.scheme != "climatiq": raise ValueError(f"Unsupported URI scheme: {uri.scheme}") path_parts = uri.path.lstrip("/").split("/") if len(path_parts) < 2 or path_parts[0] != "calculation": raise ValueError(f"Invalid resource path: {uri.path}") cache_id = path_parts[1] if cache_id not in cached_results: raise ValueError(f"Calculation result not found: {cache_id}") return json.dumps(cached_results[cache_id], indent=2) @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: """ List available prompts. """ return [ types.Prompt( name="climate-impact-explanation", description="Explains the climate impact of a specific emission calculation", arguments=[ types.PromptArgument( name="calculation_id", description="ID of the cached calculation result to explain", required=True, ), types.PromptArgument( name="detail_level", description="Level of detail (basic/detailed)", required=False, ) ], ) ] @server.get_prompt() async def handle_get_prompt( name: str, arguments: dict[str, str] | None ) -> types.GetPromptResult: """ Generate a prompt explaining climate impact. """ if name != "climate-impact-explanation": raise ValueError(f"Unknown prompt: {name}") if not arguments or "calculation_id" not in arguments: raise ValueError("Missing required argument: calculation_id") calculation_id = arguments["calculation_id"] if calculation_id not in cached_results: raise ValueError(f"Calculation result not found: {calculation_id}") result = cached_results[calculation_id] detail_level = arguments.get("detail_level", "basic") # Format the result information for the prompt co2e = result.get("co2e", 0) co2e_unit = result.get("co2e_unit", "kg") emission_factor = result.get("emission_factor", {}) activity_data = result.get("activity_data", {}) detail_text = "" if detail_level == "detailed": detail_text = "\n\nPlease provide a detailed analysis including comparisons to everyday activities and detailed explanation of the environmental impact." return types.GetPromptResult( description=f"Explaining climate impact of {emission_factor.get('name', 'the activity')}", messages=[ types.PromptMessage( role="user", content=types.TextContent( type="text", text=f"Please explain the climate impact of this emission calculation in simple terms:{detail_text}\n\n" f"Activity: {emission_factor.get('name', 'Unknown activity')}\n" f"Value: {activity_data.get('activity_value', 'Unknown')} {activity_data.get('activity_unit', '')}\n" f"Carbon footprint: {co2e} {co2e_unit} CO2e\n" f"Region: {emission_factor.get('region', 'Unknown')}\n" f"Year: {emission_factor.get('year', 'Unknown')}\n" f"Source: {emission_factor.get('source', 'Unknown')}" ), ) ], ) @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools for interacting with the Climatiq API. """ return get_tool_definitions() @server.call_tool() 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, ) ]