openclaw平替之nanobot源码解析（七）：Gateway与多渠道集成暗

背景StreamJsonRpc 是微软官方维护的用于 .NET 和 TypeScript 的 JSON-RPC 通信库以其强大的类型安全、自动代理生成和成熟的异常处理机制著称。在 HagiCode 项目中为了通过 ACP (Agent Communication Protocol) 与外部 AI 工具如 iflow CLI、OpenCode CLI进行通信并消除早期自定义 JSON-RPC 实现带来的维护成本和潜在 Bug项目决定集成 StreamJsonRpc。然而在集成过程中遇到了流式 JSON-RPC 特有的挑战特别是在处理代理目标绑定和泛型参数识别时。为了解决这些痛点我们做了一个大胆的决定整个构建系统推倒重来。这个决定带来的变化可能比你想象的还要大——稍后我会具体说。关于 HagiCode先介绍一下本文的主角项目如果你在开发中遇到过这些烦恼多项目、多技术栈构建脚本维护成本高CI/CD 流水线配置繁琐每次改都要查文档跨平台兼容性问题层出不穷想让 AI 帮忙写代码但现有工具不够智能那么我们正在做的 HagiCode 可能你会感兴趣。HagiCode 是什么一款 AI 驱动的代码智能助手支持多语言、跨平台的代码生成与优化内置游戏化机制让编码不再枯燥为什么在这里提它本文分享的 StreamJsonRpc 集成方案正是我们在开发 HagiCode 过程中实践总结出来的。如果你觉得这套工程化方案有价值说明我们的技术品味还不错——那么 HagiCode 本身也值得关注一下。想了当前项目处于 ACP 协议集成的关键阶段面临着以下几个技术痛点和架构挑战1. 自定义实现的局限原有的 JSON-RPC 实现位于?src/HagiCode.ClaudeHelper/AcpImp/包含?JsonRpcEndpoint?和?ClientSideConnection?等组件。维护这套自定义代码成本高且缺乏成熟库的高级功能如进度报告、取消支持。2. StreamJsonRpc 集成障碍在尝试将现有的?CallbackProxyTarget?模式迁移到 StreamJsonRpc 时发现?_rpc.AddLocalRpcTarget(target)?方法无法识别通过代理模式创建的目标。具体表现为StreamJsonRpc 无法自动将泛型类型?T?的属性拆分为 RPC 方法参数导致服务器端无法正确处理客户端发起的方法调用。3. 架构分层混乱现有的?ClientSideConnection?混合了传输层WebSocket/Stdio、协议层JSON-RPC和业务层ACP Agent 接口导致职责不清且存在?AcpAgentCallbackRpcAdapter?方法绑定缺失的问题。4. 日志缺失WebSocket 传输层缺少对原始 JSON 内容的日志输出导致在调试 RPC 通信问题时难以定位是序列化问题还是网络问题。解决针对上述问题我们采用了以下系统化的解决方案从架构重构、库集成和调试增强三个维度进行优化1. 全面迁移至 StreamJsonRpc移除旧代码删除?JsonRpcEndpoint.cs、AgentSideConnection.cs?及相关的自定义序列化转换器JsonRpcMessageJsonConverter?等。集成官方库引入?StreamJsonRpc?NuGet 包利用其?JsonRpc?类处理核心通信逻辑。抽象传输层定义?IAcpTransport?接口统一处理?WebSocket?和?Stdio?两种传输模式确保协议层与传输层解耦。// IAcpTransport 接口定义public interface IAcpTransport{Task SendAsync(string message, CancellationToken cancellationToken default);Task ReceiveAsync(CancellationToken cancellationToken default);Task CloseAsync(CancellationToken cancellationToken default);}// WebSocket 传输实现public class WebSocketTransport : IAcpTransport{private readonly WebSocket _webSocket;public WebSocketTransport(WebSocket webSocket){_webSocket webSocket;}// 实现发送和接收方法// ...}// Stdio 传输实现public class StdioTransport : IAcpTransport{private readonly StreamReader _reader;private readonly StreamWriter _writer;public StdioTransport(StreamReader reader, StreamWriter writer){_reader reader;_writer writer;}// 实现发送和接收方法// ...}2. 修复代理目标识别问题分析?CallbackProxyTarget检查现有的动态代理生成逻辑确定 StreamJsonRpc 无法识别的根本原因通常是因为代理对象没有公开实际的方法签名或者使用了 StreamJsonRpc 不支持的参数类型。重构参数传递将泛型属性拆分为明确的 RPC 方法参数。不再依赖动态属性而是定义具体的 Request/Response DTO数据传输对象确保 StreamJsonRpc 能通过反射正确识别方法签名。// 原有的泛型属性方式public class CallbackProxyTarget{public Func Callback { get; set; }}// 重构后的具体方法方式public class ReadTextFileRequest{public string FilePath { get; set; }}public class ReadTextFileResponse{public string Content { get; set; }}public interface IAcpAgentCallback{Task ReadTextFileAsync(ReadTextFileRequest request);// 其他方法...}使用?Attach?替代?AddLocalRpcTarget在某些复杂场景下手动代理?JsonRpc?对象并处理?RpcConnection?可能比直接添加目标更灵活。3. 实现方法绑定与日志增强实现?AcpAgentCallbackRpcAdapter确保该组件显式实现 StreamJsonRpc 的代理接口将 ACP 协议定义的方法如?ReadTextFileAsync映射到 StreamJsonRpc 的回调处理器上。集成日志记录在 WebSocket 或 Stdio 的消息处理管道中拦截并记录 JSON-RPC 请求和响应的原始文本。利用?ILogger?在解析前和序列化后输出原始 payload以便排查格式错误。// 日志增强的传输包装器public class LoggingAcpTransport : IAcpTransport{private readonly IAcpTransport _innerTransport;private readonly ILogger _logger;public LoggingAcpTransport(IAcpTransport innerTransport, ILogger logger){_innerTransport innerTransport;_logger logger;}public async Task SendAsync(string message, CancellationToken cancellationToken default){_logger.LogTrace(Sending message: {Message}, message);await _innerTransport.SendAsync(message, cancellationToken);}public async Task ReceiveAsync(CancellationToken cancellationToken default){var message await _innerTransport.ReceiveAsync(cancellationToken);_logger.LogTrace(Received message: {Message}, message);return message;}public async Task CloseAsync(CancellationToken cancellationToken default){_logger.LogDebug(Closing connection);await _innerTransport.CloseAsync(cancellationToken);}}4. 架构分层重构传输层 (AcpRpcClient)封装 StreamJsonRpc 连接负责?InvokeAsync?和连接生命周期管理。public class AcpRpcClient : IDisposable{private readonly JsonRpc _rpc;private readonly IAcpTransport _transport;public AcpRpcClient(IAcpTransport transport){_transport transport;_rpc new JsonRpc(new StreamRpcTransport(transport));_rpc.StartListening();}public async Task InvokeAsync(string methodName, object parameters){return await _rpc.InvokeAsync(methodName, parameters);}public void Dispose(){_rpc.Dispose();_transport.Dispose();}// StreamRpcTransport 是对 IAcpTransport 的 StreamJsonRpc 适配器private class StreamRpcTransport : IDuplexPipe{// 实现 IDuplexPipe 接口// ...}}协议层 (IAcpAgentClient?/?IAcpAgentCallback)定义清晰的 client-to-agent 和 agent-to-client 接口移除?Func?这种循环依赖的工厂模式改用依赖注入或直接注册回调。实践基于 StreamJsonRpc 的最佳实践和项目经验以下是实施过程中的关键建议1. 强类型 DTO 优于动态对象StreamJsonRpc 的核心优势在于强类型。不要使用?dynamic?或?JObject?传递参数。应为每个 RPC 方法定义明确的 C# POCO 类作为参数。这不仅解决了代理目标识别问题还能在编译时发现类型错误。示例将?CallbackProxyTarget?中的泛型属性替换为?ReadTextFileRequest?和?WriteTextFileRequest?等具体类。2. 显式声明 Method Name使用?[JsonRpcMethod]?特性显式指定 RPC 方法名称不要依赖默认的方法名映射。这可以防止因命名风格差异如 PascalCase vs camelCase导致的调用失败。public interface IAcpAgentCallback{[JsonRpcMethod(readTextFile)]Task ReadTextFileAsync(ReadTextFileRequest request);[JsonRpcMethod(writeTextFile)]Task WriteTextFileAsync(WriteTextFileRequest request);}3. 利用连接状态回调StreamJsonRpc 提供了?JsonRpc.ConnectionLost?事件。务必监听此事件以处理进程意外退出或网络断开的情况这比单纯依赖 Orleans 的 Grain 失效检测更及时。_rpc.ConnectionLost (sender, e) {_logger.LogError(RPC connection lost: {Reason}, e.ToString());// 处理重连逻辑或通知用户};4. 日志分层记录Trace 级别记录完整的 JSON Request/Response 原文。Debug 级别记录方法调用栈和参数摘要。注意确保日志中不包含敏感的 Authorization Token 或大文件内容的 Base64 编码。5. 处理流式传输的特殊性StreamJsonRpc 原生支持?IAsyncEnumerable。在实现 ACP 的流式 Prompt 响应时应直接使用?IAsyncEnumerable?而不是自定义的分页逻辑。这能极大简化流式处理的代码量。public interface IAcpAgentCallback{[JsonRpcMethod(streamText)]IAsyncEnumerable StreamTextAsync(StreamTextRequest request);}6. 适配器模式 (Adapter Pattern)保持?ACPSession?和?ClientSideConnection?的分离。ACPSession?应专注于 Orleans 的状态管理和业务逻辑如消息入队通过组合而非继承的方式使用 StreamJsonRpc 连接对象。总结通过全面集成 StreamJsonRpcHagiCode 项目成功解决了原自定义实现的维护成本高、功能局限性和架构分层混乱等问题。关键改进包括采用强类型 DTO 替代动态属性提高了代码的可维护性和可靠性实现了传输层抽象和协议层分离提升了架构的清晰性增强了日志记录功能便于排查通信问题继贺谋治