# ======== from tools/argocd_application_diff_tool/ ======== """Argo CD application diff/drift investigation tool.""" from __future__ import annotations from typing import Any from core.tool_framework.base import BaseTool from integrations.argocd.client import make_argocd_client class ArgoCDApplicationDiffTool(BaseTool): """Fetch Argo CD server-side diff data for an application.""" name = "argocd_application_diff" source = "argocd" description = ( "Fetch Argo CD server-side diff output and report whether live cluster state " "has drifted from the desired GitOps state." ) use_cases = [ "Detecting GitOps drift during an incident", "Checking whether an OutOfSync application has Kubernetes object diffs", "Correlating deployment drift with application health degradation", ] requires = ["base_url", "application_name"] injected_params = ["base_url", "password", "username"] input_schema = { "type": "object", "properties": { "base_url": {"type": "string", "description": "Argo CD base URL"}, "bearer_token": {"type": "string", "default": "", "description": "Argo CD API token"}, "username": {"type": "string", "default": "", "description": "Argo CD username"}, "password": {"type": "string", "default": "", "description": "Argo CD password"}, "application_name": {"type": "string", "description": "Application name"}, "project": {"type": "string", "default": "", "description": "Optional Argo CD project"}, "app_namespace": { "type": "string", "default": "", "description": "Optional app namespace", }, "verify_ssl": { "type": "boolean", "default": True, "description": "Verify TLS certificates", }, }, "required": ["base_url", "application_name"], } outputs = { "drift_detected": "True when Argo CD reports one or more object diffs", "diffs": "Sanitized server-side diff records", "diff_count": "Number of diff records returned", } def is_available(self, sources: dict) -> bool: argocd = sources.get("argocd", {}) return bool( argocd.get("connection_verified") and argocd.get("base_url") and argocd.get("application_name") and (argocd.get("bearer_token") or (argocd.get("username") and argocd.get("password"))) ) def extract_params(self, sources: dict) -> dict[str, Any]: argocd = sources["argocd"] return { "base_url": argocd.get("base_url", ""), "bearer_token": argocd.get("bearer_token", ""), "username": argocd.get("username", ""), "password": argocd.get("password", ""), "application_name": argocd.get("application_name", ""), "project": argocd.get("project", ""), "app_namespace": argocd.get("app_namespace", ""), "verify_ssl": argocd.get("verify_ssl", True), } def run( self, base_url: str, application_name: str, bearer_token: str = "", username: str = "", password: str = "", project: str = "", app_namespace: str = "", verify_ssl: bool = True, **_kwargs: Any, ) -> dict[str, Any]: client = make_argocd_client( base_url, bearer_token, username, password, project=project, app_namespace=app_namespace, verify_ssl=verify_ssl, ) if client is None: return { "source": "argocd", "available": False, "error": "Argo CD integration is not configured (missing base_url or auth).", "application_name": application_name, "drift_detected": False, "diffs": [], "diff_count": 0, } with client: result = client.get_application_diff( application_name, project=project, app_namespace=app_namespace, ) if not result.get("success"): return { "source": "argocd", "available": False, "error": result.get("error", "unknown error"), "application_name": application_name, "drift_detected": False, "diffs": [], "diff_count": 0, } return { "source": "argocd", "available": True, **result, } argocd_application_diff = ArgoCDApplicationDiffTool() # ======== from tools/argocd_application_status_tool/ ======== """Argo CD application status investigation tool.""" from core.tool_framework.base import BaseTool class ArgoCDApplicationStatusTool(BaseTool): """Fetch Argo CD application sync and health status.""" name = "argocd_application_status" source = "argocd" description = ( "Fetch Argo CD application sync status, health status, current revision, " "and recent deployment history." ) use_cases = [ "Checking whether a GitOps application is OutOfSync or Degraded", "Correlating an incident with a recent Argo CD deployment revision", "Listing visible Argo CD applications when an alert omits the application name", ] requires = ["base_url"] injected_params = ["base_url", "password", "username"] input_schema = { "type": "object", "properties": { "base_url": {"type": "string", "description": "Argo CD base URL"}, "bearer_token": {"type": "string", "default": "", "description": "Argo CD API token"}, "username": {"type": "string", "default": "", "description": "Argo CD username"}, "password": {"type": "string", "default": "", "description": "Argo CD password"}, "application_name": { "type": "string", "default": "", "description": "Application name", }, "project": {"type": "string", "default": "", "description": "Optional Argo CD project"}, "app_namespace": { "type": "string", "default": "", "description": "Optional app namespace", }, "verify_ssl": { "type": "boolean", "default": True, "description": "Verify TLS certificates", }, }, "required": ["base_url"], } outputs = { "application": "Application status summary when application_name is provided", "applications": "Application list when application_name is omitted", "recent_history": "Recent Argo CD deployment history entries", } def is_available(self, sources: dict) -> bool: argocd = sources.get("argocd", {}) return bool( argocd.get("connection_verified") and argocd.get("base_url") and (argocd.get("bearer_token") or (argocd.get("username") and argocd.get("password"))) ) def extract_params(self, sources: dict) -> dict[str, Any]: argocd = sources["argocd"] return { "base_url": argocd.get("base_url", ""), "bearer_token": argocd.get("bearer_token", ""), "username": argocd.get("username", ""), "password": argocd.get("password", ""), "application_name": argocd.get("application_name", ""), "project": argocd.get("project", ""), "app_namespace": argocd.get("app_namespace", ""), "verify_ssl": argocd.get("verify_ssl", True), } def run( self, base_url: str, bearer_token: str = "", username: str = "", password: str = "", application_name: str = "", project: str = "", app_namespace: str = "", verify_ssl: bool = True, **_kwargs: Any, ) -> dict[str, Any]: client = make_argocd_client( base_url, bearer_token, username, password, project=project, app_namespace=app_namespace, verify_ssl=verify_ssl, ) if client is None: return { "source": "argocd", "available": False, "error": "Argo CD integration is not configured (missing base_url or auth).", "application": {}, "applications": [], "recent_history": [], } with client: if application_name: result = client.get_application_summary( application_name, project=project, app_namespace=app_namespace, ) else: result = client.list_applications(projects=[project] if project else None) if not result.get("success"): return { "source": "argocd", "available": False, "error": result.get("error", "unknown error"), "application": {}, "applications": [], "recent_history": [], } return { "source": "argocd", "available": True, **result, } argocd_application_status = ArgoCDApplicationStatusTool()