SuzieQ MCP Server

Instructions

Runs a SuzieQ 'summarize' query via its REST API.

Args:
    table: The name of the SuzieQ table to summarize (e.g., 'device', 'bgp', 'interface', 'route').
    filters: An optional dictionary of filter parameters for the SuzieQ query
             (e.g., {"hostname": "leaf01", "vrf": "default"}).
             Keys should match SuzieQ filter names. Values can be strings or lists of strings.
             If no filters are needed, this can be None, null, or an empty dictionary.

Returns:
    A JSON string representing the summarized result from the SuzieQ API,
    or a JSON string with an error message.

Implementation Reference

• server.py:121-147 (handler)

  The handler function for the 'run_suzieq_summarize' tool, decorated with @mcp.tool() for registration. It processes input parameters, calls the helper to query the SuzieQ API with the 'summarize' verb, serializes the response to JSON, and handles serialization errors.

  async def run_suzieq_summarize(table: str, filters: Optional[Dict[str, Any]] = None) -> str:
      """
      Runs a SuzieQ 'summarize' query via its REST API.
  
      Args:
          table: The name of the SuzieQ table to summarize (e.g., 'device', 'bgp', 'interface', 'route').
          filters: An optional dictionary of filter parameters for the SuzieQ query
                   (e.g., {"hostname": "leaf01", "vrf": "default"}).
                   Keys should match SuzieQ filter names. Values can be strings or lists of strings.
                   If no filters are needed, this can be None, null, or an empty dictionary.
  
      Returns:
          A JSON string representing the summarized result from the SuzieQ API,
          or a JSON string with an error message.
      """
      actual_filters = filters if isinstance(filters, dict) else None
  
      # Call the generalized helper function with verb='summarize'
      result = await _query_suzieq_api(verb="summarize", table=table, params=actual_filters)
  
      try:
          # Serialize the result dictionary to a JSON string
          return json.dumps(result, indent=2, ensure_ascii=False)
      except TypeError as e:
          error_message = f"Error serializing SuzieQ 'summarize' response to JSON: {str(e)}"
          print(f"[ERROR] {error_message}") # Debug print
          return json.dumps({"error": error_message})

• server.py:25-90 (helper)

  Supporting helper function that performs the actual HTTP request to the SuzieQ REST API, constructs the URL as /{table}/{verb}, adds authentication via access_token query param, handles various errors, and returns the parsed JSON response or error dictionary.

  async def _query_suzieq_api(verb: str, table: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """
      Asynchronously queries the SuzieQ REST API for a given verb and table.
  
      Args:
          verb: The SuzieQ verb to execute (e.g., 'show', 'summarize').
          table: The SuzieQ table name (e.g., 'device', 'bgp', 'interface').
          params: A dictionary of query parameters for filtering (e.g., {'hostname': 'leaf01', 'vrf': 'default'}).
  
      Returns:
          A dictionary containing the API response or an error message.
      """
      # Check if configuration is missing
      if not SUZIEQ_API_ENDPOINT or not SUZIEQ_API_KEY:
          error_msg = "SuzieQ API endpoint or key not configured. Set SUZIEQ_API_ENDPOINT and SUZIEQ_API_KEY environment variables or in .env file."
          print(f"[ERROR] {error_msg}") # Log configuration error
          return {"error": error_msg}
  
      # Construct the API URL using the provided verb and table
      api_endpoint_clean = SUZIEQ_API_ENDPOINT.rstrip('/')
      # Assuming the API structure follows /{table}/{verb} pattern
      api_url = f"{api_endpoint_clean}/{table}/{verb}"
  
      # --- Authentication ---
      # Prepare parameters, adding the API key as 'access_token' query parameter
      query_params = params if params else {}
      query_params["access_token"] = SUZIEQ_API_KEY # Add key as query param
      query_params["format"] = "json" # Ensure response is JSON
      headers = {} # No custom headers needed now for auth
      # ---------------------------
  
      async with httpx.AsyncClient() as client:
          try:
              # Use headers={}, params=query_params
              print(f"[INFO] Querying SuzieQ API: {api_url} with params: {query_params}") # Debug print
              response = await client.get(api_url, headers=headers, params=query_params, timeout=30.0)
              print(f"[INFO] SuzieQ API Response Status: {response.status_code}") # Debug print
              response.raise_for_status()  # Raise an exception for bad status codes (4xx or 5xx)
  
              if response.status_code == 204:
                  return {"warning": f"Received empty response (204 No Content) from SuzieQ API for {verb} {table}"}
  
              content_type = response.headers.get('content-type', '')
              if 'application/json' in content_type:
                  return response.json()
              else:
                  error_detail = f"Unexpected content type received: {content_type}. Response text: {response.text}"
                  print(f"[ERROR] {error_detail}")
                  return {"error": error_detail}
          except httpx.HTTPStatusError as e:
              error_detail = f"HTTP error occurred: {e.response.status_code} - {e.response.text}"
              print(f"[ERROR] {error_detail}") # Debug print
              return {"error": error_detail}
          except httpx.RequestError as e:
              error_detail = f"An error occurred while requesting {e.request.url!r}: {e}"
              print(f"[ERROR] {error_detail}") # Debug print
              return {"error": error_detail}
          except json.JSONDecodeError as e:
              error_detail = f"Failed to decode JSON response from SuzieQ API: {e}. Response text: {response.text}"
              print(f"[ERROR] {error_detail}")
              return {"error": error_detail}
          except Exception as e:
              error_detail = f"An unexpected error occurred: {str(e)}"
              print(f"[ERROR] {error_detail}") # Debug print
              return {"error": error_detail}