GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-22 08:43:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(agent): add host policy foundation (#6273)

Browse Source
This commit is contained in:
InfinityPacer
2026-08-12 10:48:14 +08:00
committed by GitHub
parent a1dd259143
commit 010899f369
23 changed files with 4951 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
app/agent/__init__.py
View File

@@ -39,6 +39,7 @@ from app.agent.middleware.jobs import (
)
from app.agent.middleware.memory import MemoryMiddleware
from app.agent.middleware.patch_tool_calls import PatchToolCallsMiddleware
from app.agent.middleware.policy import AgentPolicyMiddleware
from app.agent.middleware.runtime_config import RuntimeConfigMiddleware
from app.agent.middleware.skills import SKILL_TOOL_NAME, SkillsMiddleware
from app.agent.middleware.subagents import (
@@ -50,6 +51,12 @@ from app.agent.middleware.subagents import (
from app.agent.middleware.tool_selection import ToolSelectorMiddleware
from app.agent.middleware.usage import UsageMiddleware
from app.agent.prompt import prompt_manager
from app.agent.policy import (
AuthSource,
PrincipalType,
ToolOrigin,
ToolPolicyContext,
)
from app.agent.runtime import agent_runtime_manager
from app.agent.mcp import agent_mcp_manager
from app.agent.tools.factory import MoviePilotToolFactory
@@ -727,6 +734,40 @@ class MoviePilotAgent:
"original_chat_id": None if self.is_background else self.original_chat_id,
}
def _build_policy_context(self) -> ToolPolicyContext:
"""根据宿主入口建立模型参数无法伪造的策略上下文。"""
if not self.has_message_context:
origin = ToolOrigin.BACKGROUND
principal_type = PrincipalType.BACKGROUND
auth_source = AuthSource.INTERNAL
elif self.channel == MessageChannel.Web.value and self.source in {
"openai",
"openai.responses",
"anthropic",
}:
origin = ToolOrigin.AGENT_API
principal_type = PrincipalType.SYSTEM_ADMIN_INTEGRATION
auth_source = AuthSource.API_TOKEN
else:
origin = ToolOrigin.AGENT_INTERACTIVE
principal_type = PrincipalType.HUMAN
auth_source = (
AuthSource.WEB_SESSION
if self.channel
in {MessageChannel.Web.value, MessageChannel.WebAgent.value}
else AuthSource.CHANNEL
)
return ToolPolicyContext(
session_id=self.session_id,
user_id=str(self.user_id or self.username or principal_type.value),
origin=origin,
principal_type=principal_type,
auth_source=auth_source,
agent_context=self._tool_context,
channel=self.channel,
source=self.source,
)
def _should_stream(self) -> bool:
"""
判断是否应启用流式输出：
@@ -1255,6 +1296,7 @@ class MoviePilotAgent:
# LLM 模型（用于 agent 执行）
agent_model = await self._initialize_llm(streaming=streaming)
self._sync_model_profile(agent_model)
# 供应商原生工具不进入本地 ToolNode，宿主策略只覆盖 client-side tools。
server_tools = LLMHelper.get_server_tools(agent_model)
use_local_web_search = LLMHelper.should_use_local_web_search(agent_model)
@@ -1292,11 +1334,13 @@ class MoviePilotAgent:
enabled=use_local_web_search,
)
subagent_tools.extend(await self._initialize_subagent_mcp_tools())
policy_context = self._build_policy_context()
subagent_middlewares, subagent_task_tools = create_subagent_middlewares(
model=non_streaming_model,
tools=subagent_tools,
server_tools=server_tools,
stream_handler=self.stream_handler,
policy_context=policy_context.for_subagent(),
)
max_tools = settings.LLM_MAX_TOOLS
always_include_tools = (
@@ -1324,6 +1368,8 @@ class MoviePilotAgent:
# 中间件
middlewares = [
# 宿主策略必须位于最外层，确保插件覆盖工具基类也不能绕过。
AgentPolicyMiddleware(context=policy_context),
# Skills
skills_middleware,
# Jobs 任务管理
@@ -1334,6 +1380,8 @@ class MoviePilotAgent:
RuntimeConfigMiddleware(),
# 记忆管理
MemoryMiddleware(memory_dir=str(agent_runtime_manager.memory_dir)),
# 活动日志依赖记忆上下文，并应在摘要压缩前完成读取与记录。
*([activity_log_middleware] if activity_log_middleware else []),
# 上下文压缩
SummarizationMiddleware(
model=non_streaming_model, trigger=("fraction", 0.85)
@@ -1346,12 +1394,6 @@ class MoviePilotAgent:
UsageMiddleware(on_usage=self._record_usage),
]
if self.has_message_context:
middlewares.insert(
4,
activity_log_middleware,
)
# 工具选择
if max_tools > 0:
middlewares.append(

9
app/agent/callback/__init__.py
View File

@@ -4,6 +4,7 @@ from typing import Any, Optional, Tuple
from fastapi.concurrency import run_in_threadpool
from app.agent.policy import sanitize_for_host
from app.chain import ChainBase
from app.log import logger
from app.schemas import Notification
@@ -256,10 +257,14 @@ class StreamingHandler:
"""
记录一次工具调用，供非啰嗦模式下延迟汇总输出。
"""
recorded_message = sanitize_for_host(tool_message) if tool_message else tool_message
recorded_args = sanitize_for_host(tool_kwargs or {})
if not isinstance(recorded_args, dict):
recorded_args = {}
category, target = self._classify_tool_call(
tool_name=tool_name,
tool_message=tool_message,
tool_kwargs=tool_kwargs or {},
tool_message=recorded_message,
tool_kwargs=recorded_args,
)
target_values = []
if isinstance(target, (list, tuple, set)):

55
app/agent/middleware/activity_log.py
View File

@@ -33,6 +33,7 @@ from langgraph.runtime import Runtime
from pydantic import BaseModel, Field
from app.agent.middleware.utils import append_to_system_message
from app.agent.policy import sanitize_for_host, summarize_error, summarize_result
from app.agent.tools.tags import ToolTag
from app.log import logger
@@ -181,7 +182,9 @@ def load_activity_log_index(activity_dir: str, days: int = PROMPT_LOAD_DAYS) ->
try:
content = log_path.read_text(encoding="utf-8", errors="replace")
except Exception as e:
logger.warning(f"读取活动日志索引失败 {log_path}: {e}")
logger.warning(
f"读取活动日志索引失败 {log_path}: {summarize_error(e)}"
)
continue
entry_count = len(_parse_activity_entries(date_str, content))
if entry_count:
@@ -245,7 +248,7 @@ def query_activity_logs(
try:
content = log_path.read_text(encoding="utf-8", errors="replace")
except Exception as e:
logger.warning(f"读取活动日志失败 {log_path}: {e}")
logger.warning(f"读取活动日志失败 {log_path}: {summarize_error(e)}")
continue
for entry in _parse_activity_entries(date_str, content):
if normalized_keyword and not _activity_summary_matches_keyword(
@@ -287,14 +290,16 @@ class _ActivityLogToolProvider:
limit: Optional[int] = DEFAULT_QUERY_LIMIT,
) -> str:
"""查询活动日志并返回 JSON 字符串。"""
logger.info(
"查询活动日志: keyword=%s, use_regex=%s, date=%s, days=%s, limit=%s",
keyword,
use_regex,
date,
days,
limit,
logged_args = sanitize_for_host(
{
"keyword": keyword,
"use_regex": use_regex,
"date": date,
"days": days,
"limit": limit,
}
)
logger.info(f"查询活动日志: args={logged_args}")
try:
payload = query_activity_logs(
self._activity_dir,
@@ -306,11 +311,12 @@ class _ActivityLogToolProvider:
)
return json.dumps(payload, ensure_ascii=False, indent=2)
except Exception as err:
logger.error(f"查询活动日志失败: {err}", exc_info=True)
error_summary = summarize_error(err)
logger.error(f"查询活动日志失败: {error_summary}")
return json.dumps(
{
"success": False,
"message": f"查询活动日志时发生错误: {str(err)}",
"message": f"查询活动日志时发生错误: {error_summary}",
},
ensure_ascii=False,
)
@@ -454,7 +460,7 @@ async def _summarize_with_llm(conversation_text: str) -> Optional[str]:
return None
return summary if summary else None
except Exception as e:
logger.debug(f"LLM 活动摘要生成失败: {e}")
logger.debug(f"LLM 活动摘要生成失败: {summarize_error(e)}")
return None
@@ -571,9 +577,9 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
else:
with os.fdopen(fd, "w", encoding="utf-8") as stream:
stream.write(header + entry)
logger.debug(f"Activity logged: {summary[:80]}")
logger.debug(f"Activity logged: {summarize_result(summary, max_chars=80)}")
except Exception as e:
logger.warning(f"Failed to append activity log: {e}")
logger.warning(f"Failed to append activity log: {summarize_error(e)}")
async def _cleanup_old_logs(self) -> None:
"""清理超过保留天数的旧日志文件。"""
@@ -599,7 +605,9 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
except ValueError:
continue
except Exception as e:
logger.warning(f"Failed to cleanup old activity logs: {e}")
logger.warning(
f"Failed to cleanup old activity logs: {summarize_error(e)}"
)
def _schedule_activity_recording(self, messages: list) -> None:
"""提交后台活动记录任务，不阻塞当前 Agent 会话结束。"""
@@ -615,7 +623,7 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
except asyncio.CancelledError:
logger.debug("活动日志后台记录任务已取消")
except Exception as err:
logger.warning(f"活动日志后台记录任务失败: {err}")
logger.warning(f"活动日志后台记录任务失败: {summarize_error(err)}")
async def _record_activity(self, messages: list) -> None:
"""在后台生成本轮活动摘要并写入活动日志。"""
@@ -637,7 +645,7 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
if summary:
await self._append_activity(summary)
except Exception as e:
logger.warning(f"Failed to record activity: {e}")
logger.warning(f"Failed to record activity: {summarize_error(e)}")
async def abefore_agent(
self, state: ActivityLogState, runtime: Runtime
@@ -686,9 +694,12 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
tool_args = tool_call.get("args") or {}
if not isinstance(tool_args, dict):
tool_args = {}
logged_args = sanitize_for_host(tool_args)
if not isinstance(logged_args, dict):
logged_args = {}
logger.info(
f"开始执行活动日志查询工具: keyword={tool_args.get('keyword') or '-'}, "
f"date={tool_args.get('date') or '-'}"
f"开始执行活动日志查询工具: keyword={logged_args.get('keyword') or '-'}, "
f"date={logged_args.get('date') or '-'}"
)
if self.stream_handler and getattr(self.stream_handler, "is_streaming", False):
self.stream_handler.record_tool_call(
@@ -699,7 +710,9 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
try:
result = await handler(request)
except Exception as err:
logger.error(f"活动日志查询工具执行失败: error={err}")
logger.error(
f"活动日志查询工具执行失败: error={summarize_error(err)}"
)
raise
logger.info("活动日志查询工具执行完成")
return result
@@ -714,7 +727,7 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
return None
self._schedule_activity_recording(list(messages))
except Exception as e:
logger.warning(f"Failed to record activity: {e}")
logger.warning(f"Failed to record activity: {summarize_error(e)}")
return None

72
app/agent/middleware/policy.py Normal file
View File

@@ -0,0 +1,72 @@
"""LangChain 工具调用的 MoviePilot 宿主策略中间件。"""
from collections.abc import Awaitable, Callable
from typing import Any
from langchain.agents.middleware import AgentMiddleware, ToolCallRequest
from app.agent.policy import (
DEFAULT_TOOL_POLICY_ORCHESTRATOR,
AgentToolPolicyOrchestrator,
ToolPolicyContext,
call_policy_hook,
)
class AgentPolicyMiddleware(AgentMiddleware):
"""观测进入本地 ToolNode 的 client-side 工具调用和结果。
模型供应商原生 server tools 在供应商侧执行，不经过本地 middleware，
因而不具备这里生成的 start/finish/fail 回执。
"""
def __init__(
self,
*,
context: ToolPolicyContext,
orchestrator: AgentToolPolicyOrchestrator = DEFAULT_TOOL_POLICY_ORCHESTRATOR,
) -> None:
"""绑定宿主可信上下文和共享策略编排器。"""
self.context = context
self.orchestrator = orchestrator
async def awrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[Any]],
) -> Any:
"""在 handler 外层生成 shadow 决策和 secret-safe 回执摘要。"""
tool_call = request.tool_call or {}
arguments = tool_call.get("args") or {}
if not isinstance(arguments, dict):
arguments = {}
observation = call_policy_hook(
"start",
self.orchestrator.start,
context=self.context,
tool=request.tool,
arguments=arguments,
invocation_id=tool_call.get("id"),
)
try:
result = await handler(request)
except Exception as error:
if observation is not None:
call_policy_hook(
"fail",
self.orchestrator.fail,
observation,
error,
)
raise
if observation is not None:
call_policy_hook(
"finish",
self.orchestrator.finish,
observation,
result,
)
return result
__all__ = ["AgentPolicyMiddleware"]

33
app/agent/middleware/skills.py
View File

@@ -24,6 +24,7 @@ from langgraph.runtime import Runtime
from pydantic import BaseModel, Field
from app.agent.middleware.utils import append_to_system_message
from app.agent.policy import sanitize_for_host, summarize_error
from app.agent.tools.tags import ToolTag
from app.log import logger
@@ -124,7 +125,7 @@ def _parse_skill_metadata( # noqa: C901
try:
frontmatter_data = yaml.safe_load(frontmatter_str)
except yaml.YAMLError as e:
logger.warning("Invalid YAML in %s: %s", skill_path, e)
logger.warning("Invalid YAML in %s: %s", skill_path, summarize_error(e))
return None
if not isinstance(frontmatter_data, dict):
@@ -339,7 +340,7 @@ def _extract_version(skill_md: Path) -> int:
try:
content = skill_md.read_text(encoding="utf-8", errors="replace")
except Exception as err:
logger.debug(f"读取技能版本失败: {err}")
logger.debug(f"读取技能版本失败: {summarize_error(err)}")
return 0
match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
if not match:
@@ -397,7 +398,11 @@ def _sync_bundled_skills(bundled_dir: Path, target_dir: Path) -> None:
"已自动复制内置技能 '%s' -> '%s'", skill_src.name, skill_dst
)
except Exception as e:
logger.warning("复制内置技能 '%s' 失败: %s", skill_src.name, e)
logger.warning(
"复制内置技能 '%s' 失败: %s",
sanitize_for_host(skill_src.name),
summarize_error(e),
)
continue
# 目标已存在，比较版本号
@@ -424,7 +429,11 @@ def _sync_bundled_skills(bundled_dir: Path, target_dir: Path) -> None:
bundled_version,
)
except Exception as e:
logger.warning("更新内置技能 '%s' 失败: %s", skill_src.name, e)
logger.warning(
"更新内置技能 '%s' 失败: %s",
sanitize_for_host(skill_src.name),
summarize_error(e),
)
class _SkillToolProvider:
@@ -519,7 +528,7 @@ class _SkillToolProvider:
async def load_skill(self, name: str) -> str:
"""加载指定 Skill 的完整说明并返回 JSON 字符串。"""
logger.info(f"加载 Skill: name={name}")
logger.info(f"加载 Skill: name={sanitize_for_host(name)}")
try:
skill = await self._find_skill(name)
if not skill:
@@ -547,11 +556,12 @@ class _SkillToolProvider:
}
)
except Exception as err:
logger.error(f"加载 Skill 失败: {err}", exc_info=True)
error_summary = summarize_error(err)
logger.error(f"加载 Skill 失败: {error_summary}")
return json.dumps(
{
"success": False,
"message": f"加载 Skill 时发生错误: {str(err)}",
"message": f"加载 Skill 时发生错误: {error_summary}",
},
ensure_ascii=False,
)
@@ -623,7 +633,7 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no
try:
_sync_bundled_skills(bundled, target)
except Exception as e:
logger.warning("同步内置技能失败: %s", e)
logger.warning(f"同步内置技能失败: {summarize_error(e)}")
def _load_skills_metadata(self) -> list[SkillMetadata]:
"""同步加载当前配置目录中的 Skill 元数据。"""
@@ -728,8 +738,11 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no
tool_args = tool_call.get("args") or {}
if not isinstance(tool_args, dict):
tool_args = {}
logged_args = sanitize_for_host(tool_args)
if not isinstance(logged_args, dict):
logged_args = {}
logger.info(
f"开始执行 Skill 工具: name={tool_args.get('name') or '-'}"
f"开始执行 Skill 工具: name={logged_args.get('name') or '-'}"
)
if self.stream_handler and getattr(self.stream_handler, "is_streaming", False):
self.stream_handler.record_tool_call(
@@ -740,7 +753,7 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no
try:
result = await handler(request)
except Exception as err:
logger.error(f"Skill 工具执行失败: error={err}")
logger.error(f"Skill 工具执行失败: error={summarize_error(err)}")
raise
logger.info("Skill 工具执行完成")
return result

99
app/agent/middleware/subagents.py
View File

@@ -24,7 +24,16 @@ from langchain_core.tools import BaseTool, StructuredTool
from pydantic import BaseModel, Field
from app.agent.llm import LLMHelper
from app.agent.middleware.policy import AgentPolicyMiddleware
from app.agent.middleware.utils import append_to_system_message
from app.agent.policy import (
AuthSource,
PrincipalType,
ToolOrigin,
ToolPolicyContext,
sanitize_for_host,
summarize_error,
)
from app.agent.runtime import SubAgentDefinition, agent_runtime_manager
from app.agent.tools.tags import ToolTag
from app.log import logger
@@ -93,6 +102,36 @@ Requirements:
"""
def _default_subagent_policy_context(tools: list[BaseTool]) -> ToolPolicyContext:
"""从已注入工具继承会话归属，确保独立构造的子图也经过宿主策略。"""
for tool in tools:
session_id = getattr(tool, "_session_id", None)
user_id = getattr(tool, "_user_id", None)
if not session_id and not user_id:
continue
agent_context = getattr(tool, "_agent_context", None)
if not isinstance(agent_context, dict):
agent_context = {}
return ToolPolicyContext(
session_id=str(session_id or "subagent"),
user_id=str(user_id or "subagent"),
origin=ToolOrigin.SUBAGENT,
principal_type=PrincipalType.SUBAGENT,
auth_source=AuthSource.INTERNAL,
agent_context=agent_context,
channel=getattr(tool, "_channel", None),
source=getattr(tool, "_source", None),
)
return ToolPolicyContext(
session_id="subagent",
user_id="subagent",
origin=ToolOrigin.SUBAGENT,
principal_type=PrincipalType.SUBAGENT,
auth_source=AuthSource.INTERNAL,
agent_context={},
)
@dataclass(frozen=True)
class _SubAgentProfile:
"""子代理运行时定义。"""
@@ -378,12 +417,14 @@ class _SubAgentAgentProvider:
profiles: tuple[_SubAgentProfile, ...],
tools: list[BaseTool],
server_tools: Optional[list[dict[str, Any]]] = None,
policy_context: Optional[ToolPolicyContext] = None,
) -> None:
"""初始化子代理执行器。"""
self._model = model
self._profiles = {profile.name: profile for profile in profiles}
self._tools = tools
self._server_tools = server_tools or []
self._policy_context = policy_context or _default_subagent_policy_context(tools)
self._agents = {}
self._default_agent_name = "general-purpose"
@@ -409,6 +450,7 @@ class _SubAgentAgentProvider:
tools=[*subagent_tools, *self._server_tools],
system_prompt=profile.prompt,
name=profile.name,
middleware=[AgentPolicyMiddleware(context=self._policy_context)],
)
self._agents[profile.name] = agent
return profile.name, agent
@@ -444,7 +486,7 @@ class _SubAgentAgentProvider:
except Exception as err:
logger.error(
f"子代理调用失败: subagent_type={agent_name}, "
f"task_id={log_task_id}, error={err}"
f"task_id={log_task_id}, error={summarize_error(err)}"
)
raise
final_text = _extract_final_text(result)
@@ -468,6 +510,7 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware):
system_prompt: str = SUBAGENT_PARENT_PROMPT,
task_description: str = SUBAGENT_TASK_DESCRIPTION,
stream_handler: Any = None,
policy_context: Optional[ToolPolicyContext] = None,
) -> None:
"""初始化同步子代理中间件。"""
self.system_prompt = system_prompt
@@ -477,6 +520,7 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware):
profiles=profiles,
tools=tools,
server_tools=server_tools,
policy_context=policy_context,
)
self.tools = [
StructuredTool.from_function(
@@ -527,9 +571,12 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware):
return await handler(request)
tool_args = _extract_tool_call_args(request)
logged_args = sanitize_for_host(tool_args)
if not isinstance(logged_args, dict):
logged_args = {}
logger.info(
f"开始执行子代理工具: tool_name={tool_name}, "
f"subagent_type={tool_args.get('subagent_type') or '-'}"
f"subagent_type={logged_args.get('subagent_type') or '-'}"
)
_record_subagent_tool_call(
stream_handler=self.stream_handler,
@@ -539,7 +586,10 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware):
try:
result = await handler(request)
except Exception as err:
logger.error(f"子代理工具执行失败: tool_name={tool_name}, error={err}")
logger.error(
f"子代理工具执行失败: tool_name={tool_name}, "
f"error={summarize_error(err)}"
)
raise
logger.info(f"子代理工具执行完成: tool_name={tool_name}")
return result
@@ -557,6 +607,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
server_tools: Optional[list[dict[str, Any]]] = None,
task_description: str = SUBAGENT_CONTROL_DESCRIPTION,
stream_handler: Any = None,
policy_context: Optional[ToolPolicyContext] = None,
) -> None:
"""初始化异步子代理调度中间件。"""
self.stream_handler = stream_handler
@@ -565,6 +616,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
profiles=profiles,
tools=tools,
server_tools=server_tools,
policy_context=policy_context,
)
self._semaphore = asyncio.Semaphore(SUBAGENT_MAX_CONCURRENT_TASKS)
self._tasks: dict[str, _SubAgentRuntimeTask] = {}
@@ -628,7 +680,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
error = record.task.exception()
if error:
payload["error"] = str(error)
payload["error"] = summarize_error(error)
return payload
result, result_truncated = _clip_text(
@@ -733,7 +785,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
)
raise
except Exception as err:
logger.error(f"子代理任务执行失败: task_id={record.task_id}, error={err}")
logger.error(
f"子代理任务执行失败: task_id={record.task_id}, "
f"error={summarize_error(err)}"
)
raise
def _mark_task_finished(self, task_id: str, task: asyncio.Task) -> None:
@@ -901,7 +956,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
)
raise
except Exception as err:
logger.error(f"管道子代理任务执行失败: task_id={record.task_id}, error={err}")
logger.error(
f"管道子代理任务执行失败: task_id={record.task_id}, "
f"error={summarize_error(err)}"
)
raise
@staticmethod
@@ -975,8 +1033,13 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
)
return records, error
except Exception as err:
error = f"第 {step_index} 个管道子代理任务执行失败: {err}"
logger.info(f"{error} task_id={record.task_id}")
error = (
f"第 {step_index} 个管道子代理任务执行失败: "
f"{summarize_error(err)}"
)
logger.info(
f"{error} task_id={record.task_id}"
)
return records, error
previous_results.append((record, result))
@@ -1003,7 +1066,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
tasks=tasks,
)
if error:
logger.info(f"子代理管控操作未启动任务: action={action}, error={error}")
logger.info(
f"子代理管控操作未启动任务: action={action}, "
f"error={sanitize_for_host(error)}"
)
return self._json_response({"success": False, "error": error})
logger.info(f"准备启动子代理任务: action={action}, tasks={len(specs)}")
@@ -1095,10 +1161,13 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
return await handler(request)
tool_args = _extract_tool_call_args(request)
logged_args = sanitize_for_host(tool_args)
if not isinstance(logged_args, dict):
logged_args = {}
logger.info(
f"开始执行子代理工具: tool_name={tool_name}, "
f"action={tool_args.get('action') or '-'}, "
f"subagent_type={tool_args.get('subagent_type') or '-'}"
f"action={logged_args.get('action') or '-'}, "
f"subagent_type={logged_args.get('subagent_type') or '-'}"
)
_record_subagent_tool_call(
stream_handler=self.stream_handler,
@@ -1108,7 +1177,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
try:
result = await handler(request)
except Exception as err:
logger.error(f"子代理工具执行失败: tool_name={tool_name}, error={err}")
logger.error(
f"子代理工具执行失败: tool_name={tool_name}, "
f"error={summarize_error(err)}"
)
raise
logger.info(f"子代理工具执行完成: tool_name={tool_name}")
return result
@@ -1120,6 +1192,7 @@ def create_subagent_middlewares(
tools: list[BaseTool],
server_tools: Optional[list[dict[str, Any]]] = None,
stream_handler: Any = None,
policy_context: Optional[ToolPolicyContext] = None,
) -> tuple[list[AgentMiddleware], list[BaseTool]]:
"""创建子代理中间件列表和任务工具列表。"""
runtime_signature = agent_runtime_manager.current_signature()
@@ -1130,6 +1203,7 @@ def create_subagent_middlewares(
tools=tools,
server_tools=server_tools or [],
stream_handler=stream_handler,
policy_context=policy_context,
)
control_middleware = SubAgentTaskControlMiddleware(
model=model,
@@ -1137,6 +1211,7 @@ def create_subagent_middlewares(
tools=tools,
server_tools=server_tools or [],
stream_handler=stream_handler,
policy_context=policy_context,
)
task_tools = [

66
app/agent/policy/__init__.py Normal file
View File

@@ -0,0 +1,66 @@
"""MoviePilot Agent 宿主策略公共内部入口。"""
from app.agent.policy.contracts import (
ActionEffect,
ActionPolicy,
AuthSource,
ConfirmationMode,
ExecutionOutcome,
ExecutionReceipt,
MigrationState,
PolicyDecision,
PolicyObservation,
PolicyPrincipal,
PrincipalRole,
PrincipalType,
RecoveryMode,
ResultSensitivity,
ToolInvocation,
ToolOrigin,
ToolPolicyContext,
)
from app.agent.policy.orchestrator import (
DEFAULT_TOOL_POLICY_ORCHESTRATOR,
AgentToolPolicyOrchestrator,
call_policy_hook,
)
from app.agent.policy.registry import DEFAULT_TOOL_POLICY_REGISTRY, ToolPolicyRegistry
from app.agent.policy.sanitizer import (
REDACTED_VALUE,
sanitize_for_host,
stable_type_name,
summarize_error,
summarize_input,
summarize_result,
)
__all__ = [
"ActionEffect",
"ActionPolicy",
"AgentToolPolicyOrchestrator",
"AuthSource",
"ConfirmationMode",
"DEFAULT_TOOL_POLICY_ORCHESTRATOR",
"DEFAULT_TOOL_POLICY_REGISTRY",
"ExecutionOutcome",
"ExecutionReceipt",
"MigrationState",
"PolicyDecision",
"PolicyObservation",
"PolicyPrincipal",
"PrincipalRole",
"PrincipalType",
"REDACTED_VALUE",
"RecoveryMode",
"ResultSensitivity",
"ToolInvocation",
"ToolOrigin",
"ToolPolicyContext",
"ToolPolicyRegistry",
"call_policy_hook",
"sanitize_for_host",
"stable_type_name",
"summarize_error",
"summarize_input",
"summarize_result",
]

247
app/agent/policy/contracts.py Normal file
View File

@@ -0,0 +1,247 @@
"""MoviePilot Agent 宿主策略的内部契约。"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Mapping, MutableMapping, Optional
class ToolOrigin(str, Enum):
"""工具调用的宿主可信入口。"""
AGENT_INTERACTIVE = "agent_interactive"
AGENT_API = "agent_api"
OPERATOR_DIRECT = "operator_direct"
BACKGROUND = "background"
SUBAGENT = "subagent"
class PrincipalType(str, Enum):
"""调用主体类型，用于区分人、管理员集成和内部运行时。"""
HUMAN = "human"
SYSTEM_ADMIN_INTEGRATION = "system_admin_integration"
SCOPED_AGENT = "scoped_agent"
BACKGROUND = "background"
SUBAGENT = "subagent"
class AuthSource(str, Enum):
"""主体身份的宿主认证来源。"""
CHANNEL = "channel"
WEB_SESSION = "web_session"
API_TOKEN = "api_token"
INTERNAL = "internal"
AGENT_TOKEN = "agent_token"
class PrincipalRole(str, Enum):
"""策略授权使用的角色层级。"""
USER = "user"
CHANNEL_ADMIN = "channel_admin"
SYSTEM_ADMIN = "system_admin"
SYSTEM_INTERNAL = "system_internal"
class ActionEffect(str, Enum):
"""工具调用的实际副作用类别。"""
SAFE_READ = "safe_read"
SENSITIVE_READ = "sensitive_read"
REVERSIBLE_WRITE = "reversible_write"
DESTRUCTIVE_WRITE = "destructive_write"
EXTERNAL_SIDE_EFFECT = "external_side_effect"
ARBITRARY_EXECUTION = "arbitrary_execution"
UNKNOWN = "unknown"
class ConfirmationMode(str, Enum):
"""动作在完成授权后所需的确认方式。"""
NONE = "none"
REQUIRED = "required"
UNSUPPORTED = "unsupported"
class RecoveryMode(str, Enum):
"""动作可提供的执行恢复保证。"""
NONE = "none"
TRANSACTION = "transaction"
BEFORE_STATE = "before_state"
RECOVERABLE_DELETE = "recoverable_delete"
IDEMPOTENT = "idempotent"
RECONCILE = "reconcile"
MANUAL_ONLY = "manual_only"
class ResultSensitivity(str, Enum):
"""工具结果进入模型、记忆和日志时的敏感等级。"""
NORMAL = "normal"
PRIVATE = "private"
SECRET = "secret"
UNKNOWN = "unknown"
class MigrationState(str, Enum):
"""工具策略从兼容观测迁移到宿主执行的状态。"""
ENFORCED = "enforced"
LEGACY_SHADOW = "legacy_shadow"
class ExecutionOutcome(str, Enum):
"""P1-G1 handler 生命周期终态；成功不代表业务授权或副作用已完成。"""
SUCCEEDED = "succeeded"
FAILED = "failed"
@dataclass(frozen=True)
class PolicyPrincipal:
"""由可信入口建立、不可由工具参数覆盖的调用主体。"""
principal_id: str
principal_type: PrincipalType
auth_source: AuthSource
role: PrincipalRole
scopes: tuple[str, ...] = ()
@dataclass(frozen=True)
class ToolInvocation:
"""一次进入宿主策略层的规范化工具调用。"""
invocation_id: str
tool_name: str
arguments: Mapping[str, Any]
principal: PolicyPrincipal
session_id: str
origin: ToolOrigin
channel: Optional[str] = None
source: Optional[str] = None
@dataclass(frozen=True)
class ActionPolicy:
"""参数级动作策略及其兼容迁移状态。"""
effect: ActionEffect
required_role: PrincipalRole
confirmation: ConfirmationMode
recovery: RecoveryMode
result_sensitivity: ResultSensitivity
migration_state: MigrationState
policy_version: str = "p1-g1-v1"
interactive_allowed: bool = True
machine_allowed: bool = True
background_allowed: bool = True
subagent_allowed: bool = True
@dataclass(frozen=True)
class PolicyDecision:
"""宿主策略层决定；shadow allow 仅表示新策略不拦截，旧门禁仍是授权事实源。"""
allowed: bool
confirmation_required: bool
shadow: bool
reason_code: str
@dataclass(frozen=True)
class PolicyObservation:
"""调用开始时生成、供完成或失败回执复用的观测对象。"""
invocation: ToolInvocation
policy: ActionPolicy
decision: PolicyDecision
input_summary: str
started_at: float
@dataclass(frozen=True)
class ExecutionReceipt:
"""P1-G1 的非持久化脱敏回执 envelope。"""
invocation_id: str
tool_name: str
origin: ToolOrigin
decision: PolicyDecision
outcome: ExecutionOutcome
input_summary: str
result_summary: Optional[str] = None
error_summary: Optional[str] = None
duration_ms: int = 0
@dataclass(frozen=True)
class ToolPolicyContext:
"""宿主入口上下文；管理员状态引用会随缓存图的每轮执行刷新。"""
session_id: str
user_id: str
origin: ToolOrigin
principal_type: PrincipalType
auth_source: AuthSource
agent_context: MutableMapping[str, Any] = field(repr=False, compare=False)
channel: Optional[str] = None
source: Optional[str] = None
@property
def principal(self) -> PolicyPrincipal:
"""根据当前宿主上下文生成本次调用主体。"""
if self.principal_type in {PrincipalType.BACKGROUND, PrincipalType.SUBAGENT}:
default_role = PrincipalRole.SYSTEM_INTERNAL
else:
default_role = PrincipalRole.USER
role = (
PrincipalRole.SYSTEM_ADMIN
if bool(self.agent_context.get("is_admin"))
else default_role
)
raw_scopes = self.agent_context.get("policy_scopes") or ()
scopes = tuple(str(scope) for scope in raw_scopes if scope)
return PolicyPrincipal(
principal_id=str(self.user_id or self.principal_type.value),
principal_type=self.principal_type,
auth_source=self.auth_source,
role=role,
scopes=scopes,
)
def for_subagent(self) -> "ToolPolicyContext":
"""保留用户与会话归属，并切换为子代理可信来源。"""
return ToolPolicyContext(
session_id=self.session_id,
user_id=self.user_id,
origin=ToolOrigin.SUBAGENT,
principal_type=PrincipalType.SUBAGENT,
auth_source=AuthSource.INTERNAL,
agent_context=self.agent_context,
channel=self.channel,
source=self.source,
)
__all__ = [
"ActionEffect",
"ActionPolicy",
"AuthSource",
"ConfirmationMode",
"ExecutionOutcome",
"ExecutionReceipt",
"MigrationState",
"PolicyDecision",
"PolicyObservation",
"PolicyPrincipal",
"PrincipalRole",
"PrincipalType",
"RecoveryMode",
"ResultSensitivity",
"ToolInvocation",
"ToolOrigin",
"ToolPolicyContext",
]

191
app/agent/policy/orchestrator.py Normal file
View File

@@ -0,0 +1,191 @@
"""Agent 工具策略观测、脱敏回执与共享执行边界。"""
import time
import uuid
from collections.abc import Callable
from typing import Any, Mapping, Optional, TypeVar
from langchain_core.messages import ToolMessage
from pydantic import ValidationError
from app.agent.policy.contracts import (
ExecutionOutcome,
ExecutionReceipt,
MigrationState,
PolicyDecision,
PolicyObservation,
ToolInvocation,
ToolPolicyContext,
)
from app.agent.policy.registry import DEFAULT_TOOL_POLICY_REGISTRY, ToolPolicyRegistry
from app.agent.policy.sanitizer import (
stable_type_name,
summarize_error,
summarize_input,
summarize_result,
)
from app.log import logger
_HookResult = TypeVar("_HookResult")
def call_policy_hook(
phase: str,
hook: Callable[..., _HookResult],
*args: Any,
**kwargs: Any,
) -> Optional[_HookResult]:
"""以 fail-open 方式调用 P1-G1 观测 hook，故障只记录稳定类型。"""
try:
return hook(*args, **kwargs)
except Exception as error:
try:
logger.warning(
f"Agent工具策略观测失败: phase={phase}, "
f"error_type={stable_type_name(error)}"
)
except Exception:
pass
return None
def _normalize_policy_arguments(tool: Any, arguments: Mapping[str, Any]) -> dict[str, Any]:
"""为策略生成 Pydantic 规范化副本，不改变真实执行参数。"""
raw_arguments = dict(arguments or {})
args_schema = getattr(tool, "args_schema", None)
if not args_schema:
return raw_arguments
try:
validated = args_schema.model_validate(raw_arguments)
return validated.model_dump(mode="json")
except (AttributeError, TypeError, ValueError, ValidationError):
# 实际 handler 仍负责既有参数错误语义；策略观测按原始值保守处理。
return raw_arguments
def _result_payload(result: Any) -> Any:
"""从 LangChain 工具消息中提取模型可见结果供脱敏摘要使用。"""
if isinstance(result, ToolMessage):
return result.content
return result
class AgentToolPolicyOrchestrator:
"""让 Agent middleware 与 direct manager 复用同一策略生命周期。"""
def __init__(self, registry: ToolPolicyRegistry = DEFAULT_TOOL_POLICY_REGISTRY) -> None:
"""绑定固定工具迁移注册表。"""
self.registry = registry
def start(
self,
*,
context: ToolPolicyContext,
tool: Any,
arguments: Mapping[str, Any],
invocation_id: Optional[str] = None,
) -> PolicyObservation:
"""解析调用策略，并创建不影响现有 allow 行为的观测对象。"""
tool_name = str(getattr(tool, "name", None) or "unknown_tool")
normalized_arguments = _normalize_policy_arguments(tool, arguments)
policy = self.registry.resolve(
tool_name=tool_name,
arguments=normalized_arguments,
requires_admin=bool(getattr(tool, "_require_admin", False)),
)
if policy.migration_state is MigrationState.LEGACY_SHADOW:
decision = PolicyDecision(
allowed=True,
confirmation_required=False,
shadow=True,
reason_code="legacy_shadow_allow",
)
else:
decision = PolicyDecision(
allowed=True,
confirmation_required=False,
shadow=False,
reason_code="safe_read_allow",
)
invocation = ToolInvocation(
invocation_id=invocation_id or uuid.uuid4().hex,
tool_name=tool_name,
arguments=normalized_arguments,
principal=context.principal,
session_id=context.session_id,
origin=context.origin,
channel=context.channel,
source=context.source,
)
input_summary = summarize_input(normalized_arguments)
observation = PolicyObservation(
invocation=invocation,
policy=policy,
decision=decision,
input_summary=input_summary,
started_at=time.monotonic(),
)
logger.debug(
f"Agent工具策略: tool={tool_name}, origin={context.origin.value}, "
f"decision={decision.reason_code}, input={input_summary}"
)
return observation
@staticmethod
def finish(observation: PolicyObservation, result: Any) -> ExecutionReceipt:
"""生成成功回执 envelope，并只记录脱敏结果摘要。"""
result_summary = summarize_result(_result_payload(result))
receipt = ExecutionReceipt(
invocation_id=observation.invocation.invocation_id,
tool_name=observation.invocation.tool_name,
origin=observation.invocation.origin,
decision=observation.decision,
outcome=ExecutionOutcome.SUCCEEDED,
input_summary=observation.input_summary,
result_summary=result_summary,
duration_ms=max(
0,
int((time.monotonic() - observation.started_at) * 1000),
),
)
logger.info(
f"Agent工具执行完成: tool={receipt.tool_name}, "
f"origin={receipt.origin.value}, shadow={receipt.decision.shadow}, "
f"duration_ms={receipt.duration_ms}, result={result_summary}"
)
return receipt
@staticmethod
def fail(observation: PolicyObservation, error: BaseException) -> ExecutionReceipt:
"""生成失败回执 envelope，不把异常中的凭据写入日志。"""
error_summary = summarize_error(error)
receipt = ExecutionReceipt(
invocation_id=observation.invocation.invocation_id,
tool_name=observation.invocation.tool_name,
origin=observation.invocation.origin,
decision=observation.decision,
outcome=ExecutionOutcome.FAILED,
input_summary=observation.input_summary,
error_summary=error_summary,
duration_ms=max(
0,
int((time.monotonic() - observation.started_at) * 1000),
),
)
logger.error(
f"Agent工具执行失败: tool={receipt.tool_name}, "
f"origin={receipt.origin.value}, shadow={receipt.decision.shadow}, "
f"duration_ms={receipt.duration_ms}, error={error_summary}"
)
return receipt
DEFAULT_TOOL_POLICY_ORCHESTRATOR = AgentToolPolicyOrchestrator()
__all__ = [
"AgentToolPolicyOrchestrator",
"DEFAULT_TOOL_POLICY_ORCHESTRATOR",
"call_policy_hook",
]

181
app/agent/policy/registry.py Normal file
View File

@@ -0,0 +1,181 @@
"""固定工具迁移注册表与参数级策略解析。"""
from typing import Any, Mapping
from app.agent.policy.contracts import (
ActionEffect,
ActionPolicy,
ConfirmationMode,
MigrationState,
PrincipalRole,
RecoveryMode,
ResultSensitivity,
)
# 这些读取已具备清晰的无副作用语义，用于证明新宿主边界不会改变正常结果。
SAFE_READ_TOOL_NAMES = frozenset(
{
"list_slash_commands",
"query_installed_plugins",
"query_personas",
"query_schedulers",
"query_workflows",
}
)
# 其余固定工具先显式处于兼容观测状态，待领域叶子 Goal 逐个迁移。
LEGACY_SHADOW_TOOL_NAMES = frozenset(
{
"add_custom_filter_rule",
"add_download_tasks",
"add_rule_group",
"add_subscribe",
"ask_user_choice",
"browse_webpage",
"create_agent_task",
"delete_agent_task",
"delete_custom_filter_rule",
"delete_download_history",
"delete_download_tasks",
"delete_rule_group",
"delete_subscribe",
"delete_transfer_history",
"edit_file",
"execute_command",
"get_recommendations",
"get_search_results",
"install_plugin",
"list_directory",
"query_agent_tasks",
"query_builtin_filter_rules",
"query_custom_filter_rules",
"query_custom_identifiers",
"query_directory_settings",
"query_doctor_report",
"query_download_tasks",
"query_downloaders",
"query_episode_schedule",
"query_library_exists",
"query_library_latest",
"query_market_plugins",
"query_media_detail",
"query_plugin_capabilities",
"query_plugin_config",
"query_plugin_data",
"query_popular_subscribes",
"query_rule_groups",
"query_site_userdata",
"query_sites",
"query_subscribe_history",
"query_subscribe_shares",
"query_subscribes",
"query_system_settings",
"query_transfer_history",
"read_file",
"recognize_captcha",
"recognize_media",
"reload_plugin",
"run_agent_task",
"run_scheduler",
"run_slash_command",
"run_workflow",
"scrape_metadata",
"search_media",
"search_person",
"search_person_credits",
"search_subscribe",
"search_torrents",
"search_web",
"send_local_file",
"send_message",
"send_voice_message",
"switch_persona",
"test_site",
"transfer_file",
"uninstall_plugin",
"update_agent_task",
"update_custom_filter_rule",
"update_custom_identifiers",
"update_download_tasks",
"update_persona_definition",
"update_plugin_config",
"update_rule_group",
"update_site",
"update_site_cookie",
"update_subscribe",
"update_system_settings",
"write_file",
}
)
class ToolPolicyRegistry:
"""解析固定和动态工具的 P1-G1 迁移策略。"""
def __init__(
self,
*,
safe_read_tool_names: frozenset[str] = SAFE_READ_TOOL_NAMES,
legacy_shadow_tool_names: frozenset[str] = LEGACY_SHADOW_TOOL_NAMES,
) -> None:
"""建立互斥的固定工具迁移表。"""
overlap = safe_read_tool_names & legacy_shadow_tool_names
if overlap:
raise ValueError(f"工具策略迁移表存在重复项: {sorted(overlap)}")
self._safe_read_tool_names = safe_read_tool_names
self._legacy_shadow_tool_names = legacy_shadow_tool_names
@property
def builtin_tool_names(self) -> set[str]:
"""返回注册表覆盖的全部固定工具名。"""
return set(self._safe_read_tool_names | self._legacy_shadow_tool_names)
def resolve(
self,
*,
tool_name: str,
arguments: Mapping[str, Any],
requires_admin: bool,
) -> ActionPolicy:
"""根据工具名和宿主权限声明解析当前迁移策略。"""
del arguments # 参数级迁移由后续领域 Goal 逐项加入。
required_role = (
PrincipalRole.SYSTEM_ADMIN if requires_admin else PrincipalRole.USER
)
if tool_name in self._safe_read_tool_names:
return ActionPolicy(
effect=ActionEffect.SAFE_READ,
required_role=required_role,
confirmation=ConfirmationMode.NONE,
recovery=RecoveryMode.NONE,
result_sensitivity=ResultSensitivity.NORMAL,
# 角色门禁仍可能异步识别渠道管理员；G1 不复制旧授权事实源。
migration_state=(
MigrationState.LEGACY_SHADOW
if requires_admin
else MigrationState.ENFORCED
),
)
# 固定未迁移工具和动态工具都保持现有执行能力，但不得被视为安全读取。
return ActionPolicy(
effect=ActionEffect.UNKNOWN,
required_role=required_role,
confirmation=ConfirmationMode.REQUIRED,
recovery=RecoveryMode.MANUAL_ONLY,
result_sensitivity=ResultSensitivity.UNKNOWN,
migration_state=MigrationState.LEGACY_SHADOW,
)
DEFAULT_TOOL_POLICY_REGISTRY = ToolPolicyRegistry()
__all__ = [
"DEFAULT_TOOL_POLICY_REGISTRY",
"LEGACY_SHADOW_TOOL_NAMES",
"SAFE_READ_TOOL_NAMES",
"ToolPolicyRegistry",
]

1056
app/agent/policy/sanitizer.py Normal file
View File

File diff suppressed because it is too large Load Diff

69
app/agent/policy/secret_fields.py Normal file
View File

@@ -0,0 +1,69 @@
"""Agent 设置工具与宿主回执共用的敏感字段身份判定。"""
import re
from typing import Any
_MAX_FIELD_NAME_CHARS = 256
_ACRONYM_BOUNDARY_PATTERN = re.compile(r"(?<=[A-Z])(?=[A-Z][a-z])")
_CAMEL_CASE_BOUNDARY_PATTERN = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
_SECRET_FIELD_NAMES = frozenset(
{
"access_token",
"api_key",
"apikey",
"api_token",
"auth_header",
"authorization",
"client_secret",
"cookie",
"passkey",
"passwd",
"password",
"private_key",
"pwd",
"refresh_token",
"secret",
"secret_access_key",
"secret_key",
"token",
}
)
_SECRET_FIELD_ENDINGS = tuple(f"_{name}" for name in _SECRET_FIELD_NAMES)
_SECRET_SETTING_NAMES = frozenset(
{
# CookieCloud 的用户 key 没有类型后缀，但与密码共同构成端到端加密凭据。
"cookiecloud_key",
}
)
_SECRET_SETTING_ENDINGS = (
"_encrypt_key",
)
def _normalize_field_name(value: Any) -> str:
"""将短字段名规范化为 snake_case，非字符串不参与身份推导。"""
if type(value) is not str:
return ""
text = value.strip()
if len(text) > _MAX_FIELD_NAME_CHARS:
text = text[-_MAX_FIELD_NAME_CHARS:]
text = _ACRONYM_BOUNDARY_PATTERN.sub("_", text)
text = _CAMEL_CASE_BOUNDARY_PATTERN.sub("_", text)
return re.sub(r"[^a-z0-9]+", "_", text.lower()).strip("_")
def is_secret_setting_key(key: Any) -> bool:
"""按完整字段或类型后缀识别凭据，避免误伤 token 统计与过期配置。"""
normalized = _normalize_field_name(key)
if not normalized:
return False
return (
normalized in _SECRET_FIELD_NAMES
or normalized in _SECRET_SETTING_NAMES
or normalized.endswith(_SECRET_FIELD_ENDINGS)
or normalized.endswith(_SECRET_SETTING_ENDINGS)
)
__all__ = ["is_secret_setting_key"]

32
app/agent/tools/base.py
View File

@@ -11,6 +11,11 @@ from langchain_core.tools import BaseTool
from pydantic import PrivateAttr
from app.agent import StreamingHandler
from app.agent.policy.sanitizer import (
summarize_error,
summarize_input,
summarize_result,
)
from app.agent.tools.tags import ToolTag
from app.chain import ChainBase
from app.core.config import settings
@@ -39,7 +44,9 @@ def serialize_tool_result_for_agent(result: Any) -> str:
try:
return json.dumps(result, ensure_ascii=False, indent=2, default=str)
except Exception as e:
logger.warning(f"工具结果转换为JSON失败: {e}, 使用字符串表示")
logger.warning(
f"工具结果转换为JSON失败: {summarize_error(e)}, 使用字符串表示"
)
return str(result)
@@ -303,27 +310,26 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
# 未启用流式传输，不发送任何工具消息内容
pass
logger.debug(f"Executing tool {self.name} with args: {kwargs}")
logger.debug(
f"Executing tool {self.name} with input summary: {summarize_input(kwargs)}"
)
# 执行具体工具逻辑
try:
result = await self.run_with_timeout(**kwargs)
# 记录工具执行结果摘要日志
str_result = serialize_tool_result_for_agent(result)
if len(str_result) > 500:
summary = str_result[:500] + f"...(已截断，总长度: {len(str_result)})"
else:
summary = str_result
logger.info(f"Agent工具 {self.name} 执行完成，结果摘要: {summary}")
logger.info(
f"Agent工具 {self.name} 执行完成，"
f"结果摘要: {summarize_result(result)}"
)
except ToolExecutionTimeoutError as e:
error_message = str(e)
error_message = summarize_error(e)
logger.warning(error_message)
result = error_message
except Exception as e:
error_message = f"工具执行异常 ({type(e).__name__}): {str(e)}"
logger.error(f"Tool {self.name} execution failed: {e}", exc_info=True)
error_message = f"工具执行异常: {summarize_error(e)}"
logger.error(f"Tool {self.name} execution failed: {summarize_error(e)}")
result = error_message
return format_tool_result_for_agent(
@@ -625,7 +631,7 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
return False
except Exception as e:
logger.error(f"检查权限失败: {e}")
logger.error(f"检查权限失败: {summarize_error(e)}")
return False

21
app/agent/tools/impl/_system_setting_utils.py
View File

@@ -3,6 +3,7 @@
from dataclasses import dataclass
from typing import Any, Optional
from app.agent.policy.secret_fields import is_secret_setting_key
from app.core.config import Settings
from app.schemas.types import SystemConfigKey
@@ -368,26 +369,6 @@ def get_default_list_match_field(setting_key: str) -> Optional[str]:
return LIST_ITEM_MATCH_FIELD_DEFAULTS.get(setting_key)
SECRET_KEYWORDS = (
"api_key",
"apikey",
"token",
"secret",
"password",
"passwd",
"cookie",
"authorization",
"refresh_token",
"access_token",
)
def is_secret_setting_key(key: str) -> bool:
"""判断设置键名是否疑似敏感字段。"""
normalized = _normalize_token(key)
return any(keyword in normalized for keyword in SECRET_KEYWORDS)
def redact_secret_value(value: Any, *, redact_scalar: bool = False) -> Any:
"""递归脱敏配置值中的密钥、Cookie、Token 等敏感字段。"""
if isinstance(value, dict):

67
app/agent/tools/manager.py
View File

@@ -3,6 +3,16 @@ import threading
import uuid
from typing import Any, Dict, List, Optional
from app.agent.policy import (
DEFAULT_TOOL_POLICY_ORCHESTRATOR,
AgentToolPolicyOrchestrator,
AuthSource,
PrincipalType,
ToolOrigin,
ToolPolicyContext,
call_policy_hook,
summarize_error,
)
from app.agent.tools.base import ToolExecutionTimeoutError, format_tool_result_for_agent
from app.agent.tools.factory import MoviePilotToolFactory
from app.core.plugin import PluginManager
@@ -30,6 +40,7 @@ class MoviePilotToolsManager:
user_id: str = "api_user",
session_id: str = uuid.uuid4(),
is_admin: bool = True,
policy_orchestrator: Optional[AgentToolPolicyOrchestrator] = None,
):
"""
初始化工具管理器
@@ -41,6 +52,19 @@ class MoviePilotToolsManager:
self.user_id = user_id
self.session_id = session_id
self.is_admin = is_admin
self.policy_orchestrator = (
policy_orchestrator or DEFAULT_TOOL_POLICY_ORCHESTRATOR
)
self._policy_context = ToolPolicyContext(
session_id=session_id,
user_id=user_id,
origin=ToolOrigin.OPERATOR_DIRECT,
principal_type=PrincipalType.SYSTEM_ADMIN_INTEGRATION,
auth_source=AuthSource.API_TOKEN,
channel=None,
source="api",
agent_context={"is_admin": is_admin},
)
self.tools: List[Any] = []
self._tools_lock = threading.Lock()
self._plugin_agent_tools_revision = -1
@@ -74,7 +98,7 @@ class MoviePilotToolsManager:
self._plugin_agent_tools_revision = plugin_tools_revision
logger.info(f"成功加载 {len(self.tools)} 个工具")
except Exception as e:
logger.error(f"加载工具失败: {e}", exc_info=True)
logger.error(f"加载工具失败: {summarize_error(e)}")
self.tools = []
self._plugin_agent_tools_revision = -1
@@ -231,7 +255,7 @@ class MoviePilotToolsManager:
schema = args_schema.model_json_schema()
properties = schema.get("properties", {})
except Exception as e:
logger.warning(f"获取工具schema失败: {e}")
logger.warning(f"获取工具schema失败: {summarize_error(e)}")
return arguments
# 规范化参数
@@ -286,6 +310,7 @@ class MoviePilotToolsManager:
)
return error_msg
observation = None
try:
permission_error = self._check_tool_permission(tool_instance)
if permission_error:
@@ -293,38 +318,50 @@ class MoviePilotToolsManager:
# 规范化参数类型
normalized_arguments = self._normalize_arguments(tool_instance, arguments)
self._policy_context.agent_context["is_admin"] = self.is_admin
observation = call_policy_hook(
"start",
self.policy_orchestrator.start,
context=self._policy_context,
tool=tool_instance,
arguments=normalized_arguments,
)
# 调用工具的run方法。HTTP/MCP 工具调用不会经过 BaseTool._arun，
# 因此这里也必须复用同一套返回值格式化和兜底截断逻辑。
result = await tool_instance.run_with_timeout(**normalized_arguments)
# 记录工具执行结果摘要日志
str_result = format_tool_result_for_agent(
result,
tool_name=tool_name,
max_chars=getattr(tool_instance, "result_max_chars", None),
)
if len(str_result) > 500:
summary = str_result[:500] + f"...(已截断，总长度: {len(str_result)})"
else:
summary = str_result
logger.info(f"Agent工具 {tool_name} 执行完成，结果摘要: {summary}")
return str_result
except ToolExecutionTimeoutError as e:
logger.warning(str(e))
if observation:
call_policy_hook("fail", self.policy_orchestrator.fail, observation, e)
logger.warning(summarize_error(e))
return format_tool_result_for_agent(
str(e),
summarize_error(e),
tool_name=tool_name,
max_chars=getattr(tool_instance, "result_max_chars", None),
)
except Exception as e:
logger.error(f"调用工具 {tool_name} 时发生错误: {e}", exc_info=True)
if observation:
call_policy_hook("fail", self.policy_orchestrator.fail, observation, e)
error_summary = summarize_error(e)
logger.error(f"调用工具 {tool_name} 时发生错误: {error_summary}")
error_msg = json.dumps(
{"error": f"调用工具 '{tool_name}' 时发生错误: {str(e)}"},
{"error": f"调用工具 '{tool_name}' 时发生错误: {error_summary}"},
ensure_ascii=False,
)
return error_msg
if observation:
call_policy_hook(
"finish",
self.policy_orchestrator.finish,
observation,
str_result,
)
return str_result
@staticmethod
def _convert_to_json_schema(args_schema: Any) -> Dict[str, Any]: