Add autonomous Agent task scheduling

app/agent/__init__.py

@@ -58,6 +58,7 @@ from app.chain import ChainBase
 from app.core.config import settings
 from app.core.event import eventmanager
 from app.db.agentchat_oper import AgentChatOper
+from app.db.agenttask_oper import AgentTaskOper
 from app.db.user_oper import UserOper
 from app.log import logger
 from app.schemas import AgentLLMProviderEventData, AgentTokensUsageEventData, Notification, NotificationType
@@ -2123,6 +2124,119 @@ class AgentManager:
             await agent.cleanup()
             memory_manager.clear_memory(session_id, user_id)
 
+    async def execute_scheduled_task(self, task_id: int) -> tuple[bool, str]:
+        """
+        按持久化上下文唤醒 Agent 执行自主定时任务并向用户回传结果。
+
+        :param task_id: Agent 定时任务 ID
+        :return: 执行是否成功及结果摘要
+        """
+        if not settings.AI_AGENT_ENABLE:
+            return False, "AI Agent 未启用"
+        oper = AgentTaskOper()
+        task = oper.get(task_id)
+        if not task or not task.enabled:
+            return False, "Agent 定时任务不存在或已停用"
+        if not oper.mark_running(task_id):
+            return False, "Agent 定时任务当前不可执行"
+
+        task_message = (
+            f"定时任务已按计划触发。请立即完成下面的任务，不要只确认收到，"
+            f"也不要重复创建同一个定时任务。\n\n"
+            f"任务名称：{task.name}\n"
+            f"任务内容：{task.content}\n\n"
+            "完成后请直接向用户报告本次执行结果；如果无法完成，请说明原因。"
+        )
+        has_message_context = bool(task.channel and task.source)
+        success = True
+        result = ""
+        try:
+            result = await self.process_message(
+                session_id=task.session_id,
+                user_id=task.user_id,
+                message=task_message,
+                channel=task.channel,
+                source=task.source,
+                username=task.username,
+                original_chat_id=task.original_chat_id,
+                reply_mode=(
+                    ReplyMode.DISPATCH
+                    if has_message_context
+                    else ReplyMode.CAPTURE_ONLY
+                ),
+                allow_message_tools=has_message_context,
+                wait_for_completion=True,
+            )
+            result_text = str(result or "").strip()
+            success = bool(result_text) and not result_text.startswith(
+                (AGENT_EXECUTION_ERROR_PREFIX, "处理消息时发生错误")
+            )
+            if not result_text:
+                result = "定时任务已执行，但 Agent 未返回结果"
+                await AgentChain().async_post_message(
+                    Notification(
+                        channel=task.channel if has_message_context else None,
+                        source=task.source if has_message_context else None,
+                        mtype=NotificationType.Agent,
+                        userid=task.user_id if has_message_context else None,
+                        username=(
+                            task.username
+                            if has_message_context
+                            else settings.SUPERUSER
+                        ),
+                        original_chat_id=task.original_chat_id,
+                        title=f"定时任务：{task.name}",
+                        text=result,
+                        save_history=False,
+                    )
+                )
+            elif not has_message_context:
+                await AgentChain().async_post_message(
+                    Notification(
+                        mtype=NotificationType.Agent,
+                        username=settings.SUPERUSER,
+                        title=f"定时任务：{task.name}",
+                        text=result_text,
+                        save_history=False,
+                    )
+                )
+        except Exception as err:
+            success = False
+            result = f"Agent 定时任务执行失败：{str(err)}"
+            logger.error(f"Agent 定时任务 {task_id} 执行失败: {str(err)}")
+            await AgentChain().async_post_message(
+                Notification(
+                    channel=task.channel if has_message_context else None,
+                    source=task.source if has_message_context else None,
+                    mtype=NotificationType.Agent,
+                    userid=task.user_id if has_message_context else None,
+                    username=(
+                        task.username
+                        if has_message_context
+                        else settings.SUPERUSER
+                    ),
+                    original_chat_id=task.original_chat_id,
+                    title=f"定时任务执行失败：{task.name}",
+                    text=result,
+                    save_history=False,
+                )
+            )
+        finally:
+            current_task = oper.get(task_id)
+            oper.finish(
+                task_id=task_id,
+                success=success,
+                result=str(result or ""),
+                disable=bool(
+                    current_task
+                    and task.trigger_type == "date"
+                    and current_task.trigger_type == task.trigger_type
+                    and current_task.run_at == task.run_at
+                ),
+            )
+
+        return success, str(result or "任务执行完成")
+
     @staticmethod
     def _build_heartbeat_prompt() -> str:
         """使用程序内置 System Tasks 定义构建心跳任务提示词。"""

app/agent/middleware/jobs.py

@@ -204,9 +204,11 @@ You have a scheduled jobs system for user-requested delayed or recurring work.
 {jobs_list}
 
 Rules:
-- Create jobs only when the user asks for delayed, recurring, reminder, or monitoring behavior.
+- For new delayed, recurring, reminder, or monitoring work, use the dedicated
-- Do not create jobs for immediate one-time work or work already handled by MoviePilot schedulers.
+  `create_agent_task`, `query_agent_tasks`, `update_agent_task`, and
-- Each job lives in its own directory with a `JOB.md`; read the listed file before executing or updating an active job.
+  `delete_agent_task` tools. Do not create or edit JOB.md files for new tasks.
+- Do not create tasks for immediate one-time work or work already handled by MoviePilot schedulers.
+- Entries listed above are legacy JOB.md tasks. Read their files only when a heartbeat asks you to execute them.
 - During heartbeat checks, act only on `pending` or `in_progress` jobs, update status/last_run/logs, and leave recurring jobs `pending` after each run.
 </jobs_system>
 """
@@ -230,7 +232,7 @@ class JobsMiddleware(AgentMiddleware[JobsState, ContextT, ResponseT]):  # noqa
     def _format_jobs_list(jobs: list[JobMetadata]) -> str:
         """格式化任务元数据列表用于系统提示词。"""
         if not jobs:
-            return "(No active jobs. You can create jobs when users request periodic or scheduled tasks.)"
+            return "(No active legacy JOB.md tasks. Use create_agent_task for new scheduled work.)"
 
         lines = []
         for job in jobs:

app/agent/prompt/System Core Prompt.txt

@@ -24,6 +24,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel
 - Do not stop for approval on read-only operations.
 - If the user has not explicitly requested an operation that changes system behavior, ask for confirmation before proceeding. This includes modifying system settings, updating plugin configuration, reloading plugins, running restart/stop/start commands, or triggering slash commands such as `/restart`.
 - Always get explicit consent before destructive or high-impact actions such as starting downloads, deleting subscriptions, deleting download tasks or files, removing history, installing/uninstalling plugins, changing site authentication, changing scheduler or workflow execution state, restarting services, or stopping services.
+- When the user explicitly asks for delayed, recurring, reminder, or monitoring work, use `create_agent_task` instead of promising to remember it or writing a JOB.md file. Use a `date` trigger with `delay_minutes` for requests such as "in 30 minutes", an exact `date` trigger for other single future runs, and a five-field `cron` trigger for recurring work. Manage existing autonomous tasks with `query_agent_tasks`, `update_agent_task`, and `delete_agent_task`.
 - If the user explicitly requested the exact write action, perform the smallest correct change and then validate the result.
 - If a requested action is ambiguous between read-only inspection and state change, inspect first and ask a short confirmation question before the state-changing step.
 </confirmation_policy>

app/agent/tools/factory.py

@@ -42,8 +42,12 @@ from app.agent.tools.impl.send_message import SendMessageTool
 from app.agent.tools.impl.ask_user_choice import AskUserChoiceTool
 from app.agent.tools.impl.send_local_file import SendLocalFileTool
 from app.agent.tools.impl.send_voice_message import SendVoiceMessageTool
+from app.agent.tools.impl.create_agent_task import CreateAgentTaskTool
+from app.agent.tools.impl.delete_agent_task import DeleteAgentTaskTool
+from app.agent.tools.impl.query_agent_tasks import QueryAgentTasksTool
 from app.agent.tools.impl.query_schedulers import QuerySchedulersTool
 from app.agent.tools.impl.run_scheduler import RunSchedulerTool
+from app.agent.tools.impl.update_agent_task import UpdateAgentTaskTool
 from app.agent.tools.impl.query_workflows import QueryWorkflowsTool
 from app.agent.tools.impl.run_workflow import RunWorkflowTool
 from app.agent.tools.impl.query_personas import QueryPersonasTool
@@ -141,6 +145,10 @@ class MoviePilotToolFactory:
         QueryTransferHistoryTool,
         TransferFileTool,
         SendMessageTool,
+        CreateAgentTaskTool,
+        QueryAgentTasksTool,
+        UpdateAgentTaskTool,
+        DeleteAgentTaskTool,
         QuerySchedulersTool,
         RunSchedulerTool,
         QueryWorkflowsTool,
@@ -181,6 +189,8 @@ class MoviePilotToolFactory:
         "edit_file",
         "execute_command",
         "ask_user_choice",
+        "create_agent_task",
+        "query_agent_tasks",
     )
 
     @staticmethod

app/agent/tools/impl/create_agent_task.py

@@ -0,0 +1,150 @@
+import json
+from datetime import datetime, timedelta
+from typing import Literal, Optional, Type
+
+import pytz
+from pydantic import BaseModel, Field, model_validator
+
+from app.agent.tools.base import MoviePilotTool
+from app.agent.tools.tags import ToolTag
+from app.core.config import settings
+from app.db.agentchat_oper import AgentChatOper
+from app.db.agenttask_oper import AgentTaskOper
+from app.utils.timer import TimerUtils
+
+
+class CreateAgentTaskInput(BaseModel):
+    """创建 Agent 自主定时任务的输入参数。"""
+
+    name: str = Field(
+        ...,
+        min_length=1,
+        max_length=100,
+        description="Short task name shown in task management and execution reports.",
+    )
+    content: str = Field(
+        ...,
+        min_length=1,
+        max_length=10000,
+        description="Complete instructions that the agent must execute when the task fires.",
+    )
+    trigger_type: Literal["date", "cron"] = Field(
+        ...,
+        description="Use 'date' for one exact future run or 'cron' for recurring work.",
+    )
+    trigger: Optional[str] = Field(
+        None,
+        min_length=1,
+        max_length=200,
+        description=(
+            "For date, an ISO 8601 local or timezone-aware time such as "
+            "2026-07-19 20:30:00; for cron, a standard five-field expression "
+            "(minute hour day month weekday). The MoviePilot system timezone is used."
+        ),
+    )
+    delay_minutes: Optional[int] = Field(
+        None,
+        ge=1,
+        le=525600,
+        description=(
+            "For a one-time date task expressed as 'in N minutes', provide this instead "
+            "of trigger. MoviePilot calculates and persists the exact future run time."
+        ),
+    )
+
+    @model_validator(mode="after")
+    def validate_trigger(self) -> "CreateAgentTaskInput":
+        """校验任务触发配置并统一格式。"""
+        self.name = self.name.strip()
+        self.content = self.content.strip()
+        if not self.name or not self.content:
+            raise ValueError("name 和 content 不能只包含空白字符")
+        if self.trigger_type == "date":
+            if (self.trigger is None) == (self.delay_minutes is None):
+                raise ValueError("date 任务必须且只能提供 trigger 或 delay_minutes 之一")
+            if self.delay_minutes is not None:
+                timezone = pytz.timezone(settings.TZ)
+                self.trigger = (
+                    datetime.now(timezone) + timedelta(minutes=self.delay_minutes)
+                ).isoformat(timespec="seconds")
+        elif self.trigger is None or self.delay_minutes is not None:
+            raise ValueError("cron 任务必须提供 trigger，且不能提供 delay_minutes")
+        self.trigger_type, self.trigger = TimerUtils.normalize_schedule_trigger(
+            trigger_type=self.trigger_type,
+            trigger_value=self.trigger,
+            timezone_name=settings.TZ,
+            require_future=True,
+        )
+        return self
+
+
+class CreateAgentTaskTool(MoviePilotTool):
+    """创建可精确唤醒当前 Agent 会话的自主定时任务。"""
+
+    name: str = "create_agent_task"
+    tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin]
+    description: str = (
+        "Create a persistent autonomous agent task only when the user explicitly asks "
+        "for delayed, scheduled, recurring, reminder, or monitoring work. Use trigger_type "
+        "'date' with delay_minutes for requests such as 'check in 30 minutes', an exact "
+        "trigger time for other one-time work, and 'cron' for recurring schedules. When "
+        "fired, MoviePilot wakes the agent in this conversation, executes content, and "
+        "sends the result to the user."
+    )
+    args_schema: Type[BaseModel] = CreateAgentTaskInput
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs: object) -> Optional[str]:
+        """生成创建定时任务的提示消息。"""
+        return f"创建自主定时任务：{kwargs.get('name', '')}"
+
+    def _create_task(self, payload: CreateAgentTaskInput) -> dict:
+        """持久化任务并立即注册到运行时调度器。"""
+        from app.scheduler import Scheduler
+
+        chat = AgentChatOper().get(
+            session_id=self._session_id,
+            user_id=self._user_id,
+        )
+        task = AgentTaskOper().add(
+            name=payload.name.strip(),
+            content=payload.content.strip(),
+            trigger_type=payload.trigger_type,
+            cron_expression=payload.trigger if payload.trigger_type == "cron" else None,
+            run_at=payload.trigger if payload.trigger_type == "date" else None,
+            user_id=str(self._user_id),
+            username=self._username or (chat.username if chat else None),
+            session_id=str(self._session_id),
+            channel=self._channel or (chat.channel if chat else None),
+            source=self._source or (chat.source if chat else None),
+            original_chat_id=chat.original_chat_id if chat else None,
+        )
+        scheduler = Scheduler()
+        next_run_at = scheduler.update_agent_task_job(task.id)
+        return AgentTaskOper.to_dict(
+            task,
+            next_run_at=next_run_at,
+            timezone=settings.TZ,
+        )
+
+    async def run(
+            self,
+            name: str,
+            content: str,
+            trigger_type: str,
+            trigger: Optional[str] = None,
+            delay_minutes: Optional[int] = None,
+            **kwargs: object,
+    ) -> str:
+        """创建 Agent 自主定时任务。"""
+        if not settings.AI_AGENT_ENABLE:
+            return "AI Agent 未启用，无法创建自主定时任务"
+        payload = CreateAgentTaskInput(
+            name=name,
+            content=content,
+            trigger_type=trigger_type,
+            trigger=trigger,
+            delay_minutes=delay_minutes,
+        )
+        task = await self.run_blocking("db", self._create_task, payload)
+        return json.dumps(task, ensure_ascii=False, indent=2)

app/agent/tools/impl/delete_agent_task.py

@@ -0,0 +1,50 @@
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.agent.tools.tags import ToolTag
+from app.db.agenttask_oper import AgentTaskOper
+
+
+class DeleteAgentTaskInput(BaseModel):
+    """删除 Agent 自主定时任务的输入参数。"""
+
+    task_id: int = Field(..., ge=1, description="ID of the task to permanently delete.")
+
+
+class DeleteAgentTaskTool(MoviePilotTool):
+    """永久删除 Agent 自主定时任务。"""
+
+    name: str = "delete_agent_task"
+    tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin]
+    description: str = (
+        "Permanently delete an autonomous agent task and remove its runtime schedule. "
+        "Use update_agent_task with enabled=false when the user only wants to pause it."
+    )
+    args_schema: Type[BaseModel] = DeleteAgentTaskInput
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs: object) -> Optional[str]:
+        """生成删除定时任务的提示消息。"""
+        return f"删除自主定时任务：{kwargs.get('task_id', '')}"
+
+    def _delete_task(self, task_id: int) -> bool:
+        """删除当前用户的任务并移除运行时调度。"""
+        from app.scheduler import Scheduler
+
+        deleted = AgentTaskOper().delete(
+            task_id=task_id,
+            user_id=str(self._user_id),
+        )
+        if deleted:
+            Scheduler().remove_agent_task_job(task_id)
+        return deleted
+
+    async def run(self, task_id: int, **kwargs: object) -> str:
+        """删除 Agent 自主定时任务。"""
+        payload = DeleteAgentTaskInput(task_id=task_id)
+        deleted = await self.run_blocking("db", self._delete_task, payload.task_id)
+        if not deleted:
+            return f"Agent 定时任务 {task_id} 不存在或不属于当前用户"
+        return f"Agent 定时任务 {task_id} 已删除"

app/agent/tools/impl/query_agent_tasks.py

@@ -0,0 +1,86 @@
+import json
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.agent.tools.tags import ToolTag
+from app.core.config import settings
+from app.db.agenttask_oper import AgentTaskOper
+
+
+class QueryAgentTasksInput(BaseModel):
+    """查询 Agent 自主定时任务的输入参数。"""
+
+    task_id: Optional[int] = Field(
+        None,
+        ge=1,
+        description="Optional task ID. Omit it to list tasks owned by the current user.",
+    )
+    enabled: Optional[bool] = Field(
+        None,
+        description="Optional enabled-state filter used when listing tasks.",
+    )
+
+
+class QueryAgentTasksTool(MoviePilotTool):
+    """查询当前用户创建的 Agent 自主定时任务。"""
+
+    name: str = "query_agent_tasks"
+    tags: list[str] = [ToolTag.Read, ToolTag.Scheduler, ToolTag.Admin]
+    description: str = (
+        "Query persistent autonomous agent tasks, including their task content, exact "
+        "date or cron trigger, enabled state, next run time, and latest execution result."
+    )
+    args_schema: Type[BaseModel] = QueryAgentTasksInput
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs: object) -> Optional[str]:
+        """生成查询定时任务的提示消息。"""
+        task_id = kwargs.get("task_id")
+        return f"查询自主定时任务：{task_id}" if task_id else "查询自主定时任务"
+
+    def _query_tasks(
+            self,
+            task_id: Optional[int],
+            enabled: Optional[bool],
+    ) -> list[dict]:
+        """读取当前用户的任务及运行时下一次触发时间。"""
+        from app.scheduler import Scheduler
+
+        oper = AgentTaskOper()
+        if task_id:
+            task = oper.get(task_id=task_id, user_id=str(self._user_id))
+            tasks = [task] if task else []
+        else:
+            tasks = oper.list(user_id=str(self._user_id), enabled=enabled)
+        scheduler = Scheduler()
+        result = []
+        for task in tasks:
+            data = oper.to_dict(
+                task,
+                next_run_at=scheduler.get_agent_task_next_run(task.id),
+                timezone=settings.TZ,
+            )
+            result.append(data)
+        return result
+
+    async def run(
+            self,
+            task_id: Optional[int] = None,
+            enabled: Optional[bool] = None,
+            **kwargs: object,
+    ) -> str:
+        """查询 Agent 自主定时任务。"""
+        payload = QueryAgentTasksInput(task_id=task_id, enabled=enabled)
+        tasks = await self.run_blocking(
+            "db",
+            self._query_tasks,
+            payload.task_id,
+            payload.enabled,
+        )
+        return json.dumps(
+            {"total": len(tasks), "tasks": tasks},
+            ensure_ascii=False,
+            indent=2,
+        )

app/agent/tools/impl/update_agent_task.py

@@ -0,0 +1,188 @@
+import json
+from datetime import datetime, timedelta
+from typing import Literal, Optional, Type
+
+import pytz
+from pydantic import BaseModel, Field, model_validator
+
+from app.agent.tools.base import MoviePilotTool
+from app.agent.tools.tags import ToolTag
+from app.core.config import settings
+from app.db.agenttask_oper import AgentTaskOper
+from app.utils.timer import TimerUtils
+
+
+class UpdateAgentTaskInput(BaseModel):
+    """更新 Agent 自主定时任务的输入参数。"""
+
+    task_id: int = Field(..., ge=1, description="ID of the task to update.")
+    name: Optional[str] = Field(None, min_length=1, max_length=100)
+    content: Optional[str] = Field(None, min_length=1, max_length=10000)
+    trigger_type: Optional[Literal["date", "cron"]] = Field(
+        None,
+        description="New trigger type. Must be provided together with trigger.",
+    )
+    trigger: Optional[str] = Field(
+        None,
+        min_length=1,
+        max_length=200,
+        description="New ISO 8601 date or five-field cron expression.",
+    )
+    delay_minutes: Optional[int] = Field(
+        None,
+        ge=1,
+        le=525600,
+        description=(
+            "For a one-time date task expressed as 'in N minutes', provide this instead "
+            "of trigger together with trigger_type='date'."
+        ),
+    )
+    enabled: Optional[bool] = Field(
+        None,
+        description="Set false to pause the task or true to resume it.",
+    )
+
+    @model_validator(mode="after")
+    def validate_update(self) -> "UpdateAgentTaskInput":
+        """校验更新内容和触发参数组合。"""
+        if self.name is not None:
+            self.name = self.name.strip()
+            if not self.name:
+                raise ValueError("name 不能只包含空白字符")
+        if self.content is not None:
+            self.content = self.content.strip()
+            if not self.content:
+                raise ValueError("content 不能只包含空白字符")
+        has_schedule_update = any(
+            value is not None
+            for value in (self.trigger_type, self.trigger, self.delay_minutes)
+        )
+        if has_schedule_update:
+            if self.trigger_type is None:
+                raise ValueError("修改触发配置时必须提供 trigger_type")
+            if self.trigger_type == "date":
+                if (self.trigger is None) == (self.delay_minutes is None):
+                    raise ValueError("date 任务必须且只能提供 trigger 或 delay_minutes 之一")
+                if self.delay_minutes is not None:
+                    timezone = pytz.timezone(settings.TZ)
+                    self.trigger = (
+                        datetime.now(timezone) + timedelta(minutes=self.delay_minutes)
+                    ).isoformat(timespec="seconds")
+            elif self.trigger is None or self.delay_minutes is not None:
+                raise ValueError("cron 任务必须提供 trigger，且不能提供 delay_minutes")
+        if all(
+            value is None
+            for value in (
+                self.name,
+                self.content,
+                self.trigger_type,
+                self.enabled,
+            )
+        ):
+            raise ValueError("至少需要提供一个要更新的字段")
+        return self
+
+
+class UpdateAgentTaskTool(MoviePilotTool):
+    """修改、暂停或恢复 Agent 自主定时任务。"""
+
+    name: str = "update_agent_task"
+    tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin]
+    description: str = (
+        "Update an autonomous agent task's name, instructions, exact date or cron "
+        "trigger, relative delay_minutes, or enabled state. Use enabled=false to pause "
+        "and enabled=true to resume."
+    )
+    args_schema: Type[BaseModel] = UpdateAgentTaskInput
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs: object) -> Optional[str]:
+        """生成更新定时任务的提示消息。"""
+        return f"更新自主定时任务：{kwargs.get('task_id', '')}"
+
+    def _update_task(self, payload: UpdateAgentTaskInput) -> Optional[dict]:
+        """更新当前用户的任务并刷新运行时调度。"""
+        from app.scheduler import Scheduler
+
+        oper = AgentTaskOper()
+        task = oper.get(task_id=payload.task_id, user_id=str(self._user_id))
+        if not task:
+            return None
+        if task.last_status == "running":
+            return {"error": f"Agent 定时任务 {payload.task_id} 正在执行，请稍后再修改"}
+
+        trigger_type = payload.trigger_type or task.trigger_type
+        trigger_value = payload.trigger or (
+            task.cron_expression if trigger_type == "cron" else task.run_at
+        )
+        enabled = task.enabled if payload.enabled is None else payload.enabled
+        normalized_type, normalized_trigger = TimerUtils.normalize_schedule_trigger(
+            trigger_type=trigger_type,
+            trigger_value=trigger_value,
+            timezone_name=settings.TZ,
+            require_future=bool(enabled and trigger_type == "date"),
+        )
+
+        update_payload = {}
+        if payload.name is not None:
+            update_payload["name"] = payload.name.strip()
+        if payload.content is not None:
+            update_payload["content"] = payload.content.strip()
+        if payload.trigger is not None:
+            update_payload.update(
+                {
+                    "trigger_type": normalized_type,
+                    "cron_expression": (
+                        normalized_trigger if normalized_type == "cron" else None
+                    ),
+                    "run_at": normalized_trigger if normalized_type == "date" else None,
+                    "last_status": "waiting",
+                    "last_result": None,
+                }
+            )
+        if payload.enabled is not None:
+            update_payload["enabled"] = payload.enabled
+            if payload.enabled:
+                update_payload["last_status"] = "waiting"
+
+        oper.update(
+            task_id=payload.task_id,
+            payload=update_payload,
+            user_id=str(self._user_id),
+        )
+        scheduler = Scheduler()
+        next_run_at = scheduler.update_agent_task_job(payload.task_id)
+        updated_task = oper.get(task_id=payload.task_id, user_id=str(self._user_id))
+        return oper.to_dict(
+            updated_task,
+            next_run_at=next_run_at,
+            timezone=settings.TZ,
+        )
+
+    async def run(
+            self,
+            task_id: int,
+            name: Optional[str] = None,
+            content: Optional[str] = None,
+            trigger_type: Optional[str] = None,
+            trigger: Optional[str] = None,
+            delay_minutes: Optional[int] = None,
+            enabled: Optional[bool] = None,
+            **kwargs: object,
+    ) -> str:
+        """更新 Agent 自主定时任务。"""
+        payload = UpdateAgentTaskInput(
+            task_id=task_id,
+            name=name,
+            content=content,
+            trigger_type=trigger_type,
+            trigger=trigger,
+            delay_minutes=delay_minutes,
+            enabled=enabled,
+        )
+        task = await self.run_blocking("db", self._update_task, payload)
+        if not task:
+            return f"Agent 定时任务 {task_id} 不存在或不属于当前用户"
+        if task.get("error"):
+            return task["error"]
+        return json.dumps(task, ensure_ascii=False, indent=2)

app/db/agenttask_oper.py

@@ -0,0 +1,150 @@
+from datetime import datetime
+from typing import Optional
+
+from app.db import DbOper
+from app.db.models.agenttask import AgentTask
+
+
+class AgentTaskOper(DbOper):
+    """
+    Agent 自主定时任务管理。
+    """
+
+    @staticmethod
+    def _now() -> str:
+        """生成当前数据库时间字符串。"""
+        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+    def add(self, **kwargs: object) -> AgentTask:
+        """
+        新增 Agent 定时任务。
+        """
+        now = self._now()
+        task_id = AgentTask.add_task(
+            self._db,
+            **kwargs,
+            enabled=True,
+            last_status="waiting",
+            run_count=0,
+            created_at=now,
+            updated_at=now,
+        )
+        return self.get(task_id)
+
+    def get(
+            self,
+            task_id: int,
+            user_id: Optional[str] = None,
+    ) -> Optional[AgentTask]:
+        """
+        查询单个 Agent 定时任务。
+        """
+        return AgentTask.get_for_user(self._db, task_id=task_id, user_id=user_id)
+
+    def list(
+            self,
+            user_id: Optional[str] = None,
+            enabled: Optional[bool] = None,
+    ) -> list[AgentTask]:
+        """
+        查询 Agent 定时任务列表。
+        """
+        return AgentTask.list_for_user(self._db, user_id=user_id, enabled=enabled)
+
+    def update(
+            self,
+            task_id: int,
+            payload: dict,
+            user_id: Optional[str] = None,
+    ) -> bool:
+        """
+        更新 Agent 定时任务。
+        """
+        normalized_payload = {
+            key: value
+            for key, value in payload.items()
+            if key in {
+                "name",
+                "content",
+                "trigger_type",
+                "cron_expression",
+                "run_at",
+                "enabled",
+                "last_status",
+                "last_result",
+            }
+        }
+        if not normalized_payload:
+            return False
+        normalized_payload["updated_at"] = self._now()
+        return AgentTask.update_task(
+            self._db,
+            task_id=task_id,
+            payload=normalized_payload,
+            user_id=user_id,
+        )
+
+    def delete(self, task_id: int, user_id: Optional[str] = None) -> bool:
+        """
+        删除 Agent 定时任务。
+        """
+        return AgentTask.delete_task(
+            self._db,
+            task_id=task_id,
+            user_id=user_id,
+        )
+
+    def mark_running(self, task_id: int) -> bool:
+        """
+        将 Agent 定时任务标记为运行中。
+        """
+        return AgentTask.mark_running(
+            self._db,
+            task_id=task_id,
+            run_at=self._now(),
+        )
+
+    def finish(
+            self,
+            task_id: int,
+            success: bool,
+            result: str,
+            disable: bool = False,
+    ) -> bool:
+        """
+        记录 Agent 定时任务执行结果。
+        """
+        return AgentTask.finish_task(
+            self._db,
+            task_id=task_id,
+            success=success,
+            result=(result or "")[:20000],
+            disable=disable,
+        )
+
+    @staticmethod
+    def to_dict(
+            task: AgentTask,
+            next_run_at: Optional[str] = None,
+            timezone: Optional[str] = None,
+    ) -> dict:
+        """
+        将 Agent 定时任务转换为工具可返回的结构。
+        """
+        return {
+            "id": task.id,
+            "name": task.name,
+            "content": task.content,
+            "trigger_type": task.trigger_type,
+            "cron_expression": task.cron_expression,
+            "run_at": task.run_at,
+            "timezone": timezone,
+            "enabled": bool(task.enabled),
+            "last_status": task.last_status,
+            "last_run_at": task.last_run_at,
+            "last_result": task.last_result,
+            "run_count": task.run_count or 0,
+            "next_run_at": next_run_at,
+            "created_at": task.created_at,
+            "updated_at": task.updated_at,
+        }

app/db/models/__init__.py

@@ -1,4 +1,5 @@
 from .agentchat import AgentChat
+from .agenttask import AgentTask
 from .downloadfailure import DownloadFailure
 from .downloadhistory import DownloadHistory, DownloadFiles
 from .mediaserver import MediaServerItem

app/db/models/agenttask.py

@@ -0,0 +1,170 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import Boolean, Column, Index, Integer, String, Text
+from sqlalchemy.orm import Session
+
+from app.db import Base, db_query, db_update, get_id_column
+
+
+class AgentTask(Base):
+    """
+    Agent 自主定时任务表。
+    """
+
+    id = get_id_column()
+    # 任务名称
+    name = Column(String, nullable=False)
+    # 交给 Agent 执行的完整任务内容
+    content = Column(Text, nullable=False)
+    # 触发类型：date-单次触发，cron-周期触发
+    trigger_type = Column(String, nullable=False)
+    # 标准五段 cron 表达式
+    cron_expression = Column(String)
+    # 单次触发时间，使用带时区的 ISO 8601 格式
+    run_at = Column(String)
+    # 是否继续接受调度
+    enabled = Column(Boolean, nullable=False, default=True)
+    # 创建任务的用户与会话上下文
+    user_id = Column(String, nullable=False)
+    username = Column(String)
+    session_id = Column(String, nullable=False)
+    channel = Column(String)
+    source = Column(String)
+    original_chat_id = Column(String)
+    # 最近一次执行状态与结果
+    last_status = Column(String, nullable=False, default="waiting")
+    last_run_at = Column(String)
+    last_result = Column(Text)
+    run_count = Column(Integer, nullable=False, default=0)
+    created_at = Column(String, nullable=False)
+    updated_at = Column(String, nullable=False)
+
+    __table_args__ = (
+        Index("ix_agenttask_enabled", "enabled"),
+        Index("ix_agenttask_user_created", "user_id", "created_at", "id"),
+    )
+
+    @classmethod
+    @db_update
+    def add_task(cls, db: Session, **kwargs: object) -> int:
+        """
+        新增 Agent 定时任务并返回任务 ID。
+        """
+        task = cls(**kwargs)
+        db.add(task)
+        db.flush()
+        return task.id
+
+    @classmethod
+    @db_query
+    def get_for_user(
+            cls,
+            db: Session,
+            task_id: int,
+            user_id: Optional[str] = None,
+    ) -> Optional["AgentTask"]:
+        """
+        按任务 ID 和可选用户 ID 查询 Agent 定时任务。
+        """
+        query = db.query(cls).filter(cls.id == task_id)
+        if user_id is not None:
+            query = query.filter(cls.user_id == user_id)
+        return query.first()
+
+    @classmethod
+    @db_query
+    def list_for_user(
+            cls,
+            db: Session,
+            user_id: Optional[str] = None,
+            enabled: Optional[bool] = None,
+    ) -> list["AgentTask"]:
+        """
+        按用户和启用状态查询 Agent 定时任务。
+        """
+        query = db.query(cls)
+        if user_id is not None:
+            query = query.filter(cls.user_id == user_id)
+        if enabled is not None:
+            query = query.filter(cls.enabled.is_(enabled))
+        return query.order_by(cls.created_at.desc(), cls.id.desc()).all()
+
+    @classmethod
+    @db_update
+    def update_task(
+            cls,
+            db: Session,
+            task_id: int,
+            payload: dict,
+            user_id: Optional[str] = None,
+    ) -> bool:
+        """
+        按任务 ID 和可选用户 ID 更新 Agent 定时任务。
+        """
+        query = db.query(cls).filter(cls.id == task_id)
+        if user_id is not None:
+            query = query.filter(cls.user_id == user_id)
+        return bool(query.update(payload))
+
+    @classmethod
+    @db_update
+    def delete_task(
+            cls,
+            db: Session,
+            task_id: int,
+            user_id: Optional[str] = None,
+    ) -> bool:
+        """
+        按任务 ID 和可选用户 ID 删除 Agent 定时任务。
+        """
+        query = db.query(cls).filter(cls.id == task_id)
+        if user_id is not None:
+            query = query.filter(cls.user_id == user_id)
+        return bool(query.delete())
+
+    @classmethod
+    @db_update
+    def mark_running(cls, db: Session, task_id: int, run_at: str) -> bool:
+        """
+        将可执行任务标记为运行中。
+        """
+        updated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        return bool(
+            db.query(cls)
+            .filter(
+                cls.id == task_id,
+                cls.enabled.is_(True),
+                cls.last_status != "running",
+            )
+            .update(
+                {
+                    "last_status": "running",
+                    "last_run_at": run_at,
+                    "updated_at": updated_at,
+                }
+            )
+        )
+
+    @classmethod
+    @db_update
+    def finish_task(
+            cls,
+            db: Session,
+            task_id: int,
+            success: bool,
+            result: str,
+            disable: bool = False,
+    ) -> bool:
+        """
+        记录 Agent 定时任务执行结果，并按需关闭单次任务。
+        """
+        payload = {
+            "last_status": "success" if success else "failed",
+            "last_result": result,
+            "run_count": cls.run_count + 1,
+            "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        }
+        if disable:
+            payload["enabled"] = False
+        return bool(db.query(cls).filter(cls.id == task_id).update(payload))

app/scheduler.py

@@ -28,6 +28,7 @@ from app.core.config import settings, global_vars
 from app.core.event import Event, eventmanager
 from app.core.plugin import PluginManager
 from app.db import SessionFactory
+from app.db.agenttask_oper import AgentTaskOper
 from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
 from app.db.models.message import Message
 from app.db.models.siteuserdata import SiteUserData
@@ -48,6 +49,7 @@ from app.utils.timer import TimerUtils
 
 lock = threading.Lock()
 SCHEDULER_PROGRESS_PREFIX = "scheduler"
+AGENT_TASK_JOB_PREFIX = "agent-task"
 
 
 class SchedulerChain(ChainBase):
@@ -705,6 +707,10 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             # 初始化工作流服务
             self.init_workflow_jobs()
 
+            # 恢复 Agent 自主定时任务
+            if settings.AI_AGENT_ENABLE:
+                self.init_agent_task_jobs()
+
             # 初始化插件服务
             self.init_plugin_jobs()
 
@@ -980,6 +986,145 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 # 运行结束
                 self.__finish_job(job_id=job_id, success=success, error=error)
 
+    @staticmethod
+    def _get_agent_task_job_id(task_id: int) -> str:
+        """生成 Agent 自主定时任务的调度器 Job ID。"""
+        return f"{AGENT_TASK_JOB_PREFIX}-{task_id}"
+
+    def init_agent_task_jobs(self) -> None:
+        """
+        从数据库恢复所有启用的 Agent 自主定时任务。
+        """
+        oper = AgentTaskOper()
+        for task in oper.list(enabled=True):
+            if task.last_status == "running":
+                oper.update(
+                    task_id=task.id,
+                    payload={
+                        "last_status": "waiting",
+                        "last_result": "服务重启后恢复调度",
+                    },
+                )
+            self.update_agent_task_job(task.id)
+
+    def update_agent_task_job(self, task_id: int) -> Optional[str]:
+        """
+        按数据库中的最新配置新增或替换 Agent 自主定时任务。
+
+        :param task_id: Agent 定时任务 ID
+        :return: 下一次执行时间，不可调度时返回 None
+        """
+        self.remove_agent_task_job(task_id)
+        task = AgentTaskOper().get(task_id)
+        if (
+                not settings.AI_AGENT_ENABLE
+                or not task
+                or not task.enabled
+                or not self._scheduler
+        ):
+            return None
+
+        trigger_value = (
+            task.cron_expression if task.trigger_type == "cron" else task.run_at
+        )
+        try:
+            trigger = TimerUtils.build_schedule_trigger(
+                trigger_type=task.trigger_type,
+                trigger_value=trigger_value,
+                timezone_name=settings.TZ,
+            )
+        except (TypeError, ValueError) as err:
+            logger.error(f"Agent 定时任务 {task_id} 的触发配置无效：{str(err)}")
+            return None
+
+        job_id = self._get_agent_task_job_id(task_id)
+        with self._lock:
+            self._jobs[job_id] = {
+                "name": task.name,
+                "provider_name": "[Agent]",
+                "func": self.execute_agent_task,
+                "running": False,
+                "kwargs": {"task_id": task_id},
+            }
+            self._scheduler.add_job(
+                self.start,
+                trigger=trigger,
+                id=job_id,
+                name=task.name,
+                kwargs={"job_id": job_id, "task_id": task_id},
+                coalesce=True,
+                max_instances=1,
+                misfire_grace_time=None,
+                replace_existing=True,
+            )
+        return self.get_agent_task_next_run(task_id)
+
+    def remove_agent_task_job(self, task_id: int) -> None:
+        """
+        从运行时调度器移除 Agent 自主定时任务。
+
+        :param task_id: Agent 定时任务 ID
+        """
+        job_id = self._get_agent_task_job_id(task_id)
+        with self._lock:
+            self._jobs.pop(job_id, None)
+            if not self._scheduler:
+                return
+            try:
+                self._scheduler.remove_job(job_id)
+            except JobLookupError:
+                pass
+
+    def get_agent_task_next_run(self, task_id: int) -> Optional[str]:
+        """
+        查询 Agent 自主定时任务的下一次执行时间。
+
+        :param task_id: Agent 定时任务 ID
+        :return: 带时区的 ISO 8601 时间，不再执行时返回 None
+        """
+        job_id = self._get_agent_task_job_id(task_id)
+        if self._scheduler:
+            job = self._scheduler.get_job(job_id)
+            next_run_time = getattr(job, "next_run_time", None) if job else None
+            if next_run_time:
+                return next_run_time.isoformat(timespec="seconds")
+
+        task = AgentTaskOper().get(task_id)
+        if not task or not task.enabled:
+            return None
+        trigger_value = (
+            task.cron_expression if task.trigger_type == "cron" else task.run_at
+        )
+        try:
+            next_run_time = TimerUtils.get_schedule_next_run_time(
+                trigger_type=task.trigger_type,
+                trigger_value=trigger_value,
+                timezone_name=settings.TZ,
+            )
+        except (TypeError, ValueError):
+            return None
+        return (
+            next_run_time.isoformat(timespec="seconds")
+            if next_run_time
+            else None
+        )
+
+    async def execute_agent_task(self, task_id: int) -> tuple[bool, str]:
+        """
+        唤醒 Agent 执行指定自主定时任务。
+
+        :param task_id: Agent 定时任务 ID
+        :return: 执行是否成功及结果摘要
+        """
+        from app.agent import agent_manager
+
+        try:
+            return await agent_manager.execute_scheduled_task(task_id)
+        finally:
+            task = AgentTaskOper().get(task_id)
+            if task and task.trigger_type == "date" and not task.enabled:
+                self.remove_agent_task_job(task_id)
+
     def init_plugin_jobs(self):
         """
         初始化插件定时服务

app/utils/timer.py

@@ -1,9 +1,117 @@
 import datetime
 import random
-from typing import List
+from typing import List, Optional, Tuple, Union
+
+import pytz
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
 
 
 class TimerUtils:
+    """
+    定时与时间差计算工具。
+    """
+
+    SCHEDULE_TRIGGER_TYPES = ("date", "cron")
+
+    @staticmethod
+    def normalize_schedule_trigger(
+            trigger_type: str,
+            trigger_value: str,
+            timezone_name: str,
+            require_future: bool = False,
+    ) -> Tuple[str, str]:
+        """
+        校验并规范化单次时间或五段 cron 表达式。
+
+        :param trigger_type: 触发类型，支持 date 或 cron
+        :param trigger_value: 带时区或本地时间字符串，或五段 cron 表达式
+        :param timezone_name: 无显式时区时使用的系统时区
+        :param require_future: 单次任务是否必须安排在未来
+        :return: 规范化后的触发类型和触发值
+        """
+        normalized_type = str(trigger_type or "").strip().lower()
+        normalized_value = str(trigger_value or "").strip()
+        if normalized_type not in TimerUtils.SCHEDULE_TRIGGER_TYPES:
+            raise ValueError("trigger_type 仅支持 date 或 cron")
+        if not normalized_value:
+            raise ValueError("trigger 不能为空")
+
+        timezone = pytz.timezone(timezone_name)
+        if normalized_type == "cron":
+            normalized_value = " ".join(normalized_value.split())
+            if len(normalized_value.split()) != 5:
+                raise ValueError("cron 必须是标准五段表达式：分 时 日 月 周")
+            CronTrigger.from_crontab(normalized_value, timezone=timezone)
+            return normalized_type, normalized_value
+
+        try:
+            run_at = datetime.datetime.fromisoformat(normalized_value)
+        except ValueError as err:
+            raise ValueError(
+                "date 时间必须使用 ISO 8601 格式，例如 2026-07-19 20:30:00"
+            ) from err
+        if run_at.tzinfo is None:
+            run_at = timezone.localize(run_at)
+        else:
+            run_at = run_at.astimezone(timezone)
+        if require_future and run_at <= datetime.datetime.now(timezone):
+            raise ValueError("单次任务的触发时间必须晚于当前时间")
+        return normalized_type, run_at.isoformat(timespec="seconds")
+
+    @staticmethod
+    def build_schedule_trigger(
+            trigger_type: str,
+            trigger_value: str,
+            timezone_name: str,
+    ) -> Union[CronTrigger, DateTrigger]:
+        """
+        构建 APScheduler 单次或 cron 触发器。
+
+        :param trigger_type: 触发类型，支持 date 或 cron
+        :param trigger_value: 已配置的触发时间或 cron 表达式
+        :param timezone_name: 调度器使用的系统时区
+        :return: APScheduler 触发器
+        """
+        normalized_type, normalized_value = TimerUtils.normalize_schedule_trigger(
+            trigger_type=trigger_type,
+            trigger_value=trigger_value,
+            timezone_name=timezone_name,
+        )
+        timezone = pytz.timezone(timezone_name)
+        if normalized_type == "cron":
+            return CronTrigger.from_crontab(normalized_value, timezone=timezone)
+        return DateTrigger(
+            run_date=datetime.datetime.fromisoformat(normalized_value),
+            timezone=timezone,
+        )
+
+    @staticmethod
+    def get_schedule_next_run_time(
+            trigger_type: str,
+            trigger_value: str,
+            timezone_name: str,
+            now: Optional[datetime.datetime] = None,
+    ) -> Optional[datetime.datetime]:
+        """
+        计算指定触发配置的下一次执行时间。
+
+        :param trigger_type: 触发类型，支持 date 或 cron
+        :param trigger_value: 已配置的触发时间或 cron 表达式
+        :param timezone_name: 调度器使用的系统时区
+        :param now: 可选的计算基准时间
+        :return: 下一次执行时间，不再触发时返回 None
+        """
+        timezone = pytz.timezone(timezone_name)
+        current_time = now or datetime.datetime.now(timezone)
+        if current_time.tzinfo is None:
+            current_time = timezone.localize(current_time)
+        trigger = TimerUtils.build_schedule_trigger(
+            trigger_type=trigger_type,
+            trigger_value=trigger_value,
+            timezone_name=timezone_name,
+        )
+        return trigger.get_next_fire_time(None, current_time)
 
     @staticmethod
     def random_scheduler(num_executions: int = 1,

database/versions/c4e8f7a1b2d3_2_2_12.py

@@ -0,0 +1,67 @@
+"""2.2.12
+新增 Agent 自主定时任务表
+
+Revision ID: c4e8f7a1b2d3
+Revises: b7d4a9c2e6f1
+Create Date: 2026-07-19
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "c4e8f7a1b2d3"
+down_revision = "b7d4a9c2e6f1"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(inspector: sa.Inspector, table_name: str) -> bool:
+    """检查数据表是否已存在。"""
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    """升级数据库结构。"""
+    inspector = sa.inspect(op.get_bind())
+    if _has_table(inspector, "agenttask"):
+        return
+
+    op.create_table(
+        "agenttask",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("trigger_type", sa.String(), nullable=False),
+        sa.Column("cron_expression", sa.String(), nullable=True),
+        sa.Column("run_at", sa.String(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), nullable=False),
+        sa.Column("user_id", sa.String(), nullable=False),
+        sa.Column("username", sa.String(), nullable=True),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("channel", sa.String(), nullable=True),
+        sa.Column("source", sa.String(), nullable=True),
+        sa.Column("original_chat_id", sa.String(), nullable=True),
+        sa.Column("last_status", sa.String(), nullable=False),
+        sa.Column("last_run_at", sa.String(), nullable=True),
+        sa.Column("last_result", sa.Text(), nullable=True),
+        sa.Column("run_count", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.String(), nullable=False),
+        sa.Column("updated_at", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_agenttask_enabled", "agenttask", ["enabled"])
+    op.create_index(
+        "ix_agenttask_user_created",
+        "agenttask",
+        ["user_id", "created_at", "id"],
+    )
+
+
+def downgrade() -> None:
+    """回滚数据库结构。"""
+    inspector = sa.inspect(op.get_bind())
+    if not _has_table(inspector, "agenttask"):
+        return
+    op.drop_index("ix_agenttask_user_created", table_name="agenttask")
+    op.drop_index("ix_agenttask_enabled", table_name="agenttask")
+    op.drop_table("agenttask")

docs/mcp-api.md

@@ -181,6 +181,35 @@ FastAPI 异常响应保留 `detail` 字段，并在错误详情为文本时返
 
 工具的 `inputSchema` 只包含实际执行业务所需的参数，不包含用于解释调用原因的通用 `explanation` 参数，以减少 Agent 上下文消耗。
 
+#### Agent 自主定时任务工具
+
+以下工具用于管理会在指定时间重新唤醒 Agent 的持久化任务，均为管理员级工具：
+
+| 工具 | 说明 |
+| :--- | :--- |
+| `create_agent_task` | 创建单次或周期任务，并保存任务内容及当前用户、会话和消息渠道 |
+| `query_agent_tasks` | 查询任务配置、启用状态、下次执行时间及最近执行结果 |
+| `update_agent_task` | 修改任务内容或触发器，也可通过 `enabled` 暂停、恢复任务 |
+| `delete_agent_task` | 永久删除任务并立即移除运行时调度 |
+
+`trigger_type=date` 表示单次执行：“30 分钟后检查”这类相对时间传 `delay_minutes=30`，由后端计算精确时间；固定时间则传 ISO 8601 `trigger`，支持精确到秒。`trigger_type=cron` 使用标准五段 cron（分、时、日、月、周），适合周期检查。未显式携带时区的时间按 MoviePilot 的 `TZ` 配置解释。任务由内存调度器精确触发，配置持久化到数据库，服务重启后会自动恢复；触发后 Agent 在原会话中执行 `content` 并把结果发送到原消息渠道。通过无会话上下文的 MCP/CLI 创建时，结果改发到 MoviePilot 已配置的管理员通知渠道。
+
+创建单次任务的参数示例：
+
+```json
+{
+  "tool_name": "create_agent_task",
+  "arguments": {
+    "name": "检查电影资源",
+    "content": "搜索电影《示例电影》是否已有可下载资源，并报告站点、版本和大小；不要自动下载。",
+    "trigger_type": "date",
+    "delay_minutes": 30
+  }
+}
+```
+
+创建每天 20:30 执行的周期任务时，使用 `trigger_type=cron` 和 `trigger="30 20 * * *"`。
+
 **认证**: 需要API KEY，在请求头中添加 `X-API-KEY: <api_key>` 或在查询参数中添加 `apikey=<api_key>`
 
 **响应示例**:

skills/moviepilot-cli/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: moviepilot-cli
-version: 3
+version: 4
 description: >-
   Use this skill when the user asks to operate MoviePilot through the local
   `moviepilot tool` MCP CLI for normal product workflows: media search, torrent
@@ -58,7 +58,7 @@ Always run `show <command>` before calling a command — parameter names are not
 | Library | query_library_exists, query_library_latest, transfer_file, scrape_metadata, query_transfer_history |
 | Files | list_directory, query_directory_settings |
 | Sites | query_sites, query_site_userdata, test_site, update_site, update_site_cookie |
-| System | query_schedulers, run_scheduler, query_workflows, run_workflow, query_rule_groups, query_episode_schedule, send_message |
+| System | query_schedulers, run_scheduler, create_agent_task, query_agent_tasks, update_agent_task, delete_agent_task, query_workflows, run_workflow, query_rule_groups, query_episode_schedule, send_message |
 
 ## Workflows
 
@@ -186,6 +186,32 @@ Trigger a search for missing episodes (confirm with user first):
 Remove a subscription (confirm with user first):
 `moviepilot tool run delete_subscribe subscribe_id=123`
 
+### Manage Autonomous Agent Tasks
+
+Use autonomous tasks only when the user explicitly requests delayed, recurring,
+reminder, or monitoring behavior. Immediate work should run directly. Use the
+MoviePilot `TZ` setting for local times.
+
+For a relative one-time request, use `date` with `delay_minutes`; MoviePilot
+calculates and persists the exact run time:
+`moviepilot tool run create_agent_task name="检查电影资源" content="搜索电影《示例电影》是否有资源并报告，不要自动下载。" trigger_type=date delay_minutes=30`
+
+For a one-time task at a fixed time, use `date` with an ISO 8601 `trigger`:
+`moviepilot tool run create_agent_task name="今晚检查资源" content="检查目标电影是否有资源并报告。" trigger_type=date trigger="2026-07-19 20:30:00"`
+
+For recurring work, use a standard five-field cron expression. This example
+runs every day at 20:30:
+`moviepilot tool run create_agent_task name="每日资源检查" content="检查目标电影是否有资源并报告。" trigger_type=cron trigger="30 20 * * *"`
+
+List tasks and inspect `next_run_at` and the latest result:
+`moviepilot tool run query_agent_tasks`
+
+Pause or resume a task:
+`moviepilot tool run update_agent_task task_id=1 enabled=false`
+
+Delete a task only after confirming permanent removal with the user:
+`moviepilot tool run delete_agent_task task_id=1`
+
 ### Check Library and Subscriptions
 
 Run before any download or subscription to avoid duplicates.

tests/test_agent_scheduled_tasks.py

@@ -0,0 +1,340 @@
+import json
+import threading
+from datetime import datetime, timedelta
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+import pytz
+from apscheduler.schedulers.background import BackgroundScheduler
+
+from app.agent import AgentChain, AgentManager, ReplyMode
+from app.agent.tools.factory import MoviePilotToolFactory
+from app.agent.tools.impl.create_agent_task import (
+    CreateAgentTaskInput,
+    CreateAgentTaskTool,
+)
+from app.agent.tools.impl.delete_agent_task import DeleteAgentTaskTool
+from app.agent.tools.impl.query_agent_tasks import QueryAgentTasksTool
+from app.agent.tools.impl.update_agent_task import UpdateAgentTaskTool
+from app.core.config import settings
+from app.db.agenttask_oper import AgentTaskOper
+from app.scheduler import Scheduler
+from app.utils.timer import TimerUtils
+
+
+class _FakeAgentTaskScheduler:
+    """记录 Agent 定时任务工具触发的运行时调度变更。"""
+
+    def __init__(self) -> None:
+        """初始化运行时调度记录。"""
+        self.updated = []
+        self.removed = []
+
+    def update_agent_task_job(self, task_id: int) -> str:
+        """记录任务重载并返回固定的下一次执行时间。"""
+        self.updated.append(task_id)
+        return "2099-01-01T00:00:00+08:00"
+
+    def remove_agent_task_job(self, task_id: int) -> None:
+        """记录任务移除。"""
+        self.removed.append(task_id)
+
+    def get_agent_task_next_run(self, task_id: int) -> str:
+        """返回固定的下一次执行时间。"""
+        return "2099-01-01T00:00:00+08:00"
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    """限定异步用例使用项目 Agent 运行时采用的 asyncio 后端。"""
+    return "asyncio"
+
+
+@pytest.fixture(autouse=True)
+def enable_ai_agent(monkeypatch) -> None:
+    """在当前测试模块中启用 Agent 调度能力并在用例后自动还原。"""
+    monkeypatch.setattr(settings, "AI_AGENT_ENABLE", True)
+
+
+def _future_time(minutes: int = 10) -> str:
+    """生成系统时区内的未来时间字符串。"""
+    timezone = pytz.timezone(settings.TZ)
+    return (datetime.now(timezone) + timedelta(minutes=minutes)).isoformat(
+        timespec="seconds"
+    )
+
+
+def _build_tool(tool_class, user_id: str):
+    """构造带当前用户消息上下文的 Agent 工具。"""
+    tool = tool_class(session_id=f"session-{user_id}", user_id=user_id)
+    tool.set_message_attr(
+        channel="Telegram",
+        source="telegram-test",
+        username="admin",
+    )
+    tool.set_agent_context({"is_admin": True})
+    return tool
+
+
+def test_timer_utils_validates_date_and_cron_triggers() -> None:
+    """自主任务时间工具应规范化单次时间并校验五段 cron。"""
+    trigger_type, trigger = TimerUtils.normalize_schedule_trigger(
+        trigger_type="date",
+        trigger_value=_future_time(),
+        timezone_name=settings.TZ,
+        require_future=True,
+    )
+    assert trigger_type == "date"
+    assert datetime.fromisoformat(trigger).tzinfo is not None
+
+    trigger_type, trigger = TimerUtils.normalize_schedule_trigger(
+        trigger_type="cron",
+        trigger_value="  30   20  * * * ",
+        timezone_name=settings.TZ,
+    )
+    assert trigger_type == "cron"
+    assert trigger == "30 20 * * *"
+
+    with pytest.raises(ValueError, match="标准五段"):
+        TimerUtils.normalize_schedule_trigger(
+            trigger_type="cron",
+            trigger_value="30 20 * *",
+            timezone_name=settings.TZ,
+        )
+
+
+def test_agent_task_tools_are_registered_with_relative_delay_schema() -> None:
+    """工具工厂应公开完整任务管理工具，并声明相对分钟参数。"""
+    tool_names = {
+        tool_class.model_fields["name"].default
+        for tool_class in MoviePilotToolFactory.BUILTIN_TOOL_CLASSES
+    }
+    assert {
+        "create_agent_task",
+        "query_agent_tasks",
+        "update_agent_task",
+        "delete_agent_task",
+    }.issubset(tool_names)
+    assert "delay_minutes" in CreateAgentTaskInput.model_json_schema()["properties"]
+
+
+def test_agent_task_oper_persists_and_scopes_tasks() -> None:
+    """AgentTaskOper 应持久化任务并按创建用户隔离查询和修改。"""
+    user_id = f"user-{uuid4().hex}"
+    oper = AgentTaskOper()
+    task = oper.add(
+        name="检查资源",
+        content="检查电影资源",
+        trigger_type="cron",
+        cron_expression="0 */2 * * *",
+        run_at=None,
+        user_id=user_id,
+        username="admin",
+        session_id=f"session-{user_id}",
+        channel="Telegram",
+        source="telegram-test",
+        original_chat_id="chat-1",
+    )
+
+    assert oper.get(task.id, user_id=user_id).content == "检查电影资源"
+    assert oper.get(task.id, user_id="another-user") is None
+    assert [item.id for item in oper.list(user_id=user_id)] == [task.id]
+
+    assert oper.update(
+        task_id=task.id,
+        user_id=user_id,
+        payload={"content": "检查更新后的资源", "unknown": "ignored"},
+    )
+    assert oper.get(task.id).content == "检查更新后的资源"
+    assert oper.mark_running(task.id)
+    assert not oper.mark_running(task.id)
+    assert oper.finish(task.id, success=True, result="完成")
+    assert not oper.delete(task.id, user_id="another-user")
+    assert oper.delete(task.id, user_id=user_id)
+
+
+def test_scheduler_registers_and_removes_agent_task_job() -> None:
+    """Scheduler 应把数据库任务注册为精确 APScheduler Job 并可动态移除。"""
+    user_id = f"user-{uuid4().hex}"
+    task = AgentTaskOper().add(
+        name="定时检查",
+        content="检查资源",
+        trigger_type="date",
+        cron_expression=None,
+        run_at=_future_time(),
+        user_id=user_id,
+        username="admin",
+        session_id=f"session-{user_id}",
+        channel="Telegram",
+        source="telegram-test",
+        original_chat_id="chat-1",
+    )
+    scheduler = object.__new__(Scheduler)
+    scheduler._lock = threading.RLock()
+    scheduler._jobs = {}
+    scheduler._scheduler = BackgroundScheduler(timezone=settings.TZ)
+
+    next_run_at = scheduler.update_agent_task_job(task.id)
+    job_id = scheduler._get_agent_task_job_id(task.id)
+
+    assert next_run_at
+    assert scheduler._scheduler.get_job(job_id) is not None
+    assert scheduler._jobs[job_id]["kwargs"] == {"task_id": task.id}
+
+    scheduler.remove_agent_task_job(task.id)
+    assert scheduler._scheduler.get_job(job_id) is None
+    assert job_id not in scheduler._jobs
+
+
+@pytest.mark.anyio
+async def test_agent_task_tools_manage_persistent_schedule(monkeypatch) -> None:
+    """Agent 管理工具应完成创建、查询、暂停、修改和删除闭环。"""
+    user_id = f"user-{uuid4().hex}"
+    fake_scheduler = _FakeAgentTaskScheduler()
+    monkeypatch.setattr("app.scheduler.Scheduler", lambda: fake_scheduler)
+
+    create_tool = _build_tool(CreateAgentTaskTool, user_id)
+    created = json.loads(
+        await create_tool.run(
+            name="十分钟后检查",
+            content="检查示例电影是否有资源，不要自动下载",
+            trigger_type="date",
+            delay_minutes=10,
+        )
+    )
+    task_id = created["id"]
+    assert created["enabled"] is True
+    assert datetime.fromisoformat(created["run_at"]) > datetime.now(
+        pytz.timezone(settings.TZ)
+    )
+    assert created["next_run_at"] == "2099-01-01T00:00:00+08:00"
+    assert fake_scheduler.updated == [task_id]
+
+    query_tool = _build_tool(QueryAgentTasksTool, user_id)
+    queried = json.loads(await query_tool.run(task_id=task_id))
+    assert queried["total"] == 1
+    assert queried["tasks"][0]["content"].startswith("检查示例电影")
+
+    update_tool = _build_tool(UpdateAgentTaskTool, user_id)
+    updated = json.loads(
+        await update_tool.run(
+            task_id=task_id,
+            content="检查示例电影是否有 4K 资源，不要自动下载",
+            trigger_type="cron",
+            trigger="*/15 * * * *",
+            enabled=True,
+        )
+    )
+    assert updated["trigger_type"] == "cron"
+    assert updated["cron_expression"] == "*/15 * * * *"
+    assert updated["run_at"] is None
+
+    assert AgentTaskOper().mark_running(task_id)
+    running_update = await update_tool.run(task_id=task_id, enabled=False)
+    assert running_update == f"Agent 定时任务 {task_id} 正在执行，请稍后再修改"
+    AgentTaskOper().finish(task_id, success=True, result="完成")
+
+    deleted = await _build_tool(DeleteAgentTaskTool, user_id).run(task_id=task_id)
+    assert deleted == f"Agent 定时任务 {task_id} 已删除"
+    assert fake_scheduler.removed == [task_id]
+
+
+@pytest.mark.anyio
+async def test_agent_manager_executes_task_in_original_session() -> None:
+    """定时触发应复用原 Agent 会话和渠道，并在单次执行后停用任务。"""
+    user_id = f"user-{uuid4().hex}"
+    task = AgentTaskOper().add(
+        name="检查电影资源",
+        content="搜索示例电影是否已有资源",
+        trigger_type="date",
+        cron_expression=None,
+        run_at=_future_time(),
+        user_id=user_id,
+        username="admin",
+        session_id=f"session-{user_id}",
+        channel="Telegram",
+        source="telegram-test",
+        original_chat_id="chat-123",
+    )
+    manager = AgentManager()
+    manager.process_message = AsyncMock(return_value="已找到 2 个资源")
+
+    success, result = await manager.execute_scheduled_task(task.id)
+
+    assert success is True
+    assert result == "已找到 2 个资源"
+    kwargs = manager.process_message.await_args.kwargs
+    assert kwargs["session_id"] == task.session_id
+    assert kwargs["user_id"] == user_id
+    assert kwargs["channel"] == "Telegram"
+    assert kwargs["source"] == "telegram-test"
+    assert kwargs["original_chat_id"] == "chat-123"
+    assert kwargs["reply_mode"] == ReplyMode.DISPATCH
+    assert kwargs["wait_for_completion"] is True
+    assert "搜索示例电影是否已有资源" in kwargs["message"]
+
+    completed = AgentTaskOper().get(task.id)
+    assert completed.enabled is False
+    assert completed.last_status == "success"
+    assert completed.last_result == "已找到 2 个资源"
+    assert completed.run_count == 1
+
+    recurring_task = AgentTaskOper().add(
+        name="周期检查电影资源",
+        content="继续检查示例电影是否已有资源",
+        trigger_type="cron",
+        cron_expression="*/30 * * * *",
+        run_at=None,
+        user_id=user_id,
+        username="admin",
+        session_id=f"session-{user_id}",
+        channel="Telegram",
+        source="telegram-test",
+        original_chat_id="chat-123",
+    )
+
+    success, _ = await manager.execute_scheduled_task(recurring_task.id)
+
+    assert success is True
+    recurring_completed = AgentTaskOper().get(recurring_task.id)
+    assert recurring_completed.enabled is True
+    assert recurring_completed.last_status == "success"
+    assert recurring_completed.run_count == 1
+
+
+@pytest.mark.anyio
+async def test_agent_manager_dispatches_contextless_result_to_admin(
+        monkeypatch,
+) -> None:
+    """无原消息渠道的 MCP 任务应捕获结果并发送到管理员通知渠道。"""
+    user_id = f"api-{uuid4().hex}"
+    task = AgentTaskOper().add(
+        name="后台检查资源",
+        content="检查示例电影是否已有资源",
+        trigger_type="cron",
+        cron_expression="0 * * * *",
+        run_at=None,
+        user_id=user_id,
+        username="API Client",
+        session_id=f"session-{user_id}",
+        channel=None,
+        source="api",
+        original_chat_id=None,
+    )
+    manager = AgentManager()
+    manager.process_message = AsyncMock(return_value="后台检查完成")
+    post_message = AsyncMock()
+    monkeypatch.setattr(AgentChain, "async_post_message", post_message)
+
+    success, result = await manager.execute_scheduled_task(task.id)
+
+    assert success is True
+    assert result == "后台检查完成"
+    kwargs = manager.process_message.await_args.kwargs
+    assert kwargs["reply_mode"] == ReplyMode.CAPTURE_ONLY
+    assert kwargs["allow_message_tools"] is False
+    notification = post_message.await_args.args[0]
+    assert notification.userid is None
+    assert notification.username == settings.SUPERUSER
+    assert notification.text == "后台检查完成"

tests/test_builtin_skill_boundaries.py

@@ -23,7 +23,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
     expected_versions = {
         "database-operation": "3",
         "moviepilot-api": "2",
-        "moviepilot-cli": "3",
+        "moviepilot-cli": "4",
         "moviepilot-update": "3",
     }