Add extensible log handler

agentstack/cli/agentstack_data.py

@@ -3,7 +3,7 @@
 from typing import Optional
 
 from agentstack.utils import clean_input, get_version
-from agentstack.logger import log
+from agentstack import log
 
 
 class ProjectMetadata:

agentstack/cli/cli.py

@@ -16,7 +16,7 @@
     ProjectStructure,
     CookiecutterData,
 )
-from agentstack.logger import log
+from agentstack import log
 from agentstack import conf
 from agentstack.conf import ConfigFile
 from agentstack.utils import get_package_path

agentstack/cli/run.py

@@ -15,7 +15,7 @@
 MAIN_MODULE_NAME = "main"
 
 
-def _format_friendy_error_message(exception: Exception):
+def _format_friendly_error_message(exception: Exception):
     """
     Projects will throw various errors, especially on first runs, so we catch
     them here and print a more helpful message.
@@ -84,7 +84,7 @@ def _import_project_module(path: Path):
     assert spec.loader is not None  # appease type checker
 
     project_module = importlib.util.module_from_spec(spec)
-    sys.path.append(str((path / MAIN_FILENAME).parent))
+    sys.path.insert(0, str((path / MAIN_FILENAME).parent))
     spec.loader.exec_module(project_module)
     return project_module
 
@@ -124,6 +124,6 @@ def run_project(command: str = 'run', debug: bool = False, cli_args: Optional[st
         if debug:
             raise exception
         print(term_color("\nAn error occurred while running your project:\n", 'red'))
-        print(_format_friendy_error_message(exception))
+        print(_format_friendly_error_message(exception))
         print(term_color("\nRun `agentstack run --debug` for a full traceback.", 'blue'))
         sys.exit(1)

agentstack/conf.py

@@ -9,6 +9,9 @@
 DEFAULT_FRAMEWORK = "crewai"
 CONFIG_FILENAME = "agentstack.json"
 
+DEBUG: bool = False
+
+# The path to the project directory ie. working directory.
 PATH: Path = Path()
 
 
@@ -18,6 +21,15 @@ def set_path(path: Union[str, Path, None]):
     PATH = Path(path) if path else Path()
 
 
+def set_debug(debug: bool):
+    """
+    Set the debug flag in the project's configuration for the session; does not
+    get saved to the project's configuration file.
+    """
+    global DEBUG
+    DEBUG = debug
+
+
 def get_framework() -> Optional[str]:
     """The framework used in the project. Will be available after PATH has been set
     and if we are inside a project directory.

agentstack/generation/tool_generation.py

@@ -6,11 +6,11 @@
 import ast
 
 from agentstack import conf
+from agentstack import log
 from agentstack.conf import ConfigFile
 from agentstack.exceptions import ValidationError
 from agentstack import frameworks
 from agentstack import packaging
-from agentstack.utils import term_color
 from agentstack.tools import ToolConfig
 from agentstack.generation import asttools
 from agentstack.generation.files import EnvFile
@@ -83,7 +83,7 @@ def add_tool(tool_name: str, agents: Optional[list[str]] = []):
     tool = ToolConfig.from_tool_name(tool_name)
 
     if tool_name in agentstack_config.tools:
-        print(term_color(f'Tool {tool_name} is already installed', 'blue'))
+        log.notify(f'Tool {tool_name} is already installed')
     else:  # handle install
         tool_file_path = tool.get_impl_file_path(agentstack_config.framework)
 
@@ -97,7 +97,7 @@ def add_tool(tool_name: str, agents: Optional[list[str]] = []):
             with ToolsInitFile(conf.PATH / TOOLS_INIT_FILENAME) as tools_init:
                 tools_init.add_import_for_tool(tool, agentstack_config.framework)
         except ValidationError as e:
-            print(term_color(f"Error adding tool:\n{e}", 'red'))
+            log.error(f"Error adding tool:\n{e}")
 
         if tool.env:  # add environment variables which don't exist
             with EnvFile() as env:
@@ -117,20 +117,19 @@ def add_tool(tool_name: str, agents: Optional[list[str]] = []):
     if not agents:  # If no agents are specified, add the tool to all agents
         agents = frameworks.get_agent_names()
     for agent_name in agents:
-        print(f'Adding tool {tool.name} to agent {agent_name}')
+        log.info(f'Adding tool {tool.name} to agent {agent_name}')
         frameworks.add_tool(tool, agent_name)
 
-    print(term_color(f'🔨 Tool {tool.name} added to agentstack project successfully', 'green'))
+    log.success(f'🔨 Tool {tool.name} added to agentstack project successfully')
     if tool.cta:
-        print(term_color(f'🪩  {tool.cta}', 'blue'))
+        log.notify(f'🪩  {tool.cta}')
 
 
 def remove_tool(tool_name: str, agents: Optional[list[str]] = []):
     agentstack_config = ConfigFile()
 
     if tool_name not in agentstack_config.tools:
-        print(term_color(f'Tool {tool_name} is not installed', 'red'))
-        sys.exit(1)
+        raise ValidationError(f'Tool {tool_name} is not installed')
 
     tool = ToolConfig.from_tool_name(tool_name)
     if tool.packages:
@@ -140,13 +139,13 @@ def remove_tool(tool_name: str, agents: Optional[list[str]] = []):
     try:
         os.remove(conf.PATH / f'src/tools/{tool.module_name}.py')
     except FileNotFoundError:
-        print(f'"src/tools/{tool.module_name}.py" not found')
+        log.warning(f'"src/tools/{tool.module_name}.py" not found')
 
     try:  # Edit the user's project tool init file to exclude the tool
         with ToolsInitFile(conf.PATH / TOOLS_INIT_FILENAME) as tools_init:
             tools_init.remove_import_for_tool(tool, agentstack_config.framework)
     except ValidationError as e:
-        print(term_color(f"Error removing tool:\n{e}", 'red'))
+        log.error(f"Error removing tool:\n{e}")
 
     # Edit the framework entrypoint file to exclude the tool in the agent definition
     if not agents:  # If no agents are specified, remove the tool from all agents
@@ -161,8 +160,4 @@ def remove_tool(tool_name: str, agents: Optional[list[str]] = []):
     with agentstack_config as config:
         config.tools.remove(tool.name)
 
-    print(
-        term_color(f'🔨 Tool {tool_name}', 'green'),
-        term_color('removed', 'red'),
-        term_color('from agentstack project successfully', 'green'),
-    )
+    log.success(f'🔨 Tool {tool_name} removed from agentstack project successfully')

agentstack/log.py

@@ -0,0 +1,185 @@
+"""
+`agentstack.log`
+
+DEBUG:    Detailed technical information, typically of interest when diagnosing problems.
+TOOL_USE: A message to indicate the use of a tool.
+THINKING: Information about an internal monologue or reasoning.
+INFO:     Useful information about the state of the application.
+NOTIFY:   A notification or update.
+SUCCESS:  An indication of a successful operation.
+RESPONSE: A response to a request.
+WARNING:  An indication that something unexpected happened, but not severe.
+ERROR:    An indication that something went wrong, and the application may not be able to continue.
+
+TODO TOOL_USE and THINKING are below INFO; this is intentional for now.
+
+TODO would be cool to intercept all messages from the framework and redirect
+them through this logger. This would allow us to capture all messages and display
+them in the console and filter based on priority.
+
+TODO With agentstack serve, we can direct all messages to the API, too.
+"""
+
+from typing import IO, Optional, Callable
+import os, sys
+import io
+import logging
+from agentstack import conf
+from agentstack.utils import term_color
+
+__all__ = [
+    'set_stdout',
+    'set_stderr',
+    'debug',
+    'tool_use',
+    'thinking',
+    'info',
+    'notify',
+    'success',
+    'response',
+    'warning',
+    'error',
+]
+
+LOG_NAME: str = 'agentstack'
+LOG_FILENAME: str = 'agentstack.log'
+
+# define additional log levels to accommodate other messages inside the app
+DEBUG = logging.DEBUG  # 10
+TOOL_USE = 16
+THINKING = 18
+INFO = logging.INFO  # 20
+NOTIFY = 22
+SUCCESS = 24
+RESPONSE = 26
+WARNING = logging.WARNING  # 30
+ERROR = logging.ERROR  # 40
+
+logging.addLevelName(THINKING, 'THINKING')
+logging.addLevelName(TOOL_USE, 'TOOL_USE')
+logging.addLevelName(NOTIFY, 'NOTIFY')
+logging.addLevelName(SUCCESS, 'SUCCESS')
+logging.addLevelName(RESPONSE, 'RESPONSE')
+
+# `instance` is lazy so we have time to set up handlers
+instance: Optional[logging.Logger] = None
+
+stdout: IO = io.StringIO()
+stderr: IO = io.StringIO()
+
+
+def set_stdout(stream: IO):
+    """
+    Redirect standard output messages to the given stream.
+    In practice, if a shell is available, pass: `sys.stdout`.
+    But, this can be any stream that implements the `write` method.
+    """
+    global stdout, instance
+    stdout = stream
+    instance = None  # force re-initialization
+
+
+def set_stderr(stream: IO):
+    """
+    Redirect standard error messages to the given stream.
+    In practice, if a shell is available, pass: `sys.stderr`.
+    But, this can be any stream that implements the `write` method.
+    """
+    global stderr, instance
+    stderr = stream
+    instance = None  # force re-initialization
+
+
+def _create_handler(levelno: int) -> Callable:
+    """Get the logging handler for the given log level."""
+
+    def handler(msg, *args, **kwargs):
+        global instance
+        if instance is None:
+            instance = _build_logger()
+        return instance.log(levelno, msg, *args, **kwargs)
+
+    return handler
+
+
+debug = _create_handler(DEBUG)
+tool_use = _create_handler(TOOL_USE)
+thinking = _create_handler(THINKING)
+info = _create_handler(INFO)
+notify = _create_handler(NOTIFY)
+success = _create_handler(SUCCESS)
+response = _create_handler(RESPONSE)
+warning = _create_handler(WARNING)
+error = _create_handler(ERROR)
+
+
+class ConsoleFormatter(logging.Formatter):
+    """Formats log messages for display in the console."""
+
+    default_format = logging.Formatter('%(message)s')
+    formats = {
+        DEBUG: logging.Formatter('DEBUG: %(message)s'),
+        SUCCESS: logging.Formatter(term_color('%(message)s', 'green')),
+        NOTIFY: logging.Formatter(term_color('%(message)s', 'blue')),
+        WARNING: logging.Formatter(term_color('%(message)s', 'yellow')),
+        ERROR: logging.Formatter(term_color('%(message)s', 'red')),
+    }
+
+    def format(self, record: logging.LogRecord) -> str:
+        template = self.formats.get(record.levelno, self.default_format)
+        return template.format(record)
+
+
+class FileFormatter(logging.Formatter):
+    """Formats log messages for display in a log file."""
+
+    default_format = logging.Formatter('%(levelname)s: %(message)s')
+    formats = {
+        DEBUG: logging.Formatter('DEBUG (%(asctime)s):\n %(pathname)s:%(lineno)d\n %(message)s'),
+    }
+
+    def format(self, record: logging.LogRecord) -> str:
+        template = self.formats.get(record.levelno, self.default_format)
+        return template.format(record)
+
+
+def _build_logger() -> logging.Logger:
+    """
+    Build the logger with the appropriate handlers.
+    All log messages are written to the log file.
+    Errors and above are written to stderr if a stream has been configured.
+    Warnings and below are written to stdout if a stream has been configured.
+    """
+    # global stdout, stderr
+
+    log = logging.getLogger(LOG_NAME)
+    # min log level set here cascades to all handlers
+    log.setLevel(DEBUG if conf.DEBUG else INFO)
+
+    # `conf.PATH`` can change during startup, so defer building the path
+    log_filename = conf.PATH / LOG_FILENAME
+    if not os.path.exists(log_filename):
+        os.makedirs(log_filename.parent, exist_ok=True)
+        log_filename.touch()
+
+    file_handler = logging.FileHandler(log_filename)
+    file_handler.setFormatter(FileFormatter())
+    file_handler.setLevel(DEBUG)
+    log.addHandler(file_handler)
+
+    # stdout handler for warnings and below
+    # `stdout` can change, so defer building the stream until we need it
+    stdout_handler = logging.StreamHandler(stdout)
+    stdout_handler.setFormatter(ConsoleFormatter())
+    stdout_handler.setLevel(DEBUG)
+    stdout_handler.addFilter(lambda record: record.levelno < ERROR)
+    log.addHandler(stdout_handler)
+
+    # stderr handler for errors and above
+    # `stderr` can change, so defer building the stream until we need it
+    stderr_handler = logging.StreamHandler(stderr)
+    stderr_handler.setFormatter(ConsoleFormatter())
+    stderr_handler.setLevel(ERROR)
+    log.addHandler(stderr_handler)
+
+    return log