pi coding-agent RPC 模式实战：用 stdin/stdout JSONL 协议构建可嵌入的无头 Agent 通道

RPC（Remote Procedure Call）模式让 pi 的 coding agent 以无头（headless）方式运行：命令通过 stdin 以 JSON 行形式进入，响应与 Agent 事件以 JSON 行形式从 stdout 流出。本文基于 RPC 官方文档 完整梳理启动方式、协议帧语义、全部命令与事件、扩展 UI 子协议和核心类型，并结合 rpc-mode.ts、rpc-types.ts 等源码实现，帮助你把 pi 嵌入到自己的应用、IDE 或自定义 UI 中。

一、启动 RPC 模式

RPC 模式通过 --mode rpc 参数启动（--mode 可选值为 text（默认）、json、rpc，见 args.ts 中的 Mode 类型定义）：

pi --mode rpc [options]

常用启动选项：

选项	说明
--provider <name>	设置 LLM 提供方（anthropic、openai、google 等）
--model <pattern>	模型模式或 ID，支持 provider/id 与可选的 :<thinking> 简写
--name <name> / -n <name>	启动时设置会话显示名
--no-session	禁用会话持久化
--session-dir <path>	自定义会话存储目录

Node.js/TypeScript 用户的选型建议：如果你在 Node.js 生态内开发，优先考虑直接从 @earendil-works/pi-coding-agent 使用 AgentSession（见 agent-session.ts），而不是 spawn 一个子进程。如果确实需要子进程方案，仓库提供了带类型的 TypeScript 客户端 rpc-client.ts 和完整交互示例 rpc-example.ts，可直接参考其 JSONL 读取与请求关联的实现。

二、协议概览：命令、响应与事件

RPC 协议由三种 JSON 行消息构成：

• 命令（Commands）：客户端写入 stdin 的 JSON 对象，每行一条；
• 响应（Responses）：type: "response" 的 JSON 对象，表示某条命令成功或失败；
• 事件（Events）：Agent 运行时以 JSON 行（JSON Lines）形式从 stdout 流出的事件。

所有命令都支持可选的 id 字段用于请求/响应关联：若提供了 id，对应的 response 会携带相同的 id。此外，bash_execution_update 事件也会携带其来源 bash 命令的 id。

这一整套消息的 TypeScript 类型都定义在 rpc-types.ts 中：RpcCommand 是全部命令的判别联合类型（每个分支都带 id?: string），RpcResponse 是全部响应类型，RpcExtensionUIRequest/RpcExtensionUIResponse 则定义扩展 UI 子协议——阅读该文件是理解整份协议最快捷的方式。

帧语义（Framing）

RPC 模式使用严格的 JSONL 语义，LF（\n）是唯一合法的记录分隔符。这对客户端实现至关重要：

• 只按 \n 切分记录；
• 可以接受可选的 \r\n 输入（剥离行尾的 \r）；
• 不要使用把 Unicode 分隔符也当作换行符的通用行读取器。

一个典型的坑：Node 的 readline 不符合 RPC 协议要求，因为它还会在 U+2028（Line Separator）和 U+2029（Paragraph Separator）处切分，而这两个字符在 JSON 字符串内部是完全合法的。

仓库中的参考实现印证了这一点。jsonl.ts 提供了两个核心工具：

// 序列化一条严格的 JSONL 记录：JSON + LF
export function serializeJsonLine(value: unknown): string {
    return `${JSON.stringify(value)}\n`;
}

attachJsonlLineReader 手动实现了一个 LF-only 的行读取器（用 StringDecoder 处理 UTF-8 跨 chunk 边界、按 \n 索引切分、剥离行尾 \r、在流 end 时冲刷残留 buffer），源码注释明确写道：“故意不使用 Node readline，因为 readline 会按 JSON 字符串内部合法的额外 Unicode 分隔符切分，从而不满足严格 JSONL 帧语义”。rpc-mode.ts 在入口处用 attachJsonlLineReader(process.stdin, ...) 挂载 stdin，用 writeRawStdout(serializeJsonLine(obj)) 输出，并配合 waitForRawStdoutBackpressure() 在写入后等待背压，避免高速事件流撑爆管道。

三、命令参考

以下命令均从 stdin 发送。除特别说明外，响应形如 {"type": "response", "command": "<cmd>", "success": true, ...}。

3.1 Prompting（提示与流程控制）

prompt

向 Agent 发送用户提示。命令响应在提示被接受（accepted）、排队（queued）或处理（handled）之后发出；事件流在接收之后继续异步流出。

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

带图片：

{"type": "prompt", "message": "What's in this image?", "images": [{"type": "image", "data": "base64-encoded-data", "mimeType": "image/png"}]}

流式进行中的行为：如果 Agent 正在流式输出，必须指定 streamingBehavior 才能排队消息：

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

• "steer"：在 Agent 运行期间排队，于当前 assistant 轮次执行完工具调用之后、下一次 LLM 调用之前送达；
• "followUp"：等待 Agent 完全结束后再送达。

Agent 正在流式且未指定 streamingBehavior 时，命令返回错误。

扩展命令：如果消息是扩展命令（如 /mycommand），即使在流式期间也会立即执行，扩展通过 pi.sendMessage() 自行管理 LLM 交互。

输入展开：Skill 命令（/skill:name）与 prompt 模板（/template）在发送/排队前会被展开。

响应：

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

success: true 表示提示已被接受、排队或立即处理；success: false 表示在接收前就被拒绝。接收之后的失败会通过正常的事件与消息流报告，而不是对同一请求 id 再发一次 response。这一点在源码中体现得很清楚：rpc-mode.ts 中 prompt 分支调用 session.prompt() 时传入 preflightResult 回调，只有 preflight 成功后才 output(success(id, "prompt"))，而后续错误走事件流。

images 字段可选，每张图片采用 ImageContent 格式：{"type": "image", "data": "base64-encoded-data", "mimeType": "image/png"}。

steer

在 Agent 运行期间排队一条“转向”消息。它在当前 assistant 轮次执行完工具调用之后、下一次 LLM 调用之前送达。支持 skill 命令与 prompt 模板展开；不允许扩展命令（请用 prompt）。

{"type": "steer", "message": "Stop and do this instead"}

带图片：

{"type": "steer", "message": "Look at this instead", "images": [{"type": "image", "data": "base64-encoded-data", "mimeType": "image/png"}]}

images 可选，格式同 prompt。响应：{"type": "response", "command": "steer", "success": true}。

如何控制转向消息的投递节奏见下文 set_steering_mode。

follow_up

排队一条待办消息，在 Agent 结束后处理。只有当 Agent 没有更多工具调用或转向消息时才会送达。支持 skill 命令与 prompt 模板展开；不允许扩展命令（请用 prompt）。

{"type": "follow_up", "message": "After you're done, also do this"}

{"type": "follow_up", "message": "Also check this image", "images": [{"type": "image", "data": "base64-encoded-data", "mimeType": "image/png"}]}

响应：{"type": "response", "command": "follow_up", "success": true}。控制投递节奏见 set_follow_up_mode。

abort

中止当前 Agent 操作：

{"type": "abort"}

{"type": "response", "command": "abort", "success": true}

clear_queue

移除已排队的 steering 与 follow-up 消息并返回其文本：

{"type": "clear_queue"}

{
  "type": "response",
  "command": "clear_queue",
  "success": true,
  "data": {
    "steering": ["Change direction"],
    "followUp": ["Summarize when finished"]
  }
}

交互式 Esc 的实现技巧：在 abort 之前先发 clear_queue，然后把返回的文本还原到客户端编辑器中。注意 abort 在队列中还有消息时会继续这些已排队的消息，所以顺序很重要。

new_session

开启新会话，可被 session_before_switch 扩展事件处理器取消：

{"type": "new_session"}

带父会话追踪（可选）：

{"type": "new_session", "parentSession": "/path/to/parent-session.jsonl"}

{"type": "response", "command": "new_session", "success": true, "data": {"cancelled": false}}

若扩展取消了切换："data": {"cancelled": true}。

3.2 State（会话状态）

get_state

获取当前会话状态：

{"type": "get_state"}

{
  "type": "response",
  "command": "get_state",
  "success": true,
  "data": {
    "model": {...},
    "thinkingLevel": "medium",
    "isStreaming": false,
    "isCompacting": false,
    "steeringMode": "all",
    "followUpMode": "one-at-a-time",
    "sessionFile": "/path/to/session.jsonl",
    "sessionId": "abc123",
    "sessionName": "my-feature-work",
    "autoCompactionEnabled": true,
    "messageCount": 5,
    "pendingMessageCount": 0
  }
}

model 字段是完整的 Model 对象或 null；sessionName 是通过 set_session_name 设置的显示名，未设置时省略。字段全集对应 rpc-types.ts 中的 RpcSessionState 接口。

get_messages

获取会话中的全部消息：

{"type": "get_messages"}

{
  "type": "response",
  "command": "get_messages",
  "success": true,
  "data": {"messages": [...]}
}

消息是 AgentMessage 对象（见 Message Types）。

3.3 Model（模型管理）

set_model

切换到指定模型：

{"type": "set_model", "provider": "anthropic", "modelId": "claude-sonnet-4-20250514"}

响应携带完整的 Model 对象：

{
  "type": "response",
  "command": "set_model",
  "success": true,
  "data": {...}
}

从源码看（rpc-mode.ts），set_model 会在 modelRuntime.getAvailableSnapshot() 中按 provider + modelId 精确查找，找不到时返回 Model not found: <provider>/<modelId> 错误。

cycle_model

循环切换到下一个可用模型；只有一个可用模型时返回 null data：

{"type": "cycle_model"}

{
  "type": "response",
  "command": "cycle_model",
  "success": true,
  "data": {
    "model": {...},
    "thinkingLevel": "medium",
    "isScoped": false
  }
}

model 字段为完整 Model 对象。

get_available_models

列出所有已配置的模型：

{"type": "get_available_models"}

{
  "type": "response",
  "command": "get_available_models",
  "success": true,
  "data": {
    "models": [...]
  }
}

models 是完整 Model 对象数组。

3.4 Thinking（推理强度）

set_thinking_level

为支持推理的模型设置 thinking 等级：

{"type": "set_thinking_level", "level": "high"}

等级取值："off"、"minimal"、"low"、"medium"、"high"、"xhigh"、"max"。

"xhigh" 与 "max" 只在所选模型支持时才暴露；部分模型（包括 GPT-5.6）两者都暴露。

{"type": "response", "command": "set_thinking_level", "success": true}

cycle_thinking_level

在可用 thinking 等级间循环；模型不支持 thinking 时返回 null data：

{"type": "cycle_thinking_level"}

{
  "type": "response",
  "command": "cycle_thinking_level",
  "success": true,
  "data": {"level": "high"}
}

get_available_thinking_levels

列出当前模型支持的 thinking 等级；不支持推理的模型返回 ["off"]：

{"type": "get_available_thinking_levels"}

{
  "type": "response",
  "command": "get_available_thinking_levels",
  "success": true,
  "data": {
    "levels": ["off", "minimal", "low", "medium", "high"]
  }
}

3.5 Queue Modes（队列投递模式）

set_steering_mode

控制 steering 消息（来自 steer）的投递方式：

{"type": "set_steering_mode", "mode": "one-at-a-time"}

模式：

• "all"：当前 assistant 轮次执行完工具调用后，一次性投递全部 steering 消息；
• "one-at-a-time"：每个完成的 assistant 轮次只投递一条 steering 消息（默认）。

{"type": "response", "command": "set_steering_mode", "success": true}

set_follow_up_mode

控制 follow-up 消息（来自 follow_up）的投递方式：

{"type": "set_follow_up_mode", "mode": "one-at-a-time"}

模式：

• "all"：Agent 结束时一次性投递全部 follow-up 消息；
• "one-at-a-time"：每次 Agent 完成只投递一条（默认）。

{"type": "response", "command": "set_follow_up_mode", "success": true}

3.6 Compaction（上下文压缩）

compact

手动压缩会话上下文以降低 token 用量：

{"type": "compact"}

带自定义指令：

{"type": "compact", "customInstructions": "Focus on code changes"}

{
  "type": "response",
  "command": "compact",
  "success": true,
  "data": {
    "summary": "Summary of conversation...",
    "firstKeptEntryId": "abc123",
    "tokensBefore": 150000,
    "estimatedTokensAfter": 32000,
    "usage": {
      "input": 32000,
      "output": 1200,
      "cacheRead": 0,
      "cacheWrite": 0,
      "totalTokens": 33200,
      "cost": {"input": 0.01, "output": 0.02, "cacheRead": 0, "cacheWrite": 0, "total": 0.03}
    },
    "details": {}
  }
}

注意：estimatedTokensAfter 是压缩后立即对重建消息上下文做的启发式估计，不是提供方精确的 token 计数；usage 报告的是生成摘要所用的一次或多次 LLM 调用，自定义 compaction 处理器可能会省略它。

set_auto_compaction

在上下文接近满载时启用/禁用自动压缩：

{"type": "set_auto_compaction", "enabled": true}

{"type": "response", "command": "set_auto_compaction", "success": true}

3.7 Retry（自动重试）

set_auto_retry

启用/禁用瞬时错误（overloaded、rate limit、5xx）的自动重试：

{"type": "set_auto_retry", "enabled": true}

{"type": "response", "command": "set_auto_retry", "success": true}

abort_retry

中止进行中的重试（取消等待并停止重试）：

{"type": "abort_retry"}

{"type": "response", "command": "abort_retry", "success": true}

3.8 Bash（直接执行 shell 命令）

bash

执行 shell 命令并把输出加入会话上下文。命令运行期间以 bash_execution_update 事件流式输出，响应携带最终结果：

{"id": "req-1", "type": "bash", "command": "ls -la"}

建议携带 id，以便把流式的 bash_execution_update 事件关联到该命令。

{
  "id": "req-1",
  "type": "response",
  "command": "bash",
  "success": true,
  "data": {
    "output": "total 48\ndrwxr-xr-x ...",
    "exitCode": 0,
    "cancelled": false,
    "truncated": false
  }
}

若输出被截断，会附带 fullOutputPath：

{
  "type": "response",
  "command": "bash",
  "success": true,
  "data": {
    "output": "truncated output...",
    "exitCode": 0,
    "cancelled": false,
    "truncated": true,
    "fullOutputPath": "/tmp/pi-bash-abc123.log"
  }
}

从源码结构看，bash 命令还支持一个文档未展开的可选字段 excludeFromContext（见 rpc-types.ts 中 RpcCommand 的 bash 分支），用于控制该次执行结果是否计入后续上下文；另外命令执行前会先经过扩展事件管道（emitUserBash），扩展若有 result 则直接采用扩展结果，否则走 session.executeBash（见 rpc-mode.ts）。

Bash 结果如何到达 LLM：bash 命令立即执行并返回 BashResult。内部会创建一条 BashExecutionMessage 并存入 Agent 的消息状态。当下一次 prompt 发送时，所有消息（包括 BashExecutionMessage）在发给 LLM 前会被转换：BashExecutionMessage 会变成如下格式的 UserMessage：

Ran `ls -la`

total 48 drwxr-xr-x ...

这意味着：

1. Bash 输出是在**下一次 prompt** 时才进入 LLM 上下文，而不是立即；
2. 可以在一次 prompt 之前连续执行多条 bash，所有输出都会一并包含。

#### abort_bash

中止正在运行的 bash 命令：

```json
{"type": "abort_bash"}

{"type": "response", "command": "abort_bash", "success": true}

3.9 Session（会话管理）

get_session_stats

获取 token 用量、成本统计与当前上下文窗口占用：

{"type": "get_session_stats"}

{
  "type": "response",
  "command": "get_session_stats",
  "success": true,
  "data": {
    "sessionFile": "/path/to/session.jsonl",
    "sessionId": "abc123",
    "userMessages": 5,
    "assistantMessages": 5,
    "toolCalls": 12,
    "toolResults": 12,
    "totalMessages": 22,
    "tokens": {
      "input": 50000,
      "output": 10000,
      "cacheRead": 40000,
      "cacheWrite": 5000,
      "total": 105000
    },
    "cost": 0.45,
    "contextUsage": {
      "tokens": 60000,
      "contextWindow": 200000,
      "percent": 30
    }
  }
}

tokens 与 cost 包含 assistant 消息、工具报告的用量，以及整个会话中 compaction/分支摘要生成所用的量。contextUsage 是用于压缩判断和页脚展示的实际当前上下文窗口估计。

contextUsage 在没有模型或上下文窗口时省略；contextUsage.tokens 与 contextUsage.percent 在压缩后、直到一条新的 post-compaction assistant 响应提供有效 usage 之前为 null。

export_html

将会话导出为 HTML 文件：

{"type": "export_html"}

自定义输出路径：

{"type": "export_html", "outputPath": "/tmp/session.html"}

{
  "type": "response",
  "command": "export_html",
  "success": true,
  "data": {"path": "/tmp/session.html"}
}

switch_session

加载另一个会话文件，可被 session_before_switch 扩展事件处理器取消：

{"type": "switch_session", "sessionPath": "/path/to/session.jsonl"}

{"type": "response", "command": "switch_session", "success": true, "data": {"cancelled": false}}

扩展取消时："data": {"cancelled": true}。

fork

从活动分支上的某条历史用户消息创建新分支（fork），可被 session_before_fork 扩展事件处理器取消。响应返回被 fork 消息的文本：

{"type": "fork", "entryId": "abc123"}

{
  "type": "response",
  "command": "fork",
  "success": true,
  "data": {"text": "The original prompt text...", "cancelled": false}
}

扩展取消时："data": {"text": "The original prompt text...", "cancelled": true}。

clone

把当前活动分支在当前位置复制为一个新会话，可被 session_before_fork 扩展事件处理器取消：

{"type": "clone"}

{
  "type": "response",
  "command": "clone",
  "success": true,
  "data": {"cancelled": false}
}

扩展取消时："data": {"cancelled": true}。从源码看，clone 的实现就是对当前 leafId 做一次 position: "at" 的 fork；空会话没有 leaf 时会返回 Cannot clone session: no current entry selected 错误（见 rpc-mode.ts）。

get_fork_messages

获取可用于 fork 的用户消息列表：

{"type": "get_fork_messages"}

{
  "type": "response",
  "command": "get_fork_messages",
  "success": true,
  "data": {
    "messages": [
      {"entryId": "abc123", "text": "First prompt..."},
      {"entryId": "def456", "text": "Second prompt..."}
    ]
  }
}

get_entries

按追加顺序获取全部会话条目（不含会话头）。会话是一个条目 id 稳定的 append-only 树，因此条目 id 可以充当持久游标：把你见过的最后一条 entry id 作为 since 传入，即可只取它之后的条目——即使客户端重启过依然有效。与 get_messages 不同，这里包含压缩前的历史和已废弃分支。

{"type": "get_entries"}

带游标：

{"type": "get_entries", "since": "abc123"}

{
  "type": "response",
  "command": "get_entries",
  "success": true,
  "data": {
    "entries": [
      {"type": "message", "id": "def456", "parentId": "abc123", "timestamp": "...", "message": {"role": "user", "...": "..."}}
    ],
    "leafId": "def456"
  }
}

leafId 是当前叶条目的 id（空会话为 null），客户端可以一次往返就判断活动分支是否移动。若 since 不匹配任何条目 id，响应为 success: false。

get_tree

以树的形式获取会话。每个节点形如 {entry, children, label?, labelTimestamp?}。正常会话有唯一根节点；孤立条目（父链断裂）也会作为根出现。

{"type": "get_tree"}

{
  "type": "response",
  "command": "get_tree",
  "success": true,
  "data": {
    "tree": [
      {
        "entry": {"type": "message", "id": "abc123", "parentId": null, "...": "..."},
        "children": [
          {"entry": {"type": "message", "id": "def456", "parentId": "abc123", "...": "..."}, "children": []}
        ]
      }
    ],
    "leafId": "def456"
  }
}

get_last_assistant_text

获取最后一条 assistant 消息的文本内容：

{"type": "get_last_assistant_text"}

{
  "type": "response",
  "command": "get_last_assistant_text",
  "success": true,
  "data": {"text": "The assistant's response..."}
}

没有 assistant 消息时返回 {"text": null}。

set_session_name

设置当前会话的显示名，便于在会话列表中识别：

{"type": "set_session_name", "name": "my-feature-work"}

{
  "type": "response",
  "command": "set_session_name",
  "success": true
}

当前会话名可通过 get_state 的 sessionName 字段获取。要在启动 RPC 模式时就设置初始名，给 pi --mode rpc 进程传 --name <name> 或 -n <name>。

3.10 Commands（可调用命令查询）

get_commands

获取可用命令列表（扩展命令、prompt 模板与 skills），这些命令可以通过 prompt 命令以 / 前缀方式调用：

{"type": "get_commands"}

{
  "type": "response",
  "command": "get_commands",
  "success": true,
  "data": {
    "commands": [
      {"name": "session-name", "description": "Set or clear session name", "source": "extension", "path": "/home/user/.pi/agent/extensions/session.ts"},
      {"name": "fix-tests", "description": "Fix failing tests", "source": "prompt", "location": "project", "path": "/home/user/myproject/.pi/agent/prompts/fix-tests.md"},
      {"name": "skill:brave-search", "description": "Web search via Brave API", "source": "skill", "location": "user", "path": "/home/user/.pi/agent/skills/brave-search/SKILL.md"}
    ]
  }
}

每个命令包含：

• name：命令名（以 /name 调用）；
• description：人类可读描述（扩展命令可选）；
• source：命令来源类型：
  • "extension"：扩展中通过 pi.registerCommand() 注册；
  • "prompt"：从 prompt 模板 .md 文件加载；
  • "skill"：从 skill 目录加载（名称带 skill: 前缀）；
• location：加载位置（可选，扩展命令没有）：
  • "user"：用户级（~/.pi/agent/）；
  • "project"：项目级（./.pi/agent/）；
  • "path"：CLI 或配置中显式指定的路径；
• path：命令源文件的绝对路径（可选）。

注意：内置 TUI 命令（/settings、/hotkeys 等）不在列表中，它们只在交互模式下处理，通过 prompt 发送也不会执行。

从源码看（rpc-mode.ts），get_commands 按三个来源依次聚合：extensionRunner.getRegisteredCommands()、session.promptTemplates、resourceLoader.getSkills().skills，skill 名称统一加上 skill: 前缀。

四、事件参考

事件在 Agent 运行期间以 JSON 行形式从 stdout 流出。事件一般不带 id 字段；bash_execution_update 会携带其来源 bash 命令的 id（若提供过）。

4.1 事件类型总览

事件	说明
agent_start	Agent 开始处理
agent_end	一次底层 agent run 完成（之后可能还有重试、压缩或排队续接）
agent_settled	Agent run 完全收敛；不再有自动重试、压缩重试或排队续接
turn_start	新一轮开始
turn_end	轮次完成（含 assistant 消息与工具结果）
message_start	消息开始
message_update	流式更新（text/thinking/toolcall 增量）
message_end	消息完成
bash_execution_update	直接 RPC bash 命令的输出块
tool_execution_start	工具开始执行
tool_execution_update	工具执行进度（流式输出）
tool_execution_end	工具完成
queue_update	待处理的 steering/follow-up 队列变化
compaction_start	压缩开始
compaction_end	压缩完成
auto_retry_start	自动重试开始（瞬时错误后）
auto_retry_end	自动重试结束（成功或最终失败）
summarization_retry_scheduled	compaction 或分支摘要摘要请求因瞬时错误被安排重试
summarization_retry_attempt_start	重试的摘要请求开始
summarization_retry_finished	摘要重试循环结束
extension_error	扩展抛出了错误

事件到 JSON 的序列化由 json-event.ts 中的 toJsonEvent 完成，rpc-mode.ts 中通过 session.subscribe 把每个会话事件转换为 JSON 行输出；当 agent_settled 到达时，还会检查扩展是否请求了关机。

4.2 agent_start / agent_end / agent_settled

agent_start 在 Agent 开始处理一条 prompt 时发出：

{"type": "agent_start"}

agent_end 在一次底层 agent run 完成时发出，包含本次 run 产生的全部消息；willRetry 为 true 时意味着随后会有自动重试：

{
  "type": "agent_end",
  "messages": [...],
  "willRetry": false
}

agent_settled 在整个会话级 run 完全收敛后发出。此时 Pi 不会再通过重试、压缩重试或排队的 follow-up 消息自动继续。客户端判断“任务彻底结束”应以 agent_settled 为准，而不是 agent_end：

{"type": "agent_settled"}

4.3 turn_start / turn_end

一个 turn 由一条 assistant 响应及其引发的工具调用与结果组成：

{"type": "turn_start"}

{
  "type": "turn_end",
  "message": {...},
  "toolResults": [...]
}

4.4 message_start / message_end

消息开始与完成时发出，message 字段是 AgentMessage：

{"type": "message_start", "message": {...}}
{"type": "message_end", "message": {...}}

4.5 message_update（流式）

在 assistant 消息流式输出期间发出，只包含增量（delta）事件，不含累计消息快照：

{
  "type": "message_update",
  "usage": {
    "input": 100,
    "output": 1,
    "cacheRead": 0,
    "cacheWrite": 0,
    "totalTokens": 101,
    "cost": {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0, "total": 0}
  },
  "assistantMessageEvent": {
    "type": "text_delta",
    "contentIndex": 0,
    "delta": "Hello "
  }
}

assistantMessageEvent 字段是以下 delta 类型之一：

类型	说明
text_start	文本内容块开始
text_delta	文本内容块增量
text_end	文本内容块结束
thinking_start	thinking 块开始
thinking_delta	thinking 内容增量
thinking_end	thinking 块结束
toolcall_start	工具调用开始（含 id 与 toolName）
toolcall_delta	工具调用参数增量
toolcall_end	工具调用结束（含完整 toolCall 对象）

流式输出一个文本响应的示例：

{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"text_start","contentIndex":0}}
{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":"Hello"}}
{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":" world"}}
{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"text_end","contentIndex":0,"content":"Hello world"}}

顶层 usage 字段是提供方报告的最新累计用量；部分提供方在流式期间不报告 usage，该字段可能一直为零直到完成。

工具调用开始的示例：

{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"toolcall_start","contentIndex":1,"id":"call_abc123","toolName":"write"}}

客户端组装规则（重要）：message_update 有意省略了旧的累计 message 字段与 assistantMessageEvent.partial。需要实时部分消息的客户端，必须用 contentIndex 从 message_start 和后续事件自行拼装；以 message_end.message 为权威版本。工具调用方面：toolcall_start 提供调用 id 与 toolName，toolcall_delta.delta 需自行缓冲作为参数，toolcall_end.toolCall 是完整调用对象。

4.6 bash_execution_update

直接 bash 命令的每个输出块都会发出一次；id 与命令的 id 一致，方便客户端把输出关联到正确命令。命令运行期间事件流会输出全部输出，即使最终 bash 响应的 output 被截断：

{
  "type": "bash_execution_update",
  "id": "req-1",
  "delta": "total 48\n"
}

4.7 tool_execution_start / tool_execution_update / tool_execution_end

工具开始、流式进度、完成执行时发出：

{
  "type": "tool_execution_start",
  "toolCallId": "call_abc123",
  "toolName": "bash",
  "args": {"command": "ls -la"}
}

执行期间 tool_execution_update 流式给出部分结果（例如 bash 输出边到边显示）：

{
  "type": "tool_execution_update",
  "toolCallId": "call_abc123",
  "toolName": "bash",
  "args": {"command": "ls -la"},
  "partialResult": {
    "content": [{"type": "text", "text": "partial output so far..."}],
    "details": {"truncation": null, "fullOutputPath": null}
  }
}

完成时：

{
  "type": "tool_execution_end",
  "toolCallId": "call_abc123",
  "toolName": "bash",
  "result": {
    "content": [{"type": "text", "text": "total 48\n..."}],
    "details": {...}
  },
  "isError": false
}

用 toolCallId 关联同一工具调用的事件序列。tool_execution_update 中的 partialResult 是到目前为止的累计输出（不是 delta），客户端每次更新直接替换显示即可。

4.8 queue_update

待处理的 steering 或 follow-up 队列变化时发出：

{
  "type": "queue_update",
  "steering": ["Focus on error handling"],
  "followUp": ["After that, summarize the result"]
}

4.9 compaction_start / compaction_end

手动或自动压缩运行时发出：

{"type": "compaction_start", "reason": "threshold"}

reason 取值为 "manual"、"threshold" 或 "overflow"。

{
  "type": "compaction_end",
  "reason": "threshold",
  "result": {
    "summary": "Summary of conversation...",
    "firstKeptEntryId": "abc123",
    "tokensBefore": 150000,
    "estimatedTokensAfter": 32000,
    "usage": {
      "input": 32000,
      "output": 1200,
      "cacheRead": 0,
      "cacheWrite": 0,
      "totalTokens": 33200,
      "cost": {"input": 0.01, "output": 0.02, "cacheRead": 0, "cacheWrite": 0, "total": 0.03}
    },
    "details": {}
  },
  "aborted": false,
  "willRetry": false
}

状态语义：

• reason 为 "overflow" 且压缩成功时，willRetry 为 true，Agent 会自动重试该 prompt；
• 压缩被中止时，result 为 null 且 aborted 为 true；
• 压缩失败（如 API 配额超限）时，result 为 null、aborted 为 false，errorMessage 包含错误描述。

4.10 auto_retry_start / auto_retry_end

瞬时错误（overloaded、rate limit、5xx）触发自动重试时发出：

{
  "type": "auto_retry_start",
  "attempt": 1,
  "maxAttempts": 3,
  "delayMs": 2000,
  "errorMessage": "529 {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}"
}

{
  "type": "auto_retry_end",
  "success": true,
  "attempt": 2
}

最终失败（超过最大重试次数）：

{
  "type": "auto_retry_end",
  "success": false,
  "attempt": 3,
  "finalError": "529 overloaded_error: Overloaded"
}

4.11 summarization_retry_*

compaction 或分支摘要在遇到提供方瞬时错误后重试时发出。这些事件与 assistant 轮次自动重试共用同一套重试配置。

{
  "type": "summarization_retry_scheduled",
  "attempt": 1,
  "maxAttempts": 3,
  "delayMs": 2000,
  "errorMessage": "terminated"
}

{
  "type": "summarization_retry_attempt_start",
  "source": "compaction",
  "reason": "threshold"
}

分支摘要场景下 source 为 "branchSummary" 且没有 reason 字段：

{
  "type": "summarization_retry_finished"
}

4.12 extension_error

扩展抛出错误时发出：

{
  "type": "extension_error",
  "extensionPath": "/path/to/extension.ts",
  "event": "tool_call",
  "error": "Error message..."
}

五、扩展 UI 子协议

扩展可以通过 ctx.ui.select()、ctx.ui.confirm() 等方法请求用户交互。在 RPC 模式下，这些调用被翻译成基础命令/事件流之上的一个请求/响应子协议。

方法分两类：

• Dialog 方法（select、confirm、input、editor）：向 stdout 发出 extension_ui_request，然后阻塞，直到客户端从 stdin 发回带匹配 id 的 extension_ui_response。
• Fire-and-forget 方法（notify、setStatus、setWidget、setTitle、set_editor_text）：向 stdout 发出 extension_ui_request，但不期望响应。客户端可以展示信息，也可以直接忽略。

如果 dialog 方法带 timeout 字段，超时后 agent 侧会自动以默认值解析，客户端无需自行跟踪超时。这一点在源码中有对应实现：rpc-mode.ts 的 createDialogPromise 为每个 dialog 请求生成一个 crypto.randomUUID() 作为 id，注册进 pendingExtensionRequests 映射，并用 setTimeout 在超时后 resolve(defaultValue)；stdin 侧收到 extension_ui_response 后按 id 找回并解析（rpc-mode.ts）。

部分 ExtensionUIContext 方法因为在 RPC 模式下没有直接 TUI 访问而不支持或降级：

• custom() 返回 undefined
• setWorkingMessage()、setWorkingIndicator()、setFooter()、setHeader()、setEditorComponent()、setToolsExpanded() 是 no-op
• getEditorText() 返回 ""
• getToolsExpanded() 返回 false
• pasteToEditor() 降级为 setEditorText()（没有粘贴/折叠处理）
• getAllThemes() 返回 []
• getTheme() 返回 undefined
• setTheme() 返回 { success: false, error: "..." }

注意：RPC 模式下 ctx.mode 为 "rpc"，ctx.hasUI 为 true（因为 dialog 与 fire-and-forget 方法通过该子协议可用）。需要真实终端的 TUI 专属功能（如 custom()）请用 ctx.mode === "tui" 做判断保护。

5.1 扩展 UI 请求（stdout）

所有请求都带 type: "extension_ui_request"、唯一 id 与 method 字段。

select

让用户从列表中选择。带 timeout（毫秒）的 dialog 会在客户端超时未响应时自动以 undefined 解析。

{
  "type": "extension_ui_request",
  "id": "uuid-1",
  "method": "select",
  "title": "Allow dangerous command?",
  "options": ["Allow", "Block"],
  "timeout": 10000
}

期望响应：带 value（选中的选项字符串）或 cancelled: true 的 extension_ui_response。

confirm

是/否确认：

{
  "type": "extension_ui_request",
  "id": "uuid-2",
  "method": "confirm",
  "title": "Clear session?",
  "message": "All messages will be lost.",
  "timeout": 5000
}

期望响应：confirmed: true/false 或 cancelled: true。

input

自由文本输入：

{
  "type": "extension_ui_request",
  "id": "uuid-3",
  "method": "input",
  "title": "Enter a value",
  "placeholder": "type something..."
}

期望响应：value（输入文本）或 cancelled: true。

editor

打开可预填内容的多行文本编辑器：

{
  "type": "extension_ui_request",
  "id": "uuid-4",
  "method": "editor",
  "title": "Edit some text",
  "prefill": "Line 1\nLine 2\nLine 3"
}

期望响应：value（编辑后的文本）或 cancelled: true。

notify

显示通知。Fire-and-forget，无需响应：

{
  "type": "extension_ui_request",
  "id": "uuid-5",
  "method": "notify",
  "message": "Command blocked by user",
  "notifyType": "warning"
}

notifyType 为 "info"、"warning" 或 "error"，省略时默认 "info"。

setStatus

设置/清除页脚状态栏中的条目。Fire-and-forget：

{
  "type": "extension_ui_request",
  "id": "uuid-6",
  "method": "setStatus",
  "statusKey": "my-ext",
  "statusText": "Turn 3 running..."
}

发送 statusText: undefined（或省略）即可清除该 key 对应的状态条目。

setWidget

设置/清除显示在编辑器上或下方的部件（若干文本行）。Fire-and-forget：

{
  "type": "extension_ui_request",
  "id": "uuid-7",
  "method": "setWidget",
  "widgetKey": "my-ext",
  "widgetLines": ["--- My Widget ---", "Line 1", "Line 2"],
  "widgetPlacement": "aboveEditor"
}

发送 widgetLines: undefined（或省略）清除部件。widgetPlacement 为 "aboveEditor"（默认）或 "belowEditor"。RPC 模式只支持字符串数组，组件工厂函数会被忽略。

setTitle

设置终端窗口/标签页标题。Fire-and-forget：

{
  "type": "extension_ui_request",
  "id": "uuid-8",
  "method": "setTitle",
  "title": "pi - my project"
}

set_editor_text

设置输入编辑器中的文本。Fire-and-forget：

{
  "type": "extension_ui_request",
  "id": "uuid-9",
  "method": "set_editor_text",
  "text": "prefilled text for the user"
}

5.2 扩展 UI 响应（stdin）

响应只针对 dialog 方法（select、confirm、input、editor）发送，id 必须与请求匹配。

数值型响应（select、input、editor）：

{"type": "extension_ui_response", "id": "uuid-1", "value": "Allow"}

确认型响应（confirm）：

{"type": "extension_ui_response", "id": "uuid-2", "confirmed": true}

取消响应（任意 dialog）：

{"type": "extension_ui_response", "id": "uuid-3", "cancelled": true}

取消时扩展收到 undefined（select/input/editor）或 false（confirm）。

处理扩展 UI 子协议的完整示例见 rpc-extension-ui.ts，可与 examples/extensions 目录下的 rpc-demo 扩展配合阅读。

六、错误处理

命令失败时返回 success: false 的响应：

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

JSON 解析失败（command 固定为 "parse"）：

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

从源码看（rpc-mode.ts），stdin 每一行先尝试 JSON.parse，失败即发出上述 parse 响应；命令处理抛出的任何异常也会被捕获并转换为 success: false 的响应，进程不会因此退出。此外还有两条生命周期细节值得客户端了解：

• stdin 关闭即退出：process.stdin 的 end 事件触发 shutdown()；
• 信号处理：注册了 SIGTERM（以及非 Windows 平台的 SIGHUP）处理器，先 killTrackedDetachedChildren() 清理跟踪中的分离子进程，再以退出码 143（SIGTERM）或 129（SIGHUP）退出。

七、核心类型参考

相关源码文件：

• packages/ai/src/types.ts — Model、UserMessage、AssistantMessage、ToolResultMessage
• packages/agent/src/types.ts — AgentMessage、AgentEvent
• messages.ts — BashExecutionMessage
• json-event.ts — JsonAgentSessionEvent
• rpc-types.ts — RPC 命令/响应类型、扩展 UI 请求/响应类型

7.1 Model

{
  "id": "claude-sonnet-4-20250514",
  "name": "Claude Sonnet 4",
  "api": "anthropic-messages",
  "provider": "anthropic",
  "baseUrl": "https://api.anthropic.com",
  "reasoning": true,
  "input": ["text", "image"],
  "contextWindow": 200000,
  "maxTokens": 16384,
  "cost": {
    "input": 3.0,
    "output": 15.0,
    "cacheRead": 0.3,
    "cacheWrite": 3.75
  }
}

7.2 Message Types

UserMessage

{
  "role": "user",
  "content": "Hello!",
  "timestamp": 1733234567890,
  "attachments": []
}

content 字段可以是字符串，也可以是 TextContent/ImageContent 块数组。

AssistantMessage

{
  "role": "assistant",
  "content": [
    {"type": "text", "text": "Hello! How can I help?"},
    {"type": "thinking", "thinking": "User is greeting me..."},
    {"type": "toolCall", "id": "call_123", "name": "bash", "arguments": {"command": "ls"}}
  ],
  "api": "anthropic-messages",
  "provider": "anthropic",
  "model": "claude-sonnet-4-20250514",
  "usage": {
    "input": 100,
    "output": 50,
    "cacheRead": 0,
    "cacheWrite": 0,
    "cost": {"input": 0.0003, "output": 0.00075, "cacheRead": 0, "cacheWrite": 0, "total": 0.00105}
  },
  "stopReason": "stop",
  "timestamp": 1733234567890
}

stopReason 取值："stop"、"length"、"toolUse"、"error"、"aborted"。

ToolResultMessage

{
  "role": "toolResult",
  "toolCallId": "call_123",
  "toolName": "bash",
  "content": [{"type": "text", "text": "total 48\ndrwxr-xr-x ..."}],
  "usage": {
    "input": 100,
    "output": 50,
    "cacheRead": 0,
    "cacheWrite": 0,
    "totalTokens": 150,
    "cost": {"input": 0.0003, "output": 0.00075, "cacheRead": 0, "cacheWrite": 0, "total": 0.00105}
  },
  "isError": false,
  "timestamp": 1733234567890
}

usage 可选，报告工具内部嵌套执行的 LLM 工作；存在时会计入会话 token 与成本总计。

BashExecutionMessage

由 bash RPC 命令创建（不是 LLM 工具调用）：

{
  "role": "bashExecution",
  "command": "ls -la",
  "output": "total 48\ndrwxr-xr-x ...",
  "exitCode": 0,
  "cancelled": false,
  "truncated": false,
  "fullOutputPath": null,
  "timestamp": 1733234567890
}

Attachment

{
  "id": "img1",
  "type": "image",
  "fileName": "photo.jpg",
  "mimeType": "image/jpeg",
  "size": 102400,
  "content": "base64-encoded-data...",
  "extractedText": null,
  "preview": null
}

八、客户端示例

8.1 基础客户端（Python）

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 prompt
send({"type": "prompt", "message": "Hello!"})

# Process events
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

8.2 交互式客户端（Node.js）

完整交互示例见 rpc-example.ts，带类型的客户端实现见 rpc-client.ts。

const { spawn } = require("child_process");
const { StringDecoder } = require("string_decoder");

const agent = spawn("pi", ["--mode", "rpc", "--no-session"]);

function attachJsonlReader(stream, onLine) {
    const decoder = new StringDecoder("utf8");
    let buffer = "";

    stream.on("data", (chunk) => {
        buffer += typeof chunk === "string" ? chunk : decoder.write(chunk);

        while (true) {
            const newlineIndex = buffer.indexOf("\n");
            if (newlineIndex === -1) break;

            let line = buffer.slice(0, newlineIndex);
            buffer = buffer.slice(newlineIndex + 1);
            if (line.endsWith("\r")) line = line.slice(0, -1);
            onLine(line);
        }
    });

    stream.on("end", () => {
        buffer += decoder.end();
        if (buffer.length > 0) {
            onLine(buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer);
        }
    });
}

attachJsonlReader(agent.stdout, (line) => {
    const event = JSON.parse(line);

    if (event.type === "message_update") {
        const { assistantMessageEvent } = event;
        if (assistantMessageEvent.type === "text_delta") {
            process.stdout.write(assistantMessageEvent.delta);
        }
    }
});

// Send prompt
agent.stdin.write(JSON.stringify({ type: "prompt", message: "Hello" }) + "\n");

// Abort on Ctrl+C
process.on("SIGINT", () => {
    agent.stdin.write(JSON.stringify({ type: "abort" }) + "\n");
});

这个 Node.js 示例手动实现的 LF-only JSONL 读取逻辑与仓库内 jsonl.ts 的 attachJsonlLineReader 完全同构，再次强调：不要用 Node readline 解析 RPC 输出。

九、实现导航

想在仓库中继续深入，建议按以下路径阅读：

文件	内容
rpc-mode.ts	RPC 模式入口 runRpcMode：命令分发（handleCommand 大 switch）、stdin JSONL 读取、扩展 UI 上下文构造、信号处理与关机流程
rpc-types.ts	RpcCommand、RpcResponse、RpcSessionState、扩展 UI 请求/响应等全部协议类型
jsonl.ts	严格 LF-only JSONL 序列化与行读取器
rpc-client.ts	子进程方案的带类型 TypeScript 客户端
json-event.ts	会话事件到 JSON 的序列化（toJsonEvent）
agent-session.ts	供 Node.js 进程内直接使用的 AgentSession API
rpc-example.ts	完整的交互式 RPC 客户端示例
rpc-extension-ui.ts	扩展 UI 子协议的处理示例
rpc.md	本文对应的官方协议文档

小结

RPC 模式是 pi coding-agent 面向嵌入场景的正式接口：一条 pi --mode rpc 命令即可把整个 coding agent（多模型切换、thinking 等级、steering/follow-up 队列、上下文压缩、自动重试、bash 直通执行、会话 fork/clone、扩展 UI 交互）开放为 stdin/stdout 上的 JSONL 协议。实现它有三个工程要点值得记住：严格 LF 帧语义（拒绝 readline）、以 agent_settled 而非 agent_end 判断 run 是否彻底收敛、以及 message_update 只传增量需由客户端用 contentIndex 自行拼装。掌握这三点后，你就可以基于 rpc-example.ts 与 rpc-client.ts 构建自己的 IDE 插件、Web 后端或自动化流水线中的 pi 通道了。