Azure Assistant MCP

Instructions

Implementation Reference

• src/azure_assistant_mcp/server.py:429-491 (handler)

  Main handler logic for the 'ask-azure' tool within the call_tool dispatcher. Generates KQL from NL question, executes it (if auto_execute), scopes subscriptions/management group, formats results.
  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:101-110 (schema)

  Input schema definition for the 'ask-azure' tool, defining parameters like question, tenant, subscriptions, and execution options.
  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"],
• src/azure_assistant_mcp/server.py:98-112 (registration)

  Tool registration for 'ask-azure' in the list_tools() function, including name, description, and schema.
  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"], }, ),
• src/azure_assistant_mcp/nl2kql.py:21-106 (helper)

  Core helper function called by 'ask-azure' handler to heuristically generate KQL query from natural language question based on keyword matching.
  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-49 (helper)

  Helper function called by 'ask-azure' handler to execute the generated KQL against Azure Resource Graph, supporting subscription or management group scoping.
  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, }