AAP MCP Server

#!/usr/bin/env python3 """ AAP Controller Project & Content 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_project_tools(mcp: FastMCP): """Register project management tools with the MCP server""" @mcp.tool() def project_content_management( action: str = Field(description="Action: list_projects, create, update, delete, update_project, get_project_status, list_playbooks, get_playbook, list_project_updates, list_organizations, create_organization, update_organization"), project_id: Optional[Union[int, float]] = Field(None, description="Project ID"), organization_id: Optional[Union[int, float]] = Field(None, description="Organization ID"), project_data: Optional[Dict[str, Any]] = Field(None, description="Project data"), organization_data: Optional[Dict[str, Any]] = Field(None, description="Organization data"), filters: Optional[Dict[str, Any]] = Field(None, description="Filters for listing") ) -> Dict[str, Any]: """ Project and content management tool. Handles projects, playbooks, organizations, and project updates. """ try: # Project Operations if action == "list_projects": params = filters or {} return get_aap_connector().get("projects/", params) elif action == "create": if not project_data: return {"error": "project_data is required"} return get_aap_connector().post("projects/", project_data) elif action == "update": if not project_id or not project_data: return {"error": "project_id and project_data are required"} return get_aap_connector().patch(f"projects/{project_id}/", project_data) elif action == "delete": if not project_id: return {"error": "project_id is required"} return get_aap_connector().delete(f"projects/{project_id}/") elif action == "update_project": if not project_id: return {"error": "project_id is required"} return get_aap_connector().post(f"projects/{project_id}/update/") elif action == "get_project_status": if not project_id: return {"error": "project_id is required"} return get_aap_connector().get(f"projects/{project_id}/") elif action == "list_playbooks": if not project_id: return {"error": "project_id is required"} return get_aap_connector().get(f"projects/{project_id}/playbooks/") elif action == "get_playbook": if not project_id: return {"error": "project_id is required"} params = filters or {} return get_aap_connector().get(f"projects/{project_id}/playbooks/", params) elif action == "list_project_updates": if project_id: params = filters or {} return get_aap_connector().get(f"projects/{project_id}/project_updates/", params) else: params = filters or {} return get_aap_connector().get("project_updates/", params) # Organization Operations elif action == "list_organizations": params = filters or {} return get_aap_connector().get("organizations/", params) elif action == "create_organization": if not organization_data: return {"error": "organization_data is required"} return get_aap_connector().post("organizations/", organization_data) elif action == "update_organization": if not organization_id or not organization_data: return {"error": "organization_id and organization_data are required"} return get_aap_connector().patch(f"organizations/{organization_id}/", organization_data) else: return {"error": f"Unknown action: {action}"} except Exception as e: return {"error": f"Project/Content management failed: {str(e)}"}