diff --git a/src/strands_agents_builder/strands.py b/src/strands_agents_builder/strands.py index 5a2ec51..4470ebf 100644 --- a/src/strands_agents_builder/strands.py +++ b/src/strands_agents_builder/strands.py @@ -4,6 +4,7 @@ """ import argparse +import logging import os # Strands @@ -14,6 +15,7 @@ from strands_agents_builder.tools import get_tools from strands_agents_builder.utils import model_utils from strands_agents_builder.utils.kb_utils import load_system_prompt, store_conversation_in_kb +from strands_agents_builder.utils.logging_utils import configure_logging from strands_agents_builder.utils.welcome_utils import render_goodbye_message, render_welcome_message os.environ["STRANDS_TOOL_CONSOLE_MODE"] = "enabled" @@ -41,8 +43,34 @@ def main(): default="{}", help="Model config as JSON string or path", ) + parser.add_argument( + "--log-level", + type=str, + default=None, + choices=list(logging.getLevelNamesMapping().keys()), + help="Set the logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)", + ) + parser.add_argument( + "--log-file", + type=str, + help="Path to log file. If not specified, logs to stderr when log-level is set", + ) args = parser.parse_args() + # Configure logging based on provided arguments + if args.log_level or args.log_file: + # Default to INFO level if log_file is provided but log_level is not + log_level = args.log_level or "INFO" + configure_logging(log_level=log_level, log_file=args.log_file) + + # Get module logger for startup messages + logger = logging.getLogger("strands_agents_builder") + logger.info(f"Strands CLI started with log level {log_level}") + if args.log_file: + logger.info(f"Log file: {os.path.abspath(args.log_file)}") + else: + logger.info("Logging to stderr") + # Get knowledge_base_id from args or environment variable knowledge_base_id = args.knowledge_base_id or os.getenv("STRANDS_KNOWLEDGE_BASE_ID") diff --git a/src/strands_agents_builder/utils/logging_utils.py b/src/strands_agents_builder/utils/logging_utils.py new file mode 100644 index 0000000..2a43c37 --- /dev/null +++ b/src/strands_agents_builder/utils/logging_utils.py @@ -0,0 +1,76 @@ +""" +Utility functions for configuring and managing logging in the Strands Agent Builder. +""" + +import logging +import os +from typing import List, Optional + + +def configure_logging( + log_level: str = "INFO", + log_file: Optional[str] = None, + log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) -> None: + """ + Configure logging for the Strands Agent Builder. + + Args: + log_level: The logging level to use (DEBUG, INFO, WARNING, ERROR, CRITICAL) + log_file: Path to the log file. If None, logs to stderr. + log_format: The format string for log messages + + Returns: + None + """ + # Convert string log level to logging constant + numeric_level = getattr(logging, log_level.upper(), None) + if not isinstance(numeric_level, int): + raise ValueError(f"Invalid log level: {log_level}") + + # Reset root logger + root = logging.getLogger() + if root.handlers: + for handler in root.handlers[:]: + root.removeHandler(handler) + + # Create handlers list + handlers: List[logging.Handler] = [] + + if log_file: + # Setup file handler + try: + log_dir = os.path.dirname(log_file) + if log_dir and not os.path.exists(log_dir): + os.makedirs(log_dir, exist_ok=True) + handlers.append(logging.FileHandler(log_file)) + except Exception as e: + print(f"Warning: Failed to create log file {log_file}: {str(e)}") + print("Falling back to stderr logging") + handlers.append(logging.StreamHandler()) + else: + # If no log file specified, use stderr + handlers.append(logging.StreamHandler()) + + # Configure root logger + logging.basicConfig( + level=numeric_level, + format=log_format, + handlers=handlers, + force=True, # Force reconfiguration + ) + + # Configure specific Strands loggers (parent loggers will handle children) + loggers = ["strands", "strands_tools", "strands_agents_builder"] + + for logger_name in loggers: + logger = logging.getLogger(logger_name) + logger.setLevel(numeric_level) + + # Log configuration information + config_logger = logging.getLogger("strands_agents_builder") + config_logger.info(f"Logging configured with level: {log_level}") + if log_file: + config_logger.info(f"Log file: {os.path.abspath(log_file)}") + else: + config_logger.info("Logging to stderr") diff --git a/tests/utils/test_logging_utils.py b/tests/utils/test_logging_utils.py new file mode 100644 index 0000000..67dd1be --- /dev/null +++ b/tests/utils/test_logging_utils.py @@ -0,0 +1,131 @@ +""" +Unit tests for the logging_utils module. +""" + +import logging +import os +import tempfile +from unittest import mock + +from strands_agents_builder.utils.logging_utils import ( + configure_logging, + get_available_log_levels, +) + + +class TestLoggingUtils: + """Tests for the logging utilities.""" + + def setup_method(self): + """Reset root logger before each test.""" + root = logging.getLogger() + for handler in root.handlers[:]: + root.removeHandler(handler) + root.setLevel(logging.WARNING) # Reset to default + + def test_get_available_log_levels(self): + """Test get_available_log_levels function.""" + levels = get_available_log_levels() + assert levels == ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + + def test_configure_logging_with_file(self): + """Test configure_logging with log file.""" + with tempfile.NamedTemporaryFile(suffix=".log") as temp: + configure_logging(log_level="DEBUG", log_file=temp.name) + root = logging.getLogger() + assert root.level == logging.DEBUG + assert any(isinstance(h, logging.FileHandler) for h in root.handlers) + + # Check specific loggers + strands_logger = logging.getLogger("strands") + assert strands_logger.level == logging.DEBUG + + def test_configure_logging_without_file(self): + """Test configure_logging without log file uses stderr.""" + configure_logging(log_level="INFO", log_file=None) + root = logging.getLogger() + assert root.level == logging.INFO + assert any(isinstance(h, logging.StreamHandler) for h in root.handlers) + + def test_configure_logging_invalid_level(self): + """Test configure_logging with invalid log level.""" + try: + configure_logging(log_level="INVALID") + # If we get here, the function didn't raise an exception + raise AssertionError("Should have raised ValueError") + except ValueError: + # Expected path - test passes + pass + + def test_create_log_directory(self): + """Test that log directory is created if it doesn't exist.""" + with tempfile.TemporaryDirectory() as temp_dir: + log_path = os.path.join(temp_dir, "logs", "test.log") + + # Ensure directory doesn't exist + log_dir = os.path.join(temp_dir, "logs") + assert not os.path.exists(log_dir) + + configure_logging(log_level="INFO", log_file=log_path) + + # Directory should now exist + assert os.path.exists(log_dir) + + # Cleanup + if os.path.exists(log_path): + os.remove(log_path) + + def test_configure_logging_with_exception_fallback_to_stderr(self): + """Test configure_logging handles exceptions during file creation and falls back to stderr.""" + with ( + mock.patch("logging.FileHandler", side_effect=PermissionError("Access denied")), + mock.patch("builtins.print") as mock_print, + ): + configure_logging(log_level="INFO", log_file="/tmp/test.log") + + # Check that warning is printed and fallback to stderr occurs + assert mock_print.call_count == 2 + assert "Warning: Failed to create log file" in mock_print.call_args_list[0][0][0] + assert "Falling back to stderr logging" in mock_print.call_args_list[1][0][0] + + # Should still have a StreamHandler for stderr + root = logging.getLogger() + assert any(isinstance(h, logging.StreamHandler) for h in root.handlers) + + def test_logger_hierarchy(self): + """Test that parent loggers are configured properly.""" + configure_logging(log_level="DEBUG") + + # Check that the main loggers are configured + strands_logger = logging.getLogger("strands") + strands_tools_logger = logging.getLogger("strands_tools") + strands_agents_builder_logger = logging.getLogger("strands_agents_builder") + + assert strands_logger.level == logging.DEBUG + assert strands_tools_logger.level == logging.DEBUG + assert strands_agents_builder_logger.level == logging.DEBUG + + def test_reset_logger_with_existing_handlers(self): + """Test that existing handlers are properly removed when reconfiguring.""" + # First set up a handler + root = logging.getLogger() + handler = logging.StreamHandler() + root.addHandler(handler) + + # Then configure logging (should reset handlers) + configure_logging(log_level="INFO") + + # Check that old handlers were removed and new ones added + # We should have exactly one handler (the new StreamHandler) + assert len(root.handlers) == 1 + assert isinstance(root.handlers[0], logging.StreamHandler) + + def test_config_logger_messages(self): + """Test that configuration messages are logged properly.""" + with tempfile.NamedTemporaryFile(suffix=".log") as temp: + # Configure logging + configure_logging(log_level="INFO", log_file=temp.name) + + # Check that the config logger exists and was configured + config_logger = logging.getLogger("strands_agents_builder") + assert config_logger.level == logging.INFO