"""
Logging configuration for Mobile Agent.
"""
import logging
import sys
from pathlib import Path
from typing import Optional
def setup_mobile_logging(
level: str = "INFO",
log_file: Optional[str] = None,
json_format: bool = False,
) -> logging.Logger:
"""
Configure logging for the mobile agent.
Args:
level: Log level (DEBUG, INFO, WARNING, ERROR)
log_file: Optional file path for log output
json_format: Use JSON format for structured logging
Returns:
Configured logger
"""
logger = logging.getLogger("comptext_mcp.mobile_agent")
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
# Clear existing handlers
logger.handlers.clear()
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
if json_format:
formatter = JsonFormatter()
else:
formatter = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# File handler (optional)
if log_file:
log_path = Path(log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
class JsonFormatter(logging.Formatter):
"""JSON log formatter for structured logging."""
def format(self, record: logging.LogRecord) -> str:
import json
from datetime import datetime
log_data = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
# Add extra fields
for key, value in record.__dict__.items():
if key not in (
"name", "msg", "args", "created", "filename", "funcName",
"levelname", "levelno", "lineno", "module", "msecs",
"pathname", "process", "processName", "relativeCreated",
"stack_info", "exc_info", "exc_text", "thread", "threadName",
"message",
):
log_data[key] = value
return json.dumps(log_data)
class AgentLogAdapter(logging.LoggerAdapter):
"""
Log adapter that adds agent context to all messages.
"""
def process(self, msg, kwargs):
extra = kwargs.get("extra", {})
extra.update(self.extra)
kwargs["extra"] = extra
# Prefix message with agent state if available
state = self.extra.get("agent_state", "")
if state:
msg = f"[{state}] {msg}"
return msg, kwargs
