【2026最新】uv + MCP + gemma2で爆速・安全！「勝手にファイル操作させない」自作AIエージェントの境地 🚀

import os
import ast
import json
import httpx
import re
import datetime
from pathlib import Path
from typing import List, Dict, Optional

# --- 設定エリア ---
MODEL_NAME = "gemma2"
OLLAMA_URL = "http://localhost:11434/api/generate"

# サブフォルダまで深くスキャンするかどうか
RECURSIVE_SCAN = True 

# 操作を許可するディレクトリ
ALLOWED_FOLDERS = ["."]

# プロジェクトのルート（AIGURUI仕様）
PROJECT_ROOT = Path(r"C:\Users\AIGURUI\Desktop\作業場")

def log(message: str, level: str = "INFO"):
    now = datetime.datetime.now().strftime("%H:%M:%S")
    icon = {"SUCCESS": "✨", "ERROR": "🛑", "AI": "🤖", "SCAN": "🔍"}.get(level, "💡")
    print(f"[{now}] {icon} {message}")

class AiguruiGuardrail:
    def __init__(self, root_path: Path, allowed_dirs: List[str]):
        self.root_path = root_path
        self.allowed_paths = [(root_path / d).resolve() for d in allowed_dirs]

    def is_write_allowed(self, target_path: str) -> bool:
        log(f"ガードレール検問中: {target_path}", "SCAN")
        try:
            full_path = (self.root_path / target_path).resolve()
            for allowed in self.allowed_paths:
                if str(full_path).startswith(str(allowed)):
                    log(f"アクセス許可を確認: {full_path}", "SUCCESS")
                    return True
            log(f"アクセス拒否: {full_path} は許可範囲外です", "ERROR")
            return False
        except Exception as e:
            log(f"パス解析エラー: {e}", "ERROR")
            return False

class RepositoryScanner:
    def __init__(self, root_path: Path):
        self.root_path = root_path

    def get_repo_map(self) -> str:
        log(f"スキャン開始: {self.root_path.absolute()}", "SCAN")
        if not self.root_path.exists():
            return "（フォルダなし）"

        repo_map = []
        exclude_names = {".git", "__pycache__", ".venv", "node_modules", ".cursor", ".vscode", "Originals", "Thumbnails"}
        file_count = 0
        
        walker = os.walk(self.root_path) if RECURSIVE_SCAN else [(str(self.root_path), [], [f.name for f in os.scandir(self.root_path) if f.is_file()])]

        for root, dirs, files in walker:
            dirs[:] = [d for d in dirs if not d.startswith('.') and d not in exclude_names]
            for file in files:
                if file.startswith('.') or file.lower() in ['desktop.ini', 'thumbs.db']:
                    continue
                file_path = Path(root) / file
                rel_path = file_path.relative_to(self.root_path)
                file_count += 1
                
                if file_path.suffix == ".py":
                    summary = self._summarize_python(file_path)
                    repo_map.append(f"📄 {rel_path} (Python)\n{summary}")
                else:
                    repo_map.append(f"📄 {rel_path} ({file_path.suffix})")

                if file_count >= 500: break
        
        log(f"スキャン完了。ファイル数: {file_count}個", "SUCCESS")
        return "\n\n".join(repo_map)

    def _summarize_python(self, file_path: Path) -> str:
        try:
            with open(file_path, "utf-8") as f:
                tree = ast.parse(f.read())
            definitions = [f"  - {n.name}" for n in tree.body if isinstance(n, (ast.FunctionDef, ast.ClassDef))]
            return "\n".join(definitions) if definitions else "  (No structure)"
        except:
            return "  (Parse Error)"

async def ask_ai(prompt: str, context: str):
    system_prompt = (
        f"あなたは凄腕エンジニアAIです。\n【現状】\n{context}\n\n"
        f"【ルール】1.回答は日本語 2.操作時はJSONフォーマットを出力すること"
    )
    payload = {"model": MODEL_NAME, "prompt": f"{system_prompt}\n\n依頼: {prompt}", "stream": False}
    async with httpx.AsyncClient(timeout=180.0) as client:
        response = await client.post(OLLAMA_URL, json=payload)
        return response.json().get("response", "")

def extract_json(text: str) -> Optional[Dict]:
    match = re.search(r'\{.*\}', text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group().replace("'", '"'))
        except:
            return None
    return None

async def main():
    log(f"AIGURUI-Agent 起動！ (Model: {MODEL_NAME})", "SUCCESS")
    guard = AiguruiGuardrail(PROJECT_ROOT, ALLOWED_FOLDERS)
    scanner = RepositoryScanner(PROJECT_ROOT)

    while True:
        user_input = input("\n🎤 命令を入力してね (exitで終了): ")
        if user_input.lower() in ["exit", "quit", "終了"]: break
        
        repo_map = scanner.get_repo_map()
        raw_response = await ask_ai(user_input, repo_map)
        
        clean_text = re.sub(r'\{.*\}', '', raw_response, flags=re.DOTALL).strip()
        if clean_text: print(f"\n🤖 AI: {clean_text}")

        command = extract_json(raw_response)
        if command and command.get("action") == "write":
            target_path = command.get("path")
            if guard.is_write_allowed(target_path):
                print(f"\n--- 📝 書き込み提案 ---\n{command.get('content')}\n---")
                if input(f"🤔 {target_path} を操作してもいい？ (y/n): ").lower() == 'y':
                    full_target = PROJECT_ROOT / target_path
                    full_target.parent.mkdir(parents=True, exist_ok=True)
                    with open(full_target, "w", encoding="utf-8") as f:
                        f.write(str(command.get('content')))
                    log("成功！", "SUCCESS")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

【ここがポイント】
相対パスを「絶対パス」に変換し、それが許可された PROJECT_ROOT 配下にあるかを文字列前方一致でチェックしているよ。これで、AIがどんなに「脱獄」しようとしても、作業場の外には一歩も出られないんだ。