diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py
index fa3d2313..320018d7 100644
--- a/apps/api/src/jobs/asset_scanner_job.py
+++ b/apps/api/src/jobs/asset_scanner_job.py
@@ -169,34 +169,30 @@ async def scan_once(
 
 async def _collect_k8s_assets() -> list[dict[str, Any]]:
     """
-    用 K8sProvider 列出全 namespace 的 pods,轉成 asset_inventory 結構。
+    直接 subprocess 執行 kubectl get pods --all-namespaces.
+
+    2026-04-19 ogt + Claude Opus 4.7 v2: 不走 K8sProvider._kubectl_get,
+    因為它把 namespace 參數當 -n 旗標,無法處理 '--all-namespaces'.
+    直接 subprocess 最可靠 + Pod 內 /usr/local/bin/kubectl 已確認可用.
 
     回傳每筆: {asset_key, asset_type, host, namespace, name, metadata, tags}
     """
-    from src.plugins.mcp.providers.k8s_provider import K8sProvider
-
-    provider = K8sProvider()
-    result = await asyncio.wait_for(
-        provider.execute(
-            tool_name="kubectl_get",
-            parameters={"namespace": "--all-namespaces", "resource": "pods"},
+    proc = await asyncio.wait_for(
+        asyncio.create_subprocess_exec(
+            "kubectl", "get", "pods", "--all-namespaces", "-o", "json",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
         ),
         timeout=_KUBECTL_TIMEOUT_SEC,
     )
-    if not result.success:
-        raise RuntimeError(f"kubectl get pods failed: {result.error}")
+    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=_KUBECTL_TIMEOUT_SEC)
+    if proc.returncode != 0:
+        raise RuntimeError(f"kubectl failed rc={proc.returncode}: {stderr.decode('utf-8', errors='replace')[:500]}")
 
-    raw = result.output
-    # k8s_provider _kubectl_get 回傳 stdout 字串 (line 299)
-    if isinstance(raw, str):
-        try:
-            payload = _json.loads(raw)
-        except _json.JSONDecodeError as e:
-            raise RuntimeError(f"kubectl JSON parse failed: {e}") from e
-    elif isinstance(raw, dict):
-        payload = raw
-    else:
-        raise RuntimeError(f"unexpected kubectl output type: {type(raw)}")
+    try:
+        payload = _json.loads(stdout.decode("utf-8", errors="replace"))
+    except _json.JSONDecodeError as e:
+        raise RuntimeError(f"kubectl JSON parse failed: {e}") from e
 
     items = payload.get("items", []) if isinstance(payload, dict) else []
     assets: list[dict[str, Any]] = []