diff --git a/app/agent/tools/manager.py b/app/agent/tools/manager.py
new file mode 100644
index 00000000..6e56cc21
--- /dev/null
+++ b/app/agent/tools/manager.py
@@ -0,0 +1,187 @@
+"""MoviePilot工具管理器
+用于HTTP API调用工具
+"""
+
+import json
+from typing import Any, Dict, List, Optional
+
+from app.agent.tools.factory import MoviePilotToolFactory
+from app.log import logger
+
+
+class ToolDefinition:
+    """工具定义"""
+
+    def __init__(self, name: str, description: str, input_schema: Dict[str, Any]):
+        self.name = name
+        self.description = description
+        self.input_schema = input_schema
+
+
+class MoviePilotToolsManager:
+    """MoviePilot工具管理器（用于HTTP API）"""
+
+    def __init__(self, user_id: str = "api_user", session_id: str = "api_session"):
+        """
+        初始化工具管理器
+        
+        Args:
+            user_id: 用户ID
+            session_id: 会话ID
+        """
+        self.user_id = user_id
+        self.session_id = session_id
+        self.tools: List[Any] = []
+        self._load_tools()
+
+    def _load_tools(self):
+        """加载所有MoviePilot工具"""
+        try:
+            # 创建工具实例
+            self.tools = MoviePilotToolFactory.create_tools(
+                session_id=self.session_id,
+                user_id=self.user_id,
+                channel=None,
+                source="api",
+                username="API Client",
+                callback_handler=None
+            )
+            logger.info(f"成功加载 {len(self.tools)} 个工具")
+        except Exception as e:
+            logger.error(f"加载工具失败: {e}", exc_info=True)
+            self.tools = []
+
+    def list_tools(self) -> List[ToolDefinition]:
+        """
+        列出所有可用的工具
+        
+        Returns:
+            工具定义列表
+        """
+        tools_list = []
+        for tool in self.tools:
+            # 获取工具的输入参数模型
+            args_schema = getattr(tool, 'args_schema', None)
+            if args_schema:
+                # 将Pydantic模型转换为JSON Schema
+                input_schema = self._convert_to_json_schema(args_schema)
+            else:
+                # 如果没有args_schema，使用基本信息
+                input_schema = {
+                    "type": "object",
+                    "properties": {},
+                    "required": []
+                }
+
+            tools_list.append(ToolDefinition(
+                name=tool.name,
+                description=tool.description or "",
+                input_schema=input_schema
+            ))
+
+        return tools_list
+
+    def get_tool(self, tool_name: str) -> Optional[Any]:
+        """
+        获取指定工具实例
+        
+        Args:
+            tool_name: 工具名称
+            
+        Returns:
+            工具实例，如果未找到返回None
+        """
+        for tool in self.tools:
+            if tool.name == tool_name:
+                return tool
+        return None
+
+    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
+        """
+        调用工具
+        
+        Args:
+            tool_name: 工具名称
+            arguments: 工具参数
+            
+        Returns:
+            工具执行结果（字符串）
+        """
+        tool_instance = self.get_tool(tool_name)
+
+        if not tool_instance:
+            error_msg = json.dumps({
+                "error": f"工具 '{tool_name}' 未找到"
+            }, ensure_ascii=False)
+            return error_msg
+
+        try:
+            # 调用工具的run方法
+            result = await tool_instance.run(**arguments)
+
+            # 确保返回字符串
+            if isinstance(result, str):
+                return result
+            else:
+                return json.dumps(result, ensure_ascii=False, indent=2)
+        except Exception as e:
+            logger.error(f"调用工具 {tool_name} 时发生错误: {e}", exc_info=True)
+            error_msg = json.dumps({
+                "error": f"调用工具 '{tool_name}' 时发生错误: {str(e)}"
+            }, ensure_ascii=False)
+            return error_msg
+
+    @staticmethod
+    def _convert_to_json_schema(args_schema: Any) -> Dict[str, Any]:
+        """
+        将Pydantic模型转换为JSON Schema
+        
+        Args:
+            args_schema: Pydantic模型类
+            
+        Returns:
+            JSON Schema字典
+        """
+        # 获取Pydantic模型的字段信息
+        schema = args_schema.model_json_schema()
+
+        # 构建JSON Schema
+        properties = {}
+        required = []
+
+        if "properties" in schema:
+            for field_name, field_info in schema["properties"].items():
+                # 转换字段类型
+                field_type = field_info.get("type", "string")
+                field_description = field_info.get("description", "")
+
+                # 处理可选字段
+                if field_name not in schema.get("required", []):
+                    # 可选字段
+                    default_value = field_info.get("default")
+                    properties[field_name] = {
+                        "type": field_type,
+                        "description": field_description
+                    }
+                    if default_value is not None:
+                        properties[field_name]["default"] = default_value
+                else:
+                    properties[field_name] = {
+                        "type": field_type,
+                        "description": field_description
+                    }
+                    required.append(field_name)
+
+                # 处理枚举类型
+                if "enum" in field_info:
+                    properties[field_name]["enum"] = field_info["enum"]
+
+                # 处理数组类型
+                if field_type == "array" and "items" in field_info:
+                    properties[field_name]["items"] = field_info["items"]
+
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required
+        }
diff --git a/app/api/apiv1.py b/app/api/apiv1.py
index 9c2501a1..8453d35a 100644
--- a/app/api/apiv1.py
+++ b/app/api/apiv1.py
@@ -2,7 +2,7 @@ from fastapi import APIRouter
 
 from app.api.endpoints import login, user, webhook, message, site, subscribe, \
     media, douban, search, plugin, tmdb, history, system, download, dashboard, \
-    transfer, mediaserver, bangumi, storage, discover, recommend, workflow, torrent
+    transfer, mediaserver, bangumi, storage, discover, recommend, workflow, torrent, mcp
 
 api_router = APIRouter()
 api_router.include_router(login.router, prefix="/login", tags=["login"])
@@ -28,3 +28,4 @@ api_router.include_router(discover.router, prefix="/discover", tags=["discover"]
 api_router.include_router(recommend.router, prefix="/recommend", tags=["recommend"])
 api_router.include_router(workflow.router, prefix="/workflow", tags=["workflow"])
 api_router.include_router(torrent.router, prefix="/torrent", tags=["torrent"])
+api_router.include_router(mcp.router, prefix="/mcp", tags=["mcp"])
diff --git a/app/api/endpoints/mcp.py b/app/api/endpoints/mcp.py
new file mode 100644
index 00000000..e36bb859
--- /dev/null
+++ b/app/api/endpoints/mcp.py
@@ -0,0 +1,161 @@
+"""工具API端点
+通过HTTP API暴露MoviePilot的智能体工具功能
+"""
+
+from typing import List, Any, Dict, Annotated
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from app import schemas
+from app.agent.tools.manager import MoviePilotToolsManager
+from app.core.security import verify_apikey
+from app.log import logger
+
+router = APIRouter()
+
+# 全局工具管理器实例（单例模式，按用户ID缓存）
+_tools_managers: Dict[str, MoviePilotToolsManager] = {}
+
+
+def get_tools_manager(user_id: str = "mcp_user", session_id: str = "mcp_session") -> MoviePilotToolsManager:
+    """
+    获取工具管理器实例（按用户ID缓存）
+    
+    Args:
+        user_id: 用户ID
+        session_id: 会话ID
+        
+    Returns:
+        MoviePilotToolsManager实例
+    """
+    global _tools_managers
+    # 使用用户ID作为缓存键
+    cache_key = f"{user_id}_{session_id}"
+    if cache_key not in _tools_managers:
+        _tools_managers[cache_key] = MoviePilotToolsManager(
+            user_id=user_id,
+            session_id=session_id
+        )
+    return _tools_managers[cache_key]
+
+
+@router.get("/tools", summary="列出所有可用工具", response_model=List[Dict[str, Any]])
+async def list_tools(
+        _: Annotated[str, Depends(verify_apikey)]
+) -> Any:
+    """
+    获取所有可用的工具列表
+    
+    返回每个工具的名称、描述和参数定义
+    """
+    try:
+        manager = get_tools_manager()
+        # 获取所有工具定义
+        tools = manager.list_tools()
+
+        # 转换为字典格式
+        tools_list = []
+        for tool in tools:
+            tool_dict = {
+                "name": tool.name,
+                "description": tool.description,
+                "inputSchema": tool.input_schema
+            }
+            tools_list.append(tool_dict)
+
+        return tools_list
+    except Exception as e:
+        logger.error(f"获取工具列表失败: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"获取工具列表失败: {str(e)}")
+
+
+@router.post("/tools/call", summary="调用工具", response_model=schemas.ToolCallResponse)
+async def call_tool(
+        request: schemas.ToolCallRequest,
+
+) -> Any:
+    """
+    调用指定的工具
+        
+    Returns:
+        工具执行结果
+    """
+    try:
+        # 使用当前用户ID创建管理器实例
+        manager = get_tools_manager()
+
+        # 调用工具
+        result_text = await manager.call_tool(request.tool_name, request.arguments)
+
+        return schemas.ToolCallResponse(
+            success=True,
+            result=result_text
+        )
+    except Exception as e:
+        logger.error(f"调用工具 {request.tool_name} 失败: {e}", exc_info=True)
+        return schemas.ToolCallResponse(
+            success=False,
+            error=f"调用工具失败: {str(e)}"
+        )
+
+
+@router.get("/tools/{tool_name}", summary="获取工具详情", response_model=Dict[str, Any])
+async def get_tool_info(
+        tool_name: str,
+        _: Annotated[str, Depends(verify_apikey)]
+) -> Any:
+    """
+    获取指定工具的详细信息
+        
+    Returns:
+        工具的详细信息，包括名称、描述和参数定义
+    """
+    try:
+        manager = get_tools_manager()
+        # 获取所有工具
+        tools = manager.list_tools()
+
+        # 查找指定工具
+        for tool in tools:
+            if tool.name == tool_name:
+                return {
+                    "name": tool.name,
+                    "description": tool.description,
+                    "inputSchema": tool.input_schema
+                }
+
+        raise HTTPException(status_code=404, detail=f"工具 '{tool_name}' 未找到")
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取工具信息失败: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"获取工具信息失败: {str(e)}")
+
+
+@router.get("/tools/{tool_name}/schema", summary="获取工具参数Schema", response_model=Dict[str, Any])
+async def get_tool_schema(
+        tool_name: str,
+        _: Annotated[str, Depends(verify_apikey)]
+) -> Any:
+    """
+    获取指定工具的参数Schema（JSON Schema格式）
+        
+    Returns:
+        工具的JSON Schema定义
+    """
+    try:
+        manager = get_tools_manager()
+        # 获取所有工具
+        tools = manager.list_tools()
+
+        # 查找指定工具
+        for tool in tools:
+            if tool.name == tool_name:
+                return tool.input_schema
+
+        raise HTTPException(status_code=404, detail=f"工具 '{tool_name}' 未找到")
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取工具Schema失败: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"获取工具Schema失败: {str(e)}")
diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py
index 69bba4f8..d04cdbb8 100644
--- a/app/schemas/__init__.py
+++ b/app/schemas/__init__.py
@@ -22,3 +22,5 @@ from .token import *
 from .transfer import *
 from .user import *
 from .workflow import *
+from .mcp import *
+
diff --git a/app/schemas/mcp.py b/app/schemas/mcp.py
new file mode 100644
index 00000000..18b49a65
--- /dev/null
+++ b/app/schemas/mcp.py
@@ -0,0 +1,16 @@
+from typing import Any, Dict, Optional
+
+from pydantic import BaseModel, Field
+
+
+class ToolCallRequest(BaseModel):
+    """工具调用请求模型"""
+    tool_name: str = Field(..., description="工具名称")
+    arguments: Dict[str, Any] = Field(default_factory=dict, description="工具参数")
+
+
+class ToolCallResponse(BaseModel):
+    """工具调用响应模型"""
+    success: bool = Field(..., description="是否成功")
+    result: Optional[str] = Field(None, description="工具执行结果")
+    error: Optional[str] = Field(None, description="错误信息")
diff --git a/docs/mcp-api.md b/docs/mcp-api.md
new file mode 100644
index 00000000..4d9f564f
--- /dev/null
+++ b/docs/mcp-api.md
@@ -0,0 +1,291 @@
+# MoviePilot 工具API文档
+
+MoviePilot的智能体工具已通过HTTP API暴露，可以通过RESTful API调用所有工具。
+
+## API端点
+
+所有工具相关的API端点都在 `/api/v1/mcp` 路径下（保持向后兼容）。
+
+### 1. 列出所有工具
+
+**GET** `/api/v1/mcp/tools`
+
+获取所有可用的MCP工具列表。
+
+**认证**: 需要API KEY，?api_key=MoviePilot设置中的API Key
+
+**响应示例**:
+```json
+[
+  {
+    "name": "add_subscribe",
+    "description": "Add media subscription to create automated download rules...",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "title": {
+          "type": "string",
+          "description": "The title of the media to subscribe to"
+        },
+        "year": {
+          "type": "string",
+          "description": "Release year of the media"
+        },
+        ...
+      },
+      "required": ["title", "year", "media_type"]
+    }
+  },
+  ...
+]
+```
+
+### 2. 调用工具
+
+**POST** `/api/v1/mcp/tools/call`
+
+调用指定的MCP工具。
+
+**认证**: 需要Bearer Token
+
+**请求体**:
+```json
+{
+  "tool_name": "add_subscribe",
+  "arguments": {
+    "title": "流浪地球",
+    "year": "2019",
+    "media_type": "电影"
+  }
+}
+```
+
+**响应示例**:
+```json
+{
+  "success": true,
+  "result": "成功添加订阅：流浪地球 (2019)",
+  "error": null
+}
+```
+
+**错误响应示例**:
+```json
+{
+  "success": false,
+  "result": null,
+  "error": "调用工具失败: 参数验证失败"
+}
+```
+
+### 3. 获取工具详情
+
+**GET** `/api/v1/mcp/tools/{tool_name}`
+
+获取指定工具的详细信息。
+
+**认证**: 需要Bearer Token
+
+**路径参数**:
+- `tool_name`: 工具名称
+
+**响应示例**:
+```json
+{
+  "name": "add_subscribe",
+  "description": "Add media subscription to create automated download rules...",
+  "inputSchema": {
+    "type": "object",
+    "properties": {
+      "title": {
+        "type": "string",
+        "description": "The title of the media to subscribe to"
+      },
+      ...
+    },
+    "required": ["title", "year", "media_type"]
+  }
+}
+```
+
+### 4. 获取工具参数Schema
+
+**GET** `/api/v1/mcp/tools/{tool_name}/schema`
+
+获取指定工具的参数Schema（JSON Schema格式）。
+
+**认证**: 需要Bearer Token
+
+**路径参数**:
+- `tool_name`: 工具名称
+
+**响应示例**:
+```json
+{
+  "type": "object",
+  "properties": {
+    "title": {
+      "type": "string",
+      "description": "The title of the media to subscribe to"
+    },
+    "year": {
+      "type": "string",
+      "description": "Release year of the media"
+    },
+    ...
+  },
+  "required": ["title", "year", "media_type"]
+}
+```
+
+## 使用示例
+
+### 使用curl调用工具
+
+```bash
+# 1. 获取访问令牌（通过登录API）
+TOKEN=$(curl -X POST "http://localhost:3001/api/v1/login/access-token" \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "username=admin&password=your_password" | jq -r '.access_token')
+
+# 2. 列出所有工具
+curl -X GET "http://localhost:3001/api/v1/mcp/tools" \
+  -H "Authorization: Bearer $TOKEN"
+
+# 3. 调用工具
+curl -X POST "http://localhost:3001/api/v1/mcp/tools/call" \
+  -H "Authorization: Bearer $TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "tool_name": "query_subscribes",
+    "arguments": {
+      "status": "all",
+      "media_type": "all"
+    }
+  }'
+
+# 4. 获取工具详情
+curl -X GET "http://localhost:3001/api/v1/mcp/tools/add_subscribe" \
+  -H "Authorization: Bearer $TOKEN"
+```
+
+### 使用Python调用
+
+```python
+import requests
+
+# 配置
+BASE_URL = "http://localhost:3001/api/v1"
+TOKEN = "your_access_token"
+HEADERS = {"Authorization": f"Bearer {TOKEN}"}
+
+# 1. 列出所有工具
+response = requests.get(f"{BASE_URL}/mcp/tools", headers=HEADERS)
+tools = response.json()
+print(f"可用工具数量: {len(tools)}")
+
+# 2. 调用工具
+tool_call = {
+    "tool_name": "add_subscribe",
+    "arguments": {
+        "title": "流浪地球",
+        "year": "2019",
+        "media_type": "电影"
+    }
+}
+response = requests.post(
+    f"{BASE_URL}/mcp/tools/call",
+    headers=HEADERS,
+    json=tool_call
+)
+result = response.json()
+print(f"执行结果: {result['result']}")
+
+# 3. 获取工具Schema
+response = requests.get(
+    f"{BASE_URL}/mcp/tools/add_subscribe/schema",
+    headers=HEADERS
+)
+schema = response.json()
+print(f"工具Schema: {schema}")
+```
+
+### 使用JavaScript/TypeScript调用
+
+```typescript
+const BASE_URL = 'http://localhost:3001/api/v1';
+const TOKEN = 'your_access_token';
+
+// 列出所有工具
+async function listTools() {
+  const response = await fetch(`${BASE_URL}/mcp/tools`, {
+    headers: {
+      'Authorization': `Bearer ${TOKEN}`
+    }
+  });
+  return await response.json();
+}
+
+// 调用工具
+async function callTool(toolName: string, arguments: Record<string, any>) {
+  const response = await fetch(`${BASE_URL}/mcp/tools/call`, {
+    method: 'POST',
+    headers: {
+      'Authorization': `Bearer ${TOKEN}`,
+      'Content-Type': 'application/json'
+    },
+    body: JSON.stringify({
+      tool_name: toolName,
+      arguments: arguments
+    })
+  });
+  return await response.json();
+}
+
+// 使用示例
+const result = await callTool('query_subscribes', {
+  status: 'all',
+  media_type: 'all'
+});
+console.log(result);
+```
+
+## 认证
+
+所有MCP API端点都需要认证。支持以下认证方式：
+
+1. **Bearer Token**: 在请求头中添加 `Authorization: Bearer <token>`
+2. **API Key**: 在请求头中添加 `X-API-KEY: <api_key>` 或在查询参数中添加 `apikey=<api_key>`
+
+获取Token的方式：
+- 通过登录API: `POST /api/v1/login/access-token`
+- 通过API Key: 在系统设置中生成API Key
+
+## 错误处理
+
+API会返回标准的HTTP状态码：
+
+- `200 OK`: 请求成功
+- `400 Bad Request`: 请求参数错误
+- `401 Unauthorized`: 未认证或Token无效
+- `404 Not Found`: 工具不存在
+- `500 Internal Server Error`: 服务器内部错误
+
+错误响应格式：
+```json
+{
+  "detail": "错误描述信息"
+}
+```
+
+## 架构说明
+
+工具API通过FastAPI端点暴露，使用HTTP协议与客户端通信。所有工具共享相同的实现，确保功能一致性。
+
+## 注意事项
+
+1. **用户上下文**: API调用会使用当前认证用户的ID作为工具执行的用户上下文
+2. **会话隔离**: 每个API请求使用独立的会话ID
+3. **参数验证**: 工具参数会根据JSON Schema进行验证
+4. **错误日志**: 所有工具调用错误都会记录到MoviePilot日志系统
+
diff --git a/requirements.in b/requirements.in
index de28a79e..b10dd956 100644
--- a/requirements.in
+++ b/requirements.in
@@ -87,4 +87,4 @@ langchain-openai==0.3.33
 langchain-google-genai==2.0.10
 langchain-deepseek==0.1.4
 langchain-experimental==0.3.4
-openai==1.108.2
\ No newline at end of file
+openai~=2.8.1
\ No newline at end of file