diff --git a/apps/api/app/api/routes/templates_editor/cover.py b/apps/api/app/api/routes/templates_editor/cover.py
index 3605845b2..e17645520 100755
--- a/apps/api/app/api/routes/templates_editor/cover.py
+++ b/apps/api/app/api/routes/templates_editor/cover.py
@@ -184,6 +184,29 @@ def editor_generate_cover(
     _, plan_svc = services
     plan = plan_svc.get_plan_or_raise(plan_id)
 
+    # 获取第一个视频的下载 URL（用于 MediaKit 抽帧）
+    primary_video_url = None
+    if body.asset_ids and body.cover_type in ("ai_frame", "ai_regenerate"):
+        try:
+            from app.database import get_db_session
+
+            from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+            from packages.shared.storage import get_shared_storage_service
+
+            with get_db_session() as session:
+                asset_repo = SQLAlchemyAssetRepository(session)
+                first_asset = asset_repo.get(body.asset_ids[0])
+                if first_asset and first_asset.storage_key:
+                    storage_svc = get_shared_storage_service()
+                    primary_video_url = storage_svc.get_download_url(first_asset.storage_key)
+                    logger.info(
+                        "获取视频URL用于封面生成: asset_id=%s url=%s",
+                        body.asset_ids[0],
+                        primary_video_url[:80] if primary_video_url else None,
+                    )
+        except Exception as e:
+            logger.warning("获取视频URL失败，将使用stub封面: %s", str(e))
+
     from packages.shared.ai_service import run_generate_cover
 
     cover_data = run_generate_cover(
@@ -191,6 +214,7 @@ def editor_generate_cover(
         asset_ids=body.asset_ids,
         cover_type=body.cover_type,
         frame_time=body.frame_time,
+        primary_video_url=primary_video_url,
     )
 
     current_config = dict(plan.config) if plan.config else {}
diff --git a/packages/config/base.py b/packages/config/base.py
index d0ef1d5ac..37ae13778 100755
--- a/packages/config/base.py
+++ b/packages/config/base.py
@@ -69,6 +69,11 @@ class SharedSettings(BaseSettings):
     doubao_timeout: int = 30
     doubao_max_retries: int = 2
 
+    # ── MediaKit (火山引擎 AI 媒体工具) ──────────────────────────────────
+    mediakit_api_key: str = ""
+    mediakit_base_url: str = "https://mediakit.cn-beijing.volces.com/api/v1"
+    mediakit_timeout: int = 60
+
     @property
     def effective_database_url(self) -> str:
         """返回实际使用的数据库 URL。
diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py
index 892f47c9f..bba27de7e 100755
--- a/packages/shared/ai_service.py
+++ b/packages/shared/ai_service.py
@@ -267,15 +267,20 @@ def _call_ai_cover_service(
     asset_ids: List[str],
     cover_type: str,
     frame_time: float | None = None,
+    primary_video_url: str | None = None,
 ) -> Dict[str, Any]:
-    """调用 AI 封面生成服务（stub）
+    """调用 AI 封面生成服务.
 
-    TODO: 接入真实 AI 服务，从视频中选帧或生成封面。
-    当前返回模拟封面数据。
+    当 cover_type 为 ai_frame 或 ai_regenerate 时，调用 MediaKit 视频截帧。
+    失败或未配置时降级为 stub 行为。
+
+    Args:
+        plan_id: 剪辑计划 ID
+        asset_ids: 素材 ID 列表
+        cover_type: 封面类型
+        frame_time: 手动选帧时间点
+        primary_video_url: 主视频的可访问 URL（用于 MediaKit 抽帧）
     """
-    # 模拟 AI 处理耗时
-    time.sleep(0.3)
-
     if cover_type == "upload":
         return {
             "type": "upload",
@@ -290,7 +295,48 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # ai_frame / ai_regenerate
+    # ai_frame / ai_regenerate - 尝试调用 MediaKit
+    if primary_video_url:
+        from packages.shared.mediakit_client import get_mediakit_client
+
+        client = get_mediakit_client()
+        if client.is_available:
+            try:
+                logger.info("调用 MediaKit 抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
+                frames = client.extract_frames(
+                    video_url=primary_video_url,
+                    strategy="SceneChange",
+                    max_frames=5,
+                )
+
+                if frames and len(frames) > 0:
+                    # 选择第一帧（SceneChange 策略的第一帧通常是最佳画面）
+                    best_frame = frames[0]
+                    image_url = best_frame.get("image_url", "")
+                    timestamp = best_frame.get("timestamp", 0.0)
+
+                    if image_url:
+                        logger.info(
+                            "MediaKit 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
+                            plan_id,
+                            timestamp,
+                            image_url[:80],
+                        )
+                        return {
+                            "type": "ai_frame",
+                            "image_url": image_url,
+                            "frame_time": round(timestamp, 1),
+                            "confidence": 0.85,
+                        }
+                    else:
+                        logger.warning("MediaKit 返回的帧无 image_url")
+
+            except Exception as e:
+                logger.exception("MediaKit 抽帧失败，降级到 stub: %s", str(e))
+
+    # 降级：stub 行为
+    logger.info("使用 stub 封面: plan_id=%s", plan_id)
+    time.sleep(0.3)
     return {
         "type": "ai_frame",
         "image_url": f"/api/v1/assets/placeholder/cover?plan={plan_id}",
@@ -351,6 +397,7 @@ def run_generate_cover(
     asset_ids: List[str],
     cover_type: str = "ai_frame",
     frame_time: float | None = None,
+    primary_video_url: str | None = None,
 ) -> Dict[str, Any]:
     """执行 AI 封面生成
 
@@ -359,21 +406,24 @@ def run_generate_cover(
         asset_ids: 素材 ID 列表（用于确定视频来源）
         cover_type: 封面类型 (ai_frame / manual / upload / ai_regenerate)
         frame_time: 手动选帧时间点（仅 manual 模式使用）
+        primary_video_url: 主视频的可访问 URL（用于 MediaKit 抽帧）
 
     Returns:
         封面数据 dict，包含 type / image_url / frame_time
     """
     logger.info(
-        "AI 封面生成: plan_id=%s type=%s assets=%d",
+        "AI 封面生成: plan_id=%s type=%s assets=%d has_video_url=%s",
         plan_id,
         cover_type,
         len(asset_ids),
+        bool(primary_video_url),
     )
     result = _call_ai_cover_service(
         plan_id=plan_id,
         asset_ids=asset_ids,
         cover_type=cover_type,
         frame_time=frame_time,
+        primary_video_url=primary_video_url,
     )
     logger.info(
         "AI 封面生成完成: plan_id=%s type=%s url=%s",
diff --git a/packages/shared/mediakit_client.py b/packages/shared/mediakit_client.py
new file mode 100755
index 000000000..4848a70f8
--- /dev/null
+++ b/packages/shared/mediakit_client.py
@@ -0,0 +1,181 @@
+"""MediaKit API 客户端（共享层）.
+
+火山引擎 AI MediaKit 提供视频抽帧、视频理解等能力。
+API 和 Worker 共用此客户端。
+
+使用方式：
+    from packages.shared.mediakit_client import get_mediakit_client
+
+    client = get_mediakit_client()
+    if client.is_available:
+        frames = client.extract_frames(video_url="https://...")
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from packages.shared.config import get_shared_settings
+
+logger = logging.getLogger(__name__)
+
+
+class MediaKitClient:
+    """MediaKit API 客户端.
+
+    封装视频抽帧、视频理解等接口，支持异步任务轮询。
+    未配置 API Key 时 is_available 为 False，调用方应降级处理。
+    """
+
+    def __init__(self) -> None:
+        settings = get_shared_settings()
+        self.api_key: str = settings.mediakit_api_key
+        self.base_url: str = settings.mediakit_base_url.rstrip("/")
+        self.timeout: int = settings.mediakit_timeout
+
+    @property
+    def is_available(self) -> bool:
+        """是否可用（配置了 API Key）."""
+        return bool(self.api_key)
+
+    def extract_frames(
+        self,
+        video_url: str,
+        strategy: str = "SceneChange",
+        max_frames: int = 10,
+        poll_interval: float = 2.0,
+        max_poll_attempts: int = 30,
+    ) -> Optional[List[Dict[str, Any]]]:
+        """调用 MediaKit 视频抽帧接口.
+
+        Args:
+            video_url: 视频 URL（需可公开访问）
+            strategy: 抽帧策略
+                - TimeInterval: 按固定时间间隔
+                - SpecifiedTime: 按指定时间点
+                - SpecifiedFrames: 首尾帧 + 指定帧数
+                - SceneChange: 场景变化检测（推荐用于封面选取）
+            max_frames: 最大返回帧数
+            poll_interval: 轮询间隔（秒）
+            max_poll_attempts: 最大轮询次数
+
+        Returns:
+            帧列表 [{"image_url": "...", "timestamp": 1.5}, ...]
+            失败返回 None
+        """
+        if not self.is_available:
+            logger.warning("MediaKit 未配置，跳过抽帧")
+            return None
+
+        # 提交抽帧任务
+        task_id = self._submit_extract_task(video_url, strategy, max_frames)
+        if not task_id:
+            return None
+
+        # 轮询任务状态
+        return self._poll_task_result(task_id, poll_interval, max_poll_attempts)
+
+    def _submit_extract_task(
+        self,
+        video_url: str,
+        strategy: str,
+        max_frames: int,
+    ) -> Optional[str]:
+        """提交抽帧任务，返回 task_id."""
+        url = f"{self.base_url}/tools/extract-frames"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+
+        payload = {
+            "video_url": video_url,
+            "strategy": strategy,
+            "max_frames": max_frames,
+        }
+
+        try:
+            response = httpx.post(url, headers=headers, json=payload, timeout=self.timeout)
+            response.raise_for_status()
+            data = response.json()
+
+            task_id = data.get("task_id")
+            if not task_id:
+                logger.error("MediaKit 抽帧任务提交失败: 无 task_id. response=%s", data)
+                return None
+
+            logger.info("MediaKit 抽帧任务已提交: task_id=%s strategy=%s", task_id, strategy)
+            return task_id
+
+        except Exception as e:
+            logger.exception("MediaKit 抽帧任务提交异常: %s", str(e))
+            return None
+
+    def _poll_task_result(
+        self,
+        task_id: str,
+        poll_interval: float,
+        max_poll_attempts: int,
+    ) -> Optional[List[Dict[str, Any]]]:
+        """轮询任务状态，返回结果."""
+        url = f"{self.base_url}/tasks/{task_id}"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+        }
+
+        for attempt in range(max_poll_attempts):
+            try:
+                response = httpx.get(url, headers=headers, timeout=self.timeout)
+                response.raise_for_status()
+                data = response.json()
+
+                status = data.get("status")
+                if status == "success":
+                    result = data.get("result", {})
+                    snapshots = result.get("snapshots", [])
+                    logger.info(
+                        "MediaKit 抽帧完成: task_id=%s frames=%d",
+                        task_id,
+                        len(snapshots),
+                    )
+                    return snapshots
+
+                elif status == "failed":
+                    error_msg = data.get("error", "unknown error")
+                    logger.error("MediaKit 抽帧任务失败: task_id=%s error=%s", task_id, error_msg)
+                    return None
+
+                # status == "processing" or "pending"
+                logger.debug(
+                    "MediaKit 抽帧进行中: task_id=%s status=%s attempt=%d/%d",
+                    task_id,
+                    status,
+                    attempt + 1,
+                    max_poll_attempts,
+                )
+                time.sleep(poll_interval)
+
+            except Exception as e:
+                logger.exception("MediaKit 轮询异常: task_id=%s error=%s", task_id, str(e))
+                time.sleep(poll_interval)
+
+        logger.error("MediaKit 抽帧超时: task_id=%s max_attempts=%d", task_id, max_poll_attempts)
+        return None
+
+
+# ── 单例管理 ────────────────────────────────────────────────────────────────
+
+
+_client_instance: Optional[MediaKitClient] = None
+
+
+def get_mediakit_client() -> MediaKitClient:
+    """获取 MediaKit 客户端单例."""
+    global _client_instance
+    if _client_instance is None:
+        _client_instance = MediaKitClient()
+    return _client_instance
diff --git a/tests/unit/test_mediakit_cover.py b/tests/unit/test_mediakit_cover.py
new file mode 100755
index 000000000..49864e136
--- /dev/null
+++ b/tests/unit/test_mediakit_cover.py
@@ -0,0 +1,320 @@
+"""MediaKit AI 封面生成单元测试.
+
+测试 #1208: AI封面接入MediaKit视频截帧
+"""
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+from packages.shared.mediakit_client import MediaKitClient, get_mediakit_client
+
+
+class TestMediaKitClient:
+    """MediaKit 客户端测试."""
+
+    def test_is_available_with_key(self):
+        """配置了 API Key 时可用."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = "test-key"
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            client = MediaKitClient()
+            assert client.is_available is True
+
+    def test_is_available_without_key(self):
+        """未配置 API Key 时不可用."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = ""
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            client = MediaKitClient()
+            assert client.is_available is False
+
+    @patch("packages.shared.mediakit_client.httpx.post")
+    @patch("packages.shared.mediakit_client.httpx.get")
+    def test_extract_frames_success(self, mock_get, mock_post):
+        """抽帧成功."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = "test-key"
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            # Mock submit task
+            mock_post.return_value.status_code = 200
+            mock_post.return_value.json.return_value = {"task_id": "task-123"}
+            mock_post.return_value.raise_for_status = Mock()
+
+            # Mock poll task
+            mock_get.return_value.status_code = 200
+            mock_get.return_value.json.return_value = {
+                "status": "success",
+                "result": {
+                    "snapshots": [
+                        {"image_url": "https://example.com/frame1.jpg", "timestamp": 2.5},
+                        {"image_url": "https://example.com/frame2.jpg", "timestamp": 5.0},
+                    ]
+                },
+            }
+            mock_get.return_value.raise_for_status = Mock()
+
+            client = MediaKitClient()
+            frames = client.extract_frames(
+                video_url="https://example.com/video.mp4",
+                strategy="SceneChange",
+                max_frames=5,
+            )
+
+            assert frames is not None
+            assert len(frames) == 2
+            assert frames[0]["image_url"] == "https://example.com/frame1.jpg"
+            assert frames[0]["timestamp"] == 2.5
+
+    @patch("packages.shared.mediakit_client.httpx.post")
+    def test_extract_frames_submit_failed(self, mock_post):
+        """提交任务失败."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = "test-key"
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            mock_post.side_effect = Exception("Network error")
+
+            client = MediaKitClient()
+            frames = client.extract_frames(
+                video_url="https://example.com/video.mp4",
+                strategy="SceneChange",
+                max_frames=5,
+            )
+
+            assert frames is None
+
+    @patch("packages.shared.mediakit_client.httpx.post")
+    @patch("packages.shared.mediakit_client.httpx.get")
+    def test_extract_frames_task_failed(self, mock_get, mock_post):
+        """任务执行失败."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = "test-key"
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            # Mock submit task
+            mock_post.return_value.status_code = 200
+            mock_post.return_value.json.return_value = {"task_id": "task-123"}
+            mock_post.return_value.raise_for_status = Mock()
+
+            # Mock poll task - failed
+            mock_get.return_value.status_code = 200
+            mock_get.return_value.json.return_value = {
+                "status": "failed",
+                "error": "Video processing failed",
+            }
+            mock_get.return_value.raise_for_status = Mock()
+
+            client = MediaKitClient()
+            frames = client.extract_frames(
+                video_url="https://example.com/video.mp4",
+                strategy="SceneChange",
+                max_frames=5,
+            )
+
+            assert frames is None
+
+    def test_extract_frames_not_available(self):
+        """客户端不可用时返回 None."""
+        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
+            mock_settings.return_value.mediakit_api_key = ""
+            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
+            mock_settings.return_value.mediakit_timeout = 60
+
+            client = MediaKitClient()
+            frames = client.extract_frames(
+                video_url="https://example.com/video.mp4",
+                strategy="SceneChange",
+                max_frames=5,
+            )
+
+            assert frames is None
+
+
+class TestAICoverService:
+    """AI 封面服务测试."""
+
+    @patch("packages.shared.mediakit_client.get_mediakit_client")
+    def test_call_ai_cover_with_mediakit_success(self, mock_get_client):
+        """MediaKit 抽帧成功."""
+        mock_client = Mock()
+        mock_client.is_available = True
+        mock_client.extract_frames.return_value = [{"image_url": "https://example.com/frame.jpg", "timestamp": 3.5}]
+        mock_get_client.return_value = mock_client
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "ai_frame"
+        assert result["image_url"] == "https://example.com/frame.jpg"
+        assert result["frame_time"] == 3.5
+        assert result["confidence"] == 0.85
+
+        mock_client.extract_frames.assert_called_once()
+
+    @patch("packages.shared.mediakit_client.get_mediakit_client")
+    def test_call_ai_cover_with_mediakit_failure_fallback(self, mock_get_client):
+        """MediaKit 失败时降级到 stub."""
+        mock_client = Mock()
+        mock_client.is_available = True
+        mock_client.extract_frames.side_effect = Exception("API error")
+        mock_get_client.return_value = mock_client
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        # 应该降级到 stub
+        assert result["type"] == "ai_frame"
+        assert "placeholder" in result["image_url"]
+
+    def test_call_ai_cover_without_video_url_fallback(self):
+        """没有视频 URL 时使用 stub."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url=None,
+        )
+
+        assert result["type"] == "ai_frame"
+        assert "placeholder" in result["image_url"]
+
+    def test_call_ai_cover_upload_type(self):
+        """upload 类型直接返回."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="upload",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "upload"
+        assert result["image_url"] == ""
+
+    def test_call_ai_cover_manual_type(self):
+        """manual 类型返回指定时间点."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="manual",
+            frame_time=5.0,
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "manual"
+        assert result["frame_time"] == 5.0
+
+    @patch("packages.shared.mediakit_client.get_mediakit_client")
+    def test_call_ai_cover_mediakit_not_available(self, mock_get_client):
+        """MediaKit 未配置时使用 stub."""
+        mock_client = Mock()
+        mock_client.is_available = False
+        mock_get_client.return_value = mock_client
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "ai_frame"
+        assert "placeholder" in result["image_url"]
+
+    @patch("packages.shared.mediakit_client.get_mediakit_client")
+    def test_call_ai_cover_empty_frames_fallback(self, mock_get_client):
+        """MediaKit 返回空帧列表时降级."""
+        mock_client = Mock()
+        mock_client.is_available = True
+        mock_client.extract_frames.return_value = []
+        mock_get_client.return_value = mock_client
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "ai_frame"
+        assert "placeholder" in result["image_url"]
+
+
+class TestGenerateCover:
+    """run_generate_cover 测试."""
+
+    @patch("packages.shared.ai_service._call_ai_cover_service")
+    def test_run_generate_cover_passes_video_url(self, mock_cover_service):
+        """正确传递 video_url 参数."""
+        mock_cover_service.return_value = {
+            "type": "ai_frame",
+            "image_url": "https://example.com/cover.jpg",
+            "frame_time": 2.0,
+            "confidence": 0.9,
+        }
+
+        from packages.shared.ai_service import run_generate_cover
+
+        result = run_generate_cover(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+            primary_video_url="https://example.com/video.mp4",
+        )
+
+        assert result["type"] == "ai_frame"
+        mock_cover_service.assert_called_once()
+        call_kwargs = mock_cover_service.call_args[1]
+        assert call_kwargs["primary_video_url"] == "https://example.com/video.mp4"
+
+    @patch("packages.shared.ai_service._call_ai_cover_service")
+    def test_run_generate_cover_without_video_url(self, mock_cover_service):
+        """不传 video_url 时默认为 None."""
+        mock_cover_service.return_value = {
+            "type": "ai_frame",
+            "image_url": "https://example.com/cover.jpg",
+            "frame_time": 2.0,
+            "confidence": 0.9,
+        }
+
+        from packages.shared.ai_service import run_generate_cover
+
+        result = run_generate_cover(
+            plan_id="plan-123",
+            asset_ids=["asset-1"],
+            cover_type="ai_frame",
+        )
+
+        assert result["type"] == "ai_frame"
+        call_kwargs = mock_cover_service.call_args[1]
+        assert call_kwargs["primary_video_url"] is None