RPC 集成

从「单向输出」到「双向对话」

JSON 模式是「pi 单向吐出事件」。如果你想要的是双向对话——发指令、等响应、中途打断、查询状态——那就该用 RPC 模式。

RPC 模式让 pi 以无头方式运行，通过 stdin/stdout 上的 JSON 协议对接。适合把 agent 嵌进其他应用、IDE 或自定义界面。

pi --mode rpc [options]

常用选项：--provider 设提供方、--model 设模型、--no-session 关掉会话持久化、--session-dir 自定义会话存储目录。

---

协议概览：三句话

• 命令：JSON 对象，一行一个，发到 stdin
• 响应：type: "response"，表示命令成功/失败
• 事件：agent 事件以 JSON 行流式输出到 stdout

所有命令都支持可选 id 字段做请求/响应关联——带上它，响应会带同一个 id，方便你配对。

---

核心命令：prompt

发一条用户提示词给 agent：

{"id": "req-1", "type": "prompt", "message": "Hello, world!"}

命令响应在提示词被接受、入队或立即处理后发出。事件在接收后继续异步流式输出。

{"id": "req-1", "type": "response", "command": "prompt", "success": true}

success: true 只表示「提示词被接受」；之后跑失败是通过事件流报告的，不会用同一个 id 再发一条响应。这是很容易理解错的点。

流式输出时，你可以指定 streamingBehavior 让消息排队："steer"（当前回合工具调用完、下次 LLM 调用前送达）或 "followUp"（等 agent 停下来才送达）。流式时不给这个字段，命令直接报错。

---

常用命令一览

RPC 暴露的命令很全，覆盖状态、模型、思考、会话等：

类别	命令
提示	prompt、steer、follow_up、abort
状态	get_state、get_messages
模型	set_model、cycle_model、get_available_models
思考	set_thinking_level、cycle_thinking_level
压缩	compact、set_auto_compaction
会话	new_session、switch_session、fork、get_tree、get_entries
执行	bash、abort_bash

比如查状态：

{"type": "get_state"}

{
  "type": "response",
  "command": "get_state",
  "success": true,
  "data": {
    "model": {...},
    "thinkingLevel": "medium",
    "isStreaming": false,
    "sessionId": "abc123"
  }
}

---

帧格式：LF 是唯一分隔符

RPC 用严格的 JSONL 语义，LF（\n）是唯一记录分隔符。这直接影响你的客户端怎么写：

• 只按 \n 切记录
• 可接受可选的 \r\n（剥掉尾部 \r）
• 别用会按 Unicode 分隔符分行的通用行读取器

特别地，官方明确点名：Node 的 readline 不符合 RPC 协议，因为它还会按 U+2028、U+2029 分行——而这两个字符在 JSON 字符串里是合法的。

---

一个完整的最小客户端（Python）

下面这段直接可抄，起了个 RPC 子进程、发一条 prompt、流式打印文本：

import subprocess
import json

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

def send(cmd):
    proc.stdin.write(json.dumps(cmd) + "\n")
    proc.stdin.flush()

def read_events():
    for line in proc.stdout:
        yield json.loads(line)

send({"type": "prompt", "message": "Hello!"})

for event in read_events():
    if event.get("type") == "message_update":
        delta = event.get("assistantMessageEvent", {})
        if delta.get("type") == "text_delta":
            print(delta["delta"], end="", flush=True)
    if event.get("type") == "agent_end":
        print()
        break

---

落地练习：用 Python 起一个 RPC 会话

把上面的最小客户端跑起来：

1. 保存为 rpc_client.py，运行它
2. 观察 stdout 上流式打印的文本，以及最终的 agent_end
3. 把消息改成一段 markdown 处理指令，验证它能调用工具返回结构化结果

怎么判断做对了？——能稳定收到 message_update 的 text_delta 流，并在 agent_end 后正常退出；2>/dev/null 之外没有解析报错。

卡住了怎么办？ 没输出？先确认 pi --mode rpc --no-session 能单独跑起来。JSON 解析报错？大概率是读了空行或非 JSON 行，加个 try/except 或在循环里跳过非 JSON 行。

---

常见坑：用 Node readline 读 RPC 输出

很多人图省事，用 Node 的 readline 逐行读 RPC 的 stdout——这是官方明说的坑。readline 会按 Unicode 行分隔符（U+2028/U+2029）分行，而这两个字符合法存在于 JSON 字符串内，一遇到就劈开一条记录、导致 JSON.parse 失败。

正确的读法是自己维护一个 buffer，只按 \n 切、剥掉尾部 \r。要么用官方 rpc-client.ts，要么照官方示例手写一个 attachJsonlReader。协议合规从帧解析做起。

---

小结

1. RPC 模式通过 stdin/stdout 上的 JSON 协议做双向对话
2. prompt 响应只表示「被接受」，结果走事件流报告
3. 命令带可选 id 做请求/响应关联
4. 帧格式：LF 是唯一分隔符，Node readline 不合规
5. 用 Python / 自定义 JSONL reader 可以轻松对接

下一节，讲比 RPC 更贴近语言层的方案：SDK 嵌入。