From d8002974a7889606e62de9aeb1e9194f3a401f04 Mon Sep 17 00:00:00 2001 From: Murat Kaan Meral Date: Sun, 25 May 2025 11:35:58 +0200 Subject: [PATCH 1/4] feat: Add logging utils --- src/strands_agents_builder/strands.py | 20 +++ .../utils/logging_utils.py | 136 ++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 src/strands_agents_builder/utils/logging_utils.py diff --git a/src/strands_agents_builder/strands.py b/src/strands_agents_builder/strands.py index 4b67e52..9bc8b62 100644 --- a/src/strands_agents_builder/strands.py +++ b/src/strands_agents_builder/strands.py @@ -5,6 +5,7 @@ import argparse import os +import logging # Strands from strands import Agent @@ -36,6 +37,7 @@ 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.welcome_utils import render_goodbye_message, render_welcome_message +from strands_agents_builder.utils.logging_utils import configure_logging, get_available_log_levels, get_logging_status # Custom tools, handlers, utils from tools import ( @@ -69,8 +71,26 @@ def main(): default="{}", help="Model config as JSON string or path", ) + parser.add_argument( + "--log-level", + type=str, + default="INFO", + choices=get_available_log_levels(), + 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, logging is disabled", + ) args = parser.parse_args() + # Configure logging + configure_logging(log_level=args.log_level, log_file=args.log_file) + if args.log_file: + logging.info(f"Strands CLI started with log level {args.log_level}") + logging.info(f"Log file: {os.path.abspath(args.log_file)}") + # 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..d8cdb76 --- /dev/null +++ b/src/strands_agents_builder/utils/logging_utils.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +""" +Utility functions for configuring and managing logging in the Strands CLI. +""" + +import logging +import os +import sys +from typing import Optional, List, Union, Dict, Any + + +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 CLI with file output. + + Args: + log_level: The logging level to use (DEBUG, INFO, WARNING, ERROR, CRITICAL) + log_file: Path to the log file. If None, logging is disabled. + 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}") + + # If no log file is specified, disable logging and return + if not log_file: + # Reset root logger + root = logging.getLogger() + if root.handlers: + for handler in root.handlers[:]: + root.removeHandler(handler) + + # Set level to CRITICAL to minimize any accidental logging + root.setLevel(logging.CRITICAL) + return + + # Create handlers + handlers: List[logging.Handler] = [] + + # 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(f"Logging will be disabled") + return + + # Configure root logger + logging.basicConfig( + level=numeric_level, + format=log_format, + handlers=handlers, + force=True, # Force reconfiguration + ) + + # Configure specific Strands loggers + loggers = ["strands", "strands.agent", "strands.models", "strands.tools", "strands_agents_builder"] + + for logger_name in loggers: + logger = logging.getLogger(logger_name) + logger.setLevel(numeric_level) + + # Log configuration information to the file + logging.info(f"Logging configured with level: {log_level}") + logging.info(f"Log file: {os.path.abspath(log_file)}") + + +def get_available_log_levels() -> List[str]: + """ + Returns a list of available logging levels. + + Returns: + List of log level names + """ + return ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + + +def get_logging_status() -> Dict[str, Any]: + """ + Get the current logging status. + + Returns: + Dict with information about the current logging configuration + """ + root_logger = logging.getLogger() + handlers = root_logger.handlers + + file_handlers = [h for h in handlers if isinstance(h, logging.FileHandler)] + file_paths = [h.baseFilename for h in file_handlers] + + status = { + "level": logging.getLevelName(root_logger.level), + "handlers": {"console": any(isinstance(h, logging.StreamHandler) for h in handlers), "files": file_paths}, + "strands_loggers": {}, + } + + # Get status of strands specific loggers + for logger_name in ["strands", "strands.agent", "strands.models", "strands.tools", "strands_agents_builder"]: + logger = logging.getLogger(logger_name) + status["strands_loggers"][logger_name] = logging.getLevelName(logger.level) + + return status + + +def set_log_level_for_module(module_name: str, log_level: Union[str, int]) -> None: + """ + Set log level for a specific module. + + Args: + module_name: The name of the module logger + log_level: The log level to set (can be string or logging constant) + + Returns: + None + """ + if isinstance(log_level, str): + numeric_level = getattr(logging, log_level.upper(), None) + if not isinstance(numeric_level, int): + raise ValueError(f"Invalid log level: {log_level}") + else: + numeric_level = log_level + + logger = logging.getLogger(module_name) + logger.setLevel(numeric_level) + logging.debug(f"Set log level for {module_name} to {log_level}") From 81fb0dfc6d03f9c8a3a0407c4f60e22a8395236e Mon Sep 17 00:00:00 2001 From: Murat Kaan Meral Date: Sun, 25 May 2025 12:48:09 +0200 Subject: [PATCH 2/4] Add unit tests for logging util --- src/strands_agents_builder/strands.py | 4 ++-- src/strands_agents_builder/utils/logging_utils.py | 5 ++--- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/strands_agents_builder/strands.py b/src/strands_agents_builder/strands.py index 9bc8b62..3955536 100644 --- a/src/strands_agents_builder/strands.py +++ b/src/strands_agents_builder/strands.py @@ -4,8 +4,8 @@ """ import argparse -import os import logging +import os # Strands from strands import Agent @@ -36,8 +36,8 @@ from strands_agents_builder.handlers.callback_handler import callback_handler 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, get_available_log_levels from strands_agents_builder.utils.welcome_utils import render_goodbye_message, render_welcome_message -from strands_agents_builder.utils.logging_utils import configure_logging, get_available_log_levels, get_logging_status # Custom tools, handlers, utils from tools import ( diff --git a/src/strands_agents_builder/utils/logging_utils.py b/src/strands_agents_builder/utils/logging_utils.py index d8cdb76..f29594a 100644 --- a/src/strands_agents_builder/utils/logging_utils.py +++ b/src/strands_agents_builder/utils/logging_utils.py @@ -5,8 +5,7 @@ import logging import os -import sys -from typing import Optional, List, Union, Dict, Any +from typing import Any, Dict, List, Optional, Union def configure_logging( @@ -53,7 +52,7 @@ def configure_logging( handlers.append(logging.FileHandler(log_file)) except Exception as e: print(f"Warning: Failed to create log file {log_file}: {str(e)}") - print(f"Logging will be disabled") + print("Logging will be disabled") return # Configure root logger From 8fbadfc393945e105f58752e7aafccee0128192b Mon Sep 17 00:00:00 2001 From: Murat Kaan Meral Date: Sat, 31 May 2025 15:30:25 +0200 Subject: [PATCH 3/4] Address all PR review comments for logging utilities MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ✅ Fixed Issues: - Updated documentation from 'Strands CLI' to 'Strands Agent Builder' - Removed shebang from utility module - Enhanced logging behavior: stderr support when only --log-level provided - Fixed logger usage: using module logger for startup messages - Optimized logger configuration: simplified to parent loggers only - Removed unused functions: get_logging_status() and set_log_level_for_module() 🚀 Improvements: - Better UX: --log-level DEBUG works without --log-file (logs to stderr) - Fallback handling: auto-fallback to stderr if file creation fails - Enhanced CLI args: improved help text and argument behavior - Comprehensive test suite: 9 unit tests covering all scenarios 📝 New Logging Behavior: - No args = no logging (clean output) - --log-level only = logs to stderr - --log-file only = logs to file (INFO level) - Both args = logs to file with specified level All tests pass ✅ --- src/strands_agents_builder/strands.py | 22 ++- .../utils/logging_utils.py | 115 +++++---------- tests/utils/test_logging_utils.py | 131 ++++++++++++++++++ 3 files changed, 179 insertions(+), 89 deletions(-) create mode 100644 tests/utils/test_logging_utils.py diff --git a/src/strands_agents_builder/strands.py b/src/strands_agents_builder/strands.py index 7201371..08b0785 100644 --- a/src/strands_agents_builder/strands.py +++ b/src/strands_agents_builder/strands.py @@ -82,22 +82,30 @@ def main(): parser.add_argument( "--log-level", type=str, - default="INFO", + default=None, choices=get_available_log_levels(), 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, logging is disabled", + help="Path to log file. If not specified, logs to stderr when log-level is set", ) args = parser.parse_args() - # Configure logging - configure_logging(log_level=args.log_level, log_file=args.log_file) - if args.log_file: - logging.info(f"Strands CLI started with log level {args.log_level}") - logging.info(f"Log file: {os.path.abspath(args.log_file)}") + # 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 index f29594a..2c555ad 100644 --- a/src/strands_agents_builder/utils/logging_utils.py +++ b/src/strands_agents_builder/utils/logging_utils.py @@ -1,11 +1,10 @@ -#!/usr/bin/env python3 """ -Utility functions for configuring and managing logging in the Strands CLI. +Utility functions for configuring and managing logging in the Strands Agent Builder. """ import logging import os -from typing import Any, Dict, List, Optional, Union +from typing import List, Optional def configure_logging( @@ -14,11 +13,11 @@ def configure_logging( log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) -> None: """ - Configure logging for the Strands CLI with file output. + 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, logging is disabled. + log_file: Path to the log file. If None, logs to stderr. log_format: The format string for log messages Returns: @@ -29,31 +28,29 @@ def configure_logging( if not isinstance(numeric_level, int): raise ValueError(f"Invalid log level: {log_level}") - # If no log file is specified, disable logging and return - if not log_file: - # Reset root logger - root = logging.getLogger() - if root.handlers: - for handler in root.handlers[:]: - root.removeHandler(handler) + # Reset root logger + root = logging.getLogger() + if root.handlers: + for handler in root.handlers[:]: + root.removeHandler(handler) - # Set level to CRITICAL to minimize any accidental logging - root.setLevel(logging.CRITICAL) - return - - # Create handlers + # Create handlers list handlers: List[logging.Handler] = [] - # 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("Logging will be disabled") - return + 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( @@ -63,16 +60,20 @@ def configure_logging( force=True, # Force reconfiguration ) - # Configure specific Strands loggers - loggers = ["strands", "strands.agent", "strands.models", "strands.tools", "strands_agents_builder"] + # 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 to the file - logging.info(f"Logging configured with level: {log_level}") - logging.info(f"Log file: {os.path.abspath(log_file)}") + # 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") def get_available_log_levels() -> List[str]: @@ -83,53 +84,3 @@ def get_available_log_levels() -> List[str]: List of log level names """ return ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] - - -def get_logging_status() -> Dict[str, Any]: - """ - Get the current logging status. - - Returns: - Dict with information about the current logging configuration - """ - root_logger = logging.getLogger() - handlers = root_logger.handlers - - file_handlers = [h for h in handlers if isinstance(h, logging.FileHandler)] - file_paths = [h.baseFilename for h in file_handlers] - - status = { - "level": logging.getLevelName(root_logger.level), - "handlers": {"console": any(isinstance(h, logging.StreamHandler) for h in handlers), "files": file_paths}, - "strands_loggers": {}, - } - - # Get status of strands specific loggers - for logger_name in ["strands", "strands.agent", "strands.models", "strands.tools", "strands_agents_builder"]: - logger = logging.getLogger(logger_name) - status["strands_loggers"][logger_name] = logging.getLevelName(logger.level) - - return status - - -def set_log_level_for_module(module_name: str, log_level: Union[str, int]) -> None: - """ - Set log level for a specific module. - - Args: - module_name: The name of the module logger - log_level: The log level to set (can be string or logging constant) - - Returns: - None - """ - if isinstance(log_level, str): - numeric_level = getattr(logging, log_level.upper(), None) - if not isinstance(numeric_level, int): - raise ValueError(f"Invalid log level: {log_level}") - else: - numeric_level = log_level - - logger = logging.getLogger(module_name) - logger.setLevel(numeric_level) - logging.debug(f"Set log level for {module_name} to {log_level}") 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 From 94d96046418e497b7836bfbffd2c22fbbf83cc84 Mon Sep 17 00:00:00 2001 From: Murat Kaan Meral Date: Wed, 11 Jun 2025 12:30:41 +0300 Subject: [PATCH 4/4] fix(logging): remove logging levels function --- src/strands_agents_builder/strands.py | 4 ++-- src/strands_agents_builder/utils/logging_utils.py | 10 ---------- 2 files changed, 2 insertions(+), 12 deletions(-) diff --git a/src/strands_agents_builder/strands.py b/src/strands_agents_builder/strands.py index 08b0785..6049ed8 100644 --- a/src/strands_agents_builder/strands.py +++ b/src/strands_agents_builder/strands.py @@ -44,7 +44,7 @@ from strands_agents_builder.handlers.callback_handler import callback_handler 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, get_available_log_levels +from strands_agents_builder.utils.logging_utils import configure_logging from strands_agents_builder.utils.welcome_utils import render_goodbye_message, render_welcome_message # Custom tools @@ -83,7 +83,7 @@ def main(): "--log-level", type=str, default=None, - choices=get_available_log_levels(), + choices=list(logging.getLevelNamesMapping().keys()), help="Set the logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)", ) parser.add_argument( diff --git a/src/strands_agents_builder/utils/logging_utils.py b/src/strands_agents_builder/utils/logging_utils.py index 2c555ad..2a43c37 100644 --- a/src/strands_agents_builder/utils/logging_utils.py +++ b/src/strands_agents_builder/utils/logging_utils.py @@ -74,13 +74,3 @@ def configure_logging( config_logger.info(f"Log file: {os.path.abspath(log_file)}") else: config_logger.info("Logging to stderr") - - -def get_available_log_levels() -> List[str]: - """ - Returns a list of available logging levels. - - Returns: - List of log level names - """ - return ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]