Climatiq MCP Server

Instructions

Implementation Reference

• src/climatiq_mcp_server/tools.py:300-389 (handler)

  Main handler function that executes the search-emission-factors tool by calling the Climatiq API search endpoint, processing results, and formatting a user-friendly summary of the top emission factors.

  async def search_emission_factors_tool(config, arguments, server, climatiq_request):
      """
      Search for emission factors in the Climatiq database.
      
      This tool allows users to search through Climatiq's extensive database of emission factors
      using keywords and filters. It helps users find the appropriate emission factors for 
      specific activities, regions, or categories. The search results include:
      - The name of the emission factor
      - Activity ID (used for custom calculations)
      - Region specificity
      - Year of publication
      - Category classification
      - Source organization
      - Available units
      
      This tool is especially useful before performing custom calculations, as it helps 
      identify the correct activity_id to use.
      """
      query = arguments.get("query")
      category = arguments.get("category", "")
      region = arguments.get("region", "")
      year = arguments.get("year", "")
      source = arguments.get("source", "")
      data_version = arguments.get("data_version", config["data_version"])
      unit_type = arguments.get("unit_type", "")
      
      if not query:
          raise ValueError("Missing search query")
          
      # Construct the request to the Climatiq API
      params = {
          "query": query,
          "data_version": data_version
      }
      
      # Add all optional filters if provided
      if category:
          params["category"] = category
      if region:
          params["region"] = region
      if year:
          params["year"] = year
      if source:
          params["source"] = source
      if unit_type:
          params["unit_type"] = unit_type
      
      try:    
          result = await climatiq_request("/data/v1/search", params, method="GET")
          
          # Process search results
          results = result.get("results", [])
          total_count = len(results)
          
          if total_count == 0:
              result_text = "No emission factors found matching your search criteria."
              return result_text, {}, None
              
          # Format a summary of the results
          result_text = f"Found {total_count} emission factors matching '{query}'.\n\n"
          
          # Display the first 5 results (or fewer if less are available)
          max_display = min(5, total_count)
          result_text += f"Here are the first {max_display} results:\n\n"
          
          for i in range(max_display):
              ef = results[i]
              result_text += f"{i+1}. {ef.get('name', 'Unnamed factor')}\n"
              result_text += f"   Activity ID: {ef.get('activity_id', 'N/A')}\n"
              result_text += f"   Region: {ef.get('region_name', ef.get('region', 'N/A'))}\n"
              result_text += f"   Year: {ef.get('year', 'N/A')}\n"
              result_text += f"   Source: {ef.get('source', 'N/A')}\n"
              if ef.get('unit'):
                  result_text += f"   Unit: {ef.get('unit', 'N/A')}\n"
              result_text += "\n"
          
          # If there are more results than we displayed
          if total_count > max_display:
              result_text += f"\n... and {total_count - max_display} more results."
              
          # Include a note about using these factors with the custom calculation tool
          result_text += "\n\nTo use one of these emission factors, copy its Activity ID and use it with the custom-emission-calculation tool."
          
          # Cache the search results
          cache_id = f"search_{query.replace(' ', '_')}_{id(result)}"
          
          return result_text, result, cache_id
      except Exception as e:
          error_msg = f"Error searching emission factors: {str(e)}"
          raise ValueError(error_msg)

• src/climatiq_mcp_server/tools.py:790-806 (schema)

  Pydantic/MCP schema definition for the search-emission-factors tool, including input parameters and validation rules.

  types.Tool(
      name="search-emission-factors",
      description="Search Climatiq's database for emission factors by keyword, category, region, year, source, and other metadata to find appropriate factors for calculations.",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {"type": "string", "description": "Search query for emission factors (e.g., 'electricity', 'flight', 'truck')"},
              "category": {"type": "string", "description": "Category of emission factors (e.g., 'Energy', 'Transport')", "default": ""},
              "region": {"type": "string", "description": "Region code to filter results (e.g., 'US', 'EU')", "default": ""},
              "year": {"type": "integer", "description": "Year of emission factors", "default": 2022},
              "source": {"type": "string", "description": "Source organization of emission factors (e.g., 'IPCC', 'EPA', 'BEIS')", "default": ""},
              "data_version": {"type": "string", "description": "Data version of emission factors", "default": ""},
              "unit_type": {"type": "string", "description": "Unit type of emission factors (e.g., 'energy', 'distance', 'weight')", "default": ""}
          },
          "required": ["query"],
      },
  ),

• src/climatiq_mcp_server/server.py:252-253 (registration)

  Tool dispatch/registration in the MCP server's handle_call_tool method, routing calls to the handler function.

  elif name == "search-emission-factors":
      result_text, result, cache_id = await search_emission_factors_tool(config, arguments, server, climatiq_request)

• src/climatiq_mcp_server/server.py:224-229 (registration)

  Registers the tool list handler which provides the tool schema via get_tool_definitions() including search-emission-factors.

  @server.list_tools()
  async def handle_list_tools() -> list[types.Tool]:
      """
      List available tools for interacting with the Climatiq API.
      """
      return get_tool_definitions()

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

  Import of the search_emission_factors_tool handler from tools.py.

  search_emission_factors_tool,