main

Simple CLI interface using the core orchestrator.

This provides a clean command-line interface that uses the extracted core orchestration logic rather than duplicating it.

MADACLIInterface

Simple command-line interface for MADA.

Uses the core MADAOrchestrator for all orchestration logic, providing a clean separation between UI and core functionality.

Source code in src/mada/interfaces/cli/main.py

class MADACLIInterface:
    """
    Simple command-line interface for MADA.

    Uses the core MADAOrchestrator for all orchestration logic,
    providing a clean separation between UI and core functionality.
    """

    def __init__(self, config: AppConfig):
        """
        Initialize the CLI with configuration.

        Args:
            config: Application configuration
        """
        self.config = config
        self.orchestrator = None
        self.session_manager = ChatSessionManager(config.database)

    def _print_history_summary(self, history: List[Dict[str, str]]):
        """
        Print a brief summary of session history for context.
        Assumes `history` is a list of dicts like:
        [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]

        Args:
            history: A list of the chat history
        """
        if not history:
            print("This session has no previous messages.")
            return

        print("\nPrevious messages in this session (most recent last):")
        # Show at most the last 5 turns
        max_messages = 10
        to_show = history[-max_messages:]
        for msg in to_show:
            role = msg.get("role", "unknown")
            content = msg.get("content", "").strip()
            # Truncate very long lines for CLI readability
            if len(content) > 200:
                content = content[:197] + "..."
            print(f"  {role}: {content}")

    def startup_session_menu(self) -> bool:
        """
        Interactive menu on startup to manage chat sessions.

        Returns:
            bool: True if a session has been selected/created and we should continue,
                False if the user chose to quit.
        """
        while True:
            print("\nChat Session Manager")
            print("=" * 50)
            sessions = self.list_sessions()

            if sessions:
                print("Existing sessions:")
                for idx, label in enumerate(sessions, start=1):
                    print(f"  {idx}. {label}")
            else:
                print("No existing sessions.")

            print("\nOptions:")
            print("  [n] New session")
            if sessions:
                print("  [s] Select session")
                print("  [d] Delete session")
                print("  [a] Delete ALL sessions")
            print("  [q] Quit")

            choice = input("\nEnter choice: ").strip().lower()

            if choice == "q":
                return False

            if choice == "n":
                self.create_new_session()
                print("Created and selected a new session.")
                return True

            if choice == "s" and sessions:
                try:
                    selection = input(
                        "Enter the number of the session to select: "
                    ).strip()
                    idx = int(selection)
                    if 1 <= idx <= len(sessions):
                        label = sessions[idx - 1]
                        history = self.select_session(label)
                        print(f"Selected session: {label}")
                        self._print_history_summary(history)
                        return True
                    else:
                        print("Invalid index.")
                except ValueError:
                    print("Please enter a valid number.")
                continue

            if choice == "d" and sessions:
                try:
                    selection = input(
                        "Enter the number of the session to delete: "
                    ).strip()
                    idx = int(selection)
                    if 1 <= idx <= len(sessions):
                        label = sessions[idx - 1]
                        confirm = (
                            input(f"Are you sure you want to delete '{label}'? [y/N]: ")
                            .strip()
                            .lower()
                        )
                        if confirm == "y":
                            self.delete_session(label)
                            print("Session deleted.")
                        else:
                            print("Deletion cancelled.")
                    else:
                        print("Invalid index.")
                except ValueError:
                    print("Please enter a valid number.")
                continue

            if choice == "a" and sessions:
                print(
                    f"\nWARNING: This will delete ALL {len(sessions)} session(s) and cannot be undone!"
                )
                confirm = (
                    input(
                        "Are you absolutely sure you want to delete ALL sessions? [y/N]: "
                    )
                    .strip()
                    .lower()
                )
                if confirm == "y":
                    # Use confirm=False since we already confirmed
                    self.session_manager.delete_all_sessions(confirm=False)
                    print("All sessions deleted.")
                else:
                    print("Deletion cancelled.")
                continue

            print("Invalid choice, please try again.")

    def list_sessions(self) -> List[str]:
        """
        List all sessions available and create a list of labels for each session.

        Returns:
            A list of labels of the form "timestamp | ID" for each session.
        """
        sessions = self.session_manager.list_sessions()
        return [
            f"{ts.isoformat(sep=' ', timespec='seconds')} | {sid}"
            for sid, ts in sessions
        ]

    def create_new_session(self):
        """
        Create a new chat session.
        """
        new_id = self.session_manager.create_session_id()
        self.session_manager.create_new_session(new_id)
        self.session_manager.select_session(new_id)

    def _extract_id_from_label(self, session_label: str) -> str:
        """
        Extract session id from label.

        Args:
            session_label (str): The label of the session of the form "timestamp | ID"

        Returns:
            The session ID.
        """
        try:
            session_id = session_label.split("|", 1)[1].strip()
        except Exception:
            session_id = session_label.strip()

        return session_id

    def select_session(self, session_label: str) -> List[Dict[str, str]]:
        """
        Given a label like 'timestamp | session_id', select that session and return its history.

        Args:
            session_label (str): The label of the chat session that the user wants to select.

        Returns:
            A list of the chat history.
        """
        if not session_label:
            return []  # nothing selected

        # Extract session id from label
        session_id = self._extract_id_from_label(session_label)

        history = self.session_manager.select_session(session_id)

        return history

    def delete_session(self, session_label: str):
        """
        Delete a chat session.

        Args:
            session_label (str): The label of the chat session that the user wants to delete.
        """
        session_id = self._extract_id_from_label(session_label)
        self.session_manager.delete_session(session_id)

    async def run(self):
        """Run the interactive CLI session."""
        print("MADA Multi-Agent Orchestrator")
        print("=" * 50)

        # Startup session menu
        should_continue = self.startup_session_menu()
        if not should_continue:
            print("No session selected. Exiting.")
            return

        # Create orchestrator
        try:
            async with MADAOrchestrator(
                model_config=self.config.model,
                database_config=self.config.database,
                session_manager=self.session_manager,
            ) as orchestrator:
                self.orchestrator = orchestrator

                # Setup agents - continue even if some MCP servers fail
                print("\nInitializing agents and MCP servers...")
                try:
                    status, tools = await orchestrator.initialize_orchestrator(
                        self.config.agents, self.config.mcp_servers
                    )
                    print(f"Status: {status}")
                    print(
                        f"Model: {self.config.model.model} from {self.config.model.provider}"
                    )

                    if tools:
                        print("\nAvailable tools:")
                        for tool in tools[:10]:  # Show first 10 tools
                            print(f"   • {tool}")
                        if len(tools) > 10:
                            print(f"   ... and {len(tools) - 10} more tools")
                    else:
                        print("\nNo MCP tools available (LLM-only agents)")

                except BaseExceptionGroup as eg:
                    print(
                        f"\nWARNING: {len(eg.exceptions)} initialization error(s) occurred. Continuing with available agents..."
                    )
                except Exception as e:
                    print(f"\nWARNING: Initialization error: {e}")
                    print("Continuing with available agents...")
                    import traceback

                    traceback.print_exc()

                print("\nChat with the agents (type 'quit' to exit)")
                print("-" * 50)

                # Interactive chat loop
                while True:
                    try:
                        user_input = input("\nYou: ").strip()

                        if user_input.lower() in ["quit", "exit", "q"]:
                            print("\nGoodbye!")
                            break

                        if not user_input:
                            continue

                        print("\nAgents:")
                        print("-" * 20)

                        # Process message through orchestrator
                        async for response_chunk in orchestrator.process_message(
                            user_input
                        ):
                            print(response_chunk, end="", flush=True)

                        print("\n")

                    except KeyboardInterrupt:
                        print("\n\nGoodbye!")
                        break
                    except Exception as e:
                        print(f"Error processing message: {e}")
                        import traceback

                        traceback.print_exc()

        except BaseExceptionGroup as eg:
            print(
                f"\nFATAL ERROR: Multiple errors creating orchestrator ({len(eg.exceptions)} errors)"
            )
            import traceback

            traceback.print_exc()
            return
        except Exception as e:
            print(f"\nFATAL ERROR: Failed to create orchestrator: {e}")
            import traceback

            traceback.print_exc()
            return

__init__(config)

Initialize the CLI with configuration.

Parameters:

Name	Type	Description	Default
config	AppConfig	Application configuration	required

Source code in src/mada/interfaces/cli/main.py

def __init__(self, config: AppConfig):
    """
    Initialize the CLI with configuration.

    Args:
        config: Application configuration
    """
    self.config = config
    self.orchestrator = None
    self.session_manager = ChatSessionManager(config.database)

create_new_session()

Create a new chat session.

Source code in src/mada/interfaces/cli/main.py

def create_new_session(self):
    """
    Create a new chat session.
    """
    new_id = self.session_manager.create_session_id()
    self.session_manager.create_new_session(new_id)
    self.session_manager.select_session(new_id)

delete_session(session_label)

Delete a chat session.

Parameters:

Name	Type	Description	Default
session_label	str	The label of the chat session that the user wants to delete.	required

Source code in src/mada/interfaces/cli/main.py

def delete_session(self, session_label: str):
    """
    Delete a chat session.

    Args:
        session_label (str): The label of the chat session that the user wants to delete.
    """
    session_id = self._extract_id_from_label(session_label)
    self.session_manager.delete_session(session_id)

list_sessions()

List all sessions available and create a list of labels for each session.

Returns:

Type	Description
List[str]	A list of labels of the form "timestamp | ID" for each session.

Source code in src/mada/interfaces/cli/main.py

def list_sessions(self) -> List[str]:
    """
    List all sessions available and create a list of labels for each session.

    Returns:
        A list of labels of the form "timestamp | ID" for each session.
    """
    sessions = self.session_manager.list_sessions()
    return [
        f"{ts.isoformat(sep=' ', timespec='seconds')} | {sid}"
        for sid, ts in sessions
    ]

run() async

Run the interactive CLI session.

Source code in src/mada/interfaces/cli/main.py

async def run(self):
    """Run the interactive CLI session."""
    print("MADA Multi-Agent Orchestrator")
    print("=" * 50)

    # Startup session menu
    should_continue = self.startup_session_menu()
    if not should_continue:
        print("No session selected. Exiting.")
        return

    # Create orchestrator
    try:
        async with MADAOrchestrator(
            model_config=self.config.model,
            database_config=self.config.database,
            session_manager=self.session_manager,
        ) as orchestrator:
            self.orchestrator = orchestrator

            # Setup agents - continue even if some MCP servers fail
            print("\nInitializing agents and MCP servers...")
            try:
                status, tools = await orchestrator.initialize_orchestrator(
                    self.config.agents, self.config.mcp_servers
                )
                print(f"Status: {status}")
                print(
                    f"Model: {self.config.model.model} from {self.config.model.provider}"
                )

                if tools:
                    print("\nAvailable tools:")
                    for tool in tools[:10]:  # Show first 10 tools
                        print(f"   • {tool}")
                    if len(tools) > 10:
                        print(f"   ... and {len(tools) - 10} more tools")
                else:
                    print("\nNo MCP tools available (LLM-only agents)")

            except BaseExceptionGroup as eg:
                print(
                    f"\nWARNING: {len(eg.exceptions)} initialization error(s) occurred. Continuing with available agents..."
                )
            except Exception as e:
                print(f"\nWARNING: Initialization error: {e}")
                print("Continuing with available agents...")
                import traceback

                traceback.print_exc()

            print("\nChat with the agents (type 'quit' to exit)")
            print("-" * 50)

            # Interactive chat loop
            while True:
                try:
                    user_input = input("\nYou: ").strip()

                    if user_input.lower() in ["quit", "exit", "q"]:
                        print("\nGoodbye!")
                        break

                    if not user_input:
                        continue

                    print("\nAgents:")
                    print("-" * 20)

                    # Process message through orchestrator
                    async for response_chunk in orchestrator.process_message(
                        user_input
                    ):
                        print(response_chunk, end="", flush=True)

                    print("\n")

                except KeyboardInterrupt:
                    print("\n\nGoodbye!")
                    break
                except Exception as e:
                    print(f"Error processing message: {e}")
                    import traceback

                    traceback.print_exc()

    except BaseExceptionGroup as eg:
        print(
            f"\nFATAL ERROR: Multiple errors creating orchestrator ({len(eg.exceptions)} errors)"
        )
        import traceback

        traceback.print_exc()
        return
    except Exception as e:
        print(f"\nFATAL ERROR: Failed to create orchestrator: {e}")
        import traceback

        traceback.print_exc()
        return

select_session(session_label)

Given a label like 'timestamp | session_id', select that session and return its history.

Parameters:

Name	Type	Description	Default
session_label	str	The label of the chat session that the user wants to select.	required

Returns:

Type	Description
List[Dict[str, str]]	A list of the chat history.

Source code in src/mada/interfaces/cli/main.py

def select_session(self, session_label: str) -> List[Dict[str, str]]:
    """
    Given a label like 'timestamp | session_id', select that session and return its history.

    Args:
        session_label (str): The label of the chat session that the user wants to select.

    Returns:
        A list of the chat history.
    """
    if not session_label:
        return []  # nothing selected

    # Extract session id from label
    session_id = self._extract_id_from_label(session_label)

    history = self.session_manager.select_session(session_id)

    return history

startup_session_menu()

Interactive menu on startup to manage chat sessions.

Returns:

Name	Type	Description
bool	bool	True if a session has been selected/created and we should continue, False if the user chose to quit.

Source code in src/mada/interfaces/cli/main.py

def startup_session_menu(self) -> bool:
    """
    Interactive menu on startup to manage chat sessions.

    Returns:
        bool: True if a session has been selected/created and we should continue,
            False if the user chose to quit.
    """
    while True:
        print("\nChat Session Manager")
        print("=" * 50)
        sessions = self.list_sessions()

        if sessions:
            print("Existing sessions:")
            for idx, label in enumerate(sessions, start=1):
                print(f"  {idx}. {label}")
        else:
            print("No existing sessions.")

        print("\nOptions:")
        print("  [n] New session")
        if sessions:
            print("  [s] Select session")
            print("  [d] Delete session")
            print("  [a] Delete ALL sessions")
        print("  [q] Quit")

        choice = input("\nEnter choice: ").strip().lower()

        if choice == "q":
            return False

        if choice == "n":
            self.create_new_session()
            print("Created and selected a new session.")
            return True

        if choice == "s" and sessions:
            try:
                selection = input(
                    "Enter the number of the session to select: "
                ).strip()
                idx = int(selection)
                if 1 <= idx <= len(sessions):
                    label = sessions[idx - 1]
                    history = self.select_session(label)
                    print(f"Selected session: {label}")
                    self._print_history_summary(history)
                    return True
                else:
                    print("Invalid index.")
            except ValueError:
                print("Please enter a valid number.")
            continue

        if choice == "d" and sessions:
            try:
                selection = input(
                    "Enter the number of the session to delete: "
                ).strip()
                idx = int(selection)
                if 1 <= idx <= len(sessions):
                    label = sessions[idx - 1]
                    confirm = (
                        input(f"Are you sure you want to delete '{label}'? [y/N]: ")
                        .strip()
                        .lower()
                    )
                    if confirm == "y":
                        self.delete_session(label)
                        print("Session deleted.")
                    else:
                        print("Deletion cancelled.")
                else:
                    print("Invalid index.")
            except ValueError:
                print("Please enter a valid number.")
            continue

        if choice == "a" and sessions:
            print(
                f"\nWARNING: This will delete ALL {len(sessions)} session(s) and cannot be undone!"
            )
            confirm = (
                input(
                    "Are you absolutely sure you want to delete ALL sessions? [y/N]: "
                )
                .strip()
                .lower()
            )
            if confirm == "y":
                # Use confirm=False since we already confirmed
                self.session_manager.delete_all_sessions(confirm=False)
                print("All sessions deleted.")
            else:
                print("Deletion cancelled.")
            continue

        print("Invalid choice, please try again.")

async_main(config_file) async

Async main entry point for CLI.

Parameters:

Name	Type	Description	Default
config_file	str	The path to the MADA configuration file.	required

Source code in src/mada/interfaces/cli/main.py

async def async_main(config_file: str):
    """
    Async main entry point for CLI.

    Args:
        config_file: The path to the MADA configuration file.
    """
    try:
        # Load configuration
        config = load_config_from_json(config_file)

        # Run CLI
        cli = MADACLIInterface(config)
        await cli.run()

    except FileNotFoundError:
        print(f"Configuration file not found: {config_file}")
        sys.exit(1)
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

main(config_file)

Run MADA in CLI mode.

CONFIG_FILE is the path to the MADA configuration file.

Source code in src/mada/interfaces/cli/main.py

@click.command(
    name="mada-cli",
    context_settings={
        "help_option_names": ["-h", "--help"],
    },
)
@click.argument(
    "config_file",
    type=str,
)
def main(config_file: str) -> None:
    """
    Run MADA in CLI mode.

    CONFIG_FILE is the path to the MADA configuration file.
    """
    asyncio.run(async_main(config_file))