diff --git a/backend/app/routers/tasks.py b/backend/app/routers/tasks.py
index 02eb734..e48aede 100644
--- a/backend/app/routers/tasks.py
+++ b/backend/app/routers/tasks.py
@@ -28,8 +28,10 @@ class TaskCreate(BaseModel):
     schedule_type: str = "simple"  # simple | cron
     time_points: Optional[List[str]] = None  # ["09:00", "14:00"]
     cron_expression: Optional[str] = None  # "0 9,14 * * *"
-    webhook_url: str
+    execution_type: str = "webhook"  # webhook | script
+    webhook_url: Optional[str] = None
     input_params: Optional[dict] = None
+    script_content: Optional[str] = None
     is_enabled: bool = True
 
 
@@ -39,8 +41,15 @@ class TaskUpdate(BaseModel):
     schedule_type: Optional[str] = None
     time_points: Optional[List[str]] = None
     cron_expression: Optional[str] = None
+    execution_type: Optional[str] = None
     webhook_url: Optional[str] = None
     input_params: Optional[dict] = None
+    script_content: Optional[str] = None
+
+
+class ScriptTestRequest(BaseModel):
+    tenant_id: str
+    script_content: str
 
 
 # API Endpoints
@@ -110,6 +119,14 @@ async def create_task(
         if not data.cron_expression:
             raise HTTPException(status_code=400, detail="CRON模式需要提供表达式")
     
+    # 验证执行配置
+    if data.execution_type == "webhook":
+        if not data.webhook_url:
+            raise HTTPException(status_code=400, detail="Webhook模式需要提供URL")
+    elif data.execution_type == "script":
+        if not data.script_content:
+            raise HTTPException(status_code=400, detail="脚本模式需要提供脚本内容")
+    
     # 插入数据库
     import json
     time_points_json = json.dumps(data.time_points) if data.time_points else None
@@ -119,9 +136,9 @@ async def create_task(
         text("""
             INSERT INTO platform_scheduled_tasks 
             (tenant_id, task_name, task_desc, schedule_type, time_points, 
-             cron_expression, webhook_url, input_params, is_enabled)
+             cron_expression, execution_type, webhook_url, input_params, script_content, is_enabled)
             VALUES (:tenant_id, :task_name, :task_desc, :schedule_type, :time_points,
-                    :cron_expression, :webhook_url, :input_params, :is_enabled)
+                    :cron_expression, :execution_type, :webhook_url, :input_params, :script_content, :is_enabled)
         """),
         {
             "tenant_id": data.tenant_id,
@@ -130,8 +147,10 @@ async def create_task(
             "schedule_type": data.schedule_type,
             "time_points": time_points_json,
             "cron_expression": data.cron_expression,
+            "execution_type": data.execution_type,
             "webhook_url": data.webhook_url,
             "input_params": input_params_json,
+            "script_content": data.script_content,
             "is_enabled": 1 if data.is_enabled else 0
         }
     )
@@ -209,9 +228,15 @@ async def update_task(
     if data.cron_expression is not None:
         updates.append("cron_expression = :cron_expression")
         params["cron_expression"] = data.cron_expression
+    if data.execution_type is not None:
+        updates.append("execution_type = :execution_type")
+        params["execution_type"] = data.execution_type
     if data.webhook_url is not None:
         updates.append("webhook_url = :webhook_url")
         params["webhook_url"] = data.webhook_url
+    if data.script_content is not None:
+        updates.append("script_content = :script_content")
+        params["script_content"] = data.script_content
     if data.input_params is not None:
         updates.append("input_params = :input_params")
         params["input_params"] = json.dumps(data.input_params)
@@ -354,3 +379,96 @@ async def get_task_logs(
         "size": size,
         "items": logs
     }
+
+
+@router.post("/test-script")
+async def test_script(
+    data: ScriptTestRequest,
+    user: User = Depends(require_operator)
+):
+    """测试执行脚本（不记录日志）"""
+    from ..services.script_executor import test_script as run_test
+    
+    if not data.script_content or not data.script_content.strip():
+        raise HTTPException(status_code=400, detail="脚本内容不能为空")
+    
+    result = await run_test(
+        tenant_id=data.tenant_id,
+        script_content=data.script_content
+    )
+    
+    return result.to_dict()
+
+
+@router.get("/sdk-docs")
+async def get_sdk_docs():
+    """获取 SDK 文档"""
+    return {
+        "description": "脚本执行 SDK 文档",
+        "methods": [
+            {
+                "name": "ai(prompt, system=None, model='gemini-2.5-flash')",
+                "description": "调用大模型生成内容",
+                "example": "result = ai('帮我写一段营销文案')"
+            },
+            {
+                "name": "dingtalk(webhook, content, msg_type='text', at_mobiles=None)",
+                "description": "发送钉钉群消息",
+                "example": "dingtalk('https://oapi.dingtalk.com/robot/send?access_token=xxx', '消息内容')"
+            },
+            {
+                "name": "wecom(webhook, content, msg_type='text')",
+                "description": "发送企业微信群消息",
+                "example": "wecom('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx', '消息内容')"
+            },
+            {
+                "name": "db(sql, params=None)",
+                "description": "执行 SQL 查询（仅支持 SELECT）",
+                "example": "rows = db('SELECT * FROM users WHERE id = :id', {'id': 1})"
+            },
+            {
+                "name": "http_get(url, headers=None, params=None)",
+                "description": "发送 HTTP GET 请求",
+                "example": "data = http_get('https://api.example.com/data')"
+            },
+            {
+                "name": "http_post(url, data=None, json_data=None, headers=None)",
+                "description": "发送 HTTP POST 请求",
+                "example": "data = http_post('https://api.example.com/submit', json_data={'key': 'value'})"
+            },
+            {
+                "name": "get_var(key, default=None)",
+                "description": "获取存储的变量（跨执行持久化）",
+                "example": "count = get_var('run_count', 0)"
+            },
+            {
+                "name": "set_var(key, value)",
+                "description": "存储变量（跨执行持久化）",
+                "example": "set_var('run_count', count + 1)"
+            },
+            {
+                "name": "log(message, level='INFO')",
+                "description": "记录日志",
+                "example": "log('任务执行完成')"
+            }
+        ],
+        "example_script": '''# 示例：每日推送 AI 生成的内容到钉钉
+import json
+
+# 获取历史数据
+history = get_var('history', [])
+
+# 调用 AI 生成内容
+prompt = f"根据以下信息生成今日营销文案：{json.dumps(history[-5:], ensure_ascii=False)}"
+content = ai(prompt, system="你是一个专业的营销文案专家")
+
+# 发送到钉钉
+dingtalk(
+    webhook="你的钉钉机器人Webhook",
+    content=content
+)
+
+# 记录日志
+log(f"已发送: {content[:50]}...")
+'''
+    }
diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py
index 976fc2f..f03c059 100644
--- a/backend/app/services/scheduler.py
+++ b/backend/app/services/scheduler.py
@@ -69,26 +69,64 @@ async def execute_task(task_id: int):
         result = db.execute(text("SELECT LAST_INSERT_ID() as id"))
         log_id = result.scalar()
         
-        # 4. 调用 webhook
-        webhook_url = task["webhook_url"]
-        input_params = task["input_params"] or {}
+        # 生成 trace_id
+        trace_id = f"task_{task_id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
         
-        async with httpx.AsyncClient(timeout=300.0) as client:
-            response = await client.post(
-                webhook_url,
-                json=input_params,
-                headers={"Content-Type": "application/json"}
-            )
+        # 4. 根据执行类型分发
+        execution_type = task.get("execution_type", "webhook")
+        
+        if execution_type == "script":
+            # 脚本执行模式
+            from .script_executor import execute_script as run_script
             
-            response_code = response.status_code
-            response_body = response.text[:5000] if response.text else ""  # 限制存储长度
-            
-            if response.is_success:
-                status = "success"
-                error_message = None
-            else:
+            script_content = task.get("script_content", "")
+            if not script_content:
                 status = "failed"
-                error_message = f"HTTP {response_code}"
+                error_message = "脚本内容为空"
+                response_code = None
+                response_body = ""
+            else:
+                script_result = await run_script(
+                    task_id=task_id,
+                    tenant_id=task["tenant_id"],
+                    script_content=script_content,
+                    trace_id=trace_id
+                )
+                
+                if script_result.success:
+                    status = "success"
+                    error_message = None
+                else:
+                    status = "failed"
+                    error_message = script_result.error
+                
+                response_code = None
+                response_body = script_result.output[:5000] if script_result.output else ""
+                
+                # 添加日志到响应体
+                if script_result.logs:
+                    response_body += "\n\n--- 执行日志 ---\n" + "\n".join(script_result.logs[-20:])
+        else:
+            # Webhook 执行模式
+            webhook_url = task["webhook_url"]
+            input_params = task["input_params"] or {}
+            
+            async with httpx.AsyncClient(timeout=300.0) as client:
+                response = await client.post(
+                    webhook_url,
+                    json=input_params,
+                    headers={"Content-Type": "application/json"}
+                )
+                
+                response_code = response.status_code
+                response_body = response.text[:5000] if response.text else ""  # 限制存储长度
+                
+                if response.is_success:
+                    status = "success"
+                    error_message = None
+                else:
+                    status = "failed"
+                    error_message = f"HTTP {response_code}"
         
         # 5. 更新执行日志
         db.execute(
diff --git a/backend/app/services/script_executor.py b/backend/app/services/script_executor.py
new file mode 100644
index 0000000..464135d
--- /dev/null
+++ b/backend/app/services/script_executor.py
@@ -0,0 +1,262 @@
+"""脚本执行器 - 安全执行 Python 脚本"""
+import asyncio
+import io
+import logging
+import sys
+import traceback
+from contextlib import redirect_stdout, redirect_stderr
+from datetime import datetime
+from typing import Any, Dict
+
+from .script_sdk import ScriptSDK
+
+logger = logging.getLogger(__name__)
+
+# 执行超时时间（秒）
+SCRIPT_TIMEOUT = 300  # 5 分钟
+
+# 禁止导入的模块
+FORBIDDEN_MODULES = {
+    'os', 'subprocess', 'sys', 'builtins', '__builtins__',
+    'importlib', 'eval', 'exec', 'compile',
+    'open', 'file', 'input',
+    'socket', 'multiprocessing', 'threading',
+    'pickle', 'marshal', 'ctypes',
+    'code', 'codeop', 'pty', 'tty',
+}
+
+
+class ScriptExecutionResult:
+    """脚本执行结果"""
+    
+    def __init__(
+        self,
+        success: bool,
+        output: str = "",
+        error: str = None,
+        logs: list = None,
+        execution_time_ms: int = 0
+    ):
+        self.success = success
+        self.output = output
+        self.error = error
+        self.logs = logs or []
+        self.execution_time_ms = execution_time_ms
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "success": self.success,
+            "output": self.output,
+            "error": self.error,
+            "logs": self.logs,
+            "execution_time_ms": self.execution_time_ms
+        }
+
+
+def create_safe_builtins() -> Dict[str, Any]:
+    """创建安全的内置函数集"""
+    import builtins
+    
+    # 允许的内置函数
+    allowed = [
+        'abs', 'all', 'any', 'ascii', 'bin', 'bool', 'bytearray', 'bytes',
+        'callable', 'chr', 'complex', 'dict', 'divmod', 'enumerate',
+        'filter', 'float', 'format', 'frozenset', 'getattr', 'hasattr',
+        'hash', 'hex', 'id', 'int', 'isinstance', 'issubclass', 'iter',
+        'len', 'list', 'map', 'max', 'min', 'next', 'object', 'oct',
+        'ord', 'pow', 'print', 'range', 'repr', 'reversed', 'round',
+        'set', 'slice', 'sorted', 'str', 'sum', 'tuple', 'type', 'zip',
+        'True', 'False', 'None',
+    ]
+    
+    safe_builtins = {}
+    for name in allowed:
+        if hasattr(builtins, name):
+            safe_builtins[name] = getattr(builtins, name)
+    
+    # 添加安全的 import 函数
+    def safe_import(name, *args, **kwargs):
+        """安全的 import 函数，只允许特定模块"""
+        allowed_modules = {
+            'json', 'datetime', 'time', 're', 'math', 'random',
+            'collections', 'itertools', 'functools', 'operator',
+            'string', 'textwrap', 'unicodedata',
+            'hashlib', 'base64', 'urllib.parse',
+        }
+        
+        if name in FORBIDDEN_MODULES:
+            raise ImportError(f"禁止导入模块: {name}")
+        
+        if name not in allowed_modules and not name.startswith('urllib.parse'):
+            raise ImportError(f"不允许导入模块: {name}，允许的模块: {', '.join(sorted(allowed_modules))}")
+        
+        return __builtins__['__import__'](name, *args, **kwargs)
+    
+    safe_builtins['__import__'] = safe_import
+    
+    return safe_builtins
+
+
+async def execute_script(
+    task_id: int,
+    tenant_id: str,
+    script_content: str,
+    trace_id: str = None
+) -> ScriptExecutionResult:
+    """
+    执行 Python 脚本
+    
+    Args:
+        task_id: 任务 ID
+        tenant_id: 租户 ID
+        script_content: 脚本内容
+        trace_id: 追踪 ID
+        
+    Returns:
+        ScriptExecutionResult: 执行结果
+    """
+    start_time = datetime.now()
+    sdk = None
+    
+    try:
+        # 创建 SDK 实例
+        sdk = ScriptSDK(tenant_id, task_id, trace_id)
+        
+        # 准备执行环境
+        script_globals = {
+            '__builtins__': create_safe_builtins(),
+            '__name__': '__script__',
+            
+            # SDK 实例
+            'sdk': sdk,
+            
+            # 快捷方法（同步包装）
+            'ai': lambda *args, **kwargs: asyncio.get_event_loop().run_until_complete(sdk.ai_chat(*args, **kwargs)),
+            'dingtalk': lambda *args, **kwargs: asyncio.get_event_loop().run_until_complete(sdk.send_dingtalk(*args, **kwargs)),
+            'wecom': lambda *args, **kwargs: asyncio.get_event_loop().run_until_complete(sdk.send_wecom(*args, **kwargs)),
+            'http_get': lambda *args, **kwargs: asyncio.get_event_loop().run_until_complete(sdk.http_get(*args, **kwargs)),
+            'http_post': lambda *args, **kwargs: asyncio.get_event_loop().run_until_complete(sdk.http_post(*args, **kwargs)),
+            
+            # 同步方法
+            'db': sdk.db_query,
+            'get_var': sdk.get_var,
+            'set_var': sdk.set_var,
+            'delete_var': sdk.delete_var,
+            'log': sdk.log,
+            
+            # 常用模块
+            'json': __import__('json'),
+            'datetime': __import__('datetime'),
+            're': __import__('re'),
+            'math': __import__('math'),
+            'random': __import__('random'),
+        }
+        
+        # 捕获输出
+        stdout = io.StringIO()
+        stderr = io.StringIO()
+        
+        sdk.log("脚本开始执行")
+        
+        # 编译并执行脚本
+        try:
+            # 编译脚本
+            code = compile(script_content, '<script>', 'exec')
+            
+            # 执行（带超时）
+            async def run_script():
+                with redirect_stdout(stdout), redirect_stderr(stderr):
+                    exec(code, script_globals)
+            
+            await asyncio.wait_for(run_script(), timeout=SCRIPT_TIMEOUT)
+            
+            execution_time = int((datetime.now() - start_time).total_seconds() * 1000)
+            sdk.log(f"脚本执行完成，耗时 {execution_time}ms")
+            
+            return ScriptExecutionResult(
+                success=True,
+                output=stdout.getvalue(),
+                logs=sdk.get_logs(),
+                execution_time_ms=execution_time
+            )
+            
+        except asyncio.TimeoutError:
+            execution_time = int((datetime.now() - start_time).total_seconds() * 1000)
+            error_msg = f"脚本执行超时（超过 {SCRIPT_TIMEOUT} 秒）"
+            sdk.log(error_msg, level="ERROR")
+            
+            return ScriptExecutionResult(
+                success=False,
+                output=stdout.getvalue(),
+                error=error_msg,
+                logs=sdk.get_logs(),
+                execution_time_ms=execution_time
+            )
+            
+        except SyntaxError as e:
+            execution_time = int((datetime.now() - start_time).total_seconds() * 1000)
+            error_msg = f"语法错误: 第 {e.lineno} 行 - {e.msg}"
+            sdk.log(error_msg, level="ERROR")
+            
+            return ScriptExecutionResult(
+                success=False,
+                output=stdout.getvalue(),
+                error=error_msg,
+                logs=sdk.get_logs(),
+                execution_time_ms=execution_time
+            )
+            
+        except Exception as e:
+            execution_time = int((datetime.now() - start_time).total_seconds() * 1000)
+            error_msg = f"{type(e).__name__}: {str(e)}"
+            
+            # 获取详细的错误堆栈
+            tb = traceback.format_exc()
+            sdk.log(f"执行错误: {error_msg}\n{tb}", level="ERROR")
+            
+            return ScriptExecutionResult(
+                success=False,
+                output=stdout.getvalue(),
+                error=error_msg,
+                logs=sdk.get_logs(),
+                execution_time_ms=execution_time
+            )
+    
+    except Exception as e:
+        execution_time = int((datetime.now() - start_time).total_seconds() * 1000)
+        error_msg = f"执行器错误: {str(e)}"
+        logger.error(f"Script executor error: {e}", exc_info=True)
+        
+        return ScriptExecutionResult(
+            success=False,
+            error=error_msg,
+            logs=sdk.get_logs() if sdk else [],
+            execution_time_ms=execution_time
+        )
+    
+    finally:
+        # 清理资源
+        if sdk:
+            sdk.cleanup()
+
+
+async def test_script(
+    tenant_id: str,
+    script_content: str
+) -> ScriptExecutionResult:
+    """
+    测试执行脚本（不记录日志到数据库）
+    
+    Args:
+        tenant_id: 租户 ID
+        script_content: 脚本内容
+        
+    Returns:
+        ScriptExecutionResult: 执行结果
+    """
+    return await execute_script(
+        task_id=0,  # 测试用 ID
+        tenant_id=tenant_id,
+        script_content=script_content,
+        trace_id=f"test_{datetime.now().strftime('%Y%m%d%H%M%S')}"
+    )
diff --git a/backend/app/services/script_sdk.py b/backend/app/services/script_sdk.py
new file mode 100644
index 0000000..7e63886
--- /dev/null
+++ b/backend/app/services/script_sdk.py
@@ -0,0 +1,443 @@
+"""脚本执行 SDK - 提供给 Python 脚本使用的内置能力"""
+import json
+import logging
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+import httpx
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from ..database import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+
+class ScriptSDK:
+    """
+    脚本执行 SDK
+    
+    提供以下能力：
+    - AI 大模型调用
+    - 钉钉/企微通知
+    - 数据库查询（只读）
+    - HTTP 请求
+    - 变量存储（跨执行持久化）
+    - 日志记录
+    """
+    
+    def __init__(self, tenant_id: str, task_id: int, trace_id: str = None):
+        self.tenant_id = tenant_id
+        self.task_id = task_id
+        self.trace_id = trace_id or f"script_{task_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
+        self._logs: List[str] = []
+        self._db: Optional[Session] = None
+    
+    def _get_db(self) -> Session:
+        """获取数据库会话"""
+        if self._db is None:
+            self._db = SessionLocal()
+        return self._db
+    
+    def _close_db(self):
+        """关闭数据库会话"""
+        if self._db:
+            self._db.close()
+            self._db = None
+    
+    # ============ AI 服务 ============
+    
+    async def ai_chat(
+        self, 
+        prompt: str, 
+        system: str = None, 
+        model: str = "gemini-2.5-flash",
+        temperature: float = 0.7,
+        max_tokens: int = 2000
+    ) -> str:
+        """
+        调用大模型
+        
+        Args:
+            prompt: 用户提示词
+            system: 系统提示词（可选）
+            model: 模型名称，默认 gemini-2.5-flash
+            temperature: 温度，默认 0.7
+            max_tokens: 最大 token 数，默认 2000
+            
+        Returns:
+            AI 生成的文本
+        """
+        # 使用 4sapi 作为 AI 服务
+        api_key = "sk-9yMCXjRGANbacz20kJY8doSNy6Rf446aYwmgGIuIXQ7DAyBw"
+        base_url = "https://4sapi.com/v1"
+        
+        messages = []
+        if system:
+            messages.append({"role": "system", "content": system})
+        messages.append({"role": "user", "content": prompt})
+        
+        try:
+            async with httpx.AsyncClient(timeout=120.0) as client:
+                response = await client.post(
+                    f"{base_url}/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {api_key}",
+                        "Content-Type": "application/json"
+                    },
+                    json={
+                        "model": model,
+                        "messages": messages,
+                        "temperature": temperature,
+                        "max_tokens": max_tokens
+                    }
+                )
+                response.raise_for_status()
+                data = response.json()
+                content = data["choices"][0]["message"]["content"]
+                self.log(f"AI 调用成功，模型: {model}，响应长度: {len(content)}")
+                return content
+        except Exception as e:
+            self.log(f"AI 调用失败: {str(e)}", level="ERROR")
+            raise
+    
+    # ============ 通知服务 ============
+    
+    async def send_dingtalk(
+        self, 
+        webhook: str, 
+        content: str, 
+        msg_type: str = "text",
+        at_mobiles: List[str] = None,
+        at_all: bool = False
+    ) -> bool:
+        """
+        发送钉钉群消息
+        
+        Args:
+            webhook: 钉钉机器人 Webhook URL
+            content: 消息内容
+            msg_type: 消息类型，text 或 markdown
+            at_mobiles: @的手机号列表
+            at_all: 是否@所有人
+            
+        Returns:
+            是否发送成功
+        """
+        try:
+            if msg_type == "text":
+                data = {
+                    "msgtype": "text",
+                    "text": {"content": content},
+                    "at": {
+                        "atMobiles": at_mobiles or [],
+                        "isAtAll": at_all
+                    }
+                }
+            else:
+                data = {
+                    "msgtype": "markdown",
+                    "markdown": {
+                        "title": content[:20] if len(content) > 20 else content,
+                        "text": content
+                    },
+                    "at": {
+                        "atMobiles": at_mobiles or [],
+                        "isAtAll": at_all
+                    }
+                }
+            
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.post(webhook, json=data)
+                result = response.json()
+                
+                if result.get("errcode") == 0:
+                    self.log(f"钉钉消息发送成功")
+                    return True
+                else:
+                    self.log(f"钉钉消息发送失败: {result.get('errmsg')}", level="ERROR")
+                    return False
+        except Exception as e:
+            self.log(f"钉钉消息发送异常: {str(e)}", level="ERROR")
+            return False
+    
+    async def send_wecom(
+        self, 
+        webhook: str, 
+        content: str, 
+        msg_type: str = "text"
+    ) -> bool:
+        """
+        发送企业微信群消息
+        
+        Args:
+            webhook: 企微机器人 Webhook URL
+            content: 消息内容
+            msg_type: 消息类型，text 或 markdown
+            
+        Returns:
+            是否发送成功
+        """
+        try:
+            if msg_type == "text":
+                data = {
+                    "msgtype": "text",
+                    "text": {"content": content}
+                }
+            else:
+                data = {
+                    "msgtype": "markdown",
+                    "markdown": {"content": content}
+                }
+            
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.post(webhook, json=data)
+                result = response.json()
+                
+                if result.get("errcode") == 0:
+                    self.log(f"企微消息发送成功")
+                    return True
+                else:
+                    self.log(f"企微消息发送失败: {result.get('errmsg')}", level="ERROR")
+                    return False
+        except Exception as e:
+            self.log(f"企微消息发送异常: {str(e)}", level="ERROR")
+            return False
+    
+    # ============ 数据库查询 ============
+    
+    def db_query(self, sql: str, params: Dict[str, Any] = None) -> List[Dict]:
+        """
+        执行 SQL 查询（只读）
+        
+        Args:
+            sql: SQL 语句（仅支持 SELECT）
+            params: 查询参数
+            
+        Returns:
+            查询结果列表
+        """
+        # 安全检查：只允许 SELECT
+        sql_upper = sql.strip().upper()
+        if not sql_upper.startswith("SELECT"):
+            raise ValueError("只允许 SELECT 查询")
+        
+        forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER", "TRUNCATE"]
+        for word in forbidden:
+            if word in sql_upper:
+                raise ValueError(f"禁止使用 {word} 语句")
+        
+        db = self._get_db()
+        try:
+            result = db.execute(text(sql), params or {})
+            rows = [dict(row) for row in result.mappings().all()]
+            self.log(f"SQL 查询成功，返回 {len(rows)} 条记录")
+            return rows
+        except Exception as e:
+            self.log(f"SQL 查询失败: {str(e)}", level="ERROR")
+            raise
+    
+    # ============ HTTP 请求 ============
+    
+    async def http_get(
+        self, 
+        url: str, 
+        headers: Dict[str, str] = None,
+        params: Dict[str, Any] = None,
+        timeout: int = 30
+    ) -> Dict:
+        """
+        发送 HTTP GET 请求
+        
+        Args:
+            url: 请求 URL
+            headers: 请求头
+            params: 查询参数
+            timeout: 超时时间（秒）
+            
+        Returns:
+            响应数据（JSON 解析后）
+        """
+        try:
+            async with httpx.AsyncClient(timeout=float(timeout)) as client:
+                response = await client.get(url, headers=headers, params=params)
+                self.log(f"HTTP GET {url} -> {response.status_code}")
+                
+                try:
+                    return response.json()
+                except:
+                    return {"text": response.text, "status_code": response.status_code}
+        except Exception as e:
+            self.log(f"HTTP GET 失败: {str(e)}", level="ERROR")
+            raise
+    
+    async def http_post(
+        self, 
+        url: str, 
+        data: Dict[str, Any] = None,
+        json_data: Dict[str, Any] = None,
+        headers: Dict[str, str] = None,
+        timeout: int = 30
+    ) -> Dict:
+        """
+        发送 HTTP POST 请求
+        
+        Args:
+            url: 请求 URL
+            data: 表单数据
+            json_data: JSON 数据
+            headers: 请求头
+            timeout: 超时时间（秒）
+            
+        Returns:
+            响应数据（JSON 解析后）
+        """
+        try:
+            async with httpx.AsyncClient(timeout=float(timeout)) as client:
+                response = await client.post(
+                    url, 
+                    data=data, 
+                    json=json_data, 
+                    headers=headers
+                )
+                self.log(f"HTTP POST {url} -> {response.status_code}")
+                
+                try:
+                    return response.json()
+                except:
+                    return {"text": response.text, "status_code": response.status_code}
+        except Exception as e:
+            self.log(f"HTTP POST 失败: {str(e)}", level="ERROR")
+            raise
+    
+    # ============ 变量存储 ============
+    
+    def get_var(self, key: str, default: Any = None) -> Any:
+        """
+        获取存储的变量
+        
+        Args:
+            key: 变量名
+            default: 默认值
+            
+        Returns:
+            变量值（JSON 解析后）
+        """
+        db = self._get_db()
+        try:
+            result = db.execute(
+                text("SELECT var_value FROM platform_script_vars WHERE tenant_id = :tid AND var_key = :key"),
+                {"tid": self.tenant_id, "key": key}
+            )
+            row = result.first()
+            if row:
+                try:
+                    return json.loads(row[0])
+                except:
+                    return row[0]
+            return default
+        except Exception as e:
+            self.log(f"获取变量失败: {str(e)}", level="ERROR")
+            return default
+    
+    def set_var(self, key: str, value: Any) -> bool:
+        """
+        存储变量（跨执行持久化）
+        
+        Args:
+            key: 变量名
+            value: 变量值（会 JSON 序列化）
+            
+        Returns:
+            是否成功
+        """
+        db = self._get_db()
+        try:
+            value_str = json.dumps(value, ensure_ascii=False) if not isinstance(value, str) else value
+            
+            # 使用 REPLACE INTO 实现 upsert
+            db.execute(
+                text("""
+                    REPLACE INTO platform_script_vars (tenant_id, var_key, var_value, updated_at)
+                    VALUES (:tid, :key, :value, NOW())
+                """),
+                {"tid": self.tenant_id, "key": key, "value": value_str}
+            )
+            db.commit()
+            self.log(f"变量已存储: {key}")
+            return True
+        except Exception as e:
+            self.log(f"存储变量失败: {str(e)}", level="ERROR")
+            return False
+    
+    def delete_var(self, key: str) -> bool:
+        """
+        删除变量
+        
+        Args:
+            key: 变量名
+            
+        Returns:
+            是否成功
+        """
+        db = self._get_db()
+        try:
+            db.execute(
+                text("DELETE FROM platform_script_vars WHERE tenant_id = :tid AND var_key = :key"),
+                {"tid": self.tenant_id, "key": key}
+            )
+            db.commit()
+            self.log(f"变量已删除: {key}")
+            return True
+        except Exception as e:
+            self.log(f"删除变量失败: {str(e)}", level="ERROR")
+            return False
+    
+    # ============ 日志 ============
+    
+    def log(self, message: str, level: str = "INFO"):
+        """
+        记录日志
+        
+        Args:
+            message: 日志内容
+            level: 日志级别（INFO, WARN, ERROR）
+        """
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        log_entry = f"[{timestamp}] [{level}] {message}"
+        self._logs.append(log_entry)
+        
+        # 同时输出到标准日志
+        if level == "ERROR":
+            logger.error(f"[Script {self.task_id}] {message}")
+        else:
+            logger.info(f"[Script {self.task_id}] {message}")
+        
+        # 写入 platform_logs
+        try:
+            db = self._get_db()
+            db.execute(
+                text("""
+                    INSERT INTO platform_logs 
+                    (trace_id, app_code, module, level, message, created_at)
+                    VALUES (:trace_id, :app_code, :module, :level, :message, NOW())
+                """),
+                {
+                    "trace_id": self.trace_id,
+                    "app_code": "000-platform",
+                    "module": "script",
+                    "level": level,
+                    "message": message[:2000]  # 限制长度
+                }
+            )
+            db.commit()
+        except Exception as e:
+            logger.warning(f"Failed to write log to database: {e}")
+    
+    def get_logs(self) -> List[str]:
+        """获取所有日志"""
+        return self._logs.copy()
+    
+    def cleanup(self):
+        """清理资源"""
+        self._close_db()
diff --git a/frontend/package.json b/frontend/package.json
index d3cd9de..87e012f 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -16,7 +16,9 @@
     "element-plus": "^2.5.0",
     "@element-plus/icons-vue": "^2.3.0",
     "echarts": "^5.4.0",
-    "dayjs": "^1.11.0"
+    "dayjs": "^1.11.0",
+    "codemirror": "^5.65.0",
+    "vue-codemirror": "^6.1.1"
   },
   "devDependencies": {
     "@vitejs/plugin-vue": "^5.0.0",
diff --git a/frontend/src/views/scheduled-tasks/index.vue b/frontend/src/views/scheduled-tasks/index.vue
index ac286cc..d1dfc0a 100644
--- a/frontend/src/views/scheduled-tasks/index.vue
+++ b/frontend/src/views/scheduled-tasks/index.vue
@@ -32,18 +32,27 @@ const form = reactive({
   schedule_type: 'simple',
   time_points: [],
   cron_expression: '',
+  execution_type: 'webhook',  // webhook | script
   webhook_url: '',
   input_params: '',
+  script_content: '',
   is_enabled: true
 })
 
 // 时间选择器
 const newTimePoint = ref('')
 
+// 脚本测试
+const testLoading = ref(false)
+const testResult = ref(null)
+
+// SDK 文档
+const sdkDocsVisible = ref(false)
+const sdkDocs = ref(null)
+
 const rules = {
   tenant_id: [{ required: true, message: '请选择租户', trigger: 'change' }],
-  task_name: [{ required: true, message: '请输入任务名称', trigger: 'blur' }],
-  webhook_url: [{ required: true, message: '请输入 Webhook URL', trigger: 'blur' }]
+  task_name: [{ required: true, message: '请输入任务名称', trigger: 'blur' }]
 }
 
 // 日志对话框
@@ -136,11 +145,14 @@ function handleCreate() {
     schedule_type: 'simple',
     time_points: [],
     cron_expression: '',
+    execution_type: 'webhook',
     webhook_url: '',
     input_params: '',
+    script_content: '',
     is_enabled: true
   })
   newTimePoint.value = ''
+  testResult.value = null
   dialogVisible.value = true
 }
 
@@ -154,11 +166,14 @@ function handleEdit(row) {
     schedule_type: row.schedule_type || 'simple',
     time_points: row.time_points || [],
     cron_expression: row.cron_expression || '',
-    webhook_url: row.webhook_url,
+    execution_type: row.execution_type || 'webhook',
+    webhook_url: row.webhook_url || '',
     input_params: row.input_params ? JSON.stringify(row.input_params, null, 2) : '',
+    script_content: row.script_content || '',
     is_enabled: row.is_enabled
   })
   newTimePoint.value = ''
+  testResult.value = null
   dialogVisible.value = true
 }
 
@@ -190,6 +205,16 @@ async function handleSubmit() {
     return
   }
   
+  // 验证执行配置
+  if (form.execution_type === 'webhook' && !form.webhook_url) {
+    ElMessage.error('请输入 Webhook URL')
+    return
+  }
+  if (form.execution_type === 'script' && !form.script_content) {
+    ElMessage.error('请输入脚本内容')
+    return
+  }
+  
   // 解析输入参数
   let inputParams = null
   if (form.input_params) {
@@ -208,7 +233,9 @@ async function handleSubmit() {
     schedule_type: form.schedule_type,
     time_points: form.schedule_type === 'simple' ? form.time_points : null,
     cron_expression: form.schedule_type === 'cron' ? form.cron_expression : null,
-    webhook_url: form.webhook_url,
+    execution_type: form.execution_type,
+    webhook_url: form.execution_type === 'webhook' ? form.webhook_url : null,
+    script_content: form.execution_type === 'script' ? form.script_content : null,
     input_params: inputParams,
     is_enabled: form.is_enabled
   }
@@ -294,6 +321,54 @@ function handleLogsPageChange(page) {
   fetchLogs()
 }
 
+// 测试脚本执行
+async function handleTestScript() {
+  if (!form.script_content) {
+    ElMessage.warning('请先输入脚本内容')
+    return
+  }
+  if (!form.tenant_id) {
+    ElMessage.warning('请先选择租户')
+    return
+  }
+  
+  testLoading.value = true
+  testResult.value = null
+  
+  try {
+    const res = await api.post('/api/scheduled-tasks/test-script', {
+      tenant_id: form.tenant_id,
+      script_content: form.script_content
+    })
+    testResult.value = res.data
+    if (res.data.success) {
+      ElMessage.success('脚本执行成功')
+    } else {
+      ElMessage.error('脚本执行失败')
+    }
+  } catch (e) {
+    testResult.value = { success: false, error: e.message || '请求失败' }
+  } finally {
+    testLoading.value = false
+  }
+}
+
+// 查看 SDK 文档
+async function handleShowSdkDocs() {
+  try {
+    const res = await api.get('/api/scheduled-tasks/sdk-docs')
+    sdkDocs.value = res.data
+    sdkDocsVisible.value = true
+  } catch (e) {
+    ElMessage.error('获取文档失败')
+  }
+}
+
+// 获取执行类型描述
+function getExecutionTypeDesc(row) {
+  return row.execution_type === 'script' ? '脚本' : 'Webhook'
+}
+
 // 快速选择租户
 function selectTenant(code) {
   if (query.tenant_id === code) {
@@ -369,6 +444,13 @@ onMounted(() => {
           </span>
         </template>
       </el-table-column>
+      <el-table-column label="执行方式" width="90">
+        <template #default="{ row }">
+          <el-tag :type="row.execution_type === 'script' ? 'success' : 'primary'" size="small">
+            {{ row.execution_type === 'script' ? '脚本' : 'Webhook' }}
+          </el-tag>
+        </template>
+      </el-table-column>
       <el-table-column label="状态" width="80">
         <template #default="{ row }">
           <el-tag :type="row.is_enabled ? 'success' : 'info'" size="small">
@@ -502,18 +584,86 @@ onMounted(() => {
           </div>
         </el-form-item>
         
-        <el-form-item label="Webhook URL" prop="webhook_url">
-          <el-input v-model="form.webhook_url" placeholder="如：https://n8n.ireborn.com.cn/webhook/xxx" />
+        <el-divider content-position="left">执行配置</el-divider>
+        
+        <el-form-item label="执行方式">
+          <el-radio-group v-model="form.execution_type">
+            <el-radio value="webhook">Webhook（调用 n8n 等）</el-radio>
+            <el-radio value="script">Python 脚本</el-radio>
+          </el-radio-group>
         </el-form-item>
         
-        <el-form-item label="输入参数">
-          <el-input 
-            v-model="form.input_params" 
-            type="textarea" 
-            :rows="4" 
-            placeholder='可选，JSON 格式，如：{"key": "value"}'
-          />
-        </el-form-item>
+        <!-- Webhook 模式 -->
+        <template v-if="form.execution_type === 'webhook'">
+          <el-form-item label="Webhook URL">
+            <el-input v-model="form.webhook_url" placeholder="如：https://n8n.ireborn.com.cn/webhook/xxx" />
+          </el-form-item>
+          
+          <el-form-item label="输入参数">
+            <el-input 
+              v-model="form.input_params" 
+              type="textarea" 
+              :rows="4" 
+              placeholder='可选，JSON 格式，如：{"key": "value"}'
+            />
+          </el-form-item>
+        </template>
+        
+        <!-- 脚本模式 -->
+        <template v-if="form.execution_type === 'script'">
+          <el-form-item label="脚本内容">
+            <div class="script-editor-header">
+              <el-button type="primary" link size="small" @click="handleShowSdkDocs">
+                查看 SDK 文档
+              </el-button>
+              <el-button 
+                type="success" 
+                size="small" 
+                :loading="testLoading"
+                @click="handleTestScript"
+              >
+                测试执行
+              </el-button>
+            </div>
+            <el-input 
+              v-model="form.script_content" 
+              type="textarea" 
+              :rows="15" 
+              placeholder="# Python 脚本
+# 可用方法：ai(), dingtalk(), wecom(), db(), http_get(), http_post()
+# get_var(), set_var(), log()
+
+content = ai('生成一段营销文案')
+dingtalk('你的webhook', content)
+log('执行完成')"
+              class="script-textarea"
+            />
+          </el-form-item>
+          
+          <!-- 测试结果 -->
+          <el-form-item v-if="testResult" label="测试结果">
+            <el-alert 
+              :type="testResult.success ? 'success' : 'error'" 
+              :closable="false"
+            >
+              <template #title>
+                {{ testResult.success ? '执行成功' : '执行失败' }}
+                <span v-if="testResult.execution_time_ms" style="margin-left: 8px; color: #909399">
+                  ({{ testResult.execution_time_ms }}ms)
+                </span>
+              </template>
+              <div v-if="testResult.error" class="test-error">{{ testResult.error }}</div>
+              <div v-if="testResult.output" class="test-output">
+                <strong>输出：</strong>
+                <pre>{{ testResult.output }}</pre>
+              </div>
+              <div v-if="testResult.logs && testResult.logs.length" class="test-logs">
+                <strong>日志：</strong>
+                <pre>{{ testResult.logs.join('\n') }}</pre>
+              </div>
+            </el-alert>
+          </el-form-item>
+        </template>
         
         <el-form-item label="启用状态">
           <el-switch v-model="form.is_enabled" />
@@ -526,6 +676,26 @@ onMounted(() => {
       </template>
     </el-dialog>
     
+    <!-- SDK 文档对话框 -->
+    <el-dialog v-model="sdkDocsVisible" title="SDK 文档" width="800px">
+      <div v-if="sdkDocs" class="sdk-docs">
+        <p>{{ sdkDocs.description }}</p>
+        
+        <h4>可用方法</h4>
+        <div v-for="method in sdkDocs.methods" :key="method.name" class="sdk-method">
+          <code>{{ method.name }}</code>
+          <p>{{ method.description }}</p>
+          <pre class="sdk-example">{{ method.example }}</pre>
+        </div>
+        
+        <h4>示例脚本</h4>
+        <pre class="sdk-example-script">{{ sdkDocs.example_script }}</pre>
+      </div>
+      <template #footer>
+        <el-button @click="sdkDocsVisible = false">关闭</el-button>
+      </template>
+    </el-dialog>
+    
     <!-- 日志对话框 -->
     <el-dialog 
       v-model="logsDialogVisible" 
@@ -642,4 +812,89 @@ onMounted(() => {
   font-size: 12px;
   margin-top: 4px;
 }
+
+/* 脚本编辑器 */
+.script-editor-header {
+  display: flex;
+  justify-content: space-between;
+  margin-bottom: 8px;
+}
+
+.script-textarea :deep(.el-textarea__inner) {
+  font-family: 'Monaco', 'Menlo', 'Consolas', monospace;
+  font-size: 13px;
+  line-height: 1.5;
+}
+
+/* 测试结果 */
+.test-error {
+  color: #f56c6c;
+  margin-top: 8px;
+}
+
+.test-output,
+.test-logs {
+  margin-top: 8px;
+}
+
+.test-output pre,
+.test-logs pre {
+  background: #f5f7fa;
+  padding: 8px;
+  border-radius: 4px;
+  font-size: 12px;
+  max-height: 150px;
+  overflow-y: auto;
+  margin-top: 4px;
+}
+
+/* SDK 文档 */
+.sdk-docs {
+  max-height: 60vh;
+  overflow-y: auto;
+}
+
+.sdk-docs h4 {
+  margin: 16px 0 8px;
+  color: #303133;
+}
+
+.sdk-method {
+  margin-bottom: 16px;
+  padding: 12px;
+  background: #f5f7fa;
+  border-radius: 8px;
+}
+
+.sdk-method code {
+  display: block;
+  font-size: 14px;
+  font-weight: 600;
+  color: #409eff;
+  margin-bottom: 8px;
+}
+
+.sdk-method p {
+  margin: 0 0 8px;
+  color: #606266;
+}
+
+.sdk-example {
+  background: #fff;
+  padding: 8px;
+  border-radius: 4px;
+  font-size: 12px;
+  margin: 0;
+  color: #303133;
+}
+
+.sdk-example-script {
+  background: #1e1e1e;
+  color: #d4d4d4;
+  padding: 16px;
+  border-radius: 8px;
+  font-size: 13px;
+  line-height: 1.5;
+  overflow-x: auto;
+}
 </style>