如何用python编写适合AI Agents调用的CLI工具

Published

Tags: AI

CLI工具越来越受欢迎,逐渐变成适合AI Agents调用工具的最佳实践。甚至有人说CLI可以替代MCP。CLI受欢迎的原因有很多种,比较light weight,适合动态加载,编写简单等。编写适合AI Agents使用的CLI和传统给人使用的CLI有些不同。具体来说,应尽量设计CLI符合如下特点:

同时应避免以下几点:

Agent-Friendly CLI的设计原则 #

默认输出要“机器可读” #

优先支持输出json格式,例如:

{
  "ok": true,
  "tasks": [
    {"id": "t1", "title": "pull data", "status": "pending"}
  ],
  "count": 1
}

避免如下输出:

Here are your pending tasks:
- pull data (id=t1)
Total: 1

Exit code必须语义清晰 #

Agent会优先看退出码,其次才看stdout/stderr

建议规则:

stderr和stdout分离 #

提供 --quiet/--verbose/--debug选项 #

对Agent来说这些选项非常有用:

支持--dry-run #

Agent在执行破坏性操作前,先dry-run验证计划。比如在执行删除记录指令时,使用--dry-run:

dbcli delete-record --id 123 --dry-run --json

输出:

{
  "ok": true,
  "action": "delete-record",
  "target": {"id": "123"},
  "will_execute": false
}

支持非交互模式(极其重要) #

Agent不会进行交互式输入,因此要避免在工具执行过程中产生用户交互性输入请求。必要时可提供--non-interactive/--yes/--force等参数避免交互。

输出稳定字段,不要随便改 #

错误也要结构化(不是只有成功才 JSON) #

错误输出参考如下:

{
  "ok": false,
  "error": {
    "code": "TASK_NOT_FOUND",
    "message": "Task 't999' was not found",
    "retryable": false
  }
}

好的错误输入便于Agent判断下一步。

保证幂等性(idempotency) #

以下是一个示范CLI代码。

#!/usr/bin/env python3
"""
Agent-friendly CLI demo (industrial-style starter template)

设计目标:
1) 适合 AI agent 调用(结构化输出、稳定错误码、可预测行为)
2) 同时保留人类可读模式(--output text)
3) 代码结构清晰,便于后续扩展到文件/数据库/API 后端

运行示例:
    python agent_task_cli.py create --id t1 --title "Buy milk"
    python agent_task_cli.py get --id t1
    python agent_task_cli.py list-tasks --status pending --limit 10
    python agent_task_cli.py delete --id t1 --yes
    python agent_task_cli.py delete --id t2 --dry-run --yes
"""

from __future__ import annotations

import json
import sys
from enum import Enum, IntEnum
from typing import Any, Dict, List, Optional

import typer
from pydantic import BaseModel, Field

# -----------------------------------------------------------------------------
# 1) CLI 应用初始化
# -----------------------------------------------------------------------------
# no_args_is_help=True:如果用户直接运行命令不带参数,自动显示帮助信息(更友好)
# add_completion=False:示例项目先关闭 shell completion,减少复杂度(可按需开启)
app = typer.Typer(
    help="Agent-friendly task CLI (demo).",
    no_args_is_help=True,
    add_completion=False,
)

# -----------------------------------------------------------------------------
# 2) “内存数据库”(演示用)
# -----------------------------------------------------------------------------
# 真实项目建议替换为:
# - SQLite(本地持久化)
# - JSON 文件(简单持久化)
# - PostgreSQL / Snowflake / API 服务(生产环境)
TASKS: Dict[str, "Task"] = {}

# -----------------------------------------------------------------------------
# 3) 枚举定义:状态、输出格式、错误码、退出码
# -----------------------------------------------------------------------------


class Status(str, Enum):
    """
    任务状态枚举。
    继承 str + Enum 的好处:
    - 可直接作为字符串参与 JSON 序列化
    - Typer / Pydantic 对这类枚举支持很好
    """
    pending = "pending"
    done = "done"


class OutputFormat(str, Enum):
    """
    输出格式枚举。
    用枚举替代 str,可以防止用户传入非法值(例如 --output banana)。
    """
    json = "json"
    text = "text"


class ExitCode(IntEnum):
    """
    进程退出码(给 shell / 脚本 / agent 判断)。
    尽量稳定,不要随意改。
    """
    OK = 0
    GENERAL_ERROR = 1
    INVALID_ARGUMENT = 2
    NOT_FOUND = 3
    PERMISSION_DENIED = 4
    RETRYABLE_ERROR = 5
    CONFLICT = 6  # 资源冲突,例如重复创建


class ErrorCode(str, Enum):
    """
    机器可读错误码(比 message 更稳定)。
    Agent 应优先看 error.code,再参考 message。
    """
    TASK_ALREADY_EXISTS = "TASK_ALREADY_EXISTS"
    TASK_NOT_FOUND = "TASK_NOT_FOUND"
    CONFIRMATION_REQUIRED = "CONFIRMATION_REQUIRED"
    INVALID_LIMIT = "INVALID_LIMIT"
    INTERNAL_ERROR = "INTERNAL_ERROR"


# -----------------------------------------------------------------------------
# 4) 数据模型定义(Pydantic)
# -----------------------------------------------------------------------------


class Task(BaseModel):
    """
    任务对象模型。
    Field(...) 用于补充描述/默认值等元信息(便于后续生成 schema)。
    """
    id: str = Field(..., description="Unique task ID")
    title: str = Field(..., description="Task title")
    status: Status = Field(default=Status.pending, description="Task status")


class ErrorDetail(BaseModel):
    """
    错误详情结构(结构化错误输出的一部分)。
    """
    code: ErrorCode
    message: str
    retryable: bool = False


class SuccessResponse(BaseModel):
    """
    通用成功响应(最小骨架)。
    具体命令通常会在外层直接构造 dict,或者未来扩展成更细的响应模型。
    """
    ok: bool = True


class ErrorResponse(BaseModel):
    """
    通用失败响应。
    """
    ok: bool = False
    error: ErrorDetail


# -----------------------------------------------------------------------------
# 5) 输出与错误处理工具函数(核心)
# -----------------------------------------------------------------------------

def _to_jsonable(value: Any) -> Any:
    """
    将对象转换为可 JSON 序列化的形式。
    当前主要处理 Pydantic 模型;其余类型原样返回。
    """
    if isinstance(value, BaseModel):
        return value.model_dump()
    return value


def _print_json_stdout(payload: Any) -> None:
    """
    将 JSON 输出到 stdout(标准输出)。
    ensure_ascii=False:保留中文可读性
    """
    print(json.dumps(payload, ensure_ascii=False))


def _print_json_stderr(payload: Any) -> None:
    """
    将 JSON 输出到 stderr(标准错误)。
    错误信息与成功结果分流,便于 agent 稳定解析 stdout。
    """
    print(json.dumps(payload, ensure_ascii=False), file=sys.stderr)


def emit_success(
    payload: Dict[str, Any],
    output: OutputFormat = OutputFormat.json,
) -> None:
    """
    统一成功输出函数。

    参数:
    - payload: 成功结果字典(建议包含 ok=True)
    - output: 输出格式(json/text)

    约定:
    - JSON 模式:输出结构化数据到 stdout
    - TEXT 模式:输出简洁人类可读文本到 stdout
    """
    # 确保有 ok 字段,方便 agent 快速判断
    if "ok" not in payload:
        payload = {"ok": True, **payload}

    # 先把 payload 中可能出现的 Pydantic 模型转成 dict
    normalized = {k: _to_jsonable(v) for k, v in payload.items()}

    if output == OutputFormat.json:
        _print_json_stdout(normalized)
        return

    # Text 模式:给人类看的简单输出(可继续美化)
    # 注意:不要把 text 模式做得过于花哨,否则未来 agent 误用 text 时更难解析
    if "task" in normalized:
        task = normalized["task"]
        print(f"OK: task id={task['id']} title={task['title']} status={task['status']}")
    elif "tasks" in normalized:
        print(f"OK: {normalized.get('count', len(normalized['tasks']))} task(s)")
        for t in normalized["tasks"]:
            print(f"- {t['id']}: {t['title']} [{t['status']}]")
    else:
        print("OK")


def fail(
    *,
    code: ErrorCode,
    message: str,
    exit_code: ExitCode,
    output: OutputFormat = OutputFormat.json,
    retryable: bool = False,
) -> None:
    """
    统一失败输出并退出。

    关键点(agent-friendly):
    1) 错误结构化(ok/error/code/message/retryable)
    2) 错误走 stderr,不污染 stdout
    3) 非 0 exit code 明确表达失败类型
    """
    err = ErrorResponse(
        ok=False,
        error=ErrorDetail(code=code, message=message, retryable=retryable),
    )
    payload = err.model_dump()

    if output == OutputFormat.json:
        _print_json_stderr(payload)
    else:
        # Text 模式下输出简洁文本错误到 stderr
        print(f"ERROR [{code.value}]: {message}", file=sys.stderr)

    raise typer.Exit(int(exit_code))


def require_positive_limit(limit: int, output: OutputFormat) -> None:
    """
    统一参数业务校验:limit 必须 > 0。
    (Typer 能做类型校验,但这种业务规则校验最好单独封装)
    """
    if limit <= 0:
        fail(
            code=ErrorCode.INVALID_LIMIT,
            message="--limit must be a positive integer",
            exit_code=ExitCode.INVALID_ARGUMENT,
            output=output,
            retryable=False,
        )


# -----------------------------------------------------------------------------
# 6) 可复用命令选项(可选优化)
# -----------------------------------------------------------------------------
# 说明:
# 这里为了让代码更清楚,仍在命令函数参数里显式写出 Option。
# 如果项目变大,可以封装公共参数对象或 helper 函数。

# -----------------------------------------------------------------------------
# 7) 命令实现
# -----------------------------------------------------------------------------


@app.command()
def create(
    task_id: str = typer.Option(..., "--id", help="Unique task ID"),
    title: str = typer.Option(..., "--title", help="Task title"),
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
    non_interactive: bool = typer.Option(
        True,
        "--non-interactive/--interactive",
        help="Disable/enable interactive prompts (agent should use non-interactive)",
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Validate and preview action without applying changes",
    ),
):
    """
    创建任务。

    agent 调用建议:
    - 默认使用 --output json(本命令已默认)
    - 自动化场景使用 --non-interactive
    - 有副作用操作前可先 --dry-run
    """
    # 这里 non_interactive 在 create 场景中并非必须,但保留是为了展示“统一命令接口风格”。
    # 在 delete/update 等有确认提示的命令中会更有用。
    _ = non_interactive  # 显式标记“目前未使用”,避免误解为遗漏

    if task_id in TASKS:
        fail(
            code=ErrorCode.TASK_ALREADY_EXISTS,
            message=f"Task '{task_id}' already exists",
            exit_code=ExitCode.CONFLICT,
            output=output,
            retryable=False,
        )

    # dry-run:只返回将要执行的动作,不做实际写入
    if dry_run:
        emit_success(
            {
                "ok": True,
                "dry_run": True,
                "action": "create",
                "task": Task(id=task_id, title=title),  # emit_success 会自动转 dict
            },
            output=output,
        )
        return

    # 实际执行创建
    TASKS[task_id] = Task(id=task_id, title=title)

    emit_success(
        {
            "ok": True,
            "dry_run": False,
            "task": TASKS[task_id],
        },
        output=output,
    )


@app.command()
def get(
    task_id: str = typer.Option(..., "--id", help="Task ID"),
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
):
    """
    获取单个任务。
    """
    task = TASKS.get(task_id)
    if task is None:
        fail(
            code=ErrorCode.TASK_NOT_FOUND,
            message=f"Task '{task_id}' was not found",
            exit_code=ExitCode.NOT_FOUND,
            output=output,
            retryable=False,
        )

    emit_success(
        {
            "ok": True,
            "task": task,
        },
        output=output,
    )


@app.command("list-tasks")
def list_tasks(
    status: Optional[Status] = typer.Option(
        None,
        "--status",
        help="Filter by status (pending/done)",
    ),
    limit: int = typer.Option(
        100,
        "--limit",
        min=1,  # Typer/Click 层面的基本约束(双保险)
        help="Maximum number of tasks to return",
    ),
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
):
    """
    列出任务(支持状态过滤、数量限制)。

    注意:
    - 函数名是 list_tasks,但命令名显式指定为 list-tasks(更符合 CLI 命名习惯)
    """
    # 业务层再做一层校验(双保险)
    require_positive_limit(limit, output)

    items: List[Task] = list(TASKS.values())

    # 可选过滤
    if status is not None:
        items = [t for t in items if t.status == status]

    # 限制返回数量(避免 payload 过大)
    items = items[:limit]

    emit_success(
        {
            "ok": True,
            "tasks": items,  # emit_success 会将每个 Task 转 dict(见下面改进)
            "count": len(items),
            "filters": {
                "status": status.value if status is not None else None,
                "limit": limit,
            },
        },
        output=output,
    )


@app.command()
def complete(
    task_id: str = typer.Option(..., "--id", help="Task ID"),
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Preview action without applying"),
):
    """
    将任务标记为完成(done)。
    """
    task = TASKS.get(task_id)
    if task is None:
        fail(
            code=ErrorCode.TASK_NOT_FOUND,
            message=f"Task '{task_id}' was not found",
            exit_code=ExitCode.NOT_FOUND,
            output=output,
            retryable=False,
        )

    if dry_run:
        preview = task.model_copy(update={"status": Status.done})
        emit_success(
            {
                "ok": True,
                "dry_run": True,
                "action": "complete",
                "task_before": task,
                "task_after": preview,
            },
            output=output,
        )
        return

    task.status = Status.done
    TASKS[task_id] = task

    emit_success(
        {
            "ok": True,
            "dry_run": False,
            "task": task,
        },
        output=output,
    )


@app.command()
def delete(
    task_id: str = typer.Option(..., "--id", help="Task ID"),
    yes: bool = typer.Option(
        False,
        "--yes",
        help="Confirm destructive action without prompt (required in non-interactive mode)",
    ),
    non_interactive: bool = typer.Option(
        True,
        "--non-interactive/--interactive",
        help="Disable/enable prompts (agent should use non-interactive)",
    ),
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Preview delete without applying"),
):
    """
    删除任务(破坏性操作)。
    这里展示更“工业级”的交互控制逻辑:
    - 非交互模式默认开启(适合 agent)
    - 非交互模式下必须显式 --yes 才允许执行删除(除非 --dry-run)
    """
    task = TASKS.get(task_id)
    if task is None:
        fail(
            code=ErrorCode.TASK_NOT_FOUND,
            message=f"Task '{task_id}' was not found",
            exit_code=ExitCode.NOT_FOUND,
            output=output,
            retryable=False,
        )

    # dry-run 允许在没有 --yes 的情况下执行,因为它不会真正删除
    if dry_run:
        emit_success(
            {
                "ok": True,
                "dry_run": True,
                "action": "delete",
                "task": task,
                "will_execute": False,
            },
            output=output,
        )
        return

    # 非交互模式:必须带 --yes,否则拒绝执行(避免 agent 误删)
    if non_interactive and not yes:
        fail(
            code=ErrorCode.CONFIRMATION_REQUIRED,
            message="Destructive action requires --yes in non-interactive mode",
            exit_code=ExitCode.INVALID_ARGUMENT,
            output=output,
            retryable=False,
        )

    # 交互模式:如果没带 --yes,则进行提示确认
    if (not non_interactive) and (not yes):
        confirmed = typer.confirm(f"Delete task '{task_id}'?", default=False)
        if not confirmed:
            fail(
                code=ErrorCode.CONFIRMATION_REQUIRED,
                message="Deletion cancelled by user",
                exit_code=ExitCode.GENERAL_ERROR,
                output=output,
                retryable=False,
            )

    # 真正执行删除
    deleted_task = TASKS.pop(task_id)

    emit_success(
        {
            "ok": True,
            "dry_run": False,
            "deleted": True,
            "task": deleted_task,
        },
        output=output,
    )


@app.command()
def version(
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
):
    """
    输出版本信息(agent 可用于能力探测/健康检查前置步骤)。
    """
    payload = {
        "ok": True,
        "name": "agent-task-cli-demo",
        "version": "0.1.0",
        "schema_version": "1",
    }
    emit_success(payload, output=output)


@app.command()
def health(
    output: OutputFormat = typer.Option(OutputFormat.json, "--output", help="Output format"),
):
    """
    健康检查命令。
    真实项目里可以检查:
    - 数据库连接
    - API 可用性
    - 凭证状态
    - 磁盘空间等
    """
    payload = {
        "ok": True,
        "status": "healthy",
        "store": "in_memory",
        "task_count": len(TASKS),
    }
    emit_success(payload, output=output)


# -----------------------------------------------------------------------------
# 8) 全局异常兜底(可选但推荐)
# -----------------------------------------------------------------------------
# 说明:
# Typer/Click 已经会处理很多参数错误并给出提示;
# 这里的兜底主要处理你业务代码里未预期的异常,避免裸 traceback 污染 agent 的解析流程。
#
# 简化做法:在 __main__ 中 try/except 包一层。
# 更复杂项目可以接入 logging,并把 traceback 写入日志文件,而 CLI 对外只给结构化错误。


def main() -> None:
    """
    程序入口包装函数,便于添加全局异常处理。
    """
    try:
        app()
    except typer.Exit:
        # 这是我们主动抛出的“正常退出”(含非零退出码),直接继续抛出给 Typer 处理
        raise
    except Exception as e:
        # 未预期异常:输出结构化错误到 stderr,并以通用错误码退出
        payload = ErrorResponse(
            ok=False,
            error=ErrorDetail(
                code=ErrorCode.INTERNAL_ERROR,
                message=f"Unhandled exception: {e}",
                retryable=False,
            ),
        ).model_dump()
        _print_json_stderr(payload)
        raise typer.Exit(int(ExitCode.GENERAL_ERROR))


# -----------------------------------------------------------------------------
# 9) 脚本入口
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    main()