Prints an event to the console. Args: event: The Event object to print. jsonl: If True, outputs structured JSONL to stdout. Otherwise, outputs human-readable text. session_id: Optional session ID to inject into the JSONL output.
(
event: Event, jsonl: bool = False, session_id: Optional[str] = None
)
| 346 | |
| 347 | |
| 348 | def _print_event( |
| 349 | event: Event, jsonl: bool = False, session_id: Optional[str] = None |
| 350 | ): |
| 351 | """Prints an event to the console. |
| 352 | |
| 353 | Args: |
| 354 | event: The Event object to print. |
| 355 | jsonl: If True, outputs structured JSONL to stdout. Otherwise, outputs |
| 356 | human-readable text. |
| 357 | session_id: Optional session ID to inject into the JSONL output. |
| 358 | """ |
| 359 | if jsonl: |
| 360 | event_dict = event.model_dump(mode='json', by_alias=True, exclude_none=True) |
| 361 | if session_id: |
| 362 | event_dict['session_id'] = session_id |
| 363 | if event.node_info and event.node_info.path: |
| 364 | event_dict['node_path'] = event.node_info.path |
| 365 | |
| 366 | # Filter out empty dictionaries in 'actions' (e.g., empty state delta) to |
| 367 | # reduce noise |
| 368 | if 'actions' in event_dict and isinstance(event_dict['actions'], dict): |
| 369 | event_dict['actions'] = { |
| 370 | k: v for k, v in event_dict['actions'].items() if v != {} |
| 371 | } |
| 372 | if not event_dict['actions']: |
| 373 | del event_dict['actions'] |
| 374 | |
| 375 | # Optimize key order for human readability in JSONL viewers |
| 376 | ordered_dict = {} |
| 377 | for k in ['author', 'session_id', 'node_path', 'id']: |
| 378 | if k in event_dict: |
| 379 | ordered_dict[k] = event_dict[k] |
| 380 | for k, v in event_dict.items(): |
| 381 | if k not in ordered_dict: |
| 382 | ordered_dict[k] = v |
| 383 | click.echo(json.dumps(ordered_dict)) |
| 384 | else: |
| 385 | # Human readable mode |
| 386 | author = event.author or 'unknown' |
| 387 | text_parts = ( |
| 388 | [p.text for p in event.content.parts if p.text] if event.content else [] |
| 389 | ) |
| 390 | if text_parts: |
| 391 | text = ''.join(text_parts) |
| 392 | click.echo(f'[{author}]: {text}') |
| 393 | elif event.long_running_tool_ids: |
| 394 | click.secho(f'[{author}]: (Paused for input...)', fg='yellow') |
| 395 | |
| 396 | |
| 397 | async def run_cli( |
no test coverage detected