Yash030/claude-code-proxy
2
1"""CLI event parser for Claude Code CLI output.2 3This parser emits an ordered stream of low-level events suitable for building a4Claude Code-like transcript in messaging UIs.5"""6 7from typing import Any8 9from loguru import logger10 11 12def parse_cli_event(event: Any, *, log_raw_cli: bool = False) -> list[dict]:13 """14 Parse a CLI event and return a structured result.15 16 Args:17 event: Raw event dictionary from CLI18 log_raw_cli: When True, log full error text from the CLI. Default is19 metadata-only (lengths / exit codes) to avoid leaking user content.20 21 Returns:22 List of parsed event dicts. Empty list if not recognized.23 """24 if not isinstance(event, dict):25 return []26 27 etype = event.get("type")28 results: list[dict[str, Any]] = []29 30 # Some CLI/proxy layers emit "system" events that are not user-visible and31 # carry no transcript content. Ignore them explicitly to avoid noisy logs.32 if etype == "system":33 return []34 35 # 1. Handle full messages (assistant/user or result)36 msg_obj = None37 if etype == "assistant" or etype == "user":38 msg_obj = event.get("message")39 elif etype == "result":40 res = event.get("result")41 if isinstance(res, dict):42 msg_obj = res.get("message")43 # Some variants put content directly on the result.44 if not msg_obj and isinstance(res.get("content"), list):45 msg_obj = {"content": res.get("content")}46 if not msg_obj:47 msg_obj = event.get("message")48 # Some variants put content directly on the event.49 if not msg_obj and isinstance(event.get("content"), list):50 msg_obj = {"content": event.get("content")}51 52 if msg_obj and isinstance(msg_obj, dict):53 content = msg_obj.get("content", [])54 if isinstance(content, list):55 # Preserve order exactly as content blocks appear.56 for c in content:57 if not isinstance(c, dict):58 continue59 ctype = c.get("type")60 if ctype == "text":61 results.append({"type": "text_chunk", "text": c.get("text", "")})62 elif ctype == "thinking":63 results.append(64 {"type": "thinking_chunk", "text": c.get("thinking", "")}65 )66 elif ctype == "tool_use":67 results.append(68 {69 "type": "tool_use",70 "id": str(c.get("id", "") or "").strip(),71 "name": c.get("name", ""),72 "input": c.get("input"),73 }74 )75 elif ctype == "tool_result":76 results.append(77 {78 "type": "tool_result",79 "tool_use_id": str(c.get("tool_use_id", "") or "").strip(),80 "content": c.get("content"),81 "is_error": bool(c.get("is_error", False)),82 }83 )84 85 if results:86 return results87 88 # 2. Handle streaming deltas89 if etype == "content_block_delta":90 delta = event.get("delta", {})91 if isinstance(delta, dict):92 if delta.get("type") == "text_delta":93 return [94 {95 "type": "text_delta",96 "index": event.get("index", -1),97 "text": delta.get("text", ""),98 }99 ]100 if delta.get("type") == "thinking_delta":101 return [102 {103 "type": "thinking_delta",104 "index": event.get("index", -1),105 "text": delta.get("thinking", ""),106 }107 ]108 if delta.get("type") == "input_json_delta":109 return [110 {111 "type": "tool_use_delta",112 "index": event.get("index", -1),113 "partial_json": delta.get("partial_json", ""),114 }115 ]116 117 # 3. Handle tool usage start118 if etype == "content_block_start":119 block = event.get("content_block", {})120 if isinstance(block, dict):121 btype = block.get("type")122 if btype == "thinking":123 return [{"type": "thinking_start", "index": event.get("index", -1)}]124 if btype == "text":125 return [{"type": "text_start", "index": event.get("index", -1)}]126 if btype == "tool_use":127 return [128 {129 "type": "tool_use_start",130 "index": event.get("index", -1),131 "id": str(block.get("id", "") or "").strip(),132 "name": block.get("name", ""),133 "input": block.get("input"),134 }135 ]136 137 # 3.5 Handle block stop (to close open streaming segments)138 if etype == "content_block_stop":139 return [{"type": "block_stop", "index": event.get("index", -1)}]140 141 # 4. Handle errors and exit142 if etype == "error":143 err = event.get("error")144 msg = err.get("message") if isinstance(err, dict) else str(err)145 if log_raw_cli:146 logger.info("CLI_PARSER: Parsed error event: {}", msg)147 else:148 mlen = len(msg) if isinstance(msg, str) else 0149 logger.info("CLI_PARSER: Parsed error event: message_chars={}", mlen)150 return [{"type": "error", "message": msg}]151 elif etype == "exit":152 code = event.get("code", 0)153 stderr = event.get("stderr")154 if code == 0:155 logger.debug(f"CLI_PARSER: Successful exit (code={code})")156 return [{"type": "complete", "status": "success"}]157 else:158 # Non-zero exit is an error159 error_msg = stderr if stderr else f"Process exited with code {code}"160 if log_raw_cli:161 logger.warning(162 "CLI_PARSER: Error exit (code={}): {}",163 code,164 error_msg,165 )166 else:167 em = error_msg if isinstance(error_msg, str) else str(error_msg)168 logger.warning(169 "CLI_PARSER: Error exit (code={}): message_chars={}",170 code,171 len(em),172 )173 return [174 {"type": "error", "message": error_msg},175 {"type": "complete", "status": "failed"},176 ]177 178 # Log unrecognized events for debugging179 if etype:180 logger.debug(f"CLI_PARSER: Unrecognized event type: {etype}")181 return []182