企业官网建设流程全解析

claude-tap实时Live模式深度解析：SSE广播如何实现AI编程代理流量实时监控？

【免费下载链接】claude-tapIntercept and inspect Coding Agent API traffic from Claude Code, Codex CLI, Gemini CLI, Cursor CLI, OpenCode, Kimi/Kimi Code, Pi, and Hermes in a local trace viewer.项目地址: https://gitcode.com/gh_mirrors/cl/claude-tap

claude-tap 是一款运行在本地的AI 编程代理流量查看器（trace viewer），它能拦截并完整记录 Claude Code、Codex CLI、Gemini CLI、Cursor CLI、OpenCode、Kimi、Pi、Hermes 等 Coding Agent 与模型 API 之间的每一次请求和响应。其中最有代表性的功能就是实时 Live 模式：一边让 Agent 干活，一边在浏览器里逐条"看到"它的 API 流量。这篇文章带你深入剖析它背后的核心机制——SSE（Server-Sent Events）广播是如何把流量实时推送到监控页面的。

claude-tap 是什么：给 AI 编程代理装上"流量仪表盘"

平时你让 Claude Code 写代码，只能看到终端里的文字输出，但它到底向模型 API 发了什么 Prompt、带了哪些工具定义、每次调用消耗多少 Token、有没有报错，完全是黑盒。

claude-tap 以本地代理的方式插入在 Agent 和 API 之间：

• 所有出入流量先经过它，被完整记录为一条"轨迹记录（trace record）"；
• 敏感请求头（如authorization、x-api-key）会被自动脱敏，不会明文落盘；
• 记录写入本地 SQLite 存储，同时可以实时推送到浏览器查看器。

下面这张图就是 claude-tap 的仪表盘：右上角绿色的Watching指示灯表示 SSE 实时通道已建立，会话列表会随着 Agent 的每一次 API 调用自动刷新。

先搞清楚两个"实时"：Live 模式与 SSE 的分工

很多新手会混淆：claude-tap 里其实存在两层 SSE，方向正好相反 🧭

上游这层由SSEReassembler完成：它逐行读取字节流，识别event:和data:帧，兼容 Anthropicmessage_start / content_block_delta、OpenAIresponse.output_text.delta、Chat Completions 的裸data:帧甚至 Gemini 的streamGenerateContent等多种协议，最终"重放"出一份完整的响应对象（见 claude_tap/sse.py#L11-L65）。

而本文的主角是下游这层——SSE 广播，下面逐步拆解。

SSE 广播四步走：一条流量如何"活"到页面上

第一步：代理捕获请求并写入轨迹

当 Agent 发起一次 API 调用，代理处理完请求后生成一条轨迹记录，交给TraceWriter。关键点在 claude_tap/trace.py#L49-L55：先落盘，再广播——写 SQLite 和推实时消息是解耦的两步，即使存储出问题，也不会阻塞代理转发，更不会让实时推送失效。

第二步：broadcast() 把记录序列化后群发

LiveViewerServer.broadcast()是整个实时系统的"广播站"，位于 claude_tap/live.py#L298-L320，它做三件事：

1. 把新记录追加进当日内存缓存（跨天自动清空，避免内存无限增长）；
2. 调用attach_cost_to_record()给记录补上按模型单价计算的费用估算，再压缩成一行 JSON；
3. 按 SSE 协议格式data: {...}\n\n写入每一个已连接的客户端，失败的连接会被自动剔除。

注意第 2 点：每条实时消息里自带 Token 用量和成本，所以监控页面不用额外查价，看到请求的同时就看到了花费 💰

第三步：/events 端点——一条永不关闭的推送管道

浏览器连接GET /events时，服务端返回Content-Type: text/event-stream的长连接（见 claude_tap/live.py#L416-L454）。这条管道有三个贴心设计：

• 补发历史：连接建立的瞬间，会先把当日已缓存的所有记录重放一遍。也就是说，你晚开 5 分钟监控页面，也不会漏掉之前的流量；
• 30 秒心跳：每 30 秒发一条: keepalive注释帧，防止中间网络设备掐断空闲连接；
• 优雅退出：客户端断开或服务关停时自动清理连接列表，无泄漏。

第四步：浏览器端 EventSource 订阅 + 去重 + 节流渲染

前端只需要一行new EventSource('/events')就能收到推送，核心逻辑在 claude_tap/viewer_assets/live_bootstrap.js#L11-L49：

• 利用 EventSource 原生的断线自动重连能力，断网恢复后无需人工干预；
• 因为重连会重放历史，前端用request_id做了去重集合，保证同一条记录不会显示两次；
• 新记录到达后延迟 50ms 合并渲染（防抖），即使 Agent 高频发请求，页面也不会卡顿。

上图就是一次实时查看的效果：左侧是按轮次（Turn）排列的请求列表，标注了模型、Token 数与耗时；右侧展开后能看到 System Prompt、用户消息、模型思考过程和工具调用的完整内容。

仪表盘实时刷新：同一套 SSE 机制的"第二路广播"

在claude-tap dashboard独立仪表盘模式下，还有一条专门的/dashboard/eventsSSE 通道。它的工作方式略有不同：

• 后台任务 claude_tap/live.py#L747-L759每 1 秒轮询一次SQLite 快照，发现会话/记录数变化就向所有仪表盘客户端广播refresh事件；
• 当前正在写数据的 Live 会话由代理直接广播record事件，两者叠加让仪表盘"即写即现"；
• 仪表盘还内置了安全校验：只有来自本机（localhost）的浏览器才拿到带 Token 的控制权限，避免局域网误开风险。

页面顶部的 Input / Cache Read / Cache Create / Total 统计条，正是每条 SSE 消息里携带的 usage 数据累加出来的，实时监控的同时也是实时的成本监控。

上手体验：3 个步骤开启实时监控

1. 安装并启动：安装 claude-tap 后，直接运行claude-tap claude -p "你的任务"（把任意受支持的 Agent 命令放在后面即可），代理启动的同时会自动打开浏览器查看器；
2. 打开独立仪表盘：运行claude-tap dashboard，即可跨会话、跨 Agent 地集中查看所有流量，右上角出现绿色 Watching 即代表实时通道就绪；
3. 切换实时与历史：查看器左上角的日期下拉框中，Live (current session)是实时模式，其余选项可回看任意一天的历史轨迹（存于本地 SQLite，支持按 Agent/状态/关键词搜索）。

整个过程无需修改 Agent 任何配置，代理地址与输出目录见 tests/e2e/test_real_proxy.py 中的真实 E2E 用例，可参考其环境变量用法。

核心文件速查

一句话总结

claude-tap 的实时 Live 模式本质是一条**"写入即广播"**的 SSE 管道：代理捕获流量 → 落盘 SQLite → 序列化群发/events→ 浏览器 EventSource 自动重连、去重、节流渲染。两层 SSE（解析上游 + 推送下游）配合历史重放与心跳保活，让"实时监控 AI 编程代理流量"既简单又可靠——这正是它作为本地 AI Agent 流量查看器最核心的竞争力。

【免费下载链接】claude-tapIntercept and inspect Coding Agent API traffic from Claude Code, Codex CLI, Gemini CLI, Cursor CLI, OpenCode, Kimi/Kimi Code, Pi, and Hermes in a local trace viewer.项目地址: https://gitcode.com/gh_mirrors/cl/claude-tap