cli.py

#!/usr/bin/env python3
"""
命令行入口
===========
用户直接在终端用自然语言提问，AI 理解意图并执行。

用法：
    python cli.py

启动后进入交互模式，输入目标，AI 自动规划并执行。
输入 quit 或 exit 退出。
"""
import sys
import os

# 确保项目根目录在 path 中
_PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)

from model.agent.debug import enable_debug, disable_debug, is_debug, log, separator
from model.agent.plan_agent import make_plan
from model.agent.executor import AgentExecutor
from model.agent.audit import format_record


def print_banner():
    print()
    print("=" * 50)
    print("  MCP Agent CLI - AI 命令行助手")
    print("  输入你的目标，AI 自动规划并执行")
    print("  输入 quit / exit 退出")
    print("  输入 debug 切换调试日志")
    print("=" * 50)
    print()


def print_plan(plan):
    """打印计划"""
    print()
    print(f"  目标: {plan.goal}")
    print(f"  计划 ID: {plan.plan_id}")
    print(f"  步骤数: {len(plan.steps)}")
    separator("计划详情")
    for step in plan.steps:
        risk_tag = f"[{step.risk_level}]" if step.risk_level != "low" else ""
        confirm_tag = " (需确认)" if step.requires_confirmation else ""
        print(f"  {step.step_id}. {step.description}")
        print(f"     工具: {step.tool_name}")
        if step.tool_args:
            for k, v in step.tool_args.items():
                val_str = str(v)
                if len(val_str) > 80:
                    val_str = val_str[:80] + "..."
                print(f"     参数 {k}: {val_str}")
        if risk_tag or confirm_tag:
            print(f"     {risk_tag}{confirm_tag}")
    print()


def print_result(record):
    """打印执行结果"""
    separator("执行结果")
    print()
    print(f"  总体状态: {record.final_status}")
    for sr in record.steps:
        status_icon = {
            "success": "[OK]",
            "failed": "[FAIL]",
            "blocked": "[BLOCKED]",
            "skipped": "[SKIP]",
        }.get(sr.status, f"[{sr.status}]")
        print(f"  步骤 {sr.step_id}: {status_icon} {sr.description}")
        if sr.result:
            result_str = str(sr.result)
            if len(result_str) > 200:
                result_str = result_str[:200] + "..."
            print(f"    结果: {result_str}")
        if sr.error:
            print(f"    错误: {sr.error}")
    print()
    separator()
    print()


def main():
    # 默认开启调试日志
    enable_debug()

    print_banner()
    log("系统", "调试日志已开启")

    executor = AgentExecutor()

    while True:
        try:
            user_input = input("你> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n  再见！")
            break

        if not user_input:
            continue

        if user_input.lower() in ("quit", "exit", "q"):
            print("  再见！")
            break

        if user_input.lower() == "debug":
            if is_debug():
                disable_debug()
                print("  调试日志已关闭")
            else:
                enable_debug()
                print("  调试日志已开启")
            continue

        # 1. AI 分析意图，生成计划
        separator("AI 分析")
        log("AI", f"开始分析用户目标: {user_input}")

        try:
            plan = make_plan(user_input)
        except Exception as e:
            print(f"  计划生成失败: {e}")
            continue

        # 2. 展示计划
        print_plan(plan)

        if not plan.steps:
            print("  没有生成任何步骤。")
            continue

        # 3. 检查是否有需要确认的高风险步骤
        high_risk_steps = [
            s for s in plan.steps
            if s.requires_confirmation and s.risk_level == "high"
        ]

        if high_risk_steps:
            print(f"  [!] 有 {len(high_risk_steps)} 个高风险步骤需要确认：")
            for s in high_risk_steps:
                print(f"    步骤 {s.step_id}: {s.description} (工具: {s.tool_name})")
            print()
            confirm = input("  是否确认执行？(y/n) > ").strip().lower()
            if confirm == "y":
                for s in high_risk_steps:
                    executor.confirm_step(plan.plan_id, s.step_id)
                log("执行", f"已确认 {len(high_risk_steps)} 个高风险步骤")
            else:
                print("  已取消执行。")
                continue

        # 4. 执行
        separator("执行")
        log("执行", f"开始执行计划 {plan.plan_id}，共 {len(plan.steps)} 步")

        try:
            record = executor.execute(plan)
        except Exception as e:
            print(f"  执行失败: {e}")
            continue

        # 5. 展示结果
        print_result(record)


if __name__ == "__main__":
    main()