Event Streaming

Overview

Mobilerun provides real-time event streaming that gives you visibility into agent execution as it happens. This allows you to build UIs, logging systems, or monitoring tools that react to agent actions in real-time. Under the hood, Mobilerun uses llama-index workflows - an event-driven orchestration system that powers the agent architecture.

Basic Usage

from mobilerun.agent.droid import MobileAgent

# Create and run agent
agent = MobileAgent(goal="Open Gmail and check inbox", config=config)
handler = agent.run()

# Stream events in real-time
async for event in handler.stream_events():
    if isinstance(event, ManagerPlanDetailsEvent):
        print(f"📋 Plan: {event.plan}")
        print(f"🎯 Current subgoal: {event.subgoal}")

    elif isinstance(event, ExecutorActionEvent):
        print(f"⚡ Action: {event.description}")
        print(f"💭 Thought: {event.thought}")

    elif isinstance(event, ScreenshotEvent):
        save_screenshot(event.screenshot, "screenshot.png")

    elif isinstance(event, FastAgentResponseEvent):
        if event.code:
            print(f"🔧 Tool calls: {event.code}")
        if event.thought:
            print(f"💭 Thought: {event.thought}")

# Wait for final result
result = await handler
print(f"✅ Success: {result.success}")
print(f"📝 Reason: {result.reason}")

Event Types

Workflow Coordination Events

Used for workflow coordination between MobileAgent and its child agents.

# Main workflow
class FastAgentExecuteEvent(Event):
    instruction: str

class FastAgentResultEvent(Event):
    success: bool
    reason: str
    instruction: str

class FinalizeEvent(Event):
    success: bool
    reason: str

class ResultEvent(StopEvent):
    success: bool
    reason: str
    steps: int
    structured_output: BaseModel | None

# Manager/Executor coordination
class ManagerInputEvent(Event): pass
class ManagerPlanEvent(Event):
    plan: str
    current_subgoal: str
    thought: str
    answer: str = ""
    success: bool | None = None

class ExecutorInputEvent(Event):
    current_subgoal: str

class ExecutorResultEvent(Event):
    action: Dict
    outcome: bool
    error: str
    summary: str

# External user message events
class ExternalUserMessageAppliedEvent(Event):
    message_ids: List[str]
    consumer: str
    step_number: int

class ExternalUserMessageDroppedEvent(Event):
    message_ids: List[str]
    reason: str
    step_number: int

Manager Events (Internal)

Internal to ManagerAgent, streamed to frontend/logging.

class ManagerContextEvent(Event): pass

class ManagerResponseEvent(Event):
    response: str
    usage: Optional[UsageResult] = None

class ManagerPlanDetailsEvent(Event):
    plan: str
    subgoal: str
    thought: str
    answer: str = ""
    memory_update: str = ""
    progress_summary: str = ""
    success: bool | None = None
    full_response: str = ""

Executor Events (Internal)

Internal to ExecutorAgent, streamed to frontend/logging.

class ExecutorContextEvent(Event):
    subgoal: str

class ExecutorResponseEvent(Event):
    response: str
    usage: Optional[UsageResult] = None

class ExecutorActionEvent(Event):
    action_json: str
    thought: str
    description: str
    full_response: str = ""

class ExecutorActionResultEvent(Event):
    action: Dict
    success: bool
    error: str
    summary: str
    thought: str = ""
    full_response: str = ""

FastAgent Events (Internal)

Internal to FastAgent, used in direct execution mode.

class FastAgentInputEvent(Event):
    pass

class FastAgentResponseEvent(Event):
    thought: str
    code: Optional[str] = None
    usage: Optional[UsageResult] = None

class FastAgentToolCallEvent(Event):
    tool_calls_repr: str

class FastAgentOutputEvent(Event):
    output: str

class FastAgentEndEvent(Event):
    success: bool
    reason: str
    tool_call_count: int = 0

Common, Visual & Telemetry Events

# Common events (mobilerun/agent/common/events.py)
class ScreenshotEvent(Event):
    screenshot: bytes

class RecordUIStateEvent(Event):
    ui_state: list[Dict[str, Any]]

class ToolExecutionEvent(Event):
    tool_name: str
    tool_args: Dict[str, Any]
    success: bool
    summary: str

# Telemetry events (when enabled)
class MobileAgentInitEvent(TelemetryEvent):
    goal: str
    llms: Dict[str, str]
    tools: str
    max_steps: int
    timeout: int
    vision: Dict[str, bool]
    reasoning: bool
    enable_tracing: bool
    debug: bool
    save_trajectories: str
    runtype: str
    custom_prompts: Optional[Dict[str, str]]

class PackageVisitEvent(TelemetryEvent):
    package_name: str
    activity_name: str
    step_number: int

class MobileAgentFinalizeEvent(TelemetryEvent):
    success: bool
    reason: str
    steps: int
    unique_packages_count: int
    unique_activities_count: int

# Usage tracking
class UsageResult(BaseModel):
    request_tokens: int
    response_tokens: int
    total_tokens: int
    requests: int

Common Patterns

Building a Live UI

async def run_with_ui(goal: str):
    agent = MobileAgent(goal=goal, config=config)
    handler = agent.run()

    async for event in handler.stream_events():
        if isinstance(event, ManagerPlanDetailsEvent):
            ui.update_plan(event.plan)
            ui.update_current_step(event.subgoal)

        elif isinstance(event, ExecutorActionEvent):
            ui.add_action_log(event.description, event.thought)

        elif isinstance(event, ScreenshotEvent):
            ui.update_screenshot(event.screenshot)

    result = await handler
    ui.show_completion(result.success, result.reason)

Tracking Token Usage

async def track_token_usage(goal: str):
    agent = MobileAgent(goal=goal, config=config)
    handler = agent.run()

    total_tokens = 0
    total_requests = 0

    async for event in handler.stream_events():
        # Check for events that contain usage information
        if hasattr(event, 'usage') and event.usage:
            total_tokens += event.usage.total_tokens
            total_requests += event.usage.requests

            print(f"LLM call - Input: {event.usage.request_tokens}, "
                  f"Output: {event.usage.response_tokens}, "
                  f"Total: {event.usage.total_tokens}")

    result = await handler
    print(f"\n📊 Total tokens used: {total_tokens}")
    print(f"📊 Total LLM requests: {total_requests}")

Logging and Monitoring

import logging

logger = logging.getLogger("mobilerun.monitor")

async def monitor_execution(goal: str):
    agent = MobileAgent(goal=goal, config=config)
    handler = agent.run()

    start_time = time.time()
    action_count = 0

    async for event in handler.stream_events():
        if isinstance(event, ExecutorActionEvent):
            action_count += 1
            logger.info(f"Action {action_count}: {event.description}")

        elif isinstance(event, FastAgentOutputEvent):
            logger.info(f"Tool execution result: {event.output}")

    result = await handler
    duration = time.time() - start_time

    logger.info(f"Task completed in {duration:.2f}s with {action_count} actions")
    logger.info(f"Result: {result.success} - {result.reason}")

Notes

Event Streaming Behavior

Events are streamed in real-time as the agent executes
Not all events are emitted in every execution (depends on mode and actions)
All events are Pydantic models with full type safety
The handler object is async - always use await handler to get the final result

Event Emission by Mode

Reasoning Mode (reasoning=True) emits:

Coordination: ManagerInputEvent, ManagerPlanEvent, ExecutorInputEvent, ExecutorResultEvent
Internal Manager: ManagerContextEvent, ManagerResponseEvent, ManagerPlanDetailsEvent
Internal Executor: ExecutorContextEvent, ExecutorResponseEvent, ExecutorActionEvent, ExecutorActionResultEvent
Tool execution: ToolExecutionEvent (after every tool dispatch)
Visual: ScreenshotEvent, RecordUIStateEvent (when enabled)

Direct Mode (reasoning=False) emits:

Coordination: FastAgentExecuteEvent, FastAgentResultEvent
Internal FastAgent: FastAgentInputEvent, FastAgentResponseEvent, FastAgentToolCallEvent, FastAgentOutputEvent, FastAgentEndEvent
Tool execution: ToolExecutionEvent (after every tool dispatch)
Visual: ScreenshotEvent, RecordUIStateEvent (when enabled)

All Modes emit:

Finalization: FinalizeEvent, ResultEvent
Telemetry: MobileAgentInitEvent, PackageVisitEvent, MobileAgentFinalizeEvent (when telemetry enabled)

Event Categories

Coordination Events - Used for workflow routing between agents (minimal data)

Located in mobilerun/agent/droid/events.py
Examples: ManagerPlanEvent, ExecutorResultEvent, ExternalUserMessageAppliedEvent

Internal Events - Used for streaming to frontend/logging (full debug data)

Located in agent-specific event files
Examples: ManagerPlanDetailsEvent, ExecutorActionEvent, FastAgentResponseEvent

Common Events - Emitted during execution (for screenshots, UI state recording, and tool tracking)

Located in mobilerun/agent/common/events.py
Examples: ScreenshotEvent, RecordUIStateEvent, ToolExecutionEvent

Telemetry Events - Captured for analytics (when enabled)

Located in mobilerun/telemetry/events.py
Examples: MobileAgentInitEvent, PackageVisitEvent, MobileAgentFinalizeEvent

Learn More

LlamaIndex Workflows - The underlying orchestration system

Introduction

Features

Guides

Concepts

SDK Reference

Overview

Basic Usage

Event Types

Common Patterns

Building a Live UI

Tracking Token Usage

Logging and Monitoring

Notes

Event Streaming Behavior

Event Emission by Mode

Event Categories

Learn More

​Overview

​Basic Usage

​Event Types

​Common Patterns

​Building a Live UI

​Tracking Token Usage

​Logging and Monitoring

​Notes

​Event Streaming Behavior

​Event Emission by Mode

​Event Categories

​Learn More

Overview

Basic Usage

Event Types

Common Patterns

Building a Live UI

Tracking Token Usage

Logging and Monitoring

Notes

Event Streaming Behavior

Event Emission by Mode

Event Categories

Learn More