AAP MCP Server

#!/usr/bin/env python3 """ AAP Gateway Monitoring & Configuration Tool """ from typing import Any, Dict, Optional, Union from fastmcp import FastMCP from pydantic import Field from connectors.gateway_connector import get_gateway_connector def register_gateway_monitoring_tools(mcp: FastMCP): """Register Gateway monitoring and configuration tools with the MCP server""" @mcp.tool() def gateway_monitoring_management( action: str = Field(description="Action: list_activity_stream, get_activity_entry, get_status, ping, list_settings, update_settings, list_http_ports, create_http_port, update_http_port, delete_http_port, get_feature_flags, get_jwt_key, list_app_urls, get_service_keys, list_service_keys, create_service_key, update_service_key, delete_service_key"), entry_id: Optional[Union[int, float]] = Field(None, description="Activity stream entry ID"), http_port_id: Optional[Union[int, float]] = Field(None, description="HTTP port ID"), service_key_id: Optional[Union[int, float]] = Field(None, description="Service key ID"), settings_data: Optional[Dict[str, Any]] = Field(None, description="Settings data"), http_port_data: Optional[Dict[str, Any]] = Field(None, description="HTTP port data"), service_key_data: Optional[Dict[str, Any]] = Field(None, description="Service key data"), filters: Optional[Dict[str, Any]] = Field(None, description="Filters for listing") ) -> Dict[str, Any]: """ Gateway monitoring and configuration tool. Handles activity streams, settings, status monitoring, and HTTP port management. """ try: client = get_gateway_connector() # Activity Stream Operations if action == "list_activity_stream": params = filters or {} return client.get("activitystream/", params) elif action == "get_activity_entry": if not entry_id: return {"error": "entry_id is required for getting activity stream entry"} return client.get(f"activitystream/{entry_id}/") # Status and Health Operations elif action == "get_status": return client.get("status/") elif action == "ping": return client.get("ping/") # Settings Operations elif action == "list_settings": params = filters or {} return client.get("settings/", params) elif action == "update_settings": if not settings_data: return {"error": "settings_data is required for updating settings"} return client.patch("settings/", settings_data) # HTTP Port Operations elif action == "list_http_ports": params = filters or {} return client.get("http_ports/", params) elif action == "create_http_port": if not http_port_data: return {"error": "http_port_data is required for creating HTTP ports"} return client.post("http_ports/", http_port_data) elif action == "update_http_port": if not http_port_id or not http_port_data: return {"error": "http_port_id and http_port_data are required"} return client.patch(f"http_ports/{http_port_id}/", http_port_data) elif action == "delete_http_port": if not http_port_id: return {"error": "http_port_id is required for deleting HTTP ports"} return client.delete(f"http_ports/{http_port_id}/") # Feature and Configuration Operations elif action == "get_feature_flags": return client.get("feature_flags_state/") elif action == "get_jwt_key": return client.get("jwt_key/") elif action == "list_app_urls": params = filters or {} return client.get("app_urls/", params) # Service Key Operations elif action == "list_service_keys": params = filters or {} return client.get("service_keys/", params) elif action == "create_service_key": if not service_key_data: return {"error": "service_key_data is required for creating service keys"} return client.post("service_keys/", service_key_data) elif action == "update_service_key": if not service_key_id or not service_key_data: return {"error": "service_key_id and service_key_data are required"} return client.patch(f"service_keys/{service_key_id}/", service_key_data) elif action == "delete_service_key": if not service_key_id: return {"error": "service_key_id is required for deleting service keys"} return client.delete(f"service_keys/{service_key_id}/") # Utility Operations elif action == "get_trigger_definition": return client.get("trigger_definition/") elif action == "get_role_metadata": return client.get("role_metadata/") elif action == "get_ui_auth": return client.get("ui_auth/") elif action == "get_legacy_auth": return client.get("legacy_auth/") else: return {"error": f"Unknown action: {action}"} except Exception as e: return {"error": f"Gateway monitoring management failed: {str(e)}"}