从零做一个自己的 CLI
引言
装过 Claude Code 或 Codex CLI 的人都有同感:终端里敲一行命令,在哪个项目文件夹都能用,不用点开浏览器,也不用先找脚本在哪。
这篇文章做两件事:第一,给已经能跑的Agent加一层CLI 外壳,让它像上述工具一样随处可用;第二,讲清楚两种流式输出在实际开发里分别适合什么场景。
文中的完整示例开源在 GitHub:https://github.com/SWUSTcyt/travel-cli(一个带工具调用与流式输出的 Agent CLI,用旅游规划作演示场景)。会一点 Python、有 API Key 就能跟着玩;安装、环境变量与运行命令见仓库根目录README.md,正文不展开pip install。
文章目录
- 从零做一个自己的 CLI
- 引言
- 一、CLI 速览:是什么、为什么要做
- 二、外层封装:Typer 小例子 + 两层结构
- 2.1 先建立直觉
- 2.2 外壳长什么样?
- 三、Agent 核心:注册工具、创建 Agent、Loop
- 3.1 注册工具:给 Agent 一双手
- 3.2 Loop:想一步、做一步、再想一步
- 3.3 外壳怎么接到核心?
- 四、两种流式:区别与开发中怎么选
- 4.1 阶段流 `--stream stage`
- 4.2 字段流 `--stream instant`
- 4.3 怎么选?一张表
- 结语
一、CLI 速览:是什么、为什么要做
CLI(Command Line Interface,命令行界面)就是用键盘输入命令、在终端里拿结果。和 GUI(点按钮的图形界面)相对:
| GUI | CLI | |
|---|---|---|
| 操作 | 鼠标点 | 键盘敲命令 |
| 例子 | 网页 ChatGPT | travel run "杭州一日游" |
一条命令通常长这样:
travel run"帮我规划杭州一日游"--streamstage │ │ │ │ 命令名 子命令 你的需求 可选参数为什么要做成 CLI?两点就够:
- 随处可用——像 Claude Code,
pip install一次后,任意目录都能敲,不必cd到脚本文件夹。 - 好分享、好复现——教程里写一行命令,读者复制就能跑。
如果只是自己试 Agent,用python main.py或input()完全没问题;要发 GitHub、写教程、给朋友用,就需要再包一层 CLI。Agent 逻辑不用重写,只是加外壳。示例项目travel-cli用「旅游规划」作演示,但外壳封装、Loop、流式选型适用于任意 Agent 场景。
二、外层封装:Typer 小例子 + 两层结构
2.1 先建立直觉
你在终端敲:travel run "杭州一日游" ↓ cli.py(Typer 外壳)—— 解析命令和参数 ↓ runner.py + loops/(Agent 核心)—— 搜网页、查地图、多轮推理重点:Typer 和 Agently没有关系。Typer 只管「用户敲了什么」;Agent 框架只管「怎么干活」。我们在现有 Agent 核心外面,套了一层通用的 Python CLI 封装。
[图:两层结构示意——上方 Typer 外壳,下方 Agent 核心]
2.2 外壳长什么样?
摘自 travel-cli 的cli.py(精简版):
importasyncioimporttyperfromtravel_cli.runnerimportrun_travel app=typer.Typer()@app.command("run")defrun_cmd(request:str,stream:str|None=None,):asyncio.run(run_travel(request,stream=stream))if__name__=="__main__":app()逐行白话(代码少,但语法容易懵):
| 代码 | 什么意思 |
|---|---|
import typer | 引入第三方库 Typer,专门用来写 CLI |
app = typer.Typer() | 创建一个 CLI「应用」对象 |
@app.command("run") | 装饰器:把紧挨着的函数注册成子命令run;你在终端敲travel run,就会进这个函数 |
request: str | 命令里"杭州一日游"这类文字,会传进这个参数 |
stream: str | None = None | 可选参数;不传就走默认批处理,传stage/instant走流式 |
asyncio.run(run_travel(...)) | Agent 内部是异步的(async def),Typer 函数是同步的,用这一行把两者接上 |
app() | 启动 CLI,开始解析你在终端输入的内容 |
再配一行pyproject.toml:
[project.scripts] travel = "travel_cli.cli:app"执行pip install -e .后,系统里就多了一个全局命令travel——和装 Claude Code 后在任意目录敲claude是同一类体验。
对比:
- 没有外壳:每次
cd进项目目录,再python xxx.py - 有外壳:任意目录
travel run "..."
三、Agent 核心:注册工具、创建 Agent、Loop
外壳只负责「接命令」。真正干活的在agent.py和loops/里。
3.1 注册工具:给 Agent 一双手
摘自agent.py:
fromagentlyimportAgentlyfromagently.builtins.actionsimportBrowse,Searchasyncdefbuild_travel_agent(workdir:str):agent=Agently.create_agent()agent.set_agent_prompt("system","你是旅游辅助助手……")Search().register_actions(agent.action)Browse().register_actions(agent.action)awaitagent.async_use_mcp(f"https://mcp.amap.com/mcp?key=...")agent.action.register_bash_sandbox_action(allowed_workdir_roots=[workdir],...)returnagent四步理解:
create_agent()—— 创建一个 Agent 实例。set_agent_prompt—— 设定角色与行为边界(示例里是旅游助手,可按业务替换)。- 注册工具—— 搜索、读网页、高德地图 MCP、受控 bash,相当于「能调用的能力」。
use_actions(在 runner 里调用)—— 从注册表里激活要用的工具。
3.2 Loop:想一步、做一步、再想一步
默认travel run走loops/batch.py。核心是Reason → Act → Reason → …循环:
flow=TriggerFlow(name="travel_agent_loop")@flow.chunkasyncdefreason(data):# 模型看用户需求和历史,决定:调工具,还是给出最终回答decision=awaitresponse.async_get_data(...)print(f"[Plan · Step{step}] …")# 终端里看到的 Plan@flow.chunkasyncdefact(data):# 按模型指定,真正执行 search / browse / 地图 等result=awaitagent.action.async_execute_action(name,kwargs)print(f"[Execute] …")# 终端里看到的 Execute终端里的[Plan]、[Execute]就是这样来的。默认模式下 Loop 全部跑完,最后汇总[最终回答]——适合「我不着急,只要结果」。
3.3 外壳怎么接到核心?
runner.py里的run_travel()做编排:
agent=awaitbuild_travel_agent(workdir)activate_travel_tools(agent)ifstreamisNone:result=awaitrun_batch_loop(agent,request)# 默认elifstream=="stage":flow=awaitbuild_stage_stream_flow(agent,request)result=awaitconsume_runtime_stream(execution,"stage")else:flow=awaitbuild_instant_stream_flow(agent,request)...项目结构(壳 vs 核):
travel_cli/ ├── cli.py ← 外壳:Typer 命令 ├── runner.py ← 编排:选哪种跑法 ├── agent.py ← 核心:工具注册 ├── loops/ ← 核心:Loop + 流式逻辑 └── display/ ← 核心:流式事件打印到终端四、两种流式:区别与开发中怎么选
Agent 任务有时要跑一两分钟。干等像「卡死了」;流式就是边跑边把进度推到终端(类似 ChatGPT 逐字输出)。
travel-cli 提供两种流式,对应两个命令行开关。
4.1 阶段流--stream stage
观察粒度:一整步 Plan、一整步 Execute。
# loops/stage_stream.py —— 每完成一步,往流里推一条事件awaitdata.async_put_into_stream({"phase":"plan",# 或 "execute""step":step,"reasoning":"...",})# display/console.py —— 终端边读边打印asyncforrawinexecution.get_async_runtime_stream(...):print_stream_event(raw)终端示例:
[Plan · Step 1] tool: 先搜索杭州攻略… [Execute] search → … [Plan · Step 2] tool: 阅读网页…实际开发适合:进度条、日志面板、运维监控——用户只需知道「现在在搜索 / 查地图」。
4.2 字段流--stream instant
观察粒度:更细——模型结构化输出里,哪个字段先写好就先推送。
asyncforstreaming_datainresponse.get_async_generator(type="instant"):ifstreaming_data.event_type=="done":awaitdata.async_put_into_stream({"phase":"plan_field","path":streaming_data.path,# 如 type、tool_name"value_preview":str(streaming_data.value)[:100],})终端会多出行如:
↳ field 1.type = tool ↳ field 1.tool_name = search [Plan · Step 1] tool: …实际开发适合:调试模型决策、精细 UI(提前显示「即将调用 search」)、可观测性要求高的场景。
4.3 怎么选?一张表
默认travel run | --stream stage | --stream instant | |
|---|---|---|---|
| 看到什么 | 跑完后汇总 | 每步 Plan / Execute | 每字段 + 每步 |
| 复杂度 | 最低 | 中等 | 较高 |
| 典型场景 | 脚本、CI、批处理 | 终端进度、日志 | 调试、精细 UI |
| 何时够用 | 只要最终结果 | 任务长、要「还在跑」 | 要看模型字段级决策 |
[图:默认 run 与--stream stage终端输出对比截图]
经验法则:先默认跑通;用户-facing 的 CLI 加stage;排查模型行为或做高级 UI 再上instant。
结语
总结一下:CLI 是外壳,Agent 是核心——Typer 让你像 Claude Code 一样随处敲命令;Agently 负责注册工具、跑 Loop。流式不是必选项:默认模式够用就上默认;要给用户「还在跑」的反馈,用--stream stage;要更细的可观测性,用--stream instant。
示例代码:https://github.com/SWUSTcyt/travel-cli ,欢迎 Star。留言区也可以说说:你想给什么样的 Agent 加 CLI?