Ipc Communication

IPC Communication Skill

Expert guidance for implementing Inter-Process Communication between Rust Orchestrator and Platform Agents using JSON protocol via stdin/stdout.

Overview

WeReply uses JSON-based IPC protocol for communication:

• Protocol: JSON messages via stdin/stdout
• Direction: Bidirectional (Orchestrator ↔ Agent)
• Message Format: Line-delimited JSON (one message per line)
• Character Encoding: UTF-8

Message Protocol Design

Message Types

use serde::{Deserialize, Serialize};
use specta::Type;

// Agent → Orchestrator 消息
#[derive(Serialize, Deserialize, Type, Debug)]
#[serde(tag = "type")]
pub enum AgentMessage {
    MessageNew {
        content: String,
        sender: String,
        timestamp: String,
    },
    CommandResponse {
        success: bool,
        #[serde(skip_serializing_if = "Option::is_none")]
        error: Option<String>,
    },
    HealthStatus {
        status: String,  // "ok", "degraded", "error"
        agent_type: String,  // "windows_wxauto", "macos_accessibility"
    },
    Heartbeat {
        timestamp: f64,
    },
    Error {
        message: String,
        #[serde(skip_serializing_if = "Option::is_none")]
        code: Option<String>,
    },
}

// Orchestrator → Agent 消息
#[derive(Serialize, Deserialize, Type, Debug)]
#[serde(tag = "type")]
pub enum OrchestratorCommand {
    WriteInput {
        content: String,
    },
    ClearInput,
    HealthCheck,
    Stop,
}

Message Format Examples

// Agent → Orchestrator: 新消息
{
  "type": "MessageNew",
  "content": "你好，最近怎么样？",
  "sender": "张三",
  "timestamp": "2024-01-23T10:30:00"
}

// Orchestrator → Agent: 写入输入框
{
  "type": "WriteInput",
  "content": "很好，谢谢！"
}

// Agent → Orchestrator: 命令响应
{
  "type": "CommandResponse",
  "success": true
}

// Agent → Orchestrator: 错误
{
  "type": "Error",
  "message": "无法访问微信窗口",
  "code": "ACCESS_DENIED"
}

Rust Orchestrator Implementation

Agent Process Manager

use tokio::process::{Child, Command};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use std::process::Stdio;

pub struct AgentProcess {
    child: Child,
    stdin: tokio::process::ChildStdin,
    stdout_reader: BufReader<tokio::process::ChildStdout>,
}

impl AgentProcess {
    pub async fn spawn(agent_path: &str) -> anyhow::Result<Self> {
        let mut child = Command::new(agent_path)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()?;

        let stdin = child.stdin.take()
            .ok_or_else(|| anyhow!("无法获取 Agent stdin"))?;

        let stdout = child.stdout.take()
            .ok_or_else(|| anyhow!("无法获取 Agent stdout"))?;

        let stdout_reader = BufReader::new(stdout);

        Ok(Self {
            child,
            stdin,
            stdout_reader,
        })
    }

    pub async fn send_command(&mut self, command: OrchestratorCommand) -> anyhow::Result<()> {
        let json = serde_json::to_string(&command)?;
        self.stdin.write_all(json.as_bytes()).await?;
        self.stdin.write_all(b"\n").await?;
        self.stdin.flush().await?;
        Ok(())
    }

    pub async fn read_message(&mut self) -> anyhow::Result<AgentMessage> {
        let mut line = String::new();
        self.stdout_reader.read_line(&mut line).await?;

        if line.is_empty() {
            return Err(anyhow!("Agent 已关闭输出"));
        }

        let message = serde_json::from_str(&line)?;
        Ok(message)
    }

    pub async fn kill(&mut self) -> anyhow::Result<()> {
        self.child.kill().await?;
        Ok(())
    }
}

Message Handler Pattern

use tokio::sync::mpsc;

pub struct AgentHandler {
    agent: AgentProcess,
    message_tx: mpsc::UnboundedSender<AgentMessage>,
}

impl AgentHandler {
    pub async fn start(agent_path: &str) -> anyhow::Result<(Self, mpsc::UnboundedReceiver<AgentMessage>)> {
        let agent = AgentProcess::spawn(agent_path).await?;
        let (message_tx, message_rx) = mpsc::unbounded_channel();

        let handler = Self {
            agent,
            message_tx,
        };

        Ok((handler, message_rx))
    }

    pub async fn run(mut self) {
        loop {
            match self.agent.read_message().await {
                Ok(message) => {
                    if self.message_tx.send(message).is_err() {
                        tracing::error!("消息接收器已关闭");
                        break;
                    }
                }
                Err(e) => {
                    tracing::error!(error = %e, "读取 Agent 消息失败");
                    break;
                }
            }
        }
    }

    pub async fn send_command(&mut self, command: OrchestratorCommand) -> anyhow::Result<()> {
        self.agent.send_command(command).await
    }
}

Error Handling with Timeout

use tokio::time::{timeout, Duration};

pub async fn send_command_with_timeout(
    agent: &mut AgentProcess,
    command: OrchestratorCommand,
    timeout_secs: u64,
) -> anyhow::Result<()> {
    timeout(
        Duration::from_secs(timeout_secs),
        agent.send_command(command)
    )
    .await
    .map_err(|_| anyhow!("发送命令超时"))?
}

pub async fn read_message_with_timeout(
    agent: &mut AgentProcess,
    timeout_secs: u64,
) -> anyhow::Result<AgentMessage> {
    timeout(
        Duration::from_secs(timeout_secs),
        agent.read_message()
    )
    .await
    .map_err(|_| anyhow!("读取消息超时"))?
}

Python Agent Implementation

Message Sender Pattern

import json
import sys
from typing import Dict, Any

class MessageSender:
    """发送消息到 Rust Orchestrator (stdout)"""

    @staticmethod
    def send_message(message: Dict[str, Any]):
        """
        发送 JSON 消息到 stdout

        Args:
            message: 消息字典，必须包含 'type' 字段
        """
        json_str = json.dumps(message, ensure_ascii=False)
        print(json_str, flush=True)

    @staticmethod
    def send_message_new(content: str, sender: str, timestamp: str):
        """发送新消息通知"""
        message = {
            "type": "MessageNew",
            "content": content,
            "sender": sender,
            "timestamp": timestamp
        }
        MessageSender.send_message(message)

    @staticmethod
    def send_command_response(success: bool, error: str = None):
        """发送命令响应"""
        message = {
            "type": "CommandResponse",
            "success": success
        }
        if error:
            message["error"] = error
        MessageSender.send_message(message)

    @staticmethod
    def send_error(error_message: str, code: str = None):
        """发送错误"""
        message = {
            "type": "Error",
            "message": error_message
        }
        if code:
            message["code"] = code
        MessageSender.send_message(message)

    @staticmethod
    def send_heartbeat(timestamp: float):
        """发送心跳"""
        message = {
            "type": "Heartbeat",
            "timestamp": timestamp
        }
        MessageSender.send_message(message)

Command Receiver Pattern

import threading
from typing import Callable, Dict, Any

class CommandReceiver:
    """从 Rust Orchestrator 接收命令 (stdin)"""

    def __init__(self):
        self.handlers: Dict[str, Callable] = {}
        self.running = True

    def register_handler(self, command_type: str, handler: Callable):
        """注册命令处理器"""
        self.handlers[command_type] = handler

    def start_listening(self):
        """开始监听命令（阻塞）"""
        try:
            for line in sys.stdin:
                if not self.running:
                    break

                try:
                    command = json.loads(line.strip())
                    self.handle_command(command)
                except json.JSONDecodeError as e:
                    MessageSender.send_error(f"JSON 解析失败: {str(e)}", "JSON_PARSE_ERROR")
                except Exception as e:
                    MessageSender.send_error(f"处理命令失败: {str(e)}", "COMMAND_HANDLER_ERROR")

        except KeyboardInterrupt:
            pass
        except Exception as e:
            MessageSender.send_error(f"命令监听错误: {str(e)}", "LISTENER_ERROR")

    def start_listening_async(self):
        """在后台线程中监听命令"""
        thread = threading.Thread(target=self.start_listening, daemon=True)
        thread.start()
        return thread

    def handle_command(self, command: Dict[str, Any]):
        """处理收到的命令"""
        command_type = command.get('type')

        if command_type not in self.handlers:
            MessageSender.send_error(f"未知命令类型: {command_type}", "UNKNOWN_COMMAND")
            return

        try:
            handler = self.handlers[command_type]
            result = handler(command)

            # 如果处理器返回 True，发送成功响应
            if result is True or result is None:
                MessageSender.send_command_response(success=True)
            elif result is False:
                MessageSender.send_command_response(success=False, error="处理失败")

        except Exception as e:
            MessageSender.send_command_response(success=False, error=str(e))

    def stop(self):
        """停止监听"""
        self.running = False

Complete Agent Example

from wechat_monitor import WeChatMonitor
from input_writer import WeChatInputWriter

class Agent:
    def __init__(self):
        self.monitor = WeChatMonitor()
        self.input_writer = WeChatInputWriter()
        self.command_receiver = CommandReceiver()

        # 注册命令处理器
        self.command_receiver.register_handler("WriteInput", self.handle_write_input)
        self.command_receiver.register_handler("ClearInput", self.handle_clear_input)
        self.command_receiver.register_handler("HealthCheck", self.handle_health_check)
        self.command_receiver.register_handler("Stop", self.handle_stop)

    def handle_write_input(self, command: Dict[str, Any]) -> bool:
        """处理写入输入框命令"""
        content = command.get('content', '')
        return self.input_writer.write_to_input(content)

    def handle_clear_input(self, command: Dict[str, Any]) -> bool:
        """处理清空输入框命令"""
        return self.input_writer.clear_input()

    def handle_health_check(self, command: Dict[str, Any]) -> bool:
        """处理健康检查"""
        message = {
            "type": "HealthStatus",
            "status": "ok",
            "agent_type": "windows_wxauto"
        }
        MessageSender.send_message(message)
        return None  # 不发送 CommandResponse

    def handle_stop(self, command: Dict[str, Any]) -> bool:
        """处理停止命令"""
        self.command_receiver.stop()
        return True

    def run(self):
        """启动 Agent"""
        # 启动命令监听（后台线程）
        self.command_receiver.start_listening_async()

        # 启动微信监听（主线程，阻塞）
        self.monitor.start_monitoring()

if __name__ == '__main__':
    agent = Agent()
    agent.run()

Swift Agent Implementation

Message Sender Pattern

import Foundation

class MessageSender {
    static func sendMessage(_ message: [String: Any]) {
        guard let jsonData = try? JSONSerialization.data(withJSONObject: message),
              let jsonString = String(data: jsonData, encoding: .utf8) else {
            return
        }

        print(jsonString)
        fflush(stdout)
    }

    static func sendMessageNew(content: String, sender: String, timestamp: String) {
        let message: [String: Any] = [
            "type": "MessageNew",
            "content": content,
            "sender": sender,
            "timestamp": timestamp
        ]
        sendMessage(message)
    }

    static func sendCommandResponse(success: Bool, error: String? = nil) {
        var message: [String: Any] = [
            "type": "CommandResponse",
            "success": success
        ]
        if let error = error {
            message["error"] = error
        }
        sendMessage(message)
    }

    static func sendError(message errorMessage: String, code: String? = nil) {
        var message: [String: Any] = [
            "type": "Error",
            "message": errorMessage
        ]
        if let code = code {
            message["code"] = code
        }
        sendMessage(message)
    }
}

Command Receiver Pattern

class CommandReceiver {
    private var handlers: [String: (([String: Any]) -> Bool)] = [:]
    private var running = true

    func registerHandler(commandType: String, handler: @escaping ([String: Any]) -> Bool) {
        handlers[commandType] = handler
    }

    func startListening() {
        while running {
            guard let line = readLine() else {
                break
            }

            guard let data = line.data(using: .utf8),
                  let command = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else {
                MessageSender.sendError(message: "JSON 解析失败", code: "JSON_PARSE_ERROR")
                continue
            }

            handleCommand(command)
        }
    }

    func startListeningAsync() {
        DispatchQueue.global(qos: .userInitiated).async {
            self.startListening()
        }
    }

    private func handleCommand(_ command: [String: Any]) {
        guard let commandType = command["type"] as? String else {
            MessageSender.sendError(message: "命令缺少 type 字段", code: "INVALID_COMMAND")
            return
        }

        guard let handler = handlers[commandType] else {
            MessageSender.sendError(message: "未知命令类型: \(commandType)", code: "UNKNOWN_COMMAND")
            return
        }

        let success = handler(command)
        MessageSender.sendCommandResponse(success: success)
    }

    func stop() {
        running = false
    }
}

Message Validation

Rust Validation

pub fn validate_agent_message(message: &AgentMessage) -> Result<(), String> {
    match message {
        AgentMessage::MessageNew { content, sender, .. } => {
            if content.is_empty() {
                return Err("消息内容为空".to_string());
            }
            if sender.is_empty() {
                return Err("发送者为空".to_string());
            }
            if content.len() > 100000 {
                return Err("消息内容过长".to_string());
            }
        }
        AgentMessage::Error { message, .. } => {
            if message.is_empty() {
                return Err("错误消息为空".to_string());
            }
        }
        _ => {}
    }
    Ok(())
}

Python Validation

def validate_command(command: Dict[str, Any]) -> bool:
    """验证命令格式"""
    # 检查必需字段
    if 'type' not in command:
        return False

    cmd_type = command['type']

    # 验证特定命令的字段
    if cmd_type == 'WriteInput':
        if 'content' not in command:
            return False
        content = command['content']
        if len(content) > 10000:  # 最大 10KB
            return False

    return True

Performance Optimization

Batch Message Processing

pub struct MessageBatcher {
    buffer: Vec<AgentMessage>,
    max_batch_size: usize,
    flush_interval: Duration,
    last_flush: Instant,
}

impl MessageBatcher {
    pub fn new(max_batch_size: usize, flush_interval: Duration) -> Self {
        Self {
            buffer: Vec::new(),
            max_batch_size,
            flush_interval,
            last_flush: Instant::now(),
        }
    }

    pub fn add_message(&mut self, message: AgentMessage) -> Option<Vec<AgentMessage>> {
        self.buffer.push(message);

        // 达到批量大小或超时，返回批次
        if self.buffer.len() >= self.max_batch_size
            || self.last_flush.elapsed() >= self.flush_interval
        {
            return Some(self.flush());
        }

        None
    }

    pub fn flush(&mut self) -> Vec<AgentMessage> {
        let messages = std::mem::take(&mut self.buffer);
        self.last_flush = Instant::now();
        messages
    }
}

Async Message Queue

use tokio::sync::mpsc;

pub struct AsyncMessageQueue {
    tx: mpsc::UnboundedSender<AgentMessage>,
    rx: mpsc::UnboundedReceiver<AgentMessage>,
}

impl AsyncMessageQueue {
    pub fn new() -> Self {
        let (tx, rx) = mpsc::unbounded_channel();
        Self { tx, rx }
    }

    pub fn sender(&self) -> mpsc::UnboundedSender<AgentMessage> {
        self.tx.clone()
    }

    pub async fn receive(&mut self) -> Option<AgentMessage> {
        self.rx.recv().await
    }
}

Testing

Mock Agent Process

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{duplex, AsyncWriteExt};

    #[tokio::test]
    async fn test_send_command() {
        let (mut client, mut server) = duplex(1024);

        // 模拟发送命令
        let command = OrchestratorCommand::WriteInput {
            content: "测试内容".to_string(),
        };
        let json = serde_json::to_string(&command).unwrap();
        client.write_all(json.as_bytes()).await.unwrap();
        client.write_all(b"\n").await.unwrap();

        // 模拟接收
        let mut buf = String::new();
        use tokio::io::AsyncBufReadExt;
        let mut reader = BufReader::new(server);
        reader.read_line(&mut buf).await.unwrap();

        let received: OrchestratorCommand = serde_json::from_str(&buf).unwrap();
        match received {
            OrchestratorCommand::WriteInput { content } => {
                assert_eq!(content, "测试内容");
            }
            _ => panic!("错误的命令类型"),
        }
    }
}

When to Use This Skill

Activate this skill when:

• Implementing IPC between Orchestrator and Agents
• Designing message protocols
• Working with stdin/stdout communication
• Handling JSON serialization/deserialization
• Implementing message validation
• Optimizing IPC performance
• Testing Agent communication
• Debugging IPC issues

Converted and distributed by TomeVault — claim your Tome and manage your conversions.