by 详细分析 ▾
运行时依赖
版本
通过 JSON 轮询实现 Kiro 实例间的实时聊天
安装命令
点击复制技能文档
# Kiro 实时聊天 ## 概述 本技能通过轮询共享 JSON 文件，实现多个 Kiro 实例之间的准实时通信。每条消息包含时间戳、发送者及内容。 ## 文件位置 `` memory/kiro-realtime.json ` ## JSON 结构 `json { "messages": [ { "id": 1, "from": "VPS Kiro", "to": "Laptop Kiro", "message": "Selam!", "timestamp": "2026-03-05T22:58:00Z", "read": false } ], "lastCheck": "2026-03-05T22:58:00Z" } ` ## 用法 ### 发送消息 `python import json import os from datetime import datetime CHAT_FILE = "memory/kiro-realtime.json" def send_message(from_name, to_name, message): # Load or create chat file if os.path.exists(CHAT_FILE): with open(CHAT_FILE, "r") as f: chat = json.load(f) else: chat = {"messages": [], "lastCheck": datetime.now().isoformat()} # Add new message msg_id = len(chat["messages"]) + 1 chat["messages"].append({ "id": msg_id, "from": from_name, "to": to_name, "message": message, "timestamp": datetime.now().isoformat(), "read": False }) # Save with open(CHAT_FILE, "w") as f: json.dump(chat, f, indent=2) return msg_id ` ### 检查新消息 `python def check_messages(my_name, since_timestamp=None): if not os.path.exists(CHAT_FILE): return [] with open(CHAT_FILE, "r") as f: chat = json.load(f) # Update last check time chat["lastCheck"] = datetime.now().isoformat() with open(CHAT_FILE, "w") as f: json.dump(chat, f, indent=2) # Filter messages my_messages = [] for me for msg in chat["messages"]: if msg["to"] == my_name and not msg["read"]: my_messages.append(msg) msg["read"] = True # Save read status with open(CHAT_FILE, "w") as f: json.dump(chat, f, indent=2) return my_messages ` ### 轮询循环（实现准实时） `python import time def poll_messages(my_name, interval=5): """Poll for new messages every N seconds""" while True: msgs = check_messages(my_name) for msg in msgs: print(f"{msg['from']}: {msg['message']}") # Process and respond here time.sleep(interval) `` ## 推荐轮询间隔 - 需要快速响应： 2–3 秒 - 正常场景： 5–10 秒 - 低带宽环境： 30–60 秒 ## 提示 - 处理后将消息标记为“已读” - 利用时间戳避免重复处理 - 如需更佳实时性，可考虑 MCP server 或 WebSocket - 单条消息请保持在 1000 字符以内