AAP MCP Server

#!/usr/bin/env python3 """ AAP Controller Inventory & Host Management Tool """ from typing import Any, Dict, Optional, Union from fastmcp import FastMCP from pydantic import Field from connectors.aap_connector import get_aap_connector def register_inventory_tools(mcp: FastMCP): """Register inventory management tools with the MCP server""" @mcp.tool() def inventory_host_management( action: str = Field(description="Action: list, create, update, delete, copy, sync, add_host, remove_host, list_hosts, update_host, get_host_facts, list_groups, create_group, update_group, delete_group, add_host_to_group, remove_host_from_group, list_sources, create_source, update_source, sync_source, variable_data, tree, script"), inventory_id: Optional[Union[int, float]] = Field(None, description="Inventory ID"), host_id: Optional[Union[int, float]] = Field(None, description="Host ID"), group_id: Optional[Union[int, float]] = Field(None, description="Group ID"), source_id: Optional[Union[int, float]] = Field(None, description="Inventory source ID"), inventory_data: Optional[Dict[str, Any]] = Field(None, description="Inventory data"), host_data: Optional[Dict[str, Any]] = Field(None, description="Host data"), group_data: Optional[Dict[str, Any]] = Field(None, description="Group data"), source_data: Optional[Dict[str, Any]] = Field(None, description="Inventory source data"), copy_data: Optional[Dict[str, Any]] = Field(None, description="Copy operation data"), filters: Optional[Dict[str, Any]] = Field(None, description="Filters for listing") ) -> Dict[str, Any]: """ Enhanced inventory and host management tool. Handles inventories, hosts, groups, sources, and bulk operations. """ try: # Inventory Operations if action == "list": params = filters or {} return get_aap_connector().get("inventories/", params) elif action == "create": if not inventory_data: return {"error": "inventory_data is required"} return get_aap_connector().post("inventories/", inventory_data) elif action == "update": if not inventory_id or not inventory_data: return {"error": "inventory_id and inventory_data are required"} return get_aap_connector().patch(f"inventories/{inventory_id}/", inventory_data) elif action == "delete": if not inventory_id: return {"error": "inventory_id is required"} return get_aap_connector().delete(f"inventories/{inventory_id}/") elif action == "copy": if not inventory_id or not copy_data: return {"error": "inventory_id and copy_data are required"} return get_aap_connector().post(f"inventories/{inventory_id}/copy/", copy_data) elif action == "sync": if not inventory_id: return {"error": "inventory_id is required"} return get_aap_connector().post(f"inventories/{inventory_id}/update_inventory_sources/") elif action == "variable_data": if not inventory_id: return {"error": "inventory_id is required"} return get_aap_connector().get(f"inventories/{inventory_id}/variable_data/") elif action == "tree": if not inventory_id: return {"error": "inventory_id is required"} return get_aap_connector().get(f"inventories/{inventory_id}/tree/") elif action == "script": if not inventory_id: return {"error": "inventory_id is required"} return get_aap_connector().get(f"inventories/{inventory_id}/script/") # Host Operations elif action == "list_hosts": if inventory_id: params = filters or {} return get_aap_connector().get(f"inventories/{inventory_id}/hosts/", params) else: params = filters or {} return get_aap_connector().get("hosts/", params) elif action == "add_host": if not inventory_id or not host_data: return {"error": "inventory_id and host_data are required"} return get_aap_connector().post(f"inventories/{inventory_id}/hosts/", host_data) elif action == "update_host": if not host_id or not host_data: return {"error": "host_id and host_data are required"} return get_aap_connector().patch(f"hosts/{host_id}/", host_data) elif action == "remove_host": if not host_id: return {"error": "host_id is required"} return get_aap_connector().delete(f"hosts/{host_id}/") elif action == "get_host_facts": if not host_id: return {"error": "host_id is required"} return get_aap_connector().get(f"hosts/{host_id}/ansible_facts/") # Group Operations elif action == "list_groups": if inventory_id: params = filters or {} return get_aap_connector().get(f"inventories/{inventory_id}/groups/", params) else: params = filters or {} return get_aap_connector().get("groups/", params) elif action == "create_group": if not inventory_id or not group_data: return {"error": "inventory_id and group_data are required"} return get_aap_connector().post(f"inventories/{inventory_id}/groups/", group_data) elif action == "update_group": if not group_id or not group_data: return {"error": "group_id and group_data are required"} return get_aap_connector().patch(f"groups/{group_id}/", group_data) elif action == "delete_group": if not group_id: return {"error": "group_id is required"} return get_aap_connector().delete(f"groups/{group_id}/") elif action == "add_host_to_group": if not group_id or not host_id: return {"error": "group_id and host_id are required"} return get_aap_connector().post(f"groups/{group_id}/hosts/", {"id": host_id}) elif action == "remove_host_from_group": if not group_id or not host_id: return {"error": "group_id and host_id are required"} return get_aap_connector().post(f"groups/{group_id}/hosts/", {"id": host_id, "disassociate": True}) # Inventory Source Operations elif action == "list_sources": if inventory_id: params = filters or {} return get_aap_connector().get(f"inventories/{inventory_id}/inventory_sources/", params) else: params = filters or {} return get_aap_connector().get("inventory_sources/", params) elif action == "create_source": if not source_data: return {"error": "source_data is required"} return get_aap_connector().post("inventory_sources/", source_data) elif action == "update_source": if not source_id or not source_data: return {"error": "source_id and source_data are required"} return get_aap_connector().patch(f"inventory_sources/{source_id}/", source_data) elif action == "sync_source": if not source_id: return {"error": "source_id is required"} return get_aap_connector().post(f"inventory_sources/{source_id}/update/") else: return {"error": f"Unknown action: {action}"} except Exception as e: return {"error": f"Inventory/Host management failed: {str(e)}"}