21 RPC 模式
21.1 RPC 模式概述
RPC 模式通过 stdin/stdout 上的 JSON 协议实现 Pi 编程助手的无头（headless）操作。适用于将 Agent 嵌入 IDE、构建自定义 UI、或集成到其他应用中。
Note
Node.js/TypeScript 用户：如果你在构建 Node.js 应用，考虑直接使用 @earendil-works/pi-coding-agent 的 AgentSession，而不是启动子进程。对于子进程方式的 TypeScript 客户端，参见 rpc-client.ts。
21.2 启动 RPC 模式
pi --mode rpc [options]常用选项：
| 选项 | 说明 |
|---|---|
--provider <name> |
设置 LLM 提供者 |
--model <pattern> |
模型模式或 ID |
--name <name> / -n |
设置会话显示名称 |
--no-session |
禁用会话持久化 |
--session-dir <path> |
自定义会话存储目录 |
21.3 协议概述
RPC 模式使用严格的 JSONL 语义：
- 命令：发送到 stdin 的 JSON 对象，每行一个
- 响应：
type: "response"的 JSON 对象，表示命令成功/失败 - 事件：Agent 事件以 JSON 行流式输出到 stdout
所有命令支持可选的 id 字段用于请求/响应关联。
21.3.1 Framing（帧协议）
RPC 模式使用 LF（\n）作为唯一的记录分隔符。
Warning
Node.js 的 readline 模块不符合 RPC 协议规范，因为它也会在 U+2028 和 U+2029 处分割，而这些是 JSON 字符串中的合法字符。请自行实现基于 \n 的分割。
客户端规则：
- 仅在
\n处分割记录 - 可接受
\r\n输入（去除尾部\r） - 不要使用通用行读取器（可能将 Unicode 分隔符视为换行）
21.4 命令
21.4.1 Prompting（提示类命令）
21.4.1.1 prompt
发送用户提示：
{"id": "req-1", "type": "prompt", "message": "Hello, world!"}带图片：
{
"type": "prompt",
"message": "What's in this image?",
"images": [{"type": "image", "data": "base64-data", "mimeType": "image/png"}]
}流式传输期间必须指定 streamingBehavior：
{"type": "prompt", "message": "New instruction", "streamingBehavior": "steer"}"steer"：在当前助手回合完成工具调用后投递"followUp"：等 Agent 完成后投递
响应：
{"id": "req-1", "type": "response", "command": "prompt", "success": true}success: true 表示提示被接受/排队/处理。success: false 表示被拒绝。
21.4.1.2 steer
{"type": "steer", "message": "Stop and do this instead"}21.4.1.3 follow_up
{"type": "follow_up", "message": "After you're done, also do this"}21.4.1.4 abort
{"type": "abort"}21.4.1.5 new_session
{"type": "new_session"}带父会话追踪：
{"type": "new_session", "parentSession": "/path/to/parent-session.jsonl"}21.4.2 State（状态类命令）
21.4.2.1 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
}
}21.4.2.2 get_messages
{"type": "get_messages"}21.4.3 Model（模型类命令）
21.4.3.1 set_model
{"type": "set_model", "provider": "anthropic", "modelId": "claude-sonnet-4-5"}21.4.3.2 cycle_model
{"type": "cycle_model"}21.4.3.3 get_available_models
{"type": "get_available_models"}21.4.4 Thinking（思考类命令）
21.4.4.1 set_thinking_level
{"type": "set_thinking_level", "level": "high"}级别："off"、"minimal"、"low"、"medium"、"high"、"xhigh"、"max"
21.4.4.2 cycle_thinking_level
{"type": "cycle_thinking_level"}21.4.4.3 get_available_thinking_levels
{"type": "get_available_thinking_levels"}21.4.5 Queue Modes（队列模式命令）
21.4.5.1 set_steering_mode
{"type": "set_steering_mode", "mode": "one-at-a-time"}模式："all"（全部投递）| "one-at-a-time"（每回合一个，默认）
21.4.5.2 set_follow_up_mode
{"type": "set_follow_up_mode", "mode": "one-at-a-time"}21.4.6 Compaction（压缩命令）
21.4.6.1 compact
{"type": "compact"}带自定义指令：
{"type": "compact", "customInstructions": "Focus on code changes"}21.4.6.2 set_auto_compaction
{"type": "set_auto_compaction", "enabled": true}21.4.7 Retry（重试命令）
21.4.7.1 retry
{"type": "retry"}21.4.8 Bash
21.4.8.1 set_bash_mode
{"type": "set_bash_mode", "enabled": true}21.4.9 Session（会话命令）
| 命令 | 说明 |
|---|---|
new_session |
新建会话 |
switch_session |
切换到指定会话文件 |
fork |
从指定条目分支 |
set_session_name |
设置会话名称 |
get_session_tree |
获取会话树结构 |
resume_session |
恢复会话 |
21.5 事件
21.5.1 Event Types（事件类型完整列表）
21.5.1.1 agent_start
Agent 开始处理提示时触发：
{"type": "agent_start"}21.5.1.2 agent_end
Agent 处理完成：
{"type": "agent_end", "messages": [...]}21.5.1.3 agent_settled
Agent 完全安定（无更多重试/压缩/跟进）：
{"type": "agent_settled"}21.5.1.4 turn_start / turn_end
每个 LLM 响应回合：
{"type": "turn_start"}
{"type": "turn_end", "message": {...}, "toolResults": [...]}21.5.1.5 message_start / message_end
消息生命周期：
{"type": "message_start", "message": {...}}
{"type": "message_end", "message": {...}}21.5.1.6 message_update (Streaming)
流式文本增量：
{"type": "message_update", "message": {...}, "assistantMessageEvent": {"type": "text_delta", "delta": "Hello"}}思考增量：
{"type": "message_update", "message": {...}, "assistantMessageEvent": {"type": "thinking_delta", "delta": "Let me think..."}}21.5.1.7 bash_execution_update
bash 命令执行中的流式输出：
{"type": "bash_execution_update", "id": "bash-1", "output": "file1.txt\nfile2.txt\n"}21.5.1.8 tool_execution_start / tool_execution_update / tool_execution_end
工具执行生命周期：
{"type": "tool_execution_start", "toolCallId": "call-1", "toolName": "read", "args": {"path": "/foo"}}
{"type": "tool_execution_update", "toolCallId": "call-1", "toolName": "read", "partialResult": "..."}
{"type": "tool_execution_end", "toolCallId": "call-1", "toolName": "read", "result": {...}, "isError": false}21.5.1.9 queue_update
队列变更时触发：
{"type": "queue_update", "steering": ["msg1"], "followUp": ["msg2"]}21.5.1.10 compaction_start / compaction_end
{"type": "compaction_start", "reason": "manual"}
{"type": "compaction_end", "reason": "manual", "result": {...}, "aborted": false}21.5.1.11 auto_retry_start / auto_retry_end
{"type": "auto_retry_start", "attempt": 1, "maxAttempts": 3, "delayMs": 1000, "errorMessage": "..."}
{"type": "auto_retry_end", "success": true, "attempt": 1}21.5.1.12 summarization_retry_scheduled / attempt_start / finished
{"type": "summarization_retry_scheduled", "attempt": 1, "maxAttempts": 2, "delayMs": 5000, "errorMessage": "..."}
{"type": "summarization_retry_attempt_start", "source": "compaction", "reason": "manual"}
{"type": "summarization_retry_finished"}21.5.1.13 extension_error
扩展错误：
{"type": "extension_error", "message": "...", "stack": "..."}21.6 Extension UI Protocol
扩展通过 RPC 协议进行 UI 交互。
21.6.1 Extension UI Requests（stdout）
当扩展调用 ctx.ui.confirm() 等方法时，Pi 向 stdout 发送 UI 请求：
{
"type": "extension_ui_request",
"id": "ui-1",
"ui_type": "confirm",
"title": "Dangerous!",
"message": "Allow rm -rf?"
}21.6.2 Extension UI Responses（stdin）
客户端处理后通过 stdin 返回：
{
"type": "extension_ui_response",
"id": "ui-1",
"result": true
}21.7 错误处理
命令执行失败时返回错误响应：
{
"type": "response",
"command": "prompt",
"success": false,
"error": "Prompt rejected by preflight"
}运行时错误通过 extension_error 事件报告。
21.8 类型定义
21.8.1 Model
interface Model {
provider: string;
id: string;
name: string;
reasoning: boolean;
input: string[];
contextWindow: number;
maxTokens: number;
cost: { input: number; output: number; cacheRead: number; cacheWrite: number };
}21.8.2 UserMessage
interface UserMessage {
role: "user";
content: ContentBlock[];
}21.8.3 AssistantMessage
interface AssistantMessage {
role: "assistant";
content: ContentBlock[];
model?: string;
thinkingLevel?: string;
}21.8.4 ToolResultMessage
interface ToolResultMessage {
role: "tool";
toolCallId: string;
toolName: string;
content: ContentBlock[];
isError: boolean;
}21.8.5 BashExecutionMessage
interface BashExecutionMessage {
role: "bash";
command: string;
stdout: string;
stderr: string;
exitCode: number;
}21.8.6 Attachment
interface Attachment {
type: "image";
data: string; // base64 编码
mimeType: string;
}21.9 示例：Python 客户端
import subprocess
import json
import sys
import threading
class PiRPCClient:
def __init__(self):
self.proc = subprocess.Popen(
["pi", "--mode", "rpc", "--no-session"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
self.req_id = 0
self.running = True
self.reader_thread = threading.Thread(target=self._read_loop, daemon=True)
self.reader_thread.start()
def _read_loop(self):
for line in self.proc.stdout:
line = line.strip()
if not line:
continue
msg = json.loads(line)
self._handle_message(msg)
def _handle_message(self, msg):
msg_type = msg.get("type")
if msg_type == "message_update":
event = msg.get("assistantMessageEvent", {})
if event.get("type") == "text_delta":
print(event["delta"], end="", flush=True)
elif msg_type == "agent_end":
print("\n--- Done ---")
elif msg_type == "response":
if not msg.get("success"):
print(f"Error: {msg.get('error')}")
def send(self, command):
self.req_id += 1
command["id"] = f"req-{self.req_id}"
self.proc.stdin.write(json.dumps(command) + "\n")
self.proc.stdin.flush()
def prompt(self, text):
self.send({"type": "prompt", "message": text})
def wait_for_completion(self):
self.proc.wait()
# 使用
client = PiRPCClient()
client.prompt("What files are in the current directory?")21.10 示例：Node.js 客户端
import { spawn } from "child_process";
const pi = spawn("pi", ["--mode", "rpc", "--no-session"], {
stdio: ["pipe", "pipe", "inherit"],
});
let buffer = "";
pi.stdout.on("data", (data) => {
buffer += data.toString();
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
if (!line.trim()) continue;
const msg = JSON.parse(line);
switch (msg.type) {
case "message_update":
const event = msg.assistantMessageEvent;
if (event?.type === "text_delta") {
process.stdout.write(event.delta);
}
break;
case "tool_execution_start":
console.log(`\n[Tool: ${msg.toolName}]`);
break;
case "agent_end":
console.log("\n--- Done ---");
break;
case "response":
if (!msg.success) {
console.error(`Error: ${msg.error}`);
}
break;
}
}
});
// 发送提示
function send(command: object) {
pi.stdin.write(JSON.stringify(command) + "\n");
}
send({ type: "prompt", message: "List all TypeScript files" });