(self, msg: Dict[str, Any])
| 191 | return {"decision": "allow"} |
| 192 | |
| 193 | def dispatch(self, msg: Dict[str, Any]) -> Optional[Dict[str, Any]]: |
| 194 | method = msg.get("method") |
| 195 | params = msg.get("params", {}) |
| 196 | req_id = msg.get("id") |
| 197 | |
| 198 | # 处理握手（使用 method 字段） |
| 199 | if method == "ahp/handshake": |
| 200 | result = self._handle_handshake(params) |
| 201 | return {"jsonrpc": "2.0", "id": req_id, "result": result} |
| 202 | |
| 203 | # 处理事件（使用 params.event_type） |
| 204 | event_type = params.get("event_type", "") |
| 205 | payload = params.get("payload", {}) |
| 206 | depth = params.get("depth", 0) |
| 207 | |
| 208 | if req_id is None: |
| 209 | self._handle_notification(event_type, payload, depth) |
| 210 | return None |
| 211 | |
| 212 | result = self._handle_request(event_type, payload, depth) |
| 213 | return {"jsonrpc": "2.0", "id": req_id, "result": result} |
| 214 | |
| 215 | def run(self): |
| 216 | log("在 stdin 上监听 JSON-RPC 2.0 消息...") |
no test coverage detected