WIN-JHFT4D3SIVT\caoxiaozhu
a3fe4d24fc
Phase 7: Built-in Hooks (audit_log, dangerous_confirmation, security_scan) Phase 8: Plugin system (PluginManager, PluginSandbox, PluginManifest) Phase 9: Skills registry (SkillRegistry, local/plugin/MCP loaders) Phase 10: TeamLeader, RemoteTransport, BackgroundTaskManager
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
"""Hook 配置持久化 - Phase 7.3"""
|
||
|
||
import json
|
||
import os
|
||
from dataclasses import asdict, dataclass
|
||
from typing import Any
|
||
|
||
from app.agents.tools.hooks.manager import get_hook_manager
|
||
|
||
|
||
@dataclass
|
||
class HookConfigEntry:
|
||
"""Hook 配置条目"""
|
||
|
||
name: str
|
||
hook_type: str
|
||
enabled: bool
|
||
tool_names: list[str] | None = None
|
||
categories: list[str] | None = None
|
||
priority: int = 0
|
||
|
||
|
||
class HookConfigPersistence:
|
||
"""Hook 配置持久化"""
|
||
|
||
def __init__(self, config_path: str | None = None):
|
||
"""
|
||
Args:
|
||
config_path: 配置文件路径，None 则使用默认路径
|
||
"""
|
||
if config_path is None:
|
||
config_path = os.path.join(
|
||
os.path.dirname(__file__), "..", "..", "..", "..", "config", "hooks.json"
|
||
)
|
||
self.config_path = config_path
|
||
|
||
def load_config(self) -> list[HookConfigEntry]:
|
||
"""从文件加载 Hook 配置"""
|
||
if not os.path.exists(self.config_path):
|
||
return []
|
||
|
||
try:
|
||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return [HookConfigEntry(**entry) for entry in data]
|
||
except Exception:
|
||
return []
|
||
|
||
def save_config(self, entries: list[HookConfigEntry]) -> bool:
|
||
"""保存 Hook 配置到文件"""
|
||
try:
|
||
os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
|
||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||
json.dump([asdict(e) for e in entries], f, indent=2, ensure_ascii=False)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
def apply_config(self) -> int:
|
||
"""应用配置到 HookManager
|
||
|
||
Returns:
|
||
应用的 Hook 数量
|
||
"""
|
||
from app.agents.tools.hooks.types import HookType
|
||
|
||
manager = get_hook_manager()
|
||
entries = self.load_config()
|
||
count = 0
|
||
|
||
for entry in entries:
|
||
if entry.enabled:
|
||
from app.agents.tools.hooks.types import HookDefinition, HookTrigger
|
||
|
||
trigger = HookTrigger(
|
||
tool_names=entry.tool_names,
|
||
categories=entry.categories,
|
||
)
|
||
|
||
# 创建空的 handler，只是注册配置
|
||
hook_def = HookDefinition(
|
||
name=entry.name,
|
||
hook_type=HookType(entry.hook_type),
|
||
trigger=trigger,
|
||
handler=lambda ctx, *args: ctx,
|
||
priority=entry.priority,
|
||
enabled=True,
|
||
)
|
||
|
||
manager.register(hook_def)
|
||
count += 1
|
||
|
||
return count
|
||
|
||
|
||
# 全局单例
|
||
_persistence: HookConfigPersistence | None = None
|
||
|
||
|
||
def get_hook_config_persistence() -> HookConfigPersistence:
|
||
"""获取全局 Hook 配置持久化实例"""
|
||
global _persistence
|
||
if _persistence is None:
|
||
_persistence = HookConfigPersistence()
|
||
return _persistence
|