Skip to main content
The Agent class is the primary entry point for the mobile-use SDK, responsible for managing device interaction and executing tasks.

Import

from minitap.mobile_use.sdk import Agent

Constructor

Agent(config: AgentConfig | None = None)

Parameters

config
AgentConfig
Custom agent configuration. If not provided, default configuration is used.

Example

from minitap.mobile_use.sdk import Agent
from minitap.mobile_use.sdk.builders import Builders
from minitap.mobile_use.sdk.types import AgentProfile

# With default configuration
agent = Agent()

# With custom configuration
profile = AgentProfile(name="default", from_file="llm-config.defaults.jsonc")
config = Builders.AgentConfig.with_default_profile(profile).build()
agent = Agent(config=config)

Methods

init

Initialize the agent by connecting to a device and starting required servers.
async def init(
    self,
    api_key: str | None = None,
    server_restart_attempts: int = 3,
    retry_count: int = 5,
    retry_wait_seconds: int = 5,
) -> bool

Parameters

api_key
str
Minitap API key for Platform features (Platform tasks, cloud mobiles). Can also be set via MINITAP_API_KEY environment variable.
server_restart_attempts
int
default:3
Maximum number of attempts to start servers if they fail
retry_count
int
default:5
Number of retries for API calls
retry_wait_seconds
int
default:5
Seconds to wait between retries

Returns

success
bool
True if initialization succeeded, False otherwise

Example

import asyncio

async def main():
    agent = Agent()

    # Initialize without API key (local mode only)
    if not await agent.init():
        print("Failed to initialize agent")
        exit(1)

    print("Agent initialized successfully")

asyncio.run(main())

Example with API Key

import asyncio

async def main():
    agent = Agent()

    # Initialize with API key for Platform features
    if not await agent.init(api_key="your-minitap-api-key"):
        print("Failed to initialize agent")
        exit(1)

    print("Agent initialized successfully with Platform support")

asyncio.run(main())
Always check the return value of init() before running tasks. Note that init() is now an async method and must be awaited.

run_task

Execute a mobile automation task asynchronously.
async def run_task(
    self,
    *,
    goal: str | None = None,
    output: type[TOutput] | str | None = None,
    profile: str | AgentProfile | None = None,
    name: str | None = None,
    request: TaskRequest[TOutput] | PlatformTaskRequest[TOutput] | None = None,
) -> str | dict | TOutput | None

Parameters

goal
str
Natural language description of what to accomplish
output
type[TOutput] | str
Type of output:
  • Pydantic model class for structured output
  • String description for output format
profile
str | AgentProfile
Agent profile to use (name or instance)
name
str
Optional name for the task (for logging/debugging)
request
TaskRequest[TOutput] | PlatformTaskRequest[TOutput]
Pre-built TaskRequest or PlatformTaskRequest (alternative to individual parameters)

Returns

result
str | dict | TOutput | None
Task result:
  • str: Simple text output
  • dict: Unstructured dictionary
  • TOutput: Instance of specified Pydantic model
  • None: Task failed or no output

Examples

result = await agent.run_task(
    goal="Open calculator and compute 5 * 7"
)
print(result)  # String output
For platform tasks, the Locked App Package is configured on the platform task itself, not in PlatformTaskRequest. For local tasks using TaskRequest, use .with_locked_app_package() in the builder.

new_task

Create a new task request builder for fluent task configuration.
def new_task(self, goal: str) -> TaskRequestBuilder[None]

Parameters

goal
str
required
Natural language description of what to accomplish

Returns

builder
TaskRequestBuilder[None]
TaskRequestBuilder instance for fluent configuration

Example

from pathlib import Path

task = (
    agent.new_task("Open Gmail and count unread emails")
    .with_name("email_count")
    .with_max_steps(400)
    .with_trace_recording(enabled=True, path=Path("/tmp/traces"))
    .build()
)

result = await agent.run_task(request=task)

clean

Clean up resources, stop servers, and reset the agent state.
async def clean(self, force: bool = False) -> None

Parameters

force
bool
default:false
Set to True to clean zombie/pre-existing mobile-use servers

Example

import asyncio

async def main():
    agent = Agent()

    try:
        await agent.init()
        await agent.run_task(goal="Some task")
    finally:
        await agent.clean()  # Always clean up

asyncio.run(main())
Use force=True if you have zombie servers from previous runs:
await agent.clean(force=True)  # Kill any existing servers
await agent.init()              # Start fresh
Note that clean() is now an async method and must be awaited.

get_screenshot

Capture a screenshot from the mobile device.
async def get_screenshot(self) -> Image.Image
This method works with both local devices and cloud mobiles:
  • For local devices: Uses ADB (Android) or xcrun (iOS) to capture screenshots directly
  • For cloud mobiles: Retrieves screenshots from the cloud mobile via the Platform API

Returns

screenshot
Image.Image
Screenshot as a PIL Image object

Raises

  • AgentNotInitializedError: If the agent is not initialized
  • CloudMobileServiceUninitializedError: If using cloud mobile without proper initialization
  • Exception: If screenshot capture fails

Example

import asyncio
from minitap.mobile_use.sdk import Agent

async def main():
    agent = Agent()

    try:
        await agent.init()

        # Capture a screenshot
        screenshot = await agent.get_screenshot()

        # Save the screenshot
        screenshot.save("device_screenshot.png")
        print("Screenshot saved!")

    finally:
        await agent.clean()

asyncio.run(main())

Example with Cloud Mobile

import asyncio
from minitap.mobile_use.sdk import Agent
from minitap.mobile_use.sdk.builders import Builders

async def main():
    # Configure for cloud mobile
    config = Builders.AgentConfig.for_cloud_mobile("my-cloud-device").build()
    agent = Agent(config=config)

    try:
        await agent.init(api_key="your-minitap-api-key")

        # Capture screenshot from cloud mobile
        screenshot = await agent.get_screenshot()
        screenshot.save("cloud_device_screenshot.png")

    finally:
        await agent.clean()

asyncio.run(main())
Use get_screenshot() for debugging, monitoring, or extracting visual data from your mobile device during automation workflows.

Complete Example

import asyncio
from pydantic import BaseModel, Field
from minitap.mobile_use.sdk import Agent
from minitap.mobile_use.sdk.types import AgentProfile
from minitap.mobile_use.sdk.builders import Builders

class WeatherInfo(BaseModel):
    temperature: float = Field(..., description="Temperature in Celsius")
    condition: str = Field(..., description="Weather condition")

async def main():
    # Configure agent
    profile = AgentProfile(name="default", from_file="llm-config.defaults.jsonc")
    config = Builders.AgentConfig.with_default_profile(profile).build()
    agent = Agent(config=config)

    try:
        # Initialize
        if not await agent.init():
            print("Failed to initialize")
            return

        # Run task with structured output
        weather = await agent.run_task(
            goal="Open weather app and check current temperature",
            output=WeatherInfo,
            name="weather_check"
        )

        if weather:
            print(f"Temperature: {weather.temperature}°C")
            print(f"Condition: {weather.condition}")

    except Exception as e:
        print(f"Error: {e}")
    finally:
        await agent.clean()

if __name__ == "__main__":
    asyncio.run(main())

Exception Handling

The Agent may raise the following exceptions:
  • AgentNotInitializedError: Agent methods called before initialization
  • DeviceNotFoundError: No device found or device disconnected
  • AgentProfileNotFoundError: Specified profile not found
  • ServerStartupError: Failed to start required servers
  • ExecutableNotFoundError: Required executable (adb, maestro, xcrun) not found
  • AgentTaskRequestError: Invalid task request configuration
  • PlatformServiceUninitializedError: Platform service not initialized (missing API key)
  • CloudMobileServiceUninitializedError: Cloud mobile service not initialized (missing API key or cloud mobile not configured)
  • AgentInvalidApiKeyError: Invalid Minitap API key
See Exceptions for details.

Next Steps