SuzieQ MCP Server

Instructions

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}