Files
2026-07-13 12:49:17 +08:00

732 lines
24 KiB
Python

# Copyright (c) 2025 ByteDance Ltd. and/or its affiliates
# SPDX-License-Identifier: MIT
"""Command Line Interface for Trae Agent."""
import asyncio
import os
import shutil
import subprocess
import sys
import traceback
from pathlib import Path
import click
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from trae_agent.agent import Agent
from trae_agent.utils.cli import CLIConsole, ConsoleFactory, ConsoleMode, ConsoleType
from trae_agent.utils.config import Config, TraeAgentConfig
# Load environment variables
_ = load_dotenv()
console = Console()
def resolve_config_file(config_file: str) -> str:
"""
Resolve config file with backward compatibility.
First tries the specified file, then falls back to JSON if YAML doesn't exist.
"""
if config_file.endswith(".yaml") or config_file.endswith(".yml"):
yaml_path = Path(config_file)
json_path = Path(config_file.replace(".yaml", ".json").replace(".yml", ".json"))
if yaml_path.exists():
return str(yaml_path)
elif json_path.exists():
console.print(f"[yellow]YAML config not found, using JSON config: {json_path}[/yellow]")
return str(json_path)
else:
console.print(
"[red]Error: Config file not found. Please specify a valid config file in the command line option --config-file[/red]"
)
sys.exit(1)
else:
return config_file
def check_docker(timeout=3):
# 1) Check whether the docker CLI is installed
if shutil.which("docker") is None:
return {
"cli": False,
"daemon": False,
"version": None,
"error": "docker CLI not found",
}
# 2) Check whether the Docker daemon is reachable (this makes a real request)
try:
cp = subprocess.run(
["docker", "version", "--format", "{{.Server.Version}}"],
capture_output=True,
text=True,
timeout=timeout,
)
if cp.returncode == 0 and cp.stdout.strip():
return {
"cli": True,
"daemon": True,
"version": cp.stdout.strip(),
"error": None,
}
else:
# The daemon may not be running or permissions may be insufficient
return {
"cli": True,
"daemon": False,
"version": None,
"error": (cp.stderr or cp.stdout).strip(),
}
except Exception as e:
return {"cli": True, "daemon": False, "version": None, "error": str(e)}
def build_with_pyinstaller():
os.system("rm -rf trae_agent/dist")
print("--- Building edit_tool ---")
subprocess.run(
[
"pyinstaller",
"--name",
"edit_tool",
"trae_agent/tools/edit_tool_cli.py",
],
check=True,
)
print("\n--- Building json_edit_tool ---")
subprocess.run(
[
"pyinstaller",
"--name",
"json_edit_tool",
"--hidden-import=jsonpath_ng",
"trae_agent/tools/json_edit_tool_cli.py",
],
check=True,
)
os.system("mkdir trae_agent/dist")
os.system("cp dist/edit_tool/edit_tool trae_agent/dist")
os.system("cp -r dist/json_edit_tool/_internal trae_agent/dist")
os.system("cp dist/json_edit_tool/json_edit_tool trae_agent/dist")
os.system("rm -rf dist")
@click.group()
@click.version_option(version="0.1.0")
def cli():
"""Trae Agent - LLM-based agent for software engineering tasks."""
pass
@cli.command()
@click.argument("task", required=False)
@click.option("--file", "-f", "file_path", help="Path to a file containing the task description.")
@click.option("--provider", "-p", help="LLM provider to use")
@click.option("--model", "-m", help="Specific model to use")
@click.option("--model-base-url", help="Base URL for the model API")
@click.option("--api-key", "-k", help="API key (or set via environment variable)")
@click.option("--max-steps", help="Maximum number of execution steps", type=int)
@click.option("--working-dir", "-w", help="Working directory for the agent")
@click.option("--must-patch", "-mp", is_flag=True, help="Whether to patch the code")
@click.option(
"--config-file",
help="Path to configuration file",
default="trae_config.yaml",
envvar="TRAE_CONFIG_FILE",
)
@click.option("--trajectory-file", "-t", help="Path to save trajectory file")
@click.option("--patch-path", "-pp", help="Path to patch file")
# --- Docker Mode Start ---
@click.option(
"--docker-image",
type=str,
default=None,
help="Specify a Docker image to run the task in a new container",
)
@click.option(
"--docker-container-id",
type=str,
default=None,
help="Attach to an existing Docker container by ID",
)
@click.option(
"--dockerfile-path",
type=click.Path(exists=True, dir_okay=False, resolve_path=True),
default=None,
help="Absolute path to a Dockerfile to build an environment",
)
@click.option(
"--docker-image-file",
type=click.Path(exists=True, dir_okay=False, resolve_path=True),
default=None,
help="Path to a local Docker image file (tar archive) to load.",
)
@click.option(
"--docker-keep",
type=bool,
default=True,
help="Keep or remove the Docker container after finishing the task",
)
# --- Docker Mode End ---
@click.option(
"--console-type",
"-ct",
default="simple",
type=click.Choice(["simple", "rich"], case_sensitive=False),
help="Type of console to use (simple or rich)",
)
@click.option(
"--agent-type",
"-at",
type=click.Choice(["trae_agent"], case_sensitive=False),
help="Type of agent to use (trae_agent)",
default="trae_agent",
)
def run(
task: str | None,
file_path: str | None,
patch_path: str,
provider: str | None = None,
model: str | None = None,
model_base_url: str | None = None,
api_key: str | None = None,
max_steps: int | None = None,
working_dir: str | None = None,
must_patch: bool = False,
config_file: str = "trae_config.yaml",
trajectory_file: str | None = None,
console_type: str | None = "simple",
agent_type: str | None = "trae_agent",
# --- Add Docker Mode ---
docker_image: str | None = None,
docker_container_id: str | None = None,
dockerfile_path: str | None = None,
docker_image_file: str | None = None,
docker_keep: bool = True,
):
"""
Run is the main function of trae. it runs a task using Trae Agent.
Args:
tasks: the task that you want your agent to solve. This is required to be in the input
model: the model expected to be use
working_dir: the working directory of the agent. This should be set either in cli or in the config file
Return:
None (it is expected to be ended after calling the run function)
"""
docker_config: dict[str, str | None] | None = None
if (
sum(
[
bool(docker_image),
bool(docker_container_id),
bool(dockerfile_path),
bool(docker_image_file),
]
)
> 1
):
console.print(
"[red]Error: --docker-image, --docker-container-id, --dockerfile-path, and --docker-image-file are mutually exclusive.[/red]"
)
sys.exit(1)
if dockerfile_path:
docker_config = {"dockerfile_path": dockerfile_path}
console.print(
f"[blue]Docker mode enabled. Building from Dockerfile: {dockerfile_path}[/blue]"
)
elif docker_image_file:
docker_config = {"docker_image_file": docker_image_file}
console.print(
f"[blue]Docker mode enabled. Loading from image file: {docker_image_file}[/blue]"
)
elif docker_container_id:
docker_config = {"container_id": docker_container_id}
console.print(
f"[blue]Docker mode enabled. Attaching to container: {docker_container_id}[/blue]"
)
elif docker_image:
docker_config = {"image": docker_image}
console.print(f"[blue]Docker mode enabled. Using image: {docker_image}[/blue]")
# --- ADDED END ---
# Apply backward compatibility for config file
config_file = resolve_config_file(config_file)
if docker_config:
check_msg = check_docker()
if check_msg["cli"] and check_msg["daemon"] and check_msg["version"]:
print("Docker is configured correctly.")
else:
print(f"Docker is configured incorrectly. {check_msg['error']}")
sys.exit(1)
if not (os.path.exists("trae_agent/dist") and os.path.exists("trae_agent/dist/_internal")):
print("Building tools of Docker mode for the first use, waiting for a few seconds...")
build_with_pyinstaller()
print("Building finished.")
if file_path:
if task:
console.print(
"[red]Error: Cannot use both a task string and the --file argument.[/red]"
)
sys.exit(1)
try:
task = Path(file_path).read_text()
except FileNotFoundError:
console.print(f"[red]Error: File not found: {file_path}[/red]")
sys.exit(1)
elif not task:
console.print(
"[red]Error: Must provide either a task string or use the --file argument.[/red]"
)
sys.exit(1)
config = Config.create(
config_file=config_file,
).resolve_config_values(
provider=provider,
model=model,
model_base_url=model_base_url,
api_key=api_key,
max_steps=max_steps,
)
if not agent_type:
console.print("[red]Error: agent_type is required.[/red]")
sys.exit(1)
# Create CLI Console
console_mode = ConsoleMode.RUN
if console_type:
selected_console_type = (
ConsoleType.SIMPLE if console_type.lower() == "simple" else ConsoleType.RICH
)
else:
selected_console_type = ConsoleFactory.get_recommended_console_type(console_mode)
cli_console = ConsoleFactory.create_console(
console_type=selected_console_type, mode=console_mode
)
# For rich console in RUN mode, set the initial task
if selected_console_type == ConsoleType.RICH and hasattr(cli_console, "set_initial_task"):
cli_console.set_initial_task(task)
# agent = Agent(agent_type, config, trajectory_file, cli_console)
if docker_config is not None:
docker_config["workspace_dir"] = working_dir # now type-safe
# Change working directory if specified
if working_dir:
try:
Path(working_dir).mkdir(parents=True, exist_ok=True)
# os.chdir(working_dir)
console.print(f"[blue]Changed working directory to: {working_dir}[/blue]")
working_dir = os.path.abspath(working_dir)
except Exception as e:
error_text = Text(f"Error changing directory: {e}", style="red")
console.print(error_text)
sys.exit(1)
else:
working_dir = os.getcwd()
console.print(f"[blue]Using current directory as working directory: {working_dir}[/blue]")
# Ensure working directory is an absolute path
if not Path(working_dir).is_absolute():
console.print(
f"[red]Working directory must be an absolute path: {working_dir}, it should start with `/`[/red]"
)
sys.exit(1)
agent = Agent(
agent_type,
config,
trajectory_file,
cli_console,
docker_config=docker_config,
docker_keep=docker_keep,
)
if not docker_config:
try:
os.chdir(working_dir)
except Exception as e:
error_text = Text(f"Error changing directory: {e}", style="red")
console.print(error_text)
sys.exit(1)
try:
task_args = {
"project_path": working_dir,
"issue": task,
"must_patch": "true" if must_patch else "false",
"patch_path": patch_path,
}
# Set up agent context for rich console if applicable
if selected_console_type == ConsoleType.RICH and hasattr(cli_console, "set_agent_context"):
cli_console.set_agent_context(agent, config.trae_agent, config_file, trajectory_file)
# Agent will handle starting the appropriate console
_ = asyncio.run(agent.run(task, task_args))
console.print(f"\n[green]Trajectory saved to: {agent.trajectory_file}[/green]")
except KeyboardInterrupt:
console.print("\n[yellow]Task execution interrupted by user[/yellow]")
console.print(f"[blue]Partial trajectory saved to: {agent.trajectory_file}[/blue]")
sys.exit(1)
except Exception as e:
try:
from docker.errors import DockerException
if isinstance(e, DockerException):
error_text = Text(f"Docker Error: {e}", style="red")
console.print(f"\n{error_text}")
console.print(
"[yellow]Please ensure the Docker daemon is running and you have the necessary permissions.[/yellow]"
)
else:
raise e
except ImportError:
error_text = Text(f"Unexpected error: {e}", style="red")
console.print(f"\n{error_text}")
console.print(traceback.format_exc())
except Exception:
error_text = Text(f"Unexpected error: {e}", style="red")
console.print(f"\n{error_text}")
console.print(traceback.format_exc())
console.print(f"[blue]Trajectory saved to: {agent.trajectory_file}[/blue]")
sys.exit(1)
@cli.command()
@click.option("--provider", "-p", help="LLM provider to use")
@click.option("--model", "-m", help="Specific model to use")
@click.option("--model-base-url", help="Base URL for the model API")
@click.option("--api-key", "-k", help="API key (or set via environment variable)")
@click.option(
"--config-file",
help="Path to configuration file",
default="trae_config.yaml",
envvar="TRAE_CONFIG_FILE",
)
@click.option("--max-steps", help="Maximum number of execution steps", type=int, default=20)
@click.option("--trajectory-file", "-t", help="Path to save trajectory file")
@click.option(
"--console-type",
"-ct",
type=click.Choice(["simple", "rich"], case_sensitive=False),
help="Type of console to use (simple or rich)",
)
@click.option(
"--agent-type",
"-at",
type=click.Choice(["trae_agent"], case_sensitive=False),
help="Type of agent to use (trae_agent)",
default="trae_agent",
)
def interactive(
provider: str | None = None,
model: str | None = None,
model_base_url: str | None = None,
api_key: str | None = None,
config_file: str = "trae_config.yaml",
max_steps: int | None = None,
trajectory_file: str | None = None,
console_type: str | None = "simple",
agent_type: str | None = "trae_agent",
):
"""
This function starts an interactive session with Trae Agent.
Args:
console_type: Type of console to use for the interactive session
"""
# Apply backward compatibility for config file
config_file = resolve_config_file(config_file)
config = Config.create(
config_file=config_file,
).resolve_config_values(
provider=provider,
model=model,
model_base_url=model_base_url,
api_key=api_key,
max_steps=max_steps,
)
if config.trae_agent:
trae_agent_config = config.trae_agent
else:
console.print("[red]Error: trae_agent configuration is required in the config file.[/red]")
sys.exit(1)
# Create CLI Console for interactive mode
console_mode = ConsoleMode.INTERACTIVE
if console_type:
selected_console_type = (
ConsoleType.SIMPLE if console_type.lower() == "simple" else ConsoleType.RICH
)
else:
selected_console_type = ConsoleFactory.get_recommended_console_type(console_mode)
cli_console = ConsoleFactory.create_console(
console_type=selected_console_type,
lakeview_config=config.lakeview,
mode=console_mode,
)
if not agent_type:
console.print("[red]Error: agent_type is required.[/red]")
sys.exit(1)
# Create agent
agent = Agent(agent_type, config, trajectory_file, cli_console)
# Get the actual trajectory file path (in case it was auto-generated)
trajectory_file = agent.trajectory_file
# For simple console, use traditional interactive loop
if selected_console_type == ConsoleType.SIMPLE:
asyncio.run(
_run_simple_interactive_loop(
agent, cli_console, trae_agent_config, config_file, trajectory_file
)
)
else:
# For rich console, start the textual app which handles interaction
asyncio.run(
_run_rich_interactive_loop(
agent, cli_console, trae_agent_config, config_file, trajectory_file
)
)
async def _run_simple_interactive_loop(
agent: Agent,
cli_console: CLIConsole,
trae_agent_config: TraeAgentConfig,
config_file: str,
trajectory_file: str | None,
):
"""Run the interactive loop for simple console."""
while True:
try:
task = cli_console.get_task_input()
if task is None:
console.print("[green]Goodbye![/green]")
break
if task.lower() == "help":
console.print(
Panel(
"""[bold]Available Commands:[/bold]
• Type any task description to execute it
• 'status' - Show agent status
• 'clear' - Clear the screen
• 'exit' or 'quit' - End the session""",
title="Help",
border_style="yellow",
)
)
continue
working_dir = cli_console.get_working_dir_input()
if task.lower() == "status":
console.print(
Panel(
f"""[bold]Provider:[/bold] {agent.agent_config.model.model_provider.provider}
[bold]Model:[/bold] {agent.agent_config.model.model}
[bold]Available Tools:[/bold] {len(agent.agent.tools)}
[bold]Config File:[/bold] {config_file}
[bold]Working Directory:[/bold] {os.getcwd()}""",
title="Agent Status",
border_style="blue",
)
)
continue
if task.lower() == "clear":
console.clear()
continue
# Set up trajectory recording for this task
console.print(f"[blue]Trajectory will be saved to: {trajectory_file}[/blue]")
task_args = {
"project_path": working_dir,
"issue": task,
"must_patch": "false",
}
# Execute the task
console.print(f"\n[blue]Executing task: {task}[/blue]")
# Start console and execute task
console_task = asyncio.create_task(cli_console.start())
execution_task = asyncio.create_task(agent.run(task, task_args))
# Wait for execution to complete
_ = await execution_task
_ = await console_task
console.print(f"\n[green]Trajectory saved to: {trajectory_file}[/green]")
except KeyboardInterrupt:
console.print("\n[yellow]Use 'exit' or 'quit' to end the session[/yellow]")
except EOFError:
console.print("\n[green]Goodbye![/green]")
break
except Exception as e:
error_text = Text(f"Error: {e}", style="red")
console.print(error_text)
async def _run_rich_interactive_loop(
agent: Agent,
cli_console: CLIConsole,
trae_agent_config: TraeAgentConfig,
config_file: str,
trajectory_file: str | None,
):
"""Run the interactive loop for rich console."""
# Set up the agent in the rich console so it can handle task execution
if hasattr(cli_console, "set_agent_context"):
cli_console.set_agent_context(agent, trae_agent_config, config_file, trajectory_file)
# Start the console UI - this will handle the entire interaction
await cli_console.start()
@cli.command()
@click.option(
"--config-file",
help="Path to configuration file",
default="trae_config.yaml",
envvar="TRAE_CONFIG_FILE",
)
@click.option("--provider", "-p", help="LLM provider to use")
@click.option("--model", "-m", help="Specific model to use")
@click.option("--model-base-url", help="Base URL for the model API")
@click.option("--api-key", "-k", help="API key (or set via environment variable)")
@click.option("--max-steps", help="Maximum number of execution steps", type=int)
def show_config(
config_file: str,
provider: str | None,
model: str | None,
model_base_url: str | None,
api_key: str | None,
max_steps: int | None,
):
"""Show current configuration settings."""
# Apply backward compatibility for config file
config_file = resolve_config_file(config_file)
config_path = Path(config_file)
if not config_path.exists():
console.print(
Panel(
f"""[yellow]No configuration file found at: {config_file}[/yellow]
Using default settings and environment variables.""",
title="Configuration Status",
border_style="yellow",
)
)
config = Config.create(
config_file=config_file,
).resolve_config_values(
provider=provider,
model=model,
model_base_url=model_base_url,
api_key=api_key,
max_steps=max_steps,
)
if config.trae_agent:
trae_agent_config = config.trae_agent
else:
console.print("[red]Error: trae_agent configuration is required in the config file.[/red]")
sys.exit(1)
# Display general settings
general_table = Table(title="General Settings")
general_table.add_column("Setting", style="cyan")
general_table.add_column("Value", style="green")
general_table.add_row(
"Default Provider",
str(trae_agent_config.model.model_provider.provider or "Not set"),
)
general_table.add_row("Max Steps", str(trae_agent_config.max_steps or "Not set"))
console.print(general_table)
# Display provider settings
provider_config = trae_agent_config.model.model_provider
provider_table = Table(title=f"{provider_config.provider.title()} Configuration")
provider_table.add_column("Setting", style="cyan")
provider_table.add_column("Value", style="green")
provider_table.add_row("Model", trae_agent_config.model.model or "Not set")
provider_table.add_row("Base URL", provider_config.base_url or "Not set")
provider_table.add_row("API Version", provider_config.api_version or "Not set")
provider_table.add_row(
"API Key",
(
f"Set ({provider_config.api_key[:4]}...{provider_config.api_key[-4:]})"
if provider_config.api_key
else "Not set"
),
)
provider_table.add_row("Max Tokens", str(trae_agent_config.model.max_tokens))
provider_table.add_row("Temperature", str(trae_agent_config.model.temperature))
provider_table.add_row("Top P", str(trae_agent_config.model.top_p))
if trae_agent_config.model.model_provider.provider == "anthropic":
provider_table.add_row("Top K", str(trae_agent_config.model.top_k))
console.print(provider_table)
@cli.command()
def tools():
"""Show available tools and their descriptions."""
from .tools import tools_registry
tools_table = Table(title="Available Tools")
tools_table.add_column("Tool Name", style="cyan")
tools_table.add_column("Description", style="green")
for tool_name in tools_registry:
try:
tool = tools_registry[tool_name]()
tools_table.add_row(tool.name, tool.description)
except Exception as e:
tools_table.add_row(tool_name, f"[red]Error loading: {e}[/red]")
console.print(tools_table)
def main():
"""Main entry point for the CLI."""
cli()
if __name__ == "__main__":
main()