AAP MCP Server

#!/usr/bin/env python3 """ AAP Controller Advanced Analytics & Reporting 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_advanced_analytics_tools(mcp: FastMCP): """Register advanced analytics and reporting tools with the MCP server""" @mcp.tool() def advanced_analytics_management( action: str = Field(description="Action: job_explorer, host_explorer, event_explorer, adoption_rate, roi_analysis, custom_reports, probe_templates, get_report, dashboard_data, host_metrics"), report_slug: Optional[str] = Field(None, description="Report slug for get_report action"), time_period: Optional[str] = Field(None, description="Time period: day, week, month, quarter, year"), filters: Optional[Dict[str, Any]] = Field(None, description="Filters for analytics queries"), params: Optional[Dict[str, Any]] = Field(None, description="Additional parameters for analytics"), host_id: Optional[Union[int, float]] = Field(None, description="Host ID for host-specific metrics"), template_id: Optional[Union[int, float]] = Field(None, description="Template ID for analysis") ) -> Dict[str, Any]: """ Advanced analytics and reporting management tool. Handles job analytics, host analysis, adoption metrics, ROI analysis, and custom reports. """ try: # Job Analytics if action == "job_explorer": params_dict = filters or {} if time_period: params_dict["period"] = time_period if params: params_dict.update(params) return get_aap_connector().get("analytics/job_explorer/", params_dict) elif action == "job_explorer_options": return get_aap_connector().get("analytics/job_explorer_options/") # Host Analytics elif action == "host_explorer": params_dict = filters or {} if time_period: params_dict["period"] = time_period if params: params_dict.update(params) return get_aap_connector().get("analytics/host_explorer/", params_dict) elif action == "host_explorer_options": return get_aap_connector().get("analytics/host_explorer_options/") elif action == "host_metrics": if host_id: params_dict = filters or {} return get_aap_connector().get(f"hosts/{host_id}/ansible_facts/", params_dict) else: params_dict = filters or {} return get_aap_connector().get("host_metrics/", params_dict) elif action == "host_metric_summary": params_dict = filters or {} return get_aap_connector().get("host_metric_summary_monthly/", params_dict) # Event Analytics elif action == "event_explorer": params_dict = filters or {} if time_period: params_dict["period"] = time_period if params: params_dict.update(params) return get_aap_connector().get("analytics/event_explorer/", params_dict) elif action == "event_explorer_options": return get_aap_connector().get("analytics/event_explorer_options/") # Adoption Rate Analysis elif action == "adoption_rate": params_dict = filters or {} if time_period: params_dict["period"] = time_period return get_aap_connector().get("analytics/adoption_rate/", params_dict) elif action == "adoption_rate_options": return get_aap_connector().get("analytics/adoption_rate_options/") # ROI Analysis elif action == "roi_analysis": params_dict = filters or {} if time_period: params_dict["period"] = time_period return get_aap_connector().get("analytics/roi_templates/", params_dict) elif action == "roi_options": return get_aap_connector().get("analytics/roi_templates_options/") # Probe Templates elif action == "probe_templates": params_dict = filters or {} return get_aap_connector().get("analytics/probe_templates/", params_dict) elif action == "probe_templates_options": return get_aap_connector().get("analytics/probe_templates_options/") elif action == "probe_template_for_hosts": params_dict = filters or {} if template_id: params_dict["template"] = template_id return get_aap_connector().get("analytics/probe_template_for_hosts/", params_dict) elif action == "probe_template_for_hosts_options": return get_aap_connector().get("analytics/probe_template_for_hosts_options/") # Custom Reports elif action == "custom_reports": params_dict = filters or {} return get_aap_connector().get("analytics/reports/", params_dict) elif action == "get_report": if not report_slug: return {"error": "report_slug is required for get_report action"} params_dict = filters or {} return get_aap_connector().get(f"analytics/report/{report_slug}/", params_dict) elif action == "report_options": return get_aap_connector().get("analytics/report_options/") # Dashboard Data elif action == "dashboard_data": return get_aap_connector().get("analytics/") elif action == "analytics_authorized": return get_aap_connector().get("analytics/authorized/") else: return {"error": f"Unknown action: {action}"} except Exception as e: return {"error": f"Advanced analytics management failed: {str(e)}"}