242 lines
6.9 KiB
Python
242 lines
6.9 KiB
Python
"""Hook API 路由 - Phase 7.5"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from app.agents.tools.hooks import HookType
|
|
from app.agents.tools.hooks.builtins import (
|
|
AuditLogHook,
|
|
DangerousConfirmationHook,
|
|
SecurityScanHook,
|
|
)
|
|
from app.agents.tools.hooks.config import (
|
|
HookConfigEntry,
|
|
get_hook_config_persistence,
|
|
)
|
|
from app.agents.tools.hooks.manager import get_hook_manager
|
|
|
|
router = APIRouter(prefix="/api/hooks", tags=["Hooks"])
|
|
|
|
|
|
class HookInfo(BaseModel):
|
|
"""Hook 信息"""
|
|
|
|
name: str
|
|
hook_type: str
|
|
description: str
|
|
builtin: bool
|
|
|
|
|
|
class HookConfigUpdate(BaseModel):
|
|
"""更新 Hook 配置"""
|
|
|
|
entries: list[HookConfigEntry]
|
|
|
|
|
|
class HookConfigResponse(BaseModel):
|
|
"""Hook 配置响应"""
|
|
|
|
entries: list[dict[str, Any]]
|
|
count: int
|
|
|
|
|
|
class HookStatusResponse(BaseModel):
|
|
"""Hook 状态响应"""
|
|
|
|
name: str
|
|
enabled: bool
|
|
hook_type: str
|
|
registered: bool
|
|
|
|
|
|
# 内置 Hook 注册表
|
|
BUILTIN_HOOKS: dict[str, dict[str, str]] = {
|
|
"audit_log": {
|
|
"name": "audit_log",
|
|
"hook_type": "pre_tool_use,post_tool_use,tool_error",
|
|
"description": "审计日志 Hook - 记录所有工具调用",
|
|
"class": "AuditLogHook",
|
|
},
|
|
"dangerous_confirmation": {
|
|
"name": "dangerous_confirmation",
|
|
"hook_type": "pre_tool_use",
|
|
"description": "危险操作确认 Hook - 拦截危险工具调用",
|
|
"class": "DangerousConfirmationHook",
|
|
},
|
|
"security_scan": {
|
|
"name": "security_scan",
|
|
"hook_type": "post_tool_use",
|
|
"description": "安全扫描 Hook - 检测敏感信息泄露",
|
|
"class": "SecurityScanHook",
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/available", response_model=list[HookInfo])
|
|
async def list_available_hooks() -> list[HookInfo]:
|
|
"""列出所有可用的内置 Hook"""
|
|
return [
|
|
HookInfo(
|
|
name=info["name"],
|
|
hook_type=info["hook_type"],
|
|
description=info["description"],
|
|
builtin=True,
|
|
)
|
|
for info in BUILTIN_HOOKS.values()
|
|
]
|
|
|
|
|
|
@router.get("/config", response_model=HookConfigResponse)
|
|
async def get_hook_config() -> HookConfigResponse:
|
|
"""获取当前 Hook 配置"""
|
|
persistence = get_hook_config_persistence()
|
|
entries = persistence.load_config()
|
|
return HookConfigResponse(
|
|
entries=[vars(e) if isinstance(e, HookConfigEntry) else e for e in entries],
|
|
count=len(entries),
|
|
)
|
|
|
|
|
|
@router.post("/config", response_model=HookConfigResponse)
|
|
async def update_hook_config(
|
|
entries: list[HookConfigEntry],
|
|
) -> HookConfigResponse:
|
|
"""更新 Hook 配置"""
|
|
persistence = get_hook_config_persistence()
|
|
success = persistence.save_config(entries)
|
|
if not success:
|
|
raise HTTPException(status_code=500, detail="Failed to save hook config")
|
|
|
|
# 应用配置到 HookManager
|
|
manager = get_hook_manager()
|
|
manager.clear() # 清除旧配置
|
|
persistence.apply_config() # 应用新配置
|
|
|
|
return HookConfigResponse(
|
|
entries=[vars(e) if isinstance(e, HookConfigEntry) else e for e in entries],
|
|
count=len(entries),
|
|
)
|
|
|
|
|
|
@router.post("/apply-config", response_model=dict[str, Any])
|
|
async def apply_hook_config() -> dict[str, Any]:
|
|
"""应用配置文件到 HookManager"""
|
|
persistence = get_hook_config_persistence()
|
|
manager = get_hook_manager()
|
|
manager.clear()
|
|
count = persistence.apply_config()
|
|
return {"applied": count, "message": f"Applied {count} hook configurations"}
|
|
|
|
|
|
@router.get("/status", response_model=list[HookStatusResponse])
|
|
async def get_hook_status() -> list[HookStatusResponse]:
|
|
"""获取所有已注册 Hook 的状态"""
|
|
manager = get_hook_manager()
|
|
all_hooks = manager.list_all()
|
|
|
|
# 按名称索引已注册的 hooks
|
|
registered: dict[str, dict[str, Any]] = {}
|
|
for hook in all_hooks:
|
|
registered[hook.name] = {
|
|
"name": hook.name,
|
|
"enabled": hook.enabled,
|
|
"hook_type": hook.hook_type.value,
|
|
"registered": True,
|
|
}
|
|
|
|
# 合并内置 Hook 信息
|
|
result: list[HookStatusResponse] = []
|
|
seen: set[str] = set()
|
|
|
|
# 先添加已注册的
|
|
for hook in all_hooks:
|
|
result.append(
|
|
HookStatusResponse(
|
|
name=hook.name,
|
|
enabled=hook.enabled,
|
|
hook_type=hook.hook_type.value,
|
|
registered=True,
|
|
)
|
|
)
|
|
seen.add(hook.name)
|
|
|
|
# 再添加内置但未注册的
|
|
for name, info in BUILTIN_HOOKS.items():
|
|
if name not in seen:
|
|
result.append(
|
|
HookStatusResponse(
|
|
name=name,
|
|
enabled=False,
|
|
hook_type=info["hook_type"],
|
|
registered=False,
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
@router.post("/{name}/enable", response_model=dict[str, str])
|
|
async def enable_hook(name: str) -> dict[str, str]:
|
|
"""启用指定 Hook"""
|
|
manager = get_hook_manager()
|
|
if manager.enable(name):
|
|
return {"status": "enabled", "name": name}
|
|
raise HTTPException(status_code=404, detail=f"Hook '{name}' not found")
|
|
|
|
|
|
@router.post("/{name}/disable", response_model=dict[str, str])
|
|
async def disable_hook(name: str) -> dict[str, str]:
|
|
"""禁用指定 Hook"""
|
|
manager = get_hook_manager()
|
|
if manager.disable(name):
|
|
return {"status": "disabled", "name": name}
|
|
raise HTTPException(status_code=404, detail=f"Hook '{name}' not found")
|
|
|
|
|
|
@router.post("/register-builtin", response_model=dict[str, str])
|
|
async def register_builtin_hook(
|
|
name: str,
|
|
hook_type: str = "pre_tool_use",
|
|
) -> dict[str, str]:
|
|
"""注册内置 Hook 到 HookManager"""
|
|
from app.agents.tools.hooks.types import HookDefinition, HookTrigger
|
|
|
|
manager = get_hook_manager()
|
|
|
|
if name == "audit_log":
|
|
hook_instance = AuditLogHook()
|
|
handler = hook_instance.pre_tool_use
|
|
hook_types = [HookType.PRE_TOOL_USE, HookType.POST_TOOL_USE, HookType.TOOL_ERROR]
|
|
elif name == "dangerous_confirmation":
|
|
hook_instance = DangerousConfirmationHook()
|
|
handler = hook_instance.pre_tool_use
|
|
hook_types = [HookType.PRE_TOOL_USE]
|
|
elif name == "security_scan":
|
|
hook_instance = SecurityScanHook()
|
|
handler = hook_instance.post_tool_use
|
|
hook_types = [HookType.POST_TOOL_USE]
|
|
else:
|
|
raise HTTPException(status_code=404, detail=f"Unknown builtin hook: {name}")
|
|
|
|
registered = []
|
|
for ht in hook_types:
|
|
hook_def = HookDefinition(
|
|
name=f"{name}_{ht.value}",
|
|
hook_type=ht,
|
|
trigger=HookTrigger(),
|
|
handler=handler,
|
|
priority=0,
|
|
enabled=True,
|
|
description=f"Built-in {name} hook",
|
|
)
|
|
manager.register(hook_def)
|
|
registered.append(ht.value)
|
|
|
|
return {
|
|
"status": "registered",
|
|
"name": name,
|
|
"hook_types": ", ".join(registered),
|
|
}
|