AAP MCP Server

#!/usr/bin/env python3 """ AAP Gateway Service Management 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_service_tools(mcp: FastMCP): """Register Gateway service management tools with the MCP server""" @mcp.tool() def gateway_service_management( action: str = Field(description="Action: list_services, create_service, update_service, delete_service, get_service, list_service_types, list_service_clusters, create_service_cluster, update_service_cluster, delete_service_cluster, list_service_nodes, create_service_node, update_service_node, delete_service_node, list_routes, create_route, update_route, delete_route, get_service_index"), service_id: Optional[Union[int, float]] = Field(None, description="Service ID"), service_type_id: Optional[Union[int, float]] = Field(None, description="Service type ID"), service_cluster_id: Optional[Union[int, float]] = Field(None, description="Service cluster ID"), service_node_id: Optional[Union[int, float]] = Field(None, description="Service node ID"), route_id: Optional[Union[int, float]] = Field(None, description="Route ID"), service_data: Optional[Dict[str, Any]] = Field(None, description="Service data"), service_cluster_data: Optional[Dict[str, Any]] = Field(None, description="Service cluster data"), service_node_data: Optional[Dict[str, Any]] = Field(None, description="Service node data"), route_data: Optional[Dict[str, Any]] = Field(None, description="Route data"), filters: Optional[Dict[str, Any]] = Field(None, description="Filters for listing") ) -> Dict[str, Any]: """ Gateway service management tool. Handles services, service types, service clusters, service nodes, and routes. """ try: client = get_gateway_connector() # Service Operations if action == "list_services": params = filters or {} return client.get("services/", params) elif action == "create_service": if not service_data: return {"error": "service_data is required for creating services"} return client.post("services/", service_data) elif action == "get_service": if not service_id: return {"error": "service_id is required for getting service details"} return client.get(f"services/{service_id}/") elif action == "update_service": if not service_id or not service_data: return {"error": "service_id and service_data are required for updating services"} return client.patch(f"services/{service_id}/", service_data) elif action == "delete_service": if not service_id: return {"error": "service_id is required for deleting services"} return client.delete(f"services/{service_id}/") # Service Type Operations elif action == "list_service_types": params = filters or {} return client.get("service_types/", params) # Service Cluster Operations elif action == "list_service_clusters": params = filters or {} return client.get("service_clusters/", params) elif action == "create_service_cluster": if not service_cluster_data: return {"error": "service_cluster_data is required for creating service clusters"} return client.post("service_clusters/", service_cluster_data) elif action == "update_service_cluster": if not service_cluster_id or not service_cluster_data: return {"error": "service_cluster_id and service_cluster_data are required"} return client.patch(f"service_clusters/{service_cluster_id}/", service_cluster_data) elif action == "delete_service_cluster": if not service_cluster_id: return {"error": "service_cluster_id is required for deleting service clusters"} return client.delete(f"service_clusters/{service_cluster_id}/") # Service Node Operations elif action == "list_service_nodes": params = filters or {} return client.get("service_nodes/", params) elif action == "create_service_node": if not service_node_data: return {"error": "service_node_data is required for creating service nodes"} return client.post("service_nodes/", service_node_data) elif action == "update_service_node": if not service_node_id or not service_node_data: return {"error": "service_node_id and service_node_data are required"} return client.patch(f"service_nodes/{service_node_id}/", service_node_data) elif action == "delete_service_node": if not service_node_id: return {"error": "service_node_id is required for deleting service nodes"} return client.delete(f"service_nodes/{service_node_id}/") # Route Operations elif action == "list_routes": params = filters or {} return client.get("routes/", params) elif action == "create_route": if not route_data: return {"error": "route_data is required for creating routes"} return client.post("routes/", route_data) elif action == "update_route": if not route_id or not route_data: return {"error": "route_id and route_data are required for updating routes"} return client.patch(f"routes/{route_id}/", route_data) elif action == "delete_route": if not route_id: return {"error": "route_id is required for deleting routes"} return client.delete(f"routes/{route_id}/") # Service Index elif action == "get_service_index": return client.get("service-index/") else: return {"error": f"Unknown action: {action}"} except Exception as e: return {"error": f"Gateway service management failed: {str(e)}"}