Files
Artemis-Eapil/artemis/sdk/examples/app_lock_messaging.py
T
somew1nd 206a9ce921 feat: initial release of ARTEMIS Android Agent & Automation Platform
ARTEMIS is an autonomous multimodal Android agent and mobile UI automation
platform engineered for physical devices and emulators. It combines state-of-the-art
multimodal Vision-Language Models (VLMs) with hierarchical closed-loop planning,
resilient self-healing, and deep Android subsystem integration.

Key Features & Architectural Highlights:
- Dual-Engine Orchestration:
  * Pro Engine: Multi-agent closed-loop (Planner -> Operator -> Validator/Failure Analyzer -> Summarizer) powered by LangGraph.
  * Flash Engine: Ultra-fast reactive loop for deterministic, low-latency UI execution (<35 steps).
- Multimodal Perception & Grounding:
  * Hybrid visual screenshot grounding, real-time OCR extraction, and XML accessibility hierarchy parsing.
  * Dynamic-First, Coordinate-Fallback locating strategy for maximum resilience against UI drift.
- Autonomous Self-Healing & Diagnostics:
  * Failure analyzer and safety net for popup dismissal, crash recovery, and auto-retry.
  * Diagnostic sub-agents for log analysis, video recording inspection, and object detection.
- Unified CLI & Python SDK:
  * Rich CLI with commands: init, doctor, run, trace, batch, bench, server, and mcp.
  * Fluent Python SDK with Pydantic structured output models and package locking.
- Multi-Provider LLM Architecture:
  * Support for Google Gemini, OpenAI, Anthropic Claude, Vertex AI, OpenRouter, xAI Grok, and local endpoints (Ollama/vLLM).
  * Centralized configuration in config/artemis.jsonc.
- Native MCP & Observability:
  * Model Context Protocol (MCP) server endpoints for AI IDEs.
  * Full trace recording with before/after screenshots, action overlays, and execution replays.

Change-Id: I430b93ee2b76b54cc5a3c6eeccd15d2cf29830a1
2026-08-13 11:46:47 -07:00

88 lines
2.7 KiB
Python

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from artemis.sdk import Agent
from pydantic import BaseModel, Field
class _CyFunctionDetectorMeta(type):
def __instancecheck__(self, instance):
name = type(instance).__name__
return (
name
in (
"cyfunction",
"cython_function_or_method",
"builtin_function_or_method",
)
or "cyfunction" in name.lower()
)
class CyFunctionDetector(metaclass=_CyFunctionDetectorMeta):
pass
class MessageResult(BaseModel):
"""Structured result from messaging task."""
model_config = {"ignored_types": (CyFunctionDetector,)}
messages_sent: int = Field(..., description="Number of messages successfully sent")
contacts: list[str] = Field(..., description="List of contacts messaged")
success: bool = Field(..., description="Whether all messages were sent successfully")
async def main() -> None:
# Create agent with default configuration
agent = Agent()
try:
await agent.init()
# Use app lock to keep execution in WhatsApp
# This ensures the agent stays in the app and relaunches if needed
task = (
agent.new_task("Send 'Happy New Year!' message to Alice, Bob, and Charlie on WhatsApp")
.with_name("send_new_year_messages")
.with_locked_app_package("com.whatsapp") # Lock to WhatsApp
.with_output_format(MessageResult)
.with_max_steps(600) # Messaging tasks may need more steps
.build()
)
print("Sending messages with app lock enabled...")
print("The agent will stay in WhatsApp and relaunch if needed.\n")
result = await agent.run_task(request=task)
if result:
print("\n=== Messaging Complete ===")
print(f"Messages sent: {result.messages_sent}")
print(f"Contacts: {', '.join(result.contacts)}")
print(f"Success: {result.success}")
else:
print("Failed to send messages")
except Exception as e:
print(f"Error: {e}")
finally:
await agent.clean()
if __name__ == "__main__":
asyncio.run(main())