run-arg-kql
Execute KQL queries against Azure Resource Graph to retrieve and analyze Azure resource information across subscriptions and tenants.
Instructions
Run a provided KQL query against Azure Resource Graph.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| kql_query | Yes | KQL to execute (must reference valid ARG tables) | |
| subscription_ids | No | Optional explicit subscription IDs | |
| tenant_name | No | Optional configured tenant name | |
| top | No | Max rows to return (default: 100) | |
| use_all_subscriptions | No | If no subscriptions are provided, attempt to auto-discover all accessible subscriptions (default: true) |
Implementation Reference
- Tool handler for 'run-arg-kql': processes input arguments, resolves tenant and subscription scope (auto-discovery or management group), executes the KQL query via execute_kql, formats and returns results.if name == "run-arg-kql": tenant_name = arguments.get("tenant_name") subs = arguments.get("subscription_ids") use_all = bool(arguments.get("use_all_subscriptions", True)) top = int(arguments.get("top", 100)) kql = arguments.get("kql_query") if not kql: return [types.TextContent(type="text", text="Error: kql_query is required")] # Auto-guess tenant if not provided if not tenant_name: tenant_name = _guess_tenant_name_from_text(kql) 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 # Only error if neither subscriptions nor MG scope is available if not subs and not (use_all and mg): return [types.TextContent(type="text", text=( "Error: 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)." ))] # Try management group scope if no subs and mg configured if use_all and mg and not subs: result = execute_kql(cred, None, kql, top=top, management_groups=[mg]) else: result = execute_kql(cred, subs, kql, top=top) if result["status"] != "success": return [types.TextContent(type="text", text=f"ARG query failed: {result.get('error','unknown error')}")] rows = result["results"] scope_line = ( f"Scope: managementGroup={mg}" if (use_all and mg and not subs) else f"Subscriptions used: {len(subs)}" ) body = [ "Azure Resource Graph Query Results:", f"Rows: {result['result_count']}", f"Tenant: {tenant_name or AZURE_CONFIG.get_default_tenant().get('name')}", scope_line, "", _format_rows(rows), "", "KQL:", result["query"][:1000], ] return [types.TextContent(type="text", text="\n".join(body))]
- src/azure_assistant_mcp/server.py:122-136 (registration)Registration of the 'run-arg-kql' tool in list_tools(), including its schema definition for input validation.types.Tool( name="run-arg-kql", description="Run a provided KQL query against Azure Resource Graph.", inputSchema={ "type": "object", "properties": { "kql_query": {"type": "string", "description": "KQL to execute (must reference valid ARG tables)"}, "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}, "top": {"type": "integer", "description": "Max rows to return (default: 100)", "default": 100} }, "required": ["kql_query"], }, ),
- Input schema definition for the 'run-arg-kql' tool.inputSchema={ "type": "object", "properties": { "kql_query": {"type": "string", "description": "KQL to execute (must reference valid ARG tables)"}, "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}, "top": {"type": "integer", "description": "Max rows to return (default: 100)", "default": 100} }, "required": ["kql_query"], }, ),
- src/azure_assistant_mcp/arg.py:7-49 (helper)Core helper function execute_kql that performs the actual Azure Resource Graph query execution using ResourceGraphClient, handles subscriptions or management groups scope, 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, }