从零做一个自己的 CLI：用 Python 给 Typer 加上 Agent 流式输出

1. 为什么我要给 Agent 套一层 Typer CLI如果你已经写过一个能跑的 Python Agent 脚本大概率经历过这个阶段每次测试都要cd到项目目录改一下main.py里的输入字符串再python main.py。想换个问题回去改代码。想给朋友演示先教他怎么配环境变量。这种体验离「工具」还差得远。CLICommand Line Interface命令行界面解决的就是这件事。它让你在任意目录敲一行travel run 帮我规划杭州两日游程序就能接住参数、跑完 Agent、把结果打回终端。装过 Claude Code 或者 Codex CLI 的人应该有体感pip install一次之后终端里随时可用不用点浏览器也不用先找脚本在哪。这篇文章聚焦一个具体问题用 Python Typer 从零搭一个带 Agent 能力的 CLI并且把流式输出在终端里的实时渲染和中断处理做扎实。适合已经会一点 Python、手里有模型 API Key、想让自己的 Agent 从「脚本」变成「命令」的人。我会给出可复制的 Typer 命令骨架、流式回调配置、本地验证步骤以及接入统一 Key/API 通道的方式最后跑通一个可交互的命令行 Agent 原型。Typer 本身只管「用户敲了什么」它和 Agent 框架没有耦合关系。你可以把它理解成一个外壳外壳负责解析命令和参数内核负责调模型、跑工具、多轮推理。两层分开写后面换模型、换工具、换流式策略外壳都不用动。2. TaoToken 前置统一 Key 与 API 通道在写代码之前先把模型服务的接入通道定下来。我这边用的是 TaoToken 的统一 Key/API 通道好处是 CLI 里只需要维护一个base_url和一个api_key不用为每个模型单独改代码。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要提前准备两样东西第一一个可用的 API Key。登录后在控制台的 API Keys 页面创建地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。创建后复制保存后面写进环境变量。第二确认你要调用的模型名。不同模型在流式输出上的行为略有差异建议先在模型对话页面手动试一次确认能正常返回地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。环境变量这样配Linux/macOS 写进~/.bashrc或~/.zshrcWindows 用系统环境变量面板export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意不要把 Key 硬编码进cli.py提交到 Git。用os.environ.get()读取缺失时给出明确报错比跑一半崩在请求里好排查得多。如果你后面要做长期编码类 Agent、需要更稳定的额度和并发可以了解下 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入细节和参数说明在文档里地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 可复制配置Typer 骨架 流式回调3.1 项目结构与依赖先建目录结构建议这样分外壳和内核分开agent_cli/ ├── pyproject.toml ├── agent_cli/ │ ├── __init__.py │ ├── cli.py # Typer 外壳 │ ├── runner.py # 编排选批处理还是流式 │ ├── agent.py # Agent 核心模型调用 │ └── display.py # 流式事件打印pyproject.toml里声明依赖和入口命令[project] name agent-cli version 0.1.0 dependencies [ typer0.12.0, openai1.30.0, rich13.0.0, ] [project.scripts] agent agent_cli.cli:app [build-system] requires [setuptools68] build-backend setuptools.build_metapip install -e .之后系统里就多了一个全局命令agent和装完 Claude Code 后任意目录敲claude是同一类体验。3.2 Typer 命令骨架cli.py的核心是三个东西Typer()应用对象、app.command()装饰器、以及把同步函数和异步内核接起来的asyncio.run()。import asyncio import typer from agent_cli.runner import run_agent app typer.Typer(help一个带流式输出的命令行 Agent) app.command(run) def run_cmd( request: str typer.Argument(..., help你的问题或任务描述), stream: bool typer.Option(False, --stream, help开启流式输出), model: str typer.Option(gpt-4o-mini, --model, help模型名), ): 向 Agent 发起一次请求。 asyncio.run(run_agent(request, streamstream, modelmodel)) if __name__ __main__: app()逐行说明几个容易懵的点。typer.Argument(...)表示这是必填的位置参数终端里agent run 你好中的你好会传进request。typer.Option(False, --stream)是可选开关不传就是False传了--stream就是True。asyncio.run()这一行是关键Agent 内核通常是async def而 Typer 注册的函数是同步的用这一行把两者桥接起来。3.3 流式回调把 token 实时推到终端流式的本质是「边生成边打印」。用 OpenAI 兼容接口时开启streamTrue后返回的是一个迭代器每个 chunk 里带着增量文本。下面这段是agent.py里的核心import os from openai import AsyncOpenAI client AsyncOpenAI( api_keyos.environ.get(TAOTOKEN_API_KEY), base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) async def stream_chat(messages, model, on_token): 流式请求每收到一个增量就调用 on_token 回调。 stream await client.chat.completions.create( modelmodel, messagesmessages, streamTrue, ) full [] async for chunk in stream: delta chunk.choices[0].delta.content if delta: full.append(delta) on_token(delta) # 实时渲染 return .join(full)on_token是一个回调函数负责把增量写到终端。用rich的Console可以避免手动处理换行和刷新from rich.console import Console console Console() def make_printer(): def on_token(text: str): console.print(text, end, soft_wrapTrue) return on_tokenend保证不换行soft_wrapTrue让长文本按终端宽度自动折行。跑起来的效果就是文字一个词一个词往外冒而不是等十几秒后一次性刷出来。3.4 中断处理CtrlC 要优雅退出流式输出最容易被忽略的是中断。用户按CtrlC时如果直接抛KeyboardInterrupt可能留下半截输出、没关闭的连接、甚至脏状态。正确做法是捕获它打印一个换行然后干净退出。import asyncio async def run_agent(request, stream, model): messages [{role: user, content: request}] printer make_printer() try: if stream: await stream_chat(messages, model, printer) else: result await batch_chat(messages, model) console.print(result) except KeyboardInterrupt: console.print(\n[yellow]已中断[/yellow]) except Exception as e: console.print(f[red]请求失败{e}[/red]) finally: console.print() # 收尾换行避免 shell 提示符黏在输出后面finally里那个换行看着不起眼但少了它中断后 shell 提示符会直接贴在最后一个字符后面体验很糟。这是我自己踩过的坑。4. 验证请求与成功结果配置写完先做最小验证。第一步确认命令注册成功pip install -e . agent --help应该看到run子命令和--stream、--model两个选项。第二步跑一次非流式请求agent run 用一句话解释什么是 CLI终端会等几秒然后一次性打印完整回答。第三步开流式agent run 写一段 100 字的 Python 简介 --stream这次应该看到文字逐段出现。如果两次都能正常返回说明 Key、base_url、模型名三者都对上了。再验证一下中断。跑一个长回答在输出到一半时按CtrlCagent run 详细讲讲 Python 的异步编程 --stream预期结果是输出立刻停止打印一行「已中断」然后干净地回到 shell 提示符没有报错堆栈。这一步过了说明流式渲染和中断处理都到位了。如果你在验证模型行为时想快速对比不同模型的流式表现可以直接在模型对话页面切换测试地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 不用每次都改代码重跑。5. 本篇常见错排查报错一openai.AuthenticationError: Incorrect API key九成是环境变量没生效。先确认当前 shell 里能读到echo $TAOTOKEN_API_KEY如果为空说明export写在了没被加载的文件里或者开了新终端没重新 source。Windows 下注意用户变量和系统变量的区别改完要重启终端。报错二TypeError: NoneType object is not iterable通常出现在async for chunk in stream这一行原因是create()调用时漏了streamTrue返回的是普通响应对象而不是迭代器。检查stream_chat里的参数有没有传对。报错三流式输出卡住不动最后一次性全出来这是缓冲问题。Python 的print在非交互环境下会缓冲rich的Console一般没这个问题但如果你用的是原生print加flushTrueprint(text, end, flushTrue)报错四RuntimeError: asyncio.run() cannot be called from a running event loop说明你在一个已经有事件循环的环境里调了asyncio.run()比如 Jupyter Notebook。CLI 场景下不会遇到但如果你把run_cmd拿去 notebook 里测就会撞上。解决办法是 notebook 里直接用await run_agent(...)不要走 Typer 那层。报错五CtrlC后进程没退出卡在finally检查finally里有没有阻塞操作比如等待网络请求完成。中断处理里只做打印和清理不要放耗时逻辑。6. 继续往下走到这里一个可交互的命令行 Agent 原型就跑通了Typer 负责接命令runner.py负责编排agent.py负责流式请求display.py负责渲染。想加工具调用就在agent.py里扩展 messages 的构造逻辑想加多轮对话就在run_agent外面套一个循环把历史消息带上。如果你打算把这个 CLI 用在长期编码或 Agent 工作流上建议把 Key 管理和额度规划单独理一下Coding Plan 的说明在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入参数和流式相关的字段细节文档里写得更全地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。新建 Key 记得去 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。最后留一个实用技巧把--stream设成默认开启再加一个--no-stream用来关掉。日常交互几乎总是想要流式批处理脚本才需要关。这样敲命令时少打一个参数手感会顺很多。