---
id: cubepi-utils
title: cubepi.utils
sidebar_position: 7
hide_table_of_contents: false
---

# `cubepi.utils`

### emit_event {#emit-event}

_function_

```python
emit_event(emit_fn: Callable, event: object) -> None
```


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/utils/emit.py#L7)


### parse_streaming_json {#parse-streaming-json}

_function_

```python
parse_streaming_json(text: str | None) -> dict
```

Parse potentially incomplete or malformed JSON from streaming.

Uses a 3-tier fallback:
  1. `json.loads(text)`
  2. `json.loads(repair_json(text))`
  3. Partial-parse: repair + close open braces/brackets/strings
  4. `\{\}` as last resort

Always returns a *dict* (or at minimum `\{\}`) so callers never need
to handle parse failures.

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/utils/json_parse.py#L169)


### repair_json {#repair-json}

_function_

```python
repair_json(text: str) -> str
```

Repair malformed JSON string literals.

Operates character-by-character, tracking whether the cursor is inside a
JSON string value.  Inside strings it:

* Escapes raw control characters (0x00-0x1F) that are not already
  escaped (`\t`, `\n`, `\r` are kept when already escaped).
* Doubles a backslash before an invalid escape character
  (e.g. `\x` becomes `\\x`), making the output parseable.
* Handles truncated backslash at end-of-string.

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/utils/json_parse.py#L39)



<!-- GENERATED by build_api_reference.py — DO NOT EDIT -->