Pi 的 RPC 模式与 JSON 事件流

2026-09-26 pi 0.87.1 实测抓取，非文档转抄

pi 是一个跑在终端里的 coding agent，但它并不只有 TUI。同一个 agent、同一套 session 与工具，可以用四种方式驱动：交互式 TUI、print、--mode json、--mode rpc。后两种是给人写程序用的——当你的脚本、Web 后端、IDE 插件需要「结构化进度」而不是一段最终文本时，就该看这一篇。

下面所有事件样本都是我在本机 pi 0.87.1 上真跑出来的（模型 deepseek-v4-pro，工作目录 /tmp/pi-demo），为了排版把过长的字段折成了 …，其余保持原样。

1 · 先选对接口

还有一条路是 SDK：进程内嵌，直接拿 TypeScript 方法和事件，没有进程边界。判断标准很简单——同语言且能接受同一个进程里跑 agent，用 SDK；跨语言、要隔离、要做自定义客户端，用 RPC。

另外：不给 --print 时，只要 stdin/stdout 不是 TTY，pi 也会自动落到 print 模式。所以 pi "…" 在管道里不会弹出 TUI。

2 · JSON 模式：一条真实的事件流

pi --mode json "只回复两个字母：OK" > events.jsonl   # 15 行

把它按类型摊开（左边是行号，中间是 type）：

 1 session         {"version":3,"id":"01a0dccb-…","cwd":"/tmp/pi-demo"}
 2 agent_start
 3 turn_start
 4 message_start     role=system    content=""  sections={preamble, tools, …}
 5 message_end       role=system
 6 message_start     role=user      [{"type":"text","text":"只回复两个字母：OK"}]
 7 message_end       role=user
 8 message_start     role=assistant content=[]  stopReason="pending" usage.totalTokens=0
 9 message_update    assistantMessageEvent={"type":"text_start","contentIndex":0}   usage.totalTokens=0
10 message_update    assistantMessageEvent={"type":"text_delta","contentIndex":0,"delta":"OK"}  usage.totalTokens=0
11 message_update    assistantMessageEvent={"type":"text_end","contentIndex":0,"content":"OK"}   usage.totalTokens=3554
12 message_end       role=assistant [{"type":"text","text":"OK"}] stopReason="stop"
13 turn_end          {message, toolResults:[]}
14 agent_end         messages=[3 条] willRetry=false
15 agent_settled

几个直接从数据里读出来的事实：

• 第一行是 session header，只在 JSON 模式出现（RPC 没有，因为它长驻，用 get_state 拿 sessionId / sessionFile）。
• system prompt 也会以 message_start/end 的形式进流里，别把它当成助手输出。
• usage 在 text_end 那一拍才从 0 跳到 3554：input 3552 + output 2。也就是说「回复两个字母」的账单是 3.5k input tokens（系统提示 + 工具定义），cost.total = 0.0047 USD。流式过程中 usage 可能一直是 0，不能当成「还没花钱」。
• 整个 agent_end 里带的是本次跑生成的全部消息，等于一份 mini transcript。

只想要完成的文本，一行 jq 就够（stderr 记得丢掉，stdout 是协议专用）：

pi --mode json "List files" 2>/dev/null | jq -c 'select(.type == "message_end")'

3 · 分帧：只按 LF 切

两个模式共用同一套 framing：一条记录 = 一个完整 JSON 对象 + \n。听起来 trivial，但这里有个真会咬人的坑：

另外三条纪律：

• 持续读 stdout。pi 尊重 stdout 反压，你不读，管道缓冲一满它就被卡住——表现是「agent 静默不动了」。
• stdout 只属于协议。诊断、日志、扩展的输出全部走 stderr。别把 stderr 当 JSON 解析，也别往 stdout 写日志。
• 写命令时同样尊重 stdin 反压。

4 · 事件词表

下面是 AgentSessionEvent 在 JSON/RPC 里会出现的全集（JSON 模式与 RPC 模式共享这套形状，RPC 另有两条独有的）。

骨架

消息与流式

工具执行

用 toolCallId 串生命周期，别按顺序猜。

状态、压缩、重试

仅 RPC 有

扩展 UI（对话框、通知）是 RPC 的另一个子协议，不属于 AgentSessionEvent。

5 · 重建流式消息

线上格式的 message_update 是纯增量：SDK 里那种累积 message 快照和 assistantMessageEvent.partial 都被剥掉了，所以流量是线性的。代价是客户端要自己拼。

嵌套的 assistantMessageEvent.type：

规则只有一句：用 contentIndex 认块；delta 拿来实时显示，一旦对应的 *_end 到了就用它覆盖；message_end.message 到了就整条覆盖。任何「自己算出来的最终文本」都是错的，权威版本在 end 里。

6 · agent_end 不等于「结束了」

这是最容易写错的一处。

prompt response        # 只代表：命令被接受 / 入队 / 立刻处理了
  ↓
agent_end              # 一次低层 run 结束
  ↓                    # 但自动重试 / 溢出恢复 / 压缩 / steer / follow-up 还能继续
agent_settled          # 这里才是「pi 不会再自己动了」

要「跑完」的语义，就等 agent_settled。上面那条最小的两次字母回复里，第 14、15 行正好说明了这对：agent_end 之后紧接着才是 agent_settled。

7 · RPC：把单向流变成双向控制

pi --mode rpc --no-session

普通 CLI 参数照常生效（--provider、--model、--session-dir、--name…）。唯一被明确拒绝的是 @file 形式的 prompt 参数—— prompt 得走 prompt 命令。

协议一共四类记录：

用 id 关联，不要按顺序猜

命令处理是异步的，响应顺序不保证。给每条可能并发的命令一个字符串 id，对应 response 会原样回显：

{"id":"req-1","type":"get_state"}
{"id":"req-1","type":"response","command":"get_state","success":true,"data":{…}}

事件一般没有 id（它描述会话，不描述某次请求）；例外是 bash_execution_update 会带上发起命令的 id，以及 extension_ui_response 会带回 extension_ui_request 的 id（它不产生普通 response）。

这是我从真进程里拿回来的 get_state 响应（截断）：

{"id":"r1","type":"response","command":"get_state","success":true,"data":{
  "model":{"id":"deepseek-v4-pro","provider":"deepseek","api":"openai-completions",
           "contextWindow":1000000,"maxTokens":384000,"cost":{…}, …},
  "thinkingLevel":"high",
  "isStreaming":false,"isCompacting":false,
  "steeringMode":"one-at-a-time","followUpMode":"one-at-a-time",
  "sessionId":"01a0dccb-5694-71b5-9605-37fc1730a673",
  "autoCompactionEnabled":true,
  "messageCount":0,"pendingMessageCount":0}}

注意 no-session 下 sessionFile 就不出现，而 sessionId 仍然有；model 是完整对象，不是字符串。

prompt 及其并发语义

{"id":"req-2","type":"prompt","message":"Review this repository"}
{"id":"req-2","type":"response","command":"prompt","success":true}

• success:true = 被接受/入队/立刻处理；false = 接受前就被拒。接受之后出的错一律回到事件流里，不会再给同一个 id 补一条 response。
• agent 正在流式时发 prompt，必须带 streamingBehavior，否则直接报错：

  {"type":"prompt","message":"New instruction","streamingBehavior":"steer"}

  "steer"：当前 turn 的工具跑完、下一次 LLM 调用之前插进去（改方向）；"followUp"：等 agent 停下来再说。也可以直接发 steer / follow_up 命令，语义相同但不带扩展命令支持。
• 可以带 images:[{type:"image",data:"<base64>",mimeType:"image/png"}]。
• skill 命令 /skill:name 和模板 /template 会在入队前展开；扩展命令（/mycommand）立刻执行，流式期间也不等。

命令分组速查

abort 会等会话真的 idle 再回 response；要做 TUI 里 Esc 那种「清空队列 + 打断 + 把文本还给输入框」，先 clear_queue（它会返回被清掉的 steering/followUp 文本），再 abort。

8 · 最小客户端

Python 版（官方示例，注意用的是二进制管道，天然只按 LF 切）：

import json, subprocess

process = subprocess.Popen(
    ["pi", "--mode", "rpc", "--no-session"],
    stdin=subprocess.PIPE, stdout=subprocess.PIPE,
)

command = {"id": "prompt-1", "type": "prompt", "message": "Hello"}
process.stdin.write(json.dumps(command).encode("utf-8") + b"\n")
process.stdin.flush()

while line := process.stdout.readline():
    record = json.loads(line)
    if record.get("type") == "message_update":
        update = record["assistantMessageEvent"]
        if update["type"] == "text_delta":
            print(update["delta"], end="", flush=True)
    elif record.get("type") == "agent_settled":
        print()
        break

process.stdin.close()
process.wait()

TypeScript 侧不必手写——包里导出了 RpcClient，它负责起子进程、关联 id、给类型化的命令方法和事件监听：

import { RpcClient } from "@earendil-works/pi-coding-agent";

const client = new RpcClient({ cliPath: "../dist/cli.js", args: ["--no-session"] });

const unsubscribe = client.onEvent((event) => {
  if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
    process.stdout.write(event.assistantMessageEvent.delta);
  } else if (event.type === "tool_execution_start") {
    process.stderr.write(`\n[tool: ${event.toolName}]\n`);
  }
});

try {
  await client.start();
  await client.promptAndWait("Explain this repository");   // 先订阅后发，无竞态
} finally {
  unsubscribe();
  await client.stop();
}

仓库里的 examples/rpc-client.ts 就是这个；它指向 dist/cli.js，所以从 checkout 跑要先 build 包。

9 · 错误与关闭

• 命令失败 = 一条 success:false 的 response：

  {"id":"req-3","type":"response","command":"set_model","success":false,
   "error":"Model not found: invalid/model"}

• JSON 写坏了 = 一条没有 id的 parse response，所以你的解析器不能在「等不到 id」时卡死：

  {"type":"response","command":"parse","success":false,
   "error":"Failed to parse command: Unexpected token..."}

• 成功只覆盖「命令被正确处理」。provider 挂了、prompt 接受后被中断，都只在消息/事件流里体现。
• 还要自己处理：子进程起不来、意外退出、stderr 诊断、取消、以及你自己的超时。
• 正常关闭 = 关掉子进程的 stdin，pi 会 dispose 当前 runtime 再退出；扩展也能请求关闭（等当前命令或 agent_settled 之后）。信号和意外退出仍要自己兜。

10 · 一页速记

• 只按 \n 切记录；readline 会因为 U+2028/2029 把记录劈开。
• stdout 只放协议，stderr 放日志；不读 stdout 会把 pi 卡住。
• 每条命令带 id，按 id 关联，别按顺序猜。
• prompt 的 success:true 只是「收到了」；跑完看 agent_settled，不是 agent_end。
• 先订阅事件再发 prompt。
• 增量 显示，*_end / message_end 覆盖；流式期间 usage 可能是 0。
• 要双向控制才上 RPC；同进程能搞定就用 SDK；只跑一次的 JSON 就够。

· 相关

• 拆解 OpenAI 的 SSE 数据流事件流的上游：模型输出的字节是怎么到达的