feat(video): add video recording and transcription tools for mobile devices

llm-config.defaults.jsonc

@@ -59,10 +59,20 @@
           "model": "gpt-5-mini"
         }
       }
+      // video_analyzer is optional - only needed when using video recording tools
+      // Requires a video-capable model (Google). To enable, uncomment below:
+      // "video_analyzer": {
+      //   "provider": "google",
+      //   "model": "gemini-3-flash-preview",
+      //   "fallback": {
+      //     "provider": "google",
+      //     "model": "gemini-2.5-flash"
+      //   }
+      // }
     }
   },
   // This is the config we recommend.
-  // To use it, copy it to llm.override.jsonc, following llm.override.template.jsonc.
+  // To use it, copy it to llm-config.override.jsonc, following llm-config.override.template.jsonc.
   "recommended": {
     "planner": {
       "provider": "minitap",
@@ -82,10 +92,10 @@
     },
     "cortex": {
       "provider": "minitap",
-      "model": "google/gemini-2.5-pro",
+      "model": "google/gemini-3-pro-preview",
       "fallback": {
         "provider": "minitap",
-        "model": "openai/gpt-5"
+        "model": "google/gemini-2.5-pro"
       }
     },
     "executor": {
@@ -121,6 +131,18 @@
           "provider": "minitap",
           "model": "openai/gpt-5-mini"
         }
+      },
+      // video_analyzer is optional - only needed when using video recording tools
+      // Requires a video-capable model. Ex of supported models:
+      // gemini-3-pro-preview, gemini-3-flash-preview, gemini-2.5-flash,
+      // gemini-2.5-flash-lite, gemini-2.5-pro, gemini-2.0-flash, gemini-2.0-flash-lite
+      "video_analyzer": {
+        "provider": "minitap",
+        "model": "google/gemini-3-flash-preview",
+        "fallback": {
+          "provider": "minitap",
+          "model": "google/gemini-2.5-flash"
+        }
       }
     }
   }

llm-config.override.template.jsonc

@@ -38,6 +38,14 @@
     "outputter": {
       "provider": "",
       "model": ""
+    },
+    "video_analyzer": {
+      "provider": "",
+      "model": "",
+      "fallback": {
+        "provider": "",
+        "model": ""
+      }
     }
   }
 }

minitap/mobile_use/agents/cortex/cortex.py

@@ -19,7 +19,11 @@
 from minitap.mobile_use.graph.state import State
 from minitap.mobile_use.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
 from minitap.mobile_use.services.telemetry import telemetry
-from minitap.mobile_use.tools.index import EXECUTOR_WRAPPERS_TOOLS, format_tools_list
+from minitap.mobile_use.tools.index import (
+    EXECUTOR_WRAPPERS_TOOLS,
+    VIDEO_RECORDING_WRAPPERS,
+    format_tools_list,
+)
 from minitap.mobile_use.utils.conversations import get_screenshot_message_for_llm
 from minitap.mobile_use.utils.decorators import wrap_with_callbacks
 from minitap.mobile_use.utils.logger import get_logger
@@ -43,6 +47,10 @@ async def __call__(self, state: State):
             self.ctx.execution_setup.get_locked_app_package() if self.ctx.execution_setup else None
         )
 
+        executor_wrappers = list(EXECUTOR_WRAPPERS_TOOLS)
+        if self.ctx.video_recording_enabled:
+            executor_wrappers.extend(VIDEO_RECORDING_WRAPPERS)
+
         system_message = Template(
             Path(__file__).parent.joinpath("cortex.md").read_text(encoding="utf-8")
         ).render(
@@ -51,7 +59,7 @@ async def __call__(self, state: State):
             subgoal_plan=state.subgoal_plan,
             current_subgoal=get_current_subgoal(state.subgoal_plan),
             executor_feedback=executor_feedback,
-            executor_tools_list=format_tools_list(ctx=self.ctx, wrappers=EXECUTOR_WRAPPERS_TOOLS),
+            executor_tools_list=format_tools_list(ctx=self.ctx, wrappers=executor_wrappers),
             locked_app_package=current_locked_app_package,
         )
         messages = [

minitap/mobile_use/agents/executor/executor.py

@@ -9,7 +9,11 @@
 from minitap.mobile_use.context import MobileUseContext
 from minitap.mobile_use.graph.state import State
 from minitap.mobile_use.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
-from minitap.mobile_use.tools.index import EXECUTOR_WRAPPERS_TOOLS, get_tools_from_wrappers
+from minitap.mobile_use.tools.index import (
+    EXECUTOR_WRAPPERS_TOOLS,
+    VIDEO_RECORDING_WRAPPERS,
+    get_tools_from_wrappers,
+)
 from minitap.mobile_use.utils.decorators import wrap_with_callbacks
 from minitap.mobile_use.utils.logger import get_logger
 
@@ -54,8 +58,13 @@ async def __call__(self, state: State):
 
         llm = get_llm(ctx=self.ctx, name="executor")
         llm_fallback = get_llm(ctx=self.ctx, name="executor", use_fallback=True)
+
+        executor_wrappers = list(EXECUTOR_WRAPPERS_TOOLS)
+        if self.ctx.video_recording_enabled:
+            executor_wrappers.extend(VIDEO_RECORDING_WRAPPERS)
+
         llm_bind_tools_kwargs: dict = {
-            "tools": get_tools_from_wrappers(self.ctx, EXECUTOR_WRAPPERS_TOOLS),
+            "tools": get_tools_from_wrappers(self.ctx, executor_wrappers),
         }
 
         # ChatGoogleGenerativeAI does not support the "parallel_tool_calls" keyword

minitap/mobile_use/agents/planner/planner.md

@@ -26,6 +26,23 @@ All actions must stay within this app (except OAuth flows).
 - **No loops**: Instead of "repeat 3 times", write 3 separate subgoals
 
 **Shortcuts**: Always prefer `launch_app` to open apps (not manual app drawer navigation), `open_link` for URLs.
+{% if video_recording_enabled %}
+### 🎥 Video Recording Pattern
+
+When a goal involves capturing, transcribing, or analyzing video content, follow this execution order:
+
+1. **Start recording BEFORE interacting with the video** — The recording must be active before any click that triggers playback
+2. **Perform video interactions** — Open/play the video and wait for it to complete
+3. **Stop recording AFTER the video ends** — Only stop once playback is fully finished
+
+⚠️ **Critical:** Each recording action (start/stop) must be its own dedicated subgoal. This prevents truncation due to short videos or inference latency.
+
+**Example subgoals for "transcribe video X":**
+1. Navigate to video X location
+2. Start screen recording
+3. Play video X and wait for completion
+4. Stop recording and extract transcription
+{% endif %}
 
 Available tools: {{ executor_tools_list }}
 

minitap/mobile_use/agents/planner/planner.py

@@ -11,7 +11,11 @@
 )
 from minitap.mobile_use.graph.state import State
 from minitap.mobile_use.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
-from minitap.mobile_use.tools.index import EXECUTOR_WRAPPERS_TOOLS, format_tools_list
+from minitap.mobile_use.tools.index import (
+    EXECUTOR_WRAPPERS_TOOLS,
+    VIDEO_RECORDING_WRAPPERS,
+    format_tools_list,
+)
 from minitap.mobile_use.utils.decorators import wrap_with_callbacks
 from minitap.mobile_use.utils.logger import get_logger
 
@@ -35,13 +39,18 @@ async def __call__(self, state: State):
         )
         current_foreground_app = get_current_foreground_package(self.ctx)
 
+        executor_wrappers = list(EXECUTOR_WRAPPERS_TOOLS)
+        if self.ctx.video_recording_enabled:
+            executor_wrappers.extend(VIDEO_RECORDING_WRAPPERS)
+
         system_message = Template(
             Path(__file__).parent.joinpath("planner.md").read_text(encoding="utf-8")
         ).render(
             platform=self.ctx.device.mobile_platform.value,
-            executor_tools_list=format_tools_list(ctx=self.ctx, wrappers=EXECUTOR_WRAPPERS_TOOLS),
+            executor_tools_list=format_tools_list(ctx=self.ctx, wrappers=executor_wrappers),
             locked_app_package=current_locked_app_package,
             current_foreground_app=current_foreground_app,
+            video_recording_enabled=self.ctx.video_recording_enabled,
         )
         human_message = Template(
             Path(__file__).parent.joinpath("human.md").read_text(encoding="utf-8")

minitap/mobile_use/agents/video_analyzer/__init__.py

@@ -0,0 +1,5 @@
+"""Video analyzer utility for analyzing video content with Gemini models."""
+
+from minitap.mobile_use.agents.video_analyzer.video_analyzer import analyze_video
+
+__all__ = ["analyze_video"]

minitap/mobile_use/agents/video_analyzer/human.md

@@ -0,0 +1,5 @@
+Please analyze the following video recording and respond to my request.
+
+---
+
+**My Request**: {{ prompt }}

minitap/mobile_use/agents/video_analyzer/video_analyzer.md

@@ -0,0 +1,37 @@
+## You are a **Video Analysis Assistant**
+
+You analyze video recordings of mobile device screens and provide accurate, detailed responses based on what you observe.
+
+---
+
+## Your Focus Areas
+
+When analyzing videos, pay attention to:
+
+- **UI elements** and their states (buttons, text fields, toggles, etc.)
+- **Text content** displayed on screen
+- **Actions that occur** (taps, scrolls, transitions, animations)
+- **Notifications or dialogs** that appear
+- **Changes in the interface** over time
+- **Audio content** if present (transcribe speech, describe sounds)
+
+---
+
+## Guidelines
+
+- **Be precise and factual** - Only describe what you can actually see or hear
+- **Note uncertainty** - If you cannot clearly see or determine something, say so
+- **Be thorough** - Capture all relevant details that relate to the user's question
+- **Use timestamps** when describing sequences of events (e.g., "At 0:05, the user taps...")
+- **Structure your response** clearly when there's a lot of information
+
+---
+
+## Response Format
+
+Adapt your response format to the user's request:
+
+- For **transcription requests**: Provide clean, readable text of what was spoken or displayed
+- For **description requests**: Give a chronological narrative of events
+- For **specific questions**: Answer directly and concisely
+- For **extraction requests**: List items clearly (e.g., notifications, text content)

minitap/mobile_use/agents/video_analyzer/video_analyzer.py

@@ -0,0 +1,111 @@
+"""
+Video Analyzer utility for analyzing video content using Gemini models.
+
+This utility sends video files to video-capable Gemini models for analysis
+and returns text descriptions based on the provided prompt.
+"""
+
+import base64
+from pathlib import Path
+
+from jinja2 import Template
+from langchain_core.messages import HumanMessage, SystemMessage
+
+from minitap.mobile_use.context import MobileUseContext
+from minitap.mobile_use.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
+from minitap.mobile_use.utils.logger import get_logger
+from minitap.mobile_use.utils.video import compress_video_for_api
+
+logger = get_logger(__name__)
+
+
+async def analyze_video(
+    ctx: MobileUseContext,
+    video_path: Path,
+    prompt: str,
+) -> str:
+    """
+    Analyze a video file using a video-capable Gemini model.
+
+    Args:
+        ctx: The MobileUseContext containing LLM configuration
+        video_path: Path to the video file (MP4)
+        prompt: The analysis prompt/question about the video
+
+    Returns:
+        Text analysis result from the model
+
+    Raises:
+        Exception: If video analysis fails
+    """
+    logger.info(f"Starting video analysis for {video_path}")
+
+    if not video_path.exists():
+        raise FileNotFoundError(f"Video file not found: {video_path}")
+
+    # Compress video if needed to fit within API limits
+    compressed_path = await compress_video_for_api(video_path)
+
+    try:
+        with open(compressed_path, "rb") as video_file:
+            video_bytes = video_file.read()
+
+        video_base64 = base64.b64encode(video_bytes).decode("utf-8")
+
+        suffix = compressed_path.suffix.lower()
+        mime_type = "video/mp4" if suffix in [".mp4", ".m4v"] else f"video/{suffix[1:]}"
+
+        system_message_content = Template(
+            Path(__file__).parent.joinpath("video_analyzer.md").read_text(encoding="utf-8")
+        ).render()
+
+        human_message_content = Template(
+            Path(__file__).parent.joinpath("human.md").read_text(encoding="utf-8")
+        ).render(prompt=prompt)
+
+        messages = [
+            SystemMessage(content=system_message_content),
+            HumanMessage(
+                content=[
+                    {
+                        "type": "text",
+                        "text": human_message_content,
+                    },
+                    {
+                        "type": "file",
+                        "source_type": "base64",
+                        "mime_type": mime_type,
+                        "data": video_base64,
+                    },
+                ]
+            ),
+        ]
+
+        llm = get_llm(ctx=ctx, name="video_analyzer", is_utils=True, temperature=0.2)
+        llm_fallback = get_llm(
+            ctx=ctx, name="video_analyzer", is_utils=True, use_fallback=True, temperature=0.2
+        )
+
+        logger.info("Sending video to LLM for analysis...")
+
+        response = await with_fallback(
+            main_call=lambda: invoke_llm_with_timeout_message(
+                llm.ainvoke(messages), timeout_seconds=120
+            ),
+            fallback_call=lambda: invoke_llm_with_timeout_message(
+                llm_fallback.ainvoke(messages), timeout_seconds=120
+            ),
+        )
+
+        content = response.content if hasattr(response, "content") else str(response)
+        result = content if isinstance(content, str) else str(content)
+        logger.info("Video analysis completed")
+
+        return result
+    finally:
+        # Clean up compressed file if it differs from original
+        if compressed_path != video_path and compressed_path.exists():
+            try:
+                compressed_path.unlink()
+            except Exception:
+                pass

minitap/mobile_use/config.py

@@ -93,7 +93,7 @@ def record_events(output_path: Path | None, events: list[str] | BaseModel | Any)
 ### LLM Configuration
 
 LLMProvider = Literal["openai", "google", "openrouter", "xai", "vertexai", "minitap"]
-LLMUtilsNode = Literal["outputter", "hopper"]
+LLMUtilsNode = Literal["outputter", "hopper", "video_analyzer"]
 LLMUtilsNodeWithFallback = LLMUtilsNode
 AgentNode = Literal[
     "planner",
@@ -158,6 +158,7 @@ def __str__(self):
 class LLMConfigUtils(BaseModel):
     outputter: LLMWithFallback
     hopper: LLMWithFallback
+    video_analyzer: LLMWithFallback | None = None
 
 
 class LLMConfig(BaseModel):
@@ -176,6 +177,8 @@ def validate_providers(self):
         self.executor.validate_provider("Executor")
         self.utils.outputter.validate_provider("Outputter")
         self.utils.hopper.validate_provider("Hopper")
+        if self.utils.video_analyzer:
+            self.utils.video_analyzer.validate_provider("VideoAnalyzer")
 
     def __str__(self):
         return f"""
@@ -187,13 +190,20 @@ def __str__(self):
 🧩 Utils:
     🔽 Hopper: {self.utils.hopper}
     📝 Outputter: {self.utils.outputter}
+    🎬 Video Analyzer: {self.utils.video_analyzer or "Not configured"}
 """
 
     def get_agent(self, item: AgentNode) -> LLMWithFallback:
         return getattr(self, item)
 
     def get_utils(self, item: LLMUtilsNode) -> LLMWithFallback:
-        return getattr(self.utils, item)
+        value = getattr(self.utils, item)
+        if value is None:
+            raise ValueError(
+                f"Utils '{item}' is not configured. "
+                f"Please add it to your LLM config or enable it via AgentConfigBuilder."
+            )
+        return value
 
 
 def get_default_llm_config() -> LLMConfig: