feat: add new query tools for enhanced subscription management

- Introduced QuerySubscribeSharesTool, QueryPopularSubscribesTool, and QuerySubscribeHistoryTool to improve subscription querying capabilities.
- Updated __all__ exports in init.py and factory.py to include the new tools.
- Enhanced QuerySubscribesTool to support media type filtering with localized descriptions.

app/agent/tools/__init__.py

@@ -6,6 +6,9 @@ from app.agent.tools.impl.add_subscribe import AddSubscribeTool
 from app.agent.tools.impl.search_torrents import SearchTorrentsTool
 from app.agent.tools.impl.add_download import AddDownloadTool
 from app.agent.tools.impl.query_subscribes import QuerySubscribesTool
+from app.agent.tools.impl.query_subscribe_shares import QuerySubscribeSharesTool
+from app.agent.tools.impl.query_popular_subscribes import QueryPopularSubscribesTool
+from app.agent.tools.impl.query_subscribe_history import QuerySubscribeHistoryTool
 from app.agent.tools.impl.delete_subscribe import DeleteSubscribeTool
 from app.agent.tools.impl.query_downloads import QueryDownloadsTool
 from app.agent.tools.impl.delete_download import DeleteDownloadTool
@@ -21,6 +24,8 @@ from app.agent.tools.impl.transfer_file import TransferFileTool
 from app.agent.tools.impl.send_message import SendMessageTool
 from app.agent.tools.impl.query_schedulers import QuerySchedulersTool
 from app.agent.tools.impl.run_scheduler import RunSchedulerTool
+from app.agent.tools.impl.query_workflows import QueryWorkflowsTool
+from app.agent.tools.impl.run_workflow import RunWorkflowTool
 from app.agent.tools.impl.update_site_cookie import UpdateSiteCookieTool
 from .factory import MoviePilotToolFactory
 
@@ -31,6 +36,9 @@ __all__ = [
     "SearchTorrentsTool",
     "AddDownloadTool",
     "QuerySubscribesTool",
+    "QuerySubscribeSharesTool",
+    "QueryPopularSubscribesTool",
+    "QuerySubscribeHistoryTool",
     "DeleteSubscribeTool",
     "QueryDownloadsTool",
     "DeleteDownloadTool",
@@ -47,5 +55,7 @@ __all__ = [
     "SendMessageTool",
     "QuerySchedulersTool",
     "RunSchedulerTool",
+    "QueryWorkflowsTool",
+    "RunWorkflowTool",
     "MoviePilotToolFactory"
 ]

app/agent/tools/factory.py

@@ -11,12 +11,17 @@ from app.agent.tools.impl.query_media_library import QueryMediaLibraryTool
 from app.agent.tools.impl.query_sites import QuerySitesTool
 from app.agent.tools.impl.test_site import TestSiteTool
 from app.agent.tools.impl.query_subscribes import QuerySubscribesTool
+from app.agent.tools.impl.query_subscribe_shares import QuerySubscribeSharesTool
+from app.agent.tools.impl.query_popular_subscribes import QueryPopularSubscribesTool
+from app.agent.tools.impl.query_subscribe_history import QuerySubscribeHistoryTool
 from app.agent.tools.impl.delete_subscribe import DeleteSubscribeTool
 from app.agent.tools.impl.search_media import SearchMediaTool
 from app.agent.tools.impl.search_torrents import SearchTorrentsTool
 from app.agent.tools.impl.send_message import SendMessageTool
 from app.agent.tools.impl.query_schedulers import QuerySchedulersTool
 from app.agent.tools.impl.run_scheduler import RunSchedulerTool
+from app.agent.tools.impl.query_workflows import QueryWorkflowsTool
+from app.agent.tools.impl.run_workflow import RunWorkflowTool
 from app.agent.tools.impl.update_site_cookie import UpdateSiteCookieTool
 from app.agent.tools.impl.delete_download import DeleteDownloadTool
 from app.agent.tools.impl.query_directories import QueryDirectoriesTool
@@ -43,6 +48,9 @@ class MoviePilotToolFactory:
             SearchTorrentsTool,
             AddDownloadTool,
             QuerySubscribesTool,
+            QuerySubscribeSharesTool,
+            QueryPopularSubscribesTool,
+            QuerySubscribeHistoryTool,
             DeleteSubscribeTool,
             QueryDownloadsTool,
             DeleteDownloadTool,
@@ -58,7 +66,9 @@ class MoviePilotToolFactory:
             TransferFileTool,
             SendMessageTool,
             QuerySchedulersTool,
-            RunSchedulerTool
+            RunSchedulerTool,
+            QueryWorkflowsTool,
+            RunWorkflowTool
         ]
         # 创建内置工具
         for ToolClass in tool_definitions:

app/agent/tools/impl/query_popular_subscribes.py

@@ -0,0 +1,136 @@
+"""查询热门订阅工具"""
+
+import json
+from typing import Optional, Type
+
+import cn2an
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.core.context import MediaInfo
+from app.helper.subscribe import SubscribeHelper
+from app.log import logger
+from app.schemas.types import MediaType
+
+
+class QueryPopularSubscribesInput(BaseModel):
+    """查询热门订阅工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
+    stype: str = Field(..., description="Media type: '电影' for films, '电视剧' for television series")
+    page: Optional[int] = Field(1, description="Page number for pagination (default: 1)")
+    count: Optional[int] = Field(30, description="Number of items per page (default: 30)")
+    min_sub: Optional[int] = Field(None, description="Minimum number of subscribers filter (optional, e.g., 5)")
+    genre_id: Optional[int] = Field(None, description="Filter by genre ID (optional)")
+    min_rating: Optional[float] = Field(None, description="Minimum rating filter (optional, e.g., 7.5)")
+    max_rating: Optional[float] = Field(None, description="Maximum rating filter (optional, e.g., 10.0)")
+    sort_type: Optional[str] = Field(None, description="Sort type (optional, e.g., 'count', 'rating')")
+
+
+class QueryPopularSubscribesTool(MoviePilotTool):
+    name: str = "query_popular_subscribes"
+    description: str = "Query popular subscriptions based on user shared data. Shows media with the most subscribers, supports filtering by genre, rating, minimum subscribers, and pagination."
+    args_schema: Type[BaseModel] = QueryPopularSubscribesInput
+
+    async def run(self, stype: str,
+                  page: Optional[int] = 1,
+                  count: Optional[int] = 30,
+                  min_sub: Optional[int] = None,
+                  genre_id: Optional[int] = None,
+                  min_rating: Optional[float] = None,
+                  max_rating: Optional[float] = None,
+                  sort_type: Optional[str] = None, **kwargs) -> str:
+        logger.info(
+            f"执行工具: {self.name}, 参数: stype={stype}, page={page}, count={count}, min_sub={min_sub}, "
+            f"genre_id={genre_id}, min_rating={min_rating}, max_rating={max_rating}, sort_type={sort_type}")
+
+        try:
+            if page is None or page < 1:
+                page = 1
+            if count is None or count < 1:
+                count = 30
+
+            subscribe_helper = SubscribeHelper()
+            subscribes = await subscribe_helper.async_get_statistic(
+                stype=stype,
+                page=page,
+                count=count,
+                genre_id=genre_id,
+                min_rating=min_rating,
+                max_rating=max_rating,
+                sort_type=sort_type
+            )
+
+            if not subscribes:
+                return "未找到热门订阅数据（可能订阅统计功能未启用）"
+
+            # 转换为MediaInfo格式并过滤
+            ret_medias = []
+            for sub in subscribes:
+                # 订阅人数
+                subscriber_count = sub.get("count", 0)
+                # 如果设置了最小订阅人数，进行过滤
+                if min_sub and subscriber_count < min_sub:
+                    continue
+
+                media = MediaInfo()
+                media.type = MediaType(sub.get("type"))
+                media.tmdb_id = sub.get("tmdbid")
+                # 处理标题
+                title = sub.get("name")
+                season = sub.get("season")
+                if season and int(season) > 1 and media.tmdb_id:
+                    # 小写数据转大写
+                    season_str = cn2an.an2cn(season, "low")
+                    title = f"{title} 第{season_str}季"
+                media.title = title
+                media.year = sub.get("year")
+                media.douban_id = sub.get("doubanid")
+                media.bangumi_id = sub.get("bangumiid")
+                media.tvdb_id = sub.get("tvdbid")
+                media.imdb_id = sub.get("imdbid")
+                media.season = sub.get("season")
+                media.overview = sub.get("description")
+                media.vote_average = sub.get("vote")
+                media.poster_path = sub.get("poster")
+                media.backdrop_path = sub.get("backdrop")
+                media.popularity = subscriber_count
+                ret_medias.append(media)
+
+            if not ret_medias:
+                return "未找到符合条件的热门订阅"
+
+            # 转换为字典格式，只保留关键信息
+            simplified_medias = []
+            for media in ret_medias:
+                media_dict = media.to_dict()
+                simplified = {
+                    "type": media_dict.get("type"),
+                    "title": media_dict.get("title"),
+                    "year": media_dict.get("year"),
+                    "tmdb_id": media_dict.get("tmdb_id"),
+                    "douban_id": media_dict.get("douban_id"),
+                    "bangumi_id": media_dict.get("bangumi_id"),
+                    "tvdb_id": media_dict.get("tvdb_id"),
+                    "imdb_id": media_dict.get("imdb_id"),
+                    "season": media_dict.get("season"),
+                    "overview": media_dict.get("overview"),
+                    "vote_average": media_dict.get("vote_average"),
+                    "poster_path": media_dict.get("poster_path"),
+                    "backdrop_path": media_dict.get("backdrop_path"),
+                    "popularity": media_dict.get("popularity"),  # 订阅人数
+                    "subscriber_count": media_dict.get("popularity")  # 明确标注为订阅人数
+                }
+                # 截断过长的描述
+                if simplified.get("overview") and len(simplified["overview"]) > 200:
+                    simplified["overview"] = simplified["overview"][:200] + "..."
+                simplified_medias.append(simplified)
+
+            result_json = json.dumps(simplified_medias, ensure_ascii=False, indent=2)
+
+            pagination_info = f"第 {page} 页，每页 {count} 条，共 {len(simplified_medias)} 条结果"
+
+            return f"{pagination_info}\n\n{result_json}"
+        except Exception as e:
+            logger.error(f"查询热门订阅失败: {e}", exc_info=True)
+            return f"查询热门订阅时发生错误: {str(e)}"
+

app/agent/tools/impl/query_subscribe_history.py

@@ -0,0 +1,100 @@
+"""查询订阅历史工具"""
+
+import json
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.db import AsyncSessionFactory
+from app.db.models.subscribehistory import SubscribeHistory
+from app.log import logger
+
+
+class QuerySubscribeHistoryInput(BaseModel):
+    """查询订阅历史工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
+    media_type: Optional[str] = Field("all", description="Filter by media type: '电影' for films, '电视剧' for television series, 'all' for all types (default: 'all')")
+    name: Optional[str] = Field(None, description="Filter by media name (partial match, optional)")
+
+
+class QuerySubscribeHistoryTool(MoviePilotTool):
+    name: str = "query_subscribe_history"
+    description: str = "Query subscription history records. Shows completed subscriptions with their details including name, type, rating, completion date, and other subscription information. Supports filtering by media type and name. Returns up to 30 records."
+    args_schema: Type[BaseModel] = QuerySubscribeHistoryInput
+
+    async def run(self, media_type: Optional[str] = "all",
+                  name: Optional[str] = None, **kwargs) -> str:
+        logger.info(f"执行工具: {self.name}, 参数: media_type={media_type}, name={name}")
+
+        try:
+            # 获取数据库会话
+            async with AsyncSessionFactory() as db:
+                # 根据类型查询
+                if media_type == "all":
+                    # 查询所有类型，需要分别查询电影和电视剧
+                    movie_history = await SubscribeHistory.async_list_by_type(db, mtype="movie", page=1, count=100)
+                    tv_history = await SubscribeHistory.async_list_by_type(db, mtype="tv", page=1, count=100)
+                    all_history = list(movie_history) + list(tv_history)
+                    # 按日期排序
+                    all_history.sort(key=lambda x: x.date or "", reverse=True)
+                else:
+                    # 查询指定类型
+                    all_history = await SubscribeHistory.async_list_by_type(db, mtype=media_type, page=1, count=100)
+                
+                # 按名称过滤
+                filtered_history = []
+                if name:
+                    name_lower = name.lower()
+                    for record in all_history:
+                        if record.name and name_lower in record.name.lower():
+                            filtered_history.append(record)
+                else:
+                    filtered_history = all_history
+                
+                if not filtered_history:
+                    return "未找到相关订阅历史记录"
+                
+                # 限制最多30条
+                total_count = len(filtered_history)
+                limited_history = filtered_history[:30]
+                
+                # 转换为字典格式，只保留关键信息
+                simplified_records = []
+                for record in limited_history:
+                    simplified = {
+                        "id": record.id,
+                        "name": record.name,
+                        "year": record.year,
+                        "type": record.type,
+                        "season": record.season,
+                        "tmdbid": record.tmdbid,
+                        "doubanid": record.doubanid,
+                        "bangumiid": record.bangumiid,
+                        "poster": record.poster,
+                        "vote": record.vote,
+                        "description": record.description[:200] + "..." if record.description and len(record.description) > 200 else record.description,
+                        "total_episode": record.total_episode,
+                        "date": record.date,
+                        "username": record.username
+                    }
+                    # 添加过滤规则信息（如果有）
+                    if record.filter:
+                        simplified["filter"] = record.filter
+                    if record.quality:
+                        simplified["quality"] = record.quality
+                    if record.resolution:
+                        simplified["resolution"] = record.resolution
+                    simplified_records.append(simplified)
+                
+                result_json = json.dumps(simplified_records, ensure_ascii=False, indent=2)
+                
+                # 如果结果被裁剪，添加提示信息
+                if total_count > 30:
+                    return f"注意：查询结果共找到 {total_count} 条，为节省上下文空间，仅显示前 30 条结果。\n\n{result_json}"
+                
+                return result_json
+        except Exception as e:
+            logger.error(f"查询订阅历史失败: {e}", exc_info=True)
+            return f"查询订阅历史时发生错误: {str(e)}"
+

app/agent/tools/impl/query_subscribe_shares.py

@@ -0,0 +1,94 @@
+"""查询订阅分享工具"""
+
+import json
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.helper.subscribe import SubscribeHelper
+from app.log import logger
+
+
+class QuerySubscribeSharesInput(BaseModel):
+    """查询订阅分享工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
+    name: Optional[str] = Field(None, description="Filter shares by media name (partial match, optional)")
+    page: Optional[int] = Field(1, description="Page number for pagination (default: 1)")
+    count: Optional[int] = Field(30, description="Number of items per page (default: 30)")
+    genre_id: Optional[int] = Field(None, description="Filter by genre ID (optional)")
+    min_rating: Optional[float] = Field(None, description="Minimum rating filter (optional, e.g., 7.5)")
+    max_rating: Optional[float] = Field(None, description="Maximum rating filter (optional, e.g., 10.0)")
+    sort_type: Optional[str] = Field(None, description="Sort type (optional, e.g., 'count', 'rating')")
+
+
+class QuerySubscribeSharesTool(MoviePilotTool):
+    name: str = "query_subscribe_shares"
+    description: str = "Query shared subscriptions from other users. Shows popular subscriptions shared by the community with filtering and pagination support."
+    args_schema: Type[BaseModel] = QuerySubscribeSharesInput
+
+    async def run(self, name: Optional[str] = None,
+                  page: Optional[int] = 1,
+                  count: Optional[int] = 30,
+                  genre_id: Optional[int] = None,
+                  min_rating: Optional[float] = None,
+                  max_rating: Optional[float] = None,
+                  sort_type: Optional[str] = None, **kwargs) -> str:
+        logger.info(
+            f"执行工具: {self.name}, 参数: name={name}, page={page}, count={count}, genre_id={genre_id}, "
+            f"min_rating={min_rating}, max_rating={max_rating}, sort_type={sort_type}")
+
+        try:
+            if page is None or page < 1:
+                page = 1
+            if count is None or count < 1:
+                count = 30
+
+            subscribe_helper = SubscribeHelper()
+            shares = await subscribe_helper.async_get_shares(
+                name=name,
+                page=page,
+                count=count,
+                genre_id=genre_id,
+                min_rating=min_rating,
+                max_rating=max_rating,
+                sort_type=sort_type
+            )
+
+            if not shares:
+                return "未找到订阅分享数据（可能订阅分享功能未启用）"
+
+            # 简化字段，只保留关键信息
+            simplified_shares = []
+            for share in shares:
+                simplified = {
+                    "id": share.get("id"),
+                    "name": share.get("name"),
+                    "year": share.get("year"),
+                    "type": share.get("type"),
+                    "season": share.get("season"),
+                    "tmdbid": share.get("tmdbid"),
+                    "doubanid": share.get("doubanid"),
+                    "bangumiid": share.get("bangumiid"),
+                    "poster": share.get("poster"),
+                    "vote": share.get("vote"),
+                    "description": share.get("description"),
+                    "share_title": share.get("share_title"),
+                    "share_comment": share.get("share_comment"),
+                    "share_user": share.get("share_user"),
+                    "fork_count": share.get("fork_count", 0)
+                }
+                # 截断过长的描述
+                if simplified.get("description") and len(simplified["description"]) > 200:
+                    simplified["description"] = simplified["description"][:200] + "..."
+                simplified_shares.append(simplified)
+
+            result_json = json.dumps(simplified_shares, ensure_ascii=False, indent=2)
+
+            pagination_info = f"第 {page} 页，每页 {count} 条，共 {len(simplified_shares)} 条结果"
+
+            return f"{pagination_info}\n\n{result_json}"
+        except Exception as e:
+            logger.error(f"查询订阅分享失败: {e}", exc_info=True)
+            return f"查询订阅分享时发生错误: {str(e)}"
+

app/agent/tools/impl/query_subscribes.py

@@ -16,7 +16,7 @@ class QuerySubscribesInput(BaseModel):
     status: Optional[str] = Field("all",
                                   description="Filter subscriptions by status: 'R' for enabled subscriptions, 'P' for disabled ones, 'all' for all subscriptions")
     media_type: Optional[str] = Field("all",
-                                      description="Filter by media type: 'movie' for films, 'tv' for television series, 'all' for all types")
+                                      description="Filter by media type: '电影' for films, '电视剧' for television series, 'all' for all types")
 
 
 class QuerySubscribesTool(MoviePilotTool):

app/agent/tools/impl/query_workflows.py

@@ -0,0 +1,107 @@
+"""查询工作流工具"""
+
+import json
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.db import AsyncSessionFactory
+from app.db.workflow_oper import WorkflowOper
+from app.log import logger
+
+
+class QueryWorkflowsInput(BaseModel):
+    """查询工作流工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
+    state: Optional[str] = Field("all", description="Filter workflows by state: 'W' for waiting, 'R' for running, 'P' for paused, 'S' for success, 'F' for failed, 'all' for all workflows (default: 'all')")
+    name: Optional[str] = Field(None, description="Filter workflows by name (partial match, optional)")
+    trigger_type: Optional[str] = Field("all", description="Filter workflows by trigger type: 'timer' for scheduled, 'event' for event-triggered, 'manual' for manual, 'all' for all types (default: 'all')")
+
+
+class QueryWorkflowsTool(MoviePilotTool):
+    name: str = "query_workflows"
+    description: str = "Query workflow list and status. Shows workflow name, description, trigger type, state, execution count, and other workflow details. Supports filtering by state, name, and trigger type."
+    args_schema: Type[BaseModel] = QueryWorkflowsInput
+
+    async def run(self, state: Optional[str] = "all",
+                  name: Optional[str] = None,
+                  trigger_type: Optional[str] = "all", **kwargs) -> str:
+        logger.info(f"执行工具: {self.name}, 参数: state={state}, name={name}, trigger_type={trigger_type}")
+
+        try:
+            # 获取数据库会话
+            async with AsyncSessionFactory() as db:
+                workflow_oper = WorkflowOper(db)
+                workflows = await workflow_oper.async_list()
+                
+                # 过滤工作流
+                filtered_workflows = []
+                for wf in workflows:
+                    # 按状态过滤
+                    if state != "all" and wf.state != state:
+                        continue
+                    
+                    # 按触发类型过滤
+                    if trigger_type != "all":
+                        if trigger_type == "timer" and wf.trigger_type not in ["timer", None]:
+                            continue
+                        elif trigger_type == "event" and wf.trigger_type != "event":
+                            continue
+                        elif trigger_type == "manual" and wf.trigger_type != "manual":
+                            continue
+                    
+                    # 按名称过滤（部分匹配）
+                    if name and wf.name and name.lower() not in wf.name.lower():
+                        continue
+                    
+                    filtered_workflows.append(wf)
+                
+                if not filtered_workflows:
+                    return "未找到相关工作流"
+                
+                # 转换为字典格式，只保留关键信息
+                simplified_workflows = []
+                for wf in filtered_workflows:
+                    # 状态说明
+                    state_map = {
+                        "W": "等待",
+                        "R": "运行中",
+                        "P": "暂停",
+                        "S": "成功",
+                        "F": "失败"
+                    }
+                    state_desc = state_map.get(wf.state, wf.state)
+                    
+                    # 触发类型说明
+                    trigger_type_map = {
+                        "timer": "定时触发",
+                        "event": "事件触发",
+                        "manual": "手动触发"
+                    }
+                    trigger_type_desc = trigger_type_map.get(wf.trigger_type, wf.trigger_type or "定时触发")
+                    
+                    simplified = {
+                        "id": wf.id,
+                        "name": wf.name,
+                        "description": wf.description,
+                        "trigger_type": trigger_type_desc,
+                        "state": state_desc,
+                        "run_count": wf.run_count,
+                        "timer": wf.timer,
+                        "event_type": wf.event_type,
+                        "add_time": wf.add_time,
+                        "last_time": wf.last_time,
+                        "current_action": wf.current_action
+                    }
+                    # 如果有结果，添加结果信息
+                    if wf.result:
+                        simplified["result"] = wf.result
+                    simplified_workflows.append(simplified)
+                
+                result_json = json.dumps(simplified_workflows, ensure_ascii=False, indent=2)
+                return result_json
+        except Exception as e:
+            logger.error(f"查询工作流失败: {e}", exc_info=True)
+            return f"查询工作流时发生错误: {str(e)}"
+

app/agent/tools/impl/run_workflow.py

@@ -0,0 +1,59 @@
+"""执行工作流工具"""
+
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.chain.workflow import WorkflowChain
+from app.db import AsyncSessionFactory
+from app.db.workflow_oper import WorkflowOper
+from app.log import logger
+
+
+class RunWorkflowInput(BaseModel):
+    """执行工作流工具的输入参数模型"""
+    explanation: str = Field(..., description="Clear explanation of why this tool is being used in the current context")
+    workflow_identifier: str = Field(..., description="Workflow identifier: can be workflow ID (integer as string) or workflow name")
+    from_begin: Optional[bool] = Field(True, description="Whether to run workflow from the beginning (default: True, if False will continue from last executed action)")
+
+
+class RunWorkflowTool(MoviePilotTool):
+    name: str = "run_workflow"
+    description: str = "Execute a specific workflow manually. Can run workflow by ID or name. Supports running from the beginning or continuing from the last executed action."
+    args_schema: Type[BaseModel] = RunWorkflowInput
+
+    async def run(self, workflow_identifier: str,
+                  from_begin: Optional[bool] = True, **kwargs) -> str:
+        logger.info(f"执行工具: {self.name}, 参数: workflow_identifier={workflow_identifier}, from_begin={from_begin}")
+
+        try:
+            # 获取数据库会话
+            async with AsyncSessionFactory() as db:
+                workflow_oper = WorkflowOper(db)
+                
+                # 尝试解析为工作流ID
+                workflow = None
+                if workflow_identifier.isdigit():
+                    # 如果是数字，尝试作为工作流ID查询
+                    workflow = await workflow_oper.async_get(int(workflow_identifier))
+                
+                # 如果不是ID或ID查询失败，尝试按名称查询
+                if not workflow:
+                    workflow = await workflow_oper.async_get_by_name(workflow_identifier)
+                
+                if not workflow:
+                    return f"未找到工作流：{workflow_identifier}，请使用 query_workflows 工具查询可用的工作流"
+                
+                # 执行工作流
+                workflow_chain = WorkflowChain()
+                state, errmsg = workflow_chain.process(workflow.id, from_begin=from_begin)
+                
+                if not state:
+                    return f"执行工作流失败：{workflow.name} (ID: {workflow.id})\n错误原因：{errmsg}"
+                else:
+                    return f"工作流执行成功：{workflow.name} (ID: {workflow.id})"
+        except Exception as e:
+            logger.error(f"执行工作流失败: {e}", exc_info=True)
+            return f"执行工作流时发生错误: {str(e)}"
+