xtrm-agent/xtrm_agent/main.py

"""Entry point — typer CLI."""

from __future__ import annotations

import asyncio
from pathlib import Path
from typing import Optional

import typer
from loguru import logger
from rich.console import Console
from rich.table import Table

app = typer.Typer(name="xtrm-agent", help="Multi-agent AI automation system")
console = Console()


@app.command()
def chat(
    message: Optional[str] = typer.Option(None, "-m", "--message", help="Single-shot message"),
    agent: Optional[str] = typer.Option(None, "--agent", help="Target agent name"),
    config_path: str = typer.Option("config.yaml", "--config", "-c", help="Config file path"),
) -> None:
    """Interactive chat REPL or single-shot message."""
    asyncio.run(_chat(message, agent, config_path))


async def _chat(message: str | None, agent: str | None, config_path: str) -> None:
    from xtrm_agent.config import load_config
    from xtrm_agent.orchestrator import Orchestrator

    config = load_config(config_path)
    orch = Orchestrator(config, interactive=True)
    await orch.setup()

    # Start orchestrator loop in background
    loop_task = asyncio.create_task(orch.run_loop())

    try:
        if message:
            # Single-shot mode
            from xtrm_agent.channels.cli import run_single_message

            outbound_queue = orch.bus.subscribe_outbound("cli")
            result = await run_single_message(orch.bus, message, agent, outbound_queue)
            console.print(result)
        else:
            # Interactive REPL
            from xtrm_agent.channels.cli import CLIChannel

            cli = CLIChannel(
                bus=orch.bus,
                default_agent=config.channels.cli.default_agent,
            )
            await cli.start()
    finally:
        loop_task.cancel()
        await orch.stop()


@app.command()
def serve(
    config_path: str = typer.Option("config.yaml", "--config", "-c", help="Config file path"),
) -> None:
    """Run all agents + Discord bot (production mode)."""
    asyncio.run(_serve(config_path))


async def _serve(config_path: str) -> None:
    from xtrm_agent.config import load_config
    from xtrm_agent.orchestrator import Orchestrator

    config = load_config(config_path)
    orch = Orchestrator(config, interactive=False)
    await orch.setup()

    tasks = [asyncio.create_task(orch.run_loop())]

    # Start Discord if enabled
    if config.channels.discord.enabled:
        from xtrm_agent.channels.discord import DiscordChannel

        discord_channel = DiscordChannel(
            bus=orch.bus,
            token_env=config.channels.discord.token_env,
            default_agent=config.channels.discord.default_agent,
            allowed_users=config.channels.discord.allowed_users,
        )
        tasks.append(asyncio.create_task(discord_channel.start()))

    logger.info("xtrm-agent serving — press Ctrl+C to stop")

    try:
        await asyncio.gather(*tasks)
    except (KeyboardInterrupt, asyncio.CancelledError):
        pass
    finally:
        await orch.stop()


@app.command()
def status(
    config_path: str = typer.Option("config.yaml", "--config", "-c", help="Config file path"),
) -> None:
    """Show configuration, agents, tools, and MCP servers."""
    from xtrm_agent.config import load_config

    config = load_config(config_path)

    console.print("[bold]xtrm-agent status[/bold]\n")

    # Providers
    table = Table(title="LLM Providers")
    table.add_column("Name")
    table.add_column("Model")
    table.add_column("Provider")
    for name, prov in config.llm.providers.items():
        table.add_row(name, prov.model, prov.provider)
    console.print(table)
    console.print()

    # Agents
    table = Table(title="Agents")
    table.add_column("Name")
    table.add_column("Path")
    for name, path in config.agents.items():
        table.add_row(name, path)
    console.print(table)
    console.print()

    # Channels
    table = Table(title="Channels")
    table.add_column("Channel")
    table.add_column("Enabled")
    table.add_column("Default Agent")
    table.add_row("CLI", str(config.channels.cli.enabled), config.channels.cli.default_agent)
    table.add_row("Discord", str(config.channels.discord.enabled), config.channels.discord.default_agent)
    console.print(table)
    console.print()

    # MCP Servers
    if config.mcp_servers:
        table = Table(title="MCP Servers")
        table.add_column("Name")
        table.add_column("Type")
        for name, srv in config.mcp_servers.items():
            srv_type = "stdio" if srv.command else "http" if srv.url else "unknown"
            table.add_row(name, srv_type)
        console.print(table)
    else:
        console.print("[dim]No MCP servers configured[/dim]")

    # Tool policies
    console.print()
    console.print(f"[bold]Tool Workspace:[/bold] {config.tools.workspace}")
    console.print(f"[bold]Auto-approve:[/bold] {', '.join(config.tools.auto_approve)}")
    console.print(f"[bold]Require approval:[/bold] {', '.join(config.tools.require_approval)}")


@app.command()
def agents(
    config_path: str = typer.Option("config.yaml", "--config", "-c", help="Config file path"),
) -> None:
    """List all agent definitions and their configuration."""
    from xtrm_agent.config import load_config, parse_agent_file

    config = load_config(config_path)

    for name, agent_path in config.agents.items():
        p = Path(agent_path)
        if not p.is_absolute():
            p = Path.cwd() / p

        console.print(f"\n[bold]{name}[/bold]")
        if p.exists():
            cfg = parse_agent_file(p)
            console.print(f"  Provider: {cfg.provider}")
            console.print(f"  Model: {cfg.model or '(default)'}")
            console.print(f"  Temperature: {cfg.temperature}")
            console.print(f"  Max iterations: {cfg.max_iterations}")
            console.print(f"  Tools: {', '.join(cfg.tools) if cfg.tools else '(all)'}")
        else:
            console.print(f"  [red]File not found: {p}[/red]")


if __name__ == "__main__":
    app()