Skip to content

API Reference

Reference documentation for the modules and classes in bfast-llm.

Complete Interface

bfast_llm.middleware

BFastLLM

Middleware for transparently compressing LLM contexts using the B-FAST binary protocol.

Source code in bfast_llm/middleware.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class BFastLLM:
    """Middleware for transparently compressing LLM contexts using the B-FAST binary protocol."""

    def __init__(
        self,
        registry: Optional[BFastRegistry] = None,
        threshold_bytes: int = 1024,
        compress_payloads: bool = True,
    ):
        """
        Initialize BFastLLM middleware.

        Args:
            registry: A BFastRegistry instance. If None, creates an in-memory registry.
            threshold_bytes: Content size threshold above which compression is triggered (default: 1KB).
            compress_payloads: Whether to apply compression to BFast payloads.
        """
        self.registry = registry or BFastRegistry()
        self.tool = BFastTool(self.registry)
        self.threshold_bytes = threshold_bytes
        self.compress_payloads = compress_payloads
        self.encoder = b_fast.BFast()

    def compress_messages(
        self, messages: List[Dict[str, Any]]
    ) -> Tuple[List[Dict[str, Any]], bool]:
        """
        Interprets messages, serializing large payloads into B-FAST binaries and swapping them with references.

        Args:
            messages: List of message dictionaries in standard Chat Completion format.

        Returns:
            A tuple of (modified_messages, was_compressed)
        """
        modified_messages = []
        was_compressed = False

        for msg in messages:
            msg.get("role")
            content = msg.get("content")

            # We compress large content in user or tool messages, or assistant tool results
            if (
                content is not None
                and isinstance(content, str)
                and len(content) >= self.threshold_bytes
            ):
                # Check if the content is valid JSON (list or dict)
                parsed_data = None
                stripped = content.strip()
                if (stripped.startswith("{") and stripped.endswith("}")) or (
                    stripped.startswith("[") and stripped.endswith("]")
                ):
                    try:
                        parsed_data = json.loads(content)
                    except json.JSONDecodeError:
                        pass

                if parsed_data is not None:
                    # Encode to B-FAST binary format
                    try:
                        binary_data = self.encoder.encode_packed(
                            parsed_data, compress=self.compress_payloads
                        )
                        # Register in our BFast cache registry
                        ref_tag = self.registry.register(binary_data)

                        # Create new message with content replaced by reference tag
                        new_msg = msg.copy()
                        new_msg["content"] = (
                            f"The data was compressed using B-FAST to save tokens:\n{ref_tag}\n"
                            f"To inspect or read this data, call the 'bfast_retrieve' tool with id."
                        )
                        modified_messages.append(new_msg)
                        was_compressed = True
                        continue
                    except Exception as e:
                        # Log error internally and keep original message if encoding fails
                        print(
                            f"[BFastLLM] Warning: Failed to serialize payload: {e}",
                            file=sys.stderr,
                        )

            # Also handle if the content is a dictionary, list, NumPy array, or Pandas DataFrame
            else:
                is_supported = False
                processed_content = content

                # Check for Pandas DataFrame (lazy check by class name)
                if content is not None and type(content).__name__ == "DataFrame":
                    try:
                        import pandas as pd

                        if isinstance(content, pd.DataFrame):
                            processed_content = content.to_dict(orient="records")
                            is_supported = True
                    except ImportError:
                        pass

                # Check for NumPy ndarray (numpy is already a dependency of bfast-py)
                elif content is not None and type(content).__name__ == "ndarray":
                    try:
                        import numpy as np

                        if isinstance(content, np.ndarray):
                            is_supported = True
                    except ImportError:
                        pass

                # Accept dict/list
                elif content is not None and isinstance(content, (dict, list)):
                    is_supported = True

                if is_supported:
                    try:
                        # Convert object/array to BFast directly
                        binary_data = self.encoder.encode_packed(
                            processed_content, compress=self.compress_payloads
                        )
                        ref_tag = self.registry.register(binary_data)
                        new_msg = msg.copy()
                        new_msg["content"] = (
                            f"The data was compressed using B-FAST to save tokens:\n{ref_tag}\n"
                            f"To inspect or read this data, call the 'bfast_retrieve' tool with id."
                        )
                        modified_messages.append(new_msg)
                        was_compressed = True
                        continue
                    except Exception as e:
                        print(
                            f"[BFastLLM] Warning: Failed to serialize object/array payload: {e}",
                            file=sys.stderr,
                        )

            modified_messages.append(msg)

        return modified_messages, was_compressed

    def get_tools(self) -> List[Dict[str, Any]]:
        """Get the retrieve tool definition to inject into the LLM call."""
        return [self.tool.get_openai_tool_definition()]

    def handle_tool_call(self, tool_call: Any) -> Optional[str]:
        """
        Handle a tool call if it is 'bfast_retrieve'.

        Args:
            tool_call: The tool call object (e.g. from OpenAI response, having 'function' attribute with 'name' and 'arguments')

        Returns:
            The string result of the retrieval if handled, otherwise None.
        """
        # Supports both object style and dictionary style tool calls
        name = getattr(tool_call, "name", None) or getattr(
            getattr(tool_call, "function", None), "name", None
        )
        arguments = getattr(tool_call, "arguments", None) or getattr(
            getattr(tool_call, "function", None), "arguments", None
        )

        if name != "bfast_retrieve":
            if isinstance(tool_call, dict):
                func = tool_call.get("function", {})
                if func.get("name") == "bfast_retrieve":
                    name = "bfast_retrieve"
                    arguments = func.get("arguments")
                else:
                    return None
            else:
                return None

        if isinstance(arguments, str):
            try:
                args = json.loads(arguments)
            except json.JSONDecodeError:
                return "Error: Invalid arguments JSON."
        else:
            args = arguments or {}

        ref_id = args.get("ref_id")
        fmt = args.get("format", "markdown")

        if not ref_id:
            return "Error: Missing 'ref_id' argument."
        return self.tool.bfast_retrieve(ref_id, format=fmt)

__init__(registry=None, threshold_bytes=1024, compress_payloads=True)

Initialize BFastLLM middleware.

Parameters:

Name Type Description Default
registry Optional[BFastRegistry]

A BFastRegistry instance. If None, creates an in-memory registry.

 None
threshold_bytes int

Content size threshold above which compression is triggered (default: 1KB).

 1024
compress_payloads bool

Whether to apply compression to BFast payloads.

 True
Source code in bfast_llm/middleware.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    registry: Optional[BFastRegistry] = None,
    threshold_bytes: int = 1024,
    compress_payloads: bool = True,
):
    """
    Initialize BFastLLM middleware.

    Args:
        registry: A BFastRegistry instance. If None, creates an in-memory registry.
        threshold_bytes: Content size threshold above which compression is triggered (default: 1KB).
        compress_payloads: Whether to apply compression to BFast payloads.
    """
    self.registry = registry or BFastRegistry()
    self.tool = BFastTool(self.registry)
    self.threshold_bytes = threshold_bytes
    self.compress_payloads = compress_payloads
    self.encoder = b_fast.BFast()

compress_messages(messages)

Interprets messages, serializing large payloads into B-FAST binaries and swapping them with references.

Parameters:

Name Type Description Default
messages List[Dict[str, Any]]

List of message dictionaries in standard Chat Completion format.

 required

Returns:

Type Description
Tuple[List[Dict[str, Any]], bool]

A tuple of (modified_messages, was_compressed)
Source code in bfast_llm/middleware.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def compress_messages(
    self, messages: List[Dict[str, Any]]
) -> Tuple[List[Dict[str, Any]], bool]:
    """
    Interprets messages, serializing large payloads into B-FAST binaries and swapping them with references.

    Args:
        messages: List of message dictionaries in standard Chat Completion format.

    Returns:
        A tuple of (modified_messages, was_compressed)
    """
    modified_messages = []
    was_compressed = False

    for msg in messages:
        msg.get("role")
        content = msg.get("content")

        # We compress large content in user or tool messages, or assistant tool results
        if (
            content is not None
            and isinstance(content, str)
            and len(content) >= self.threshold_bytes
        ):
            # Check if the content is valid JSON (list or dict)
            parsed_data = None
            stripped = content.strip()
            if (stripped.startswith("{") and stripped.endswith("}")) or (
                stripped.startswith("[") and stripped.endswith("]")
            ):
                try:
                    parsed_data = json.loads(content)
                except json.JSONDecodeError:
                    pass

            if parsed_data is not None:
                # Encode to B-FAST binary format
                try:
                    binary_data = self.encoder.encode_packed(
                        parsed_data, compress=self.compress_payloads
                    )
                    # Register in our BFast cache registry
                    ref_tag = self.registry.register(binary_data)

                    # Create new message with content replaced by reference tag
                    new_msg = msg.copy()
                    new_msg["content"] = (
                        f"The data was compressed using B-FAST to save tokens:\n{ref_tag}\n"
                        f"To inspect or read this data, call the 'bfast_retrieve' tool with id."
                    )
                    modified_messages.append(new_msg)
                    was_compressed = True
                    continue
                except Exception as e:
                    # Log error internally and keep original message if encoding fails
                    print(
                        f"[BFastLLM] Warning: Failed to serialize payload: {e}",
                        file=sys.stderr,
                    )

        # Also handle if the content is a dictionary, list, NumPy array, or Pandas DataFrame
        else:
            is_supported = False
            processed_content = content

            # Check for Pandas DataFrame (lazy check by class name)
            if content is not None and type(content).__name__ == "DataFrame":
                try:
                    import pandas as pd

                    if isinstance(content, pd.DataFrame):
                        processed_content = content.to_dict(orient="records")
                        is_supported = True
                except ImportError:
                    pass

            # Check for NumPy ndarray (numpy is already a dependency of bfast-py)
            elif content is not None and type(content).__name__ == "ndarray":
                try:
                    import numpy as np

                    if isinstance(content, np.ndarray):
                        is_supported = True
                except ImportError:
                    pass

            # Accept dict/list
            elif content is not None and isinstance(content, (dict, list)):
                is_supported = True

            if is_supported:
                try:
                    # Convert object/array to BFast directly
                    binary_data = self.encoder.encode_packed(
                        processed_content, compress=self.compress_payloads
                    )
                    ref_tag = self.registry.register(binary_data)
                    new_msg = msg.copy()
                    new_msg["content"] = (
                        f"The data was compressed using B-FAST to save tokens:\n{ref_tag}\n"
                        f"To inspect or read this data, call the 'bfast_retrieve' tool with id."
                    )
                    modified_messages.append(new_msg)
                    was_compressed = True
                    continue
                except Exception as e:
                    print(
                        f"[BFastLLM] Warning: Failed to serialize object/array payload: {e}",
                        file=sys.stderr,
                    )

        modified_messages.append(msg)

    return modified_messages, was_compressed

get_tools()

Get the retrieve tool definition to inject into the LLM call.

Source code in bfast_llm/middleware.py 
156
157
158
def get_tools(self) -> List[Dict[str, Any]]:
    """Get the retrieve tool definition to inject into the LLM call."""
    return [self.tool.get_openai_tool_definition()]

handle_tool_call(tool_call)

Handle a tool call if it is 'bfast_retrieve'.

Parameters:

Name Type Description Default
tool_call Any

The tool call object (e.g. from OpenAI response, having 'function' attribute with 'name' and 'arguments')

 required

Returns:

Type Description
Optional[str]

The string result of the retrieval if handled, otherwise None.
Source code in bfast_llm/middleware.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def handle_tool_call(self, tool_call: Any) -> Optional[str]:
    """
    Handle a tool call if it is 'bfast_retrieve'.

    Args:
        tool_call: The tool call object (e.g. from OpenAI response, having 'function' attribute with 'name' and 'arguments')

    Returns:
        The string result of the retrieval if handled, otherwise None.
    """
    # Supports both object style and dictionary style tool calls
    name = getattr(tool_call, "name", None) or getattr(
        getattr(tool_call, "function", None), "name", None
    )
    arguments = getattr(tool_call, "arguments", None) or getattr(
        getattr(tool_call, "function", None), "arguments", None
    )

    if name != "bfast_retrieve":
        if isinstance(tool_call, dict):
            func = tool_call.get("function", {})
            if func.get("name") == "bfast_retrieve":
                name = "bfast_retrieve"
                arguments = func.get("arguments")
            else:
                return None
        else:
            return None

    if isinstance(arguments, str):
        try:
            args = json.loads(arguments)
        except json.JSONDecodeError:
            return "Error: Invalid arguments JSON."
    else:
        args = arguments or {}

    ref_id = args.get("ref_id")
    fmt = args.get("format", "markdown")

    if not ref_id:
        return "Error: Missing 'ref_id' argument."
    return self.tool.bfast_retrieve(ref_id, format=fmt)

bfast_tune(client, threshold_bytes=1024, compress_payloads=True)

Tune an LLM client instance (e.g. OpenAI) to automatically apply B-FAST prompt compression and handle retrieval tool calls transparently.

Source code in bfast_llm/middleware.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def bfast_tune(client, threshold_bytes: int = 1024, compress_payloads: bool = True):
    """
    Tune an LLM client instance (e.g. OpenAI) to automatically apply B-FAST prompt
    compression and handle retrieval tool calls transparently.
    """
    original_create = client.chat.completions.create
    wrapped_create = wrap_completion(
        original_create, threshold_bytes, compress_payloads
    )

    class WrappedCompletions:
        def __init__(self, original):
            self._original = original

        def __getattr__(self, name):
            return getattr(self._original, name)

        def create(self, *args, **kwargs):
            return wrapped_create(*args, **kwargs)

    client.chat.completions = WrappedCompletions(client.chat.completions)
    return client

wrap_completion(completion_func, threshold_bytes=1024, compress_payloads=True)

Wrap any OpenAI-compatible chat completion function to automatically apply B-FAST compression and transparently handle retrieval tool calls.

Source code in bfast_llm/middleware.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def wrap_completion(
    completion_func, threshold_bytes: int = 1024, compress_payloads: bool = True
):
    """
    Wrap any OpenAI-compatible chat completion function to automatically
    apply B-FAST compression and transparently handle retrieval tool calls.
    """
    bfast_llm = BFastLLM(
        threshold_bytes=threshold_bytes, compress_payloads=compress_payloads
    )

    def wrapped(*args, **kwargs):
        messages = kwargs.get("messages")
        if not messages:
            return completion_func(*args, **kwargs)

        # Copy to avoid side-effects on original user messages list
        compressed_messages, was_compressed = bfast_llm.compress_messages(messages)
        kwargs["messages"] = list(compressed_messages)

        if was_compressed:
            tools = kwargs.get("tools", [])
            has_tool = any(
                t.get("function", {}).get("name") == "bfast_retrieve" for t in tools
            )
            if not has_tool:
                tools = list(tools) + bfast_llm.get_tools()
                kwargs["tools"] = tools
                if "tool_choice" not in kwargs:
                    kwargs["tool_choice"] = "auto"

        while True:
            response = completion_func(*args, **kwargs)

            is_dict = isinstance(response, dict)
            choices = (
                response.get("choices", [])
                if is_dict
                else getattr(response, "choices", [])
            )
            if not choices:
                return response

            message = choices[0].get("message", {}) if is_dict else choices[0].message
            tool_calls = (
                message.get("tool_calls", [])
                if is_dict
                else getattr(message, "tool_calls", None)
            )

            if not tool_calls:
                return response

            bfast_calls = []
            for tc in tool_calls:
                name = (
                    tc.get("function", {}).get("name")
                    if is_dict
                    else getattr(getattr(tc, "function", None), "name", None)
                )
                if name == "bfast_retrieve":
                    bfast_calls.append(tc)

            if not bfast_calls:
                return response

            print(f"[BFastLLM] Resolving {len(bfast_calls)} retrieval calls...")

            kwargs["messages"].append(message)

            for tc in bfast_calls:
                result = bfast_llm.handle_tool_call(tc)
                tc_id = tc.get("id") if is_dict else getattr(tc, "id", None)
                kwargs["messages"].append(
                    {
                        "role": "tool",
                        "tool_call_id": tc_id,
                        "name": "bfast_retrieve",
                        "content": result,
                    }
                )

    return wrapped

bfast_llm.summarizer

BFastSummarizer

Helper to summarize B-FAST data structure and generate token-efficient representation.

Source code in bfast_llm/summarizer.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class BFastSummarizer:
    """Helper to summarize B-FAST data structure and generate token-efficient representation."""

    @classmethod
    def summarize(cls, data: Any, max_list_items_preview: int = 3) -> str:
        """
        Generate a concise text summary of a python object decoded from B-FAST.

        Args:
            data: Any Python object
            max_list_items_preview: How many items to preview if listing individual elements

        Returns:
            A string describing the structure and summary of the data.
        """
        if data is None:
            return "Null"

        if isinstance(data, bool):
            return f"Boolean ({data})"

        if isinstance(data, (int, float)):
            return f"{type(data).__name__} ({data})"

        if isinstance(data, (datetime, date, time, UUID, Decimal)):
            return f"{type(data).__name__} ({data})"

        if isinstance(data, str):
            if len(data) <= 100:
                return f"String: '{data}'"
            return f"String (length={len(data)}): '{data[:60]}...'"

        if isinstance(data, (bytes, bytearray)):
            return f"Binary data (size={len(data)} bytes)"

        if isinstance(data, list):
            if not data:
                return "Empty List"

            # Check if it's a list of dictionaries (common for datasets)
            first_item = data[0]
            if isinstance(first_item, dict):
                keys_info = []
                for k, v in first_item.items():
                    val_type = type(v).__name__
                    keys_info.append(f"{k}: {val_type}")
                schema_str = ", ".join(keys_info)
                return f"List of {len(data)} objects. Schema: {{{schema_str}}}"

            # Check if it's a list of numbers (float/int)
            if all(isinstance(x, (int, float)) for x in data[:10]):
                nums = [float(x) for x in data]
                if nums:
                    min_val = min(nums)
                    max_val = max(nums)
                    avg_val = sum(nums) / len(nums)
                    return f"List of {len(data)} numbers (float/int). Stats: min={min_val:.4f}, max={max_val:.4f}, avg={avg_val:.4f}"

            # General list
            types = set(type(x).__name__ for x in data[:10])
            types_str = "|".join(types)
            return f"List of {len(data)} items of type ({types_str})"

        if isinstance(data, dict):
            if not data:
                return "Empty Object"

            keys_preview = []
            for k, v in list(data.items())[:10]:
                val_summary = ""
                if (
                    isinstance(v, (int, float, bool, str))
                    and not isinstance(v, str)
                    or (isinstance(v, str) and len(v) < 30)
                ):
                    val_summary = f"={repr(v)}"
                else:
                    val_summary = f" ({type(v).__name__})"
                keys_preview.append(f"'{k}'{val_summary}")

            total_keys = len(data)
            keys_str = ", ".join(keys_preview)
            if total_keys > 10:
                keys_str += f", ... (+{total_keys - 10} more)"

            return f"Object with {total_keys} keys: {{{keys_str}}}"

        return f"Unknown Type ({type(data).__name__}): {str(data)[:100]}"

summarize(data, max_list_items_preview=3) classmethod

Generate a concise text summary of a python object decoded from B-FAST.

Parameters:

Name Type Description Default
data Any

Any Python object

 required
max_list_items_preview int

How many items to preview if listing individual elements

 3

Returns:

Type Description
str

A string describing the structure and summary of the data.
Source code in bfast_llm/summarizer.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def summarize(cls, data: Any, max_list_items_preview: int = 3) -> str:
    """
    Generate a concise text summary of a python object decoded from B-FAST.

    Args:
        data: Any Python object
        max_list_items_preview: How many items to preview if listing individual elements

    Returns:
        A string describing the structure and summary of the data.
    """
    if data is None:
        return "Null"

    if isinstance(data, bool):
        return f"Boolean ({data})"

    if isinstance(data, (int, float)):
        return f"{type(data).__name__} ({data})"

    if isinstance(data, (datetime, date, time, UUID, Decimal)):
        return f"{type(data).__name__} ({data})"

    if isinstance(data, str):
        if len(data) <= 100:
            return f"String: '{data}'"
        return f"String (length={len(data)}): '{data[:60]}...'"

    if isinstance(data, (bytes, bytearray)):
        return f"Binary data (size={len(data)} bytes)"

    if isinstance(data, list):
        if not data:
            return "Empty List"

        # Check if it's a list of dictionaries (common for datasets)
        first_item = data[0]
        if isinstance(first_item, dict):
            keys_info = []
            for k, v in first_item.items():
                val_type = type(v).__name__
                keys_info.append(f"{k}: {val_type}")
            schema_str = ", ".join(keys_info)
            return f"List of {len(data)} objects. Schema: {{{schema_str}}}"

        # Check if it's a list of numbers (float/int)
        if all(isinstance(x, (int, float)) for x in data[:10]):
            nums = [float(x) for x in data]
            if nums:
                min_val = min(nums)
                max_val = max(nums)
                avg_val = sum(nums) / len(nums)
                return f"List of {len(data)} numbers (float/int). Stats: min={min_val:.4f}, max={max_val:.4f}, avg={avg_val:.4f}"

        # General list
        types = set(type(x).__name__ for x in data[:10])
        types_str = "|".join(types)
        return f"List of {len(data)} items of type ({types_str})"

    if isinstance(data, dict):
        if not data:
            return "Empty Object"

        keys_preview = []
        for k, v in list(data.items())[:10]:
            val_summary = ""
            if (
                isinstance(v, (int, float, bool, str))
                and not isinstance(v, str)
                or (isinstance(v, str) and len(v) < 30)
            ):
                val_summary = f"={repr(v)}"
            else:
                val_summary = f" ({type(v).__name__})"
            keys_preview.append(f"'{k}'{val_summary}")

        total_keys = len(data)
        keys_str = ", ".join(keys_preview)
        if total_keys > 10:
            keys_str += f", ... (+{total_keys - 10} more)"

        return f"Object with {total_keys} keys: {{{keys_str}}}"

    return f"Unknown Type ({type(data).__name__}): {str(data)[:100]}"

bfast_llm.decoder

BFastDecoder

Decoder for the B-FAST binary serialization protocol.

Source code in bfast_llm/decoder.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class BFastDecoder:
    """Decoder for the B-FAST binary serialization protocol."""

    @classmethod
    def decode(cls, data: Union[bytes, bytearray]) -> Any:
        """
        Decode B-FAST binary data to Python objects.

        Args:
            data: bytes or bytearray containing B-FAST data compressed

        Returns:
            Decoded Python object
        """
        if isinstance(data, bytearray):
            data = bytes(data)

        try:
            bf = b_fast.BFast()
            return bf.decode_packed(data, decompress=True)
        except Exception as e:
            raise BFastError(f"B-FAST decoding failed: {e}")

decode(data) classmethod

Decode B-FAST binary data to Python objects.

Parameters:

Name Type Description Default
data Union[bytes, bytearray]

bytes or bytearray containing B-FAST data compressed

 required

Returns:

Type Description
Any

Decoded Python object
Source code in bfast_llm/decoder.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@classmethod
def decode(cls, data: Union[bytes, bytearray]) -> Any:
    """
    Decode B-FAST binary data to Python objects.

    Args:
        data: bytes or bytearray containing B-FAST data compressed

    Returns:
        Decoded Python object
    """
    if isinstance(data, bytearray):
        data = bytes(data)

    try:
        bf = b_fast.BFast()
        return bf.decode_packed(data, decompress=True)
    except Exception as e:
        raise BFastError(f"B-FAST decoding failed: {e}")

BFastError

Bases: Exception

Exception raised for B-FAST parsing and decoding errors.

Source code in bfast_llm/decoder.py 
12
13
14
15
class BFastError(Exception):
    """Exception raised for B-FAST parsing and decoding errors."""

    pass

bfast_llm.registry

BFastRegistry

Registry for managing and storing B-FAST binary payloads.

Source code in bfast_llm/registry.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class BFastRegistry:
    """Registry for managing and storing B-FAST binary payloads."""

    def __init__(self, db_path: Optional[Union[str, Path]] = None):
        """
        Initialize the registry.

        Args:
            db_path: Path to SQLite file for persistent storage. If None, uses in-memory storage.
        """
        self.db_path = db_path
        self.in_memory_cache: Dict[str, bytes] = {}
        self.summaries: Dict[str, str] = {}

        if db_path is not None:
            self._init_db()

    def _init_db(self) -> None:
        db_file = Path(self.db_path)
        # Ensure parent directories exist
        db_file.parent.mkdir(parents=True, exist_ok=True)

        conn = sqlite3.connect(str(db_file))
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS bfast_payloads (
                ref_id TEXT PRIMARY KEY,
                binary_data BLOB NOT NULL,
                summary TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()
        conn.close()

    def register(self, binary_data: bytes, summary: Optional[str] = None) -> str:
        """
        Register a B-FAST binary payload in the registry.

        Args:
            binary_data: The B-FAST binary payload (bytes)
            summary: Optional pre-computed summary. If None, it will be automatically generated.

        Returns:
            A formatted XML/HTML reference tag that can be sent to the LLM.
        """
        # Generate content-hash-based reference ID to automatically deduplicate identical payloads
        hasher = hashlib.sha256(binary_data)
        ref_id = f"bfast_{hasher.hexdigest()[:12]}"

        if summary is None:
            try:
                decoded = BFastDecoder.decode(binary_data)
                summary = BFastSummarizer.summarize(decoded)
            except Exception as e:
                summary = f"BFast Payload (failed to decode: {e})"

        # Format human-readable size
        size_bytes = len(binary_data)
        if size_bytes < 1024:
            size_str = f"{size_bytes}B"
        else:
            size_str = f"{size_bytes / 1024:.2f}KB"

        # Store in database if persistent, else in-memory
        if self.db_path is not None:
            conn = sqlite3.connect(str(self.db_path))
            cursor = conn.cursor()
            cursor.execute(
                "INSERT OR REPLACE INTO bfast_payloads (ref_id, binary_data, summary) VALUES (?, ?, ?)",
                (ref_id, binary_data, summary),
            )
            conn.commit()
            conn.close()
        else:
            self.in_memory_cache[ref_id] = binary_data
            self.summaries[ref_id] = summary

        return f'<bfast-ref id="{ref_id}" size="{size_str}" summary="{summary}"/>'

    def get(self, ref_id: str) -> Optional[bytes]:
        """Retrieve the binary payload by reference ID."""
        # Clean up ID if LLM included surrounding quotes or brackets
        ref_id = ref_id.strip("'\"<>[]")

        if self.db_path is not None:
            conn = sqlite3.connect(str(self.db_path))
            cursor = conn.cursor()
            cursor.execute(
                "SELECT binary_data FROM bfast_payloads WHERE ref_id = ?", (ref_id,)
            )
            row = cursor.fetchone()
            conn.close()
            if row:
                return row[0]
        else:
            return self.in_memory_cache.get(ref_id)

        return None

    def get_summary(self, ref_id: str) -> Optional[str]:
        """Retrieve the summary of a payload by reference ID."""
        ref_id = ref_id.strip("'\"<>[]")

        if self.db_path is not None:
            conn = sqlite3.connect(str(self.db_path))
            cursor = conn.cursor()
            cursor.execute(
                "SELECT summary FROM bfast_payloads WHERE ref_id = ?", (ref_id,)
            )
            row = cursor.fetchone()
            conn.close()
            if row:
                return row[0]
        else:
            return self.summaries.get(ref_id)

        return None

    def get_decoded(self, ref_id: str) -> Any:
        """Retrieve and decode the payload by reference ID."""
        binary_data = self.get(ref_id)
        if binary_data is None:
            raise KeyError(f"Payload ID '{ref_id}' not found in registry.")
        return BFastDecoder.decode(binary_data)

    def clear(self) -> None:
        """Clear all payloads in the registry."""
        if self.db_path is not None:
            conn = sqlite3.connect(str(self.db_path))
            cursor = conn.cursor()
            cursor.execute("DELETE FROM bfast_payloads")
            conn.commit()
            conn.close()
        else:
            self.in_memory_cache.clear()
            self.summaries.clear()

__init__(db_path=None)

Initialize the registry.

Parameters:

Name Type Description Default
db_path Optional[Union[str, Path]]

Path to SQLite file for persistent storage. If None, uses in-memory storage.

 None
Source code in bfast_llm/registry.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, db_path: Optional[Union[str, Path]] = None):
    """
    Initialize the registry.

    Args:
        db_path: Path to SQLite file for persistent storage. If None, uses in-memory storage.
    """
    self.db_path = db_path
    self.in_memory_cache: Dict[str, bytes] = {}
    self.summaries: Dict[str, str] = {}

    if db_path is not None:
        self._init_db()

clear()

Clear all payloads in the registry.

Source code in bfast_llm/registry.py 
136
137
138
139
140
141
142
143
144
145
146
def clear(self) -> None:
    """Clear all payloads in the registry."""
    if self.db_path is not None:
        conn = sqlite3.connect(str(self.db_path))
        cursor = conn.cursor()
        cursor.execute("DELETE FROM bfast_payloads")
        conn.commit()
        conn.close()
    else:
        self.in_memory_cache.clear()
        self.summaries.clear()

get(ref_id)

Retrieve the binary payload by reference ID.

Source code in bfast_llm/registry.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get(self, ref_id: str) -> Optional[bytes]:
    """Retrieve the binary payload by reference ID."""
    # Clean up ID if LLM included surrounding quotes or brackets
    ref_id = ref_id.strip("'\"<>[]")

    if self.db_path is not None:
        conn = sqlite3.connect(str(self.db_path))
        cursor = conn.cursor()
        cursor.execute(
            "SELECT binary_data FROM bfast_payloads WHERE ref_id = ?", (ref_id,)
        )
        row = cursor.fetchone()
        conn.close()
        if row:
            return row[0]
    else:
        return self.in_memory_cache.get(ref_id)

    return None

get_decoded(ref_id)

Retrieve and decode the payload by reference ID.

Source code in bfast_llm/registry.py 
129
130
131
132
133
134
def get_decoded(self, ref_id: str) -> Any:
    """Retrieve and decode the payload by reference ID."""
    binary_data = self.get(ref_id)
    if binary_data is None:
        raise KeyError(f"Payload ID '{ref_id}' not found in registry.")
    return BFastDecoder.decode(binary_data)

get_summary(ref_id)

Retrieve the summary of a payload by reference ID.

Source code in bfast_llm/registry.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def get_summary(self, ref_id: str) -> Optional[str]:
    """Retrieve the summary of a payload by reference ID."""
    ref_id = ref_id.strip("'\"<>[]")

    if self.db_path is not None:
        conn = sqlite3.connect(str(self.db_path))
        cursor = conn.cursor()
        cursor.execute(
            "SELECT summary FROM bfast_payloads WHERE ref_id = ?", (ref_id,)
        )
        row = cursor.fetchone()
        conn.close()
        if row:
            return row[0]
    else:
        return self.summaries.get(ref_id)

    return None

register(binary_data, summary=None)

Register a B-FAST binary payload in the registry.

Parameters:

Name Type Description Default
binary_data bytes

The B-FAST binary payload (bytes)

 required
summary Optional[str]

Optional pre-computed summary. If None, it will be automatically generated.

 None

Returns:

Type Description
str

A formatted XML/HTML reference tag that can be sent to the LLM.
Source code in bfast_llm/registry.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def register(self, binary_data: bytes, summary: Optional[str] = None) -> str:
    """
    Register a B-FAST binary payload in the registry.

    Args:
        binary_data: The B-FAST binary payload (bytes)
        summary: Optional pre-computed summary. If None, it will be automatically generated.

    Returns:
        A formatted XML/HTML reference tag that can be sent to the LLM.
    """
    # Generate content-hash-based reference ID to automatically deduplicate identical payloads
    hasher = hashlib.sha256(binary_data)
    ref_id = f"bfast_{hasher.hexdigest()[:12]}"

    if summary is None:
        try:
            decoded = BFastDecoder.decode(binary_data)
            summary = BFastSummarizer.summarize(decoded)
        except Exception as e:
            summary = f"BFast Payload (failed to decode: {e})"

    # Format human-readable size
    size_bytes = len(binary_data)
    if size_bytes < 1024:
        size_str = f"{size_bytes}B"
    else:
        size_str = f"{size_bytes / 1024:.2f}KB"

    # Store in database if persistent, else in-memory
    if self.db_path is not None:
        conn = sqlite3.connect(str(self.db_path))
        cursor = conn.cursor()
        cursor.execute(
            "INSERT OR REPLACE INTO bfast_payloads (ref_id, binary_data, summary) VALUES (?, ?, ?)",
            (ref_id, binary_data, summary),
        )
        conn.commit()
        conn.close()
    else:
        self.in_memory_cache[ref_id] = binary_data
        self.summaries[ref_id] = summary

    return f'<bfast-ref id="{ref_id}" size="{size_str}" summary="{summary}"/>'

bfast_llm.tool

BFastTool

Implement LLM tools for interacting with B-FAST context.

Source code in bfast_llm/tool.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class BFastTool:
    """Implement LLM tools for interacting with B-FAST context."""

    def __init__(self, registry: BFastRegistry):
        self.registry = registry

    def bfast_retrieve(self, ref_id: str, format: str = "markdown") -> str:
        """
        Retrieve and decode a B-FAST payload from the registry, formatting it as requested.

        Args:
            ref_id: The reference ID of the payload (e.g., 'bfast_abc123')
            format: Output format. One of 'markdown', 'json', 'yaml', or 'csv'. Default is 'markdown'.

        Returns:
            Decoded payload formatted as a string.
        """
        try:
            decoded = self.registry.get_decoded(ref_id)
        except KeyError:
            return f"Error: Payload ID '{ref_id}' not found in registry."
        except Exception as e:
            return f"Error: Failed to decode payload '{ref_id}': {e}"

        format = format.lower()

        if format == "json":
            # Handle non-serializable types by converting to string
            def default_serializer(o):
                if hasattr(o, "isoformat"):
                    return o.isoformat()
                return str(o)

            return json.dumps(decoded, indent=2, default=default_serializer)

        elif format == "yaml":
            try:
                import yaml

                # Custom representers for special types
                def decimal_representer(dumper, data):
                    return dumper.represent_scalar("tag:yaml.org,2002:float", str(data))

                def datetime_representer(dumper, data):
                    return dumper.represent_scalar(
                        "tag:yaml.org,2002:timestamp", data.isoformat()
                    )

                def uuid_representer(dumper, data):
                    return dumper.represent_scalar("tag:yaml.org,2002:str", str(data))

                yaml.add_representer(
                    type(None),
                    lambda dumper, _: dumper.represent_scalar(
                        "tag:yaml.org,2002:null", "null"
                    ),
                )
                # Register if types are present
                from decimal import Decimal
                from datetime import datetime, date, time
                from uuid import UUID

                yaml.add_representer(Decimal, decimal_representer)
                yaml.add_representer(datetime, datetime_representer)
                yaml.add_representer(date, datetime_representer)
                yaml.add_representer(time, datetime_representer)
                yaml.add_representer(UUID, uuid_representer)

                return yaml.dump(decoded, default_flow_style=False)
            except ImportError:
                # Fallback to JSON
                return (
                    self.bfast_retrieve(ref_id, format="json")
                    + "\n\n(Note: PyYAML not installed. Fell back to JSON.)"
                )

        elif format == "csv":
            if isinstance(decoded, list) and decoded and isinstance(decoded[0], dict):
                output = io.StringIO()
                headers = list(decoded[0].keys())
                writer = csv.DictWriter(output, fieldnames=headers)
                writer.writeheader()
                for row in decoded:
                    # format non-serializable objects
                    formatted_row = {}
                    for k, v in row.items():
                        if hasattr(v, "isoformat"):
                            formatted_row[k] = v.isoformat()
                        else:
                            formatted_row[k] = str(v)
                    writer.writerow(formatted_row)
                return output.getvalue()
            else:
                return (
                    self.bfast_retrieve(ref_id, format="json")
                    + "\n\n(Note: Payload is not a list of dictionaries. Fell back to JSON.)"
                )

        # Default/Markdown
        else:
            return self._format_markdown(decoded, ref_id)

    def _format_markdown(self, decoded: Any, ref_id: str) -> str:
        """Format decoded B-FAST values nicely as Markdown."""
        summary = self.registry.get_summary(ref_id) or "BFast Payload"
        header = f"### ⚡ B-FAST Payload `{ref_id}`\n*{summary}*\n\n"

        # Table formatting for list of dicts
        if isinstance(decoded, list) and decoded and isinstance(decoded[0], dict):
            headers = list(decoded[0].keys())
            lines = []
            lines.append("| " + " | ".join(headers) + " |")
            lines.append("| " + " | ".join(["---"] * len(headers)) + " |")

            for item in decoded:
                row = []
                for h in headers:
                    val = item.get(h, "")
                    if val is None:
                        row.append("null")
                    elif isinstance(val, bool):
                        row.append("✅ Yes" if val else "❌ No")
                    elif hasattr(val, "isoformat"):
                        row.append(val.isoformat())
                    else:
                        row.append(str(val).replace("\n", " ").replace("|", "\\|"))
                lines.append("| " + " | ".join(row) + " |")

            return header + "\n".join(lines)

        # List formatting for single dict
        elif isinstance(decoded, dict):
            lines = []
            lines.append("| Key | Value (Type) |")
            lines.append("| --- | --- |")
            for k, v in decoded.items():
                val_str = ""
                if v is None:
                    val_str = "*null*"
                elif isinstance(v, bool):
                    val_str = "✅ Yes" if v else "❌ No"
                elif hasattr(v, "isoformat"):
                    val_str = f"`{v.isoformat()}`"
                else:
                    val_str = f"`{v}`"
                lines.append(f"| **{k}** | {val_str} ({type(v).__name__}) |")
            return header + "\n".join(lines)

        # Fallback for plain values
        return f"{header}```python\n{repr(decoded)}\n```"

    def get_openai_tool_definition(self) -> Dict[str, Any]:
        """Get the OpenAI/Anthropic compatible tool schema for bfast_retrieve."""
        return {
            "type": "function",
            "function": {
                "name": "bfast_retrieve",
                "description": (
                    "Retrieve the original content of a compressed B-FAST payload. "
                    "Use this when you need to inspect the full contents of data references (e.g. <bfast-ref id='...'/>)."
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "ref_id": {
                            "type": "string",
                            "description": "The unique reference ID of the payload to retrieve, starting with 'bfast_'.",
                        },
                        "format": {
                            "type": "string",
                            "enum": ["markdown", "json", "yaml", "csv"],
                            "default": "markdown",
                            "description": "The format to return the decoded data in. Default is 'markdown'.",
                        },
                    },
                    "required": ["ref_id"],
                },
            },
        }

bfast_retrieve(ref_id, format='markdown')

Retrieve and decode a B-FAST payload from the registry, formatting it as requested.

Parameters:

Name Type Description Default
ref_id str

The reference ID of the payload (e.g., 'bfast_abc123')

 required
format str

Output format. One of 'markdown', 'json', 'yaml', or 'csv'. Default is 'markdown'.

 'markdown'

Returns:

Type Description
str

Decoded payload formatted as a string.
Source code in bfast_llm/tool.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def bfast_retrieve(self, ref_id: str, format: str = "markdown") -> str:
    """
    Retrieve and decode a B-FAST payload from the registry, formatting it as requested.

    Args:
        ref_id: The reference ID of the payload (e.g., 'bfast_abc123')
        format: Output format. One of 'markdown', 'json', 'yaml', or 'csv'. Default is 'markdown'.

    Returns:
        Decoded payload formatted as a string.
    """
    try:
        decoded = self.registry.get_decoded(ref_id)
    except KeyError:
        return f"Error: Payload ID '{ref_id}' not found in registry."
    except Exception as e:
        return f"Error: Failed to decode payload '{ref_id}': {e}"

    format = format.lower()

    if format == "json":
        # Handle non-serializable types by converting to string
        def default_serializer(o):
            if hasattr(o, "isoformat"):
                return o.isoformat()
            return str(o)

        return json.dumps(decoded, indent=2, default=default_serializer)

    elif format == "yaml":
        try:
            import yaml

            # Custom representers for special types
            def decimal_representer(dumper, data):
                return dumper.represent_scalar("tag:yaml.org,2002:float", str(data))

            def datetime_representer(dumper, data):
                return dumper.represent_scalar(
                    "tag:yaml.org,2002:timestamp", data.isoformat()
                )

            def uuid_representer(dumper, data):
                return dumper.represent_scalar("tag:yaml.org,2002:str", str(data))

            yaml.add_representer(
                type(None),
                lambda dumper, _: dumper.represent_scalar(
                    "tag:yaml.org,2002:null", "null"
                ),
            )
            # Register if types are present
            from decimal import Decimal
            from datetime import datetime, date, time
            from uuid import UUID

            yaml.add_representer(Decimal, decimal_representer)
            yaml.add_representer(datetime, datetime_representer)
            yaml.add_representer(date, datetime_representer)
            yaml.add_representer(time, datetime_representer)
            yaml.add_representer(UUID, uuid_representer)

            return yaml.dump(decoded, default_flow_style=False)
        except ImportError:
            # Fallback to JSON
            return (
                self.bfast_retrieve(ref_id, format="json")
                + "\n\n(Note: PyYAML not installed. Fell back to JSON.)"
            )

    elif format == "csv":
        if isinstance(decoded, list) and decoded and isinstance(decoded[0], dict):
            output = io.StringIO()
            headers = list(decoded[0].keys())
            writer = csv.DictWriter(output, fieldnames=headers)
            writer.writeheader()
            for row in decoded:
                # format non-serializable objects
                formatted_row = {}
                for k, v in row.items():
                    if hasattr(v, "isoformat"):
                        formatted_row[k] = v.isoformat()
                    else:
                        formatted_row[k] = str(v)
                writer.writerow(formatted_row)
            return output.getvalue()
        else:
            return (
                self.bfast_retrieve(ref_id, format="json")
                + "\n\n(Note: Payload is not a list of dictionaries. Fell back to JSON.)"
            )

    # Default/Markdown
    else:
        return self._format_markdown(decoded, ref_id)

get_openai_tool_definition()

Get the OpenAI/Anthropic compatible tool schema for bfast_retrieve.

Source code in bfast_llm/tool.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def get_openai_tool_definition(self) -> Dict[str, Any]:
    """Get the OpenAI/Anthropic compatible tool schema for bfast_retrieve."""
    return {
        "type": "function",
        "function": {
            "name": "bfast_retrieve",
            "description": (
                "Retrieve the original content of a compressed B-FAST payload. "
                "Use this when you need to inspect the full contents of data references (e.g. <bfast-ref id='...'/>)."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "ref_id": {
                        "type": "string",
                        "description": "The unique reference ID of the payload to retrieve, starting with 'bfast_'.",
                    },
                    "format": {
                        "type": "string",
                        "enum": ["markdown", "json", "yaml", "csv"],
                        "default": "markdown",
                        "description": "The format to return the decoded data in. Default is 'markdown'.",
                    },
                },
                "required": ["ref_id"],
            },
        },
    }