ask-azure
Answer questions about Azure resources by generating and executing KQL queries against Azure Resource Graph.
Instructions
Answer a question by generating and running an Azure Resource Graph KQL query.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| question | Yes | Your natural-language question about Azure resources | |
| tenant_name | No | Optional configured tenant name | |
| subscription_ids | No | Optional explicit subscription IDs | |
| use_all_subscriptions | No | If no subscriptions are provided, attempt to auto-discover all accessible subscriptions (default: true) | |
| auto_execute | No | Execute the generated KQL automatically (default: true) |
Implementation Reference
- Main handler logic for the 'ask-azure' tool: processes input arguments, generates KQL query from natural language question using generate_kql, determines scope (subscriptions or management group), executes the query using execute_kql, formats results, and returns text content.
if name == "ask-azure": tenant_name = arguments.get("tenant_name") subs = arguments.get("subscription_ids") use_all = bool(arguments.get("use_all_subscriptions", True)) auto_execute = bool(arguments.get("auto_execute", True)) question = arguments.get("question") if not question: return [types.TextContent(type="text", text="Error: question is required")] kql, meta = generate_kql(question) header = [ "Proposed Azure Resource Graph Query:", f"Intent: {meta.get('intent','generic')}", "", "KQL:", kql, ] if not auto_execute: header.append("\nSet auto_execute=true to run this query.") return [types.TextContent(type="text", text="\n".join(header))] # Auto-guess tenant if not provided if not tenant_name: tenant_name = _guess_tenant_name_from_text(question) cred, default_subs = AZURE_CONFIG.get_credentials(tenant_name) mg = AZURE_CONFIG.get_management_group_id(tenant_name) if subs: pass elif use_all and mg: subs = [] # force MG usage below elif use_all: discovered = _enumerate_subscriptions_for_credential(cred) subs = discovered or default_subs else: subs = default_subs if not subs and not (use_all and mg): header.append( "\nError: No subscriptions available; provide subscription_ids, set default_subscription_id in config, or ensure subscription discovery is possible (azure-mgmt-subscription installed and SP has access)." ) return [types.TextContent(type="text", text="\n".join(header))] if use_all and mg and not subs: result = execute_kql(cred, None, kql, top=100, management_groups=[mg]) else: result = execute_kql(cred, subs, kql, top=100) if result["status"] != "success": header.append(f"\nExecution failed: {result.get('error','unknown error')}") return [types.TextContent(type="text", text="\n".join(header))] rows = result["results"] scope_line = ( f"Scope: managementGroup={mg}" if (use_all and mg and not subs) else f"Subscriptions used: {len(subs)}" ) body = [ *header, "", f"Rows: {result['result_count']}", f"Tenant: {tenant_name or AZURE_CONFIG.get_default_tenant().get('name')}", scope_line, _format_rows(rows), ] return [types.TextContent(type="text", text="\n".join(body))] - src/azure_assistant_mcp/server.py:98-112 (registration)Registration of the 'ask-azure' tool in the list_tools() function, including its description and input schema definition.
types.Tool( name="ask-azure", description="Answer a question by generating and running an Azure Resource Graph KQL query.", inputSchema={ "type": "object", "properties": { "question": {"type": "string", "description": "Your natural-language question about Azure resources"}, "tenant_name": {"type": "string", "description": "Optional configured tenant name"}, "subscription_ids": {"type": "array", "items": {"type": "string"}, "description": "Optional explicit subscription IDs"}, "use_all_subscriptions": {"type": "boolean", "description": "If no subscriptions are provided, attempt to auto-discover all accessible subscriptions (default: true)", "default": True}, "auto_execute": {"type": "boolean", "description": "Execute the generated KQL automatically (default: true)", "default": True} }, "required": ["question"], }, ), - Helper function generate_kql that heuristically maps natural language questions to predefined KQL queries based on keyword matching, returning the KQL string and metadata.
def generate_kql(question: str) -> Tuple[str, Dict[str, str]]: """Very simple heuristic NL -> KQL generator. Returns (kql_query, metadata) where metadata includes 'intent'. """ q = question.strip().lower() meta: Dict[str, str] = {"intent": "generic"} # Common VM queries if ("vm" in q or "virtual machine" in q) and ("stopped" in q or "deallocated" in q): meta["intent"] = "stopped_vms" kql = stopped_vms_kql() return kql, meta # Subscriptions list if ("subscription" in q) and ("list" in q or "all" in q or "show" in q): meta["intent"] = "list_subscriptions" kql = list_subscriptions_kql() return kql, meta # Resource groups without tags if ("resource group" in q or "resource groups" in q) and ("without tag" in q or "no tag" in q or "untagged" in q): meta["intent"] = "untagged_resource_groups" kql = untagged_resource_groups_kql() return kql, meta # Generic resource group listing if ("resource group" in q or "resource groups" in q) and ("list" in q or "show" in q or "all" in q or "any" in q or "find" in q): meta["intent"] = "list_resource_groups" kql = list_resource_groups_kql(limit=50) return kql, meta # Advisor recommendations if ("advisor" in q or "recommendation" in q): meta["intent"] = "advisor_recommendations" kql = advisor_recommendations_kql() return kql, meta # Policy non-compliance if ("policy" in q) and ("non-compliant" in q or "noncompliant" in q or "non compliant" in q or "compliance" in q): meta["intent"] = "policy_non_compliance" kql = policy_noncompliant_kql() return kql, meta # Health incidents/advisories if ("health" in q or "incident" in q or "advisories" in q or "advisory" in q): meta["intent"] = "service_health" kql = health_advisories_kql() return kql, meta # Manual changes last 30 days (resourcechanges) if ("manual" in q and "change" in q) and ("30" in q or "30d" in q or "last 30" in q or "month" in q): meta["intent"] = "manual_changes_30d" kql = manual_changes_kql(days=30) return kql, meta # Recent changes (resource level) if ("change" in q or "changed" in q or "changes" in q) and not ("resource group" in q): meta["intent"] = "resource_changes_recent" kql = resource_changes_recent_kql(days=30) return kql, meta # Recent changes (resource group / subscription level) if ("change" in q or "changed" in q or "changes" in q) and ("resource group" in q or "subscription" in q): meta["intent"] = "resource_container_changes_recent" kql = resource_container_changes_recent_kql(days=30) return kql, meta if ("public ip" in q or "internet" in q) and ("exposed" in q or "open" in q): meta["intent"] = "public_ips" kql = public_ips_kql() return kql, meta if ("storage" in q and ("unencrypted" in q or "not encrypted" in q)): meta["intent"] = "unencrypted_storage" kql = unencrypted_storage_kql() return kql, meta if ("keyvault" in q or "key vault" in q) and ("firewall" in q or "public" in q): meta["intent"] = "keyvault_network" kql = keyvault_network_kql() return kql, meta # Default generic listing kql = generic_list_resources_kql(limit=50) return kql, meta - src/azure_assistant_mcp/arg.py:7-50 (helper)Core helper function execute_kql that runs the KQL query against Azure Resource Graph using the ResourceGraphClient, supporting subscription or management group scopes, and returns structured results or error.
def execute_kql( credential, subscriptions: Optional[List[str]], kql_query: str, top: int = 100, management_groups: Optional[List[str]] = None, ) -> Dict[str, Any]: """Execute a KQL query against Azure Resource Graph. Returns a dict with keys: status, results, result_count, warnings, query """ client = ResourceGraphClient(credential=credential) # Build typed request for reliability options = QueryRequestOptions(result_format="objectArray", top=top) # Either subscriptions or management_groups must be provided. request = QueryRequest( subscriptions=subscriptions, management_groups=management_groups, query=kql_query, options=options, ) try: response = client.resources(request) # response.data is a list[dict] when result_format=objectArray rows = list(response.data or []) return { "status": "success", "results": rows, "result_count": len(rows), "warnings": [], "query": kql_query, } except Exception as e: return { "status": "error", "error": str(e), "results": [], "result_count": 0, "warnings": [], "query": kql_query, }