heaven_base.tools.registry_tool

View Source

  1from typing import Dict, Any, Optional, List, Union
  2from pydantic import BaseModel, Field
  3
  4from ..registry.registry_service import RegistryService
  5from ..baseheaventool import BaseHeavenTool, ToolArgsSchema
  6
  7import uuid
  8import json
  9from datetime import datetime
 10import copy
 11
 12      
 13class RegistryToolArgsSchema(ToolArgsSchema):
 14    arguments: Dict[str, Dict[str, Any]] = {
 15        'operation': {
 16            'name': 'operation',
 17            'type': 'str',
 18            'description': 'Operation to perform: create_registry, list_registries, get, get_all, add, update, delete, list_keys',
 19            'required': True
 20        },
 21        'registry_name': {
 22            'name': 'registry_name',
 23            'type': 'str',
 24            'description': 'Name of the registry to operate on',
 25            'required': False
 26        },
 27        'key': {
 28            'name': 'key',
 29            'type': 'str',
 30            'description': 'Key for get, add, update, delete operations',
 31            'required': False
 32        },
 33        'value_str': {
 34            'name': 'value_str',
 35            'type': 'str',
 36            'description': 'String value for add and update operations. Use this for simple string values.',
 37            'required': False
 38        },
 39        'value_dict': {
 40            'name': 'value_dict',
 41            'type': 'dict',
 42            'description': 'Dictionary/object value for add and update operations. Use this for structured data.',
 43            'required': False
 44        }
 45    }
 46
 47
 48
 49def registry_util_func(operation: str, registry_name: Optional[str] = None, 
 50                   key: Optional[str] = None, value_str: Optional[str] = None,
 51                   value_dict: Optional[Dict[str, Any]] = None) -> str:
 52        """Run the registry tool.
 53        
 54        Args:
 55            operation: Operation to perform
 56            registry_name: Name of the registry to operate on
 57            key: Key for get, add, update, delete operations
 58            value_str: String value for add and update operations
 59            value_dict: Dictionary value for add and update operations
 60            
 61        Returns:
 62            Result of the operation as a string
 63        """
 64        service = RegistryService()
 65        
 66        # Determine which value to use
 67        value = None
 68        if value_dict is not None:
 69            value = value_dict
 70        elif value_str is not None:
 71            value = value_str
 72        
 73        if operation == "create_registry":
 74            if not registry_name:
 75                return "Error: registry_name is required for create_registry operation"
 76            
 77            result = service.create_registry(registry_name)
 78            return f"Registry '{registry_name}' created: {result}"
 79        
 80        elif operation == "list_registries":
 81            registries = service.list_registries()
 82            if not registries:
 83                return "No registries found"
 84            
 85            return f"Available registries: {', '.join(registries)}"
 86        
 87        # Operations that require a registry_name
 88        if not registry_name:
 89            return "Error: registry_name is required for this operation"
 90        
 91        if operation == "get_all":
 92            items = service.get_all(registry_name)
 93            if items is None:
 94                return f"Error: Registry '{registry_name}' not found"
 95            
 96            if not items:
 97                return f"Registry '{registry_name}' is empty"
 98            
 99            return f"Items in registry '{registry_name}': {items}"
100        
101        elif operation == "list_keys":
102            keys = service.list_keys(registry_name)
103            if keys is None:
104                return f"Error: Registry '{registry_name}' not found"
105            
106            if not keys:
107                return f"Registry '{registry_name}' is empty"
108            
109            return f"Keys in registry '{registry_name}': {', '.join(keys)}"
110        
111        # Operations that require a key
112        if not key:
113            return "Error: key is required for this operation"
114        
115        if operation == "get":
116            item = service.get(registry_name, key)
117            if item is None:
118                return f"Item '{key}' not found in registry '{registry_name}'"
119            
120            return f"Item '{key}' in registry '{registry_name}': {item}"
121        
122        elif operation == "delete":
123            result = service.delete(registry_name, key)
124            return f"Item '{key}' deleted from registry '{registry_name}': {result}"
125        
126        # Operations that require a value
127        if operation in ["add", "update"]:
128            if value is None:
129                return "Error: either value_str or value_dict is required for this operation"
130            
131            if operation == "add":
132                result = service.add(registry_name, key, value)
133                return f"Item '{key}' added to registry '{registry_name}': {result}"
134            
135            elif operation == "update":
136                result = service.update(registry_name, key, value)
137                return f"Item '{key}' updated in registry '{registry_name}': {result}"
138        
139        return f"Error: Unknown operation '{operation}'"
140  
141class RegistryTool(BaseHeavenTool):
142    """Tool for managing registries."""
143    
144    name = "RegistryTool"
145    description = """Tool for managing registries in the system.
146
147    The Registry System provides key-value storage for various types of data across the system.
148    Each registry is a separate storage container that can hold multiple key-value pairs.
149
150    Operations and their required parameters:
151
152    1. create_registry:
153       - Required: registry_name
154       - Creates a new registry with the given name
155
156    2. list_registries:
157       - No parameters required
158       - Returns a list of all available registries
159
160    3. get:
161       - Required: registry_name, key
162       - Returns the value associated with the key in the specified registry
163
164    4. get_all:
165       - Required: registry_name
166       - Returns all key-value pairs in the specified registry
167
168    5. add:
169       - Required: registry_name, key, value_str OR value_dict
170       - Adds a new key-value pair to the specified registry
171       - Use value_str for simple string values, value_dict for structured data
172
173    6. update:
174       - Required: registry_name, key, value_str OR value_dict
175       - Updates an existing key with a new value in the specified registry
176       - Use value_str for simple string values, value_dict for structured data
177
178    7. delete:
179       - Required: registry_name, key
180       - Removes a key-value pair from the specified registry
181
182    8. list_keys:
183       - Required: registry_name
184       - Returns a list of all keys in the specified registry
185
186    Examples:
187    - Create a registry: operation="create_registry", registry_name="my_registry"
188    - Add a string item: operation="add", registry_name="my_registry", key="item1", value_str="value1"
189    - Add a dict item: operation="add", registry_name="my_registry", key="config", value_dict={"setting": "value"}
190    - Get an item: operation="get", registry_name="my_registry", key="item1"
191
192    Registry-reference syntax (any string field can be a pointer):
193
194    registry_key_ref=<registry_name>:<key>
195
196    • Resolves to the locator string “@<registry>/<key>” (the key itself).
197
198    • Use when a parent record just needs to point at another record.
199
200    Example:
201
202    "title": "registry_key_ref=task_registry:T123"
203
204    --> read-time result: "@task_registry/T123"
205
206    registry_object_ref=<registry_name>:<key>#/<optional/json/pointer>
207
208    • Resolves to the value stored at that key (optionally narrowed by a JSON-pointer path).
209
210    • Recursion continues on the returned value until a non-pointer is reached.
211
212    Example:
213
214    "spec": "registry_object_ref=settings_registry:colors#/header/bg"
215
216    --> read-time result: "blue" (assuming that path exists)
217
218    registry_all_ref=<registry_name>
219
220    • Resolves to the entire contents of the specified registry.
221
222    • All values in the returned registry are also resolved if they contain pointers.
223
224    Example:
225
226    "my_data": "registry_all_ref=knowledge_base"
227
228    --> read-time result: {entire contents of knowledge_base registry}
229
230    Rules & behaviour:
231
232    • Write the reference exactly as a plain string—no braces, no quotes inside.
233
234    • Reads (get, get_all) automatically resolve:
235
236    – key-refs return the locator string.
237
238    – object-refs return the full (or sliced) value.
239
240    • Pointers can chain indefinitely; cycles are detected and depth is capped at 99 hops.
241
242    • Update-guard: if you try update() on an entry whose current value is a pointer string, the operation is refused with an error telling you the target locator—modify the referenced registry/key instead, or replace the pointer string explicitly.
243    """
244    func = registry_util_func
245    args_schema = RegistryToolArgsSchema
246    is_async = False
247    
248    
249

class RegistryToolArgsSchema(heaven_base.baseheaventool.ToolArgsSchema): View Source

14class RegistryToolArgsSchema(ToolArgsSchema):
15    arguments: Dict[str, Dict[str, Any]] = {
16        'operation': {
17            'name': 'operation',
18            'type': 'str',
19            'description': 'Operation to perform: create_registry, list_registries, get, get_all, add, update, delete, list_keys',
20            'required': True
21        },
22        'registry_name': {
23            'name': 'registry_name',
24            'type': 'str',
25            'description': 'Name of the registry to operate on',
26            'required': False
27        },
28        'key': {
29            'name': 'key',
30            'type': 'str',
31            'description': 'Key for get, add, update, delete operations',
32            'required': False
33        },
34        'value_str': {
35            'name': 'value_str',
36            'type': 'str',
37            'description': 'String value for add and update operations. Use this for simple string values.',
38            'required': False
39        },
40        'value_dict': {
41            'name': 'value_dict',
42            'type': 'dict',
43            'description': 'Dictionary/object value for add and update operations. Use this for structured data.',
44            'required': False
45        }
46    }

Meta-validator for tool arguments ensuring LangChain compatibility

arguments: Dict[str, Dict[str, Any]] = {'operation': {'name': 'operation', 'type': 'str', 'description': 'Operation to perform: create_registry, list_registries, get, get_all, add, update, delete, list_keys', 'required': True}, 'registry_name': {'name': 'registry_name', 'type': 'str', 'description': 'Name of the registry to operate on', 'required': False}, 'key': {'name': 'key', 'type': 'str', 'description': 'Key for get, add, update, delete operations', 'required': False}, 'value_str': {'name': 'value_str', 'type': 'str', 'description': 'String value for add and update operations. Use this for simple string values.', 'required': False}, 'value_dict': {'name': 'value_dict', 'type': 'dict', 'description': 'Dictionary/object value for add and update operations. Use this for structured data.', 'required': False}}

Inherited Members

heaven_base.baseheaventool.ToolArgsSchema: custom_list_type; to_pydantic_schema; validate_arguments

def registry_util_func( operation: str, registry_name: Optional[str] = None, key: Optional[str] = None, value_str: Optional[str] = None, value_dict: Optional[Dict[str, Any]] = None) -> str: View Source

 50def registry_util_func(operation: str, registry_name: Optional[str] = None, 
 51                   key: Optional[str] = None, value_str: Optional[str] = None,
 52                   value_dict: Optional[Dict[str, Any]] = None) -> str:
 53        """Run the registry tool.
 54        
 55        Args:
 56            operation: Operation to perform
 57            registry_name: Name of the registry to operate on
 58            key: Key for get, add, update, delete operations
 59            value_str: String value for add and update operations
 60            value_dict: Dictionary value for add and update operations
 61            
 62        Returns:
 63            Result of the operation as a string
 64        """
 65        service = RegistryService()
 66        
 67        # Determine which value to use
 68        value = None
 69        if value_dict is not None:
 70            value = value_dict
 71        elif value_str is not None:
 72            value = value_str
 73        
 74        if operation == "create_registry":
 75            if not registry_name:
 76                return "Error: registry_name is required for create_registry operation"
 77            
 78            result = service.create_registry(registry_name)
 79            return f"Registry '{registry_name}' created: {result}"
 80        
 81        elif operation == "list_registries":
 82            registries = service.list_registries()
 83            if not registries:
 84                return "No registries found"
 85            
 86            return f"Available registries: {', '.join(registries)}"
 87        
 88        # Operations that require a registry_name
 89        if not registry_name:
 90            return "Error: registry_name is required for this operation"
 91        
 92        if operation == "get_all":
 93            items = service.get_all(registry_name)
 94            if items is None:
 95                return f"Error: Registry '{registry_name}' not found"
 96            
 97            if not items:
 98                return f"Registry '{registry_name}' is empty"
 99            
100            return f"Items in registry '{registry_name}': {items}"
101        
102        elif operation == "list_keys":
103            keys = service.list_keys(registry_name)
104            if keys is None:
105                return f"Error: Registry '{registry_name}' not found"
106            
107            if not keys:
108                return f"Registry '{registry_name}' is empty"
109            
110            return f"Keys in registry '{registry_name}': {', '.join(keys)}"
111        
112        # Operations that require a key
113        if not key:
114            return "Error: key is required for this operation"
115        
116        if operation == "get":
117            item = service.get(registry_name, key)
118            if item is None:
119                return f"Item '{key}' not found in registry '{registry_name}'"
120            
121            return f"Item '{key}' in registry '{registry_name}': {item}"
122        
123        elif operation == "delete":
124            result = service.delete(registry_name, key)
125            return f"Item '{key}' deleted from registry '{registry_name}': {result}"
126        
127        # Operations that require a value
128        if operation in ["add", "update"]:
129            if value is None:
130                return "Error: either value_str or value_dict is required for this operation"
131            
132            if operation == "add":
133                result = service.add(registry_name, key, value)
134                return f"Item '{key}' added to registry '{registry_name}': {result}"
135            
136            elif operation == "update":
137                result = service.update(registry_name, key, value)
138                return f"Item '{key}' updated in registry '{registry_name}': {result}"
139        
140        return f"Error: Unknown operation '{operation}'"

Run the registry tool.

Args: operation: Operation to perform registry_name: Name of the registry to operate on key: Key for get, add, update, delete operations value_str: String value for add and update operations value_dict: Dictionary value for add and update operations

Returns: Result of the operation as a string

class RegistryTool(heaven_base.baseheaventool.BaseHeavenTool): View Source

142class RegistryTool(BaseHeavenTool):
143    """Tool for managing registries."""
144    
145    name = "RegistryTool"
146    description = """Tool for managing registries in the system.
147
148    The Registry System provides key-value storage for various types of data across the system.
149    Each registry is a separate storage container that can hold multiple key-value pairs.
150
151    Operations and their required parameters:
152
153    1. create_registry:
154       - Required: registry_name
155       - Creates a new registry with the given name
156
157    2. list_registries:
158       - No parameters required
159       - Returns a list of all available registries
160
161    3. get:
162       - Required: registry_name, key
163       - Returns the value associated with the key in the specified registry
164
165    4. get_all:
166       - Required: registry_name
167       - Returns all key-value pairs in the specified registry
168
169    5. add:
170       - Required: registry_name, key, value_str OR value_dict
171       - Adds a new key-value pair to the specified registry
172       - Use value_str for simple string values, value_dict for structured data
173
174    6. update:
175       - Required: registry_name, key, value_str OR value_dict
176       - Updates an existing key with a new value in the specified registry
177       - Use value_str for simple string values, value_dict for structured data
178
179    7. delete:
180       - Required: registry_name, key
181       - Removes a key-value pair from the specified registry
182
183    8. list_keys:
184       - Required: registry_name
185       - Returns a list of all keys in the specified registry
186
187    Examples:
188    - Create a registry: operation="create_registry", registry_name="my_registry"
189    - Add a string item: operation="add", registry_name="my_registry", key="item1", value_str="value1"
190    - Add a dict item: operation="add", registry_name="my_registry", key="config", value_dict={"setting": "value"}
191    - Get an item: operation="get", registry_name="my_registry", key="item1"
192
193    Registry-reference syntax (any string field can be a pointer):
194
195    registry_key_ref=<registry_name>:<key>
196
197    • Resolves to the locator string “@<registry>/<key>” (the key itself).
198
199    • Use when a parent record just needs to point at another record.
200
201    Example:
202
203    "title": "registry_key_ref=task_registry:T123"
204
205    --> read-time result: "@task_registry/T123"
206
207    registry_object_ref=<registry_name>:<key>#/<optional/json/pointer>
208
209    • Resolves to the value stored at that key (optionally narrowed by a JSON-pointer path).
210
211    • Recursion continues on the returned value until a non-pointer is reached.
212
213    Example:
214
215    "spec": "registry_object_ref=settings_registry:colors#/header/bg"
216
217    --> read-time result: "blue" (assuming that path exists)
218
219    registry_all_ref=<registry_name>
220
221    • Resolves to the entire contents of the specified registry.
222
223    • All values in the returned registry are also resolved if they contain pointers.
224
225    Example:
226
227    "my_data": "registry_all_ref=knowledge_base"
228
229    --> read-time result: {entire contents of knowledge_base registry}
230
231    Rules & behaviour:
232
233    • Write the reference exactly as a plain string—no braces, no quotes inside.
234
235    • Reads (get, get_all) automatically resolve:
236
237    – key-refs return the locator string.
238
239    – object-refs return the full (or sliced) value.
240
241    • Pointers can chain indefinitely; cycles are detected and depth is capped at 99 hops.
242
243    • Update-guard: if you try update() on an entry whose current value is a pointer string, the operation is refused with an error telling you the target locator—modify the referenced registry/key instead, or replace the pointer string explicitly.
244    """
245    func = registry_util_func
246    args_schema = RegistryToolArgsSchema
247    is_async = False

Tool for managing registries.

name = 'RegistryTool'

description = 'Tool for managing registries in the system.\n\n The Registry System provides key-value storage for various types of data across the system.\n Each registry is a separate storage container that can hold multiple key-value pairs.\n\n Operations and their required parameters:\n\n 1. create_registry:\n - Required: registry_name\n - Creates a new registry with the given name\n\n 2. list_registries:\n - No parameters required\n - Returns a list of all available registries\n\n 3. get:\n - Required: registry_name, key\n - Returns the value associated with the key in the specified registry\n\n 4. get_all:\n - Required: registry_name\n - Returns all key-value pairs in the specified registry\n\n 5. add:\n - Required: registry_name, key, value_str OR value_dict\n - Adds a new key-value pair to the specified registry\n - Use value_str for simple string values, value_dict for structured data\n\n 6. update:\n - Required: registry_name, key, value_str OR value_dict\n - Updates an existing key with a new value in the specified registry\n - Use value_str for simple string values, value_dict for structured data\n\n 7. delete:\n - Required: registry_name, key\n - Removes a key-value pair from the specified registry\n\n 8. list_keys:\n - Required: registry_name\n - Returns a list of all keys in the specified registry\n\n Examples:\n - Create a registry: operation="create_registry", registry_name="my_registry"\n - Add a string item: operation="add", registry_name="my_registry", key="item1", value_str="value1"\n - Add a dict item: operation="add", registry_name="my_registry", key="config", value_dict={"setting": "value"}\n - Get an item: operation="get", registry_name="my_registry", key="item1"\n\n Registry-reference syntax (any string field can be a pointer):\n\n registry_key_ref=<registry_name>:<key>\n\n • Resolves to the locator string “@<registry>/<key>” (the key itself).\n\n • Use when a parent record just needs to point at another record.\n\n Example:\n\n "title": "registry_key_ref=task_registry:T123"\n\n --> read-time result: "@task_registry/T123"\n\n registry_object_ref=<registry_name>:<key>#/<optional/json/pointer>\n\n • Resolves to the value stored at that key (optionally narrowed by a JSON-pointer path).\n\n • Recursion continues on the returned value until a non-pointer is reached.\n\n Example:\n\n "spec": "registry_object_ref=settings_registry:colors#/header/bg"\n\n --> read-time result: "blue" (assuming that path exists)\n\n registry_all_ref=<registry_name>\n\n • Resolves to the entire contents of the specified registry.\n\n • All values in the returned registry are also resolved if they contain pointers.\n\n Example:\n\n "my_data": "registry_all_ref=knowledge_base"\n\n --> read-time result: {entire contents of knowledge_base registry}\n\n Rules & behaviour:\n\n • Write the reference exactly as a plain string—no braces, no quotes inside.\n\n • Reads (get, get_all) automatically resolve:\n\n – key-refs return the locator string.\n\n – object-refs return the full (or sliced) value.\n\n • Pointers can chain indefinitely; cycles are detected and depth is capped at 99 hops.\n\n • Update-guard: if you try update() on an entry whose current value is a pointer string, the operation is refused with an error telling you the target locator—modify the referenced registry/key instead, or replace the pointer string explicitly.\n '

def func( operation: str, registry_name: Optional[str] = None, key: Optional[str] = None, value_str: Optional[str] = None, value_dict: Optional[Dict[str, Any]] = None) -> str: View Source

 50def registry_util_func(operation: str, registry_name: Optional[str] = None, 
 51                   key: Optional[str] = None, value_str: Optional[str] = None,
 52                   value_dict: Optional[Dict[str, Any]] = None) -> str:
 53        """Run the registry tool.
 54        
 55        Args:
 56            operation: Operation to perform
 57            registry_name: Name of the registry to operate on
 58            key: Key for get, add, update, delete operations
 59            value_str: String value for add and update operations
 60            value_dict: Dictionary value for add and update operations
 61            
 62        Returns:
 63            Result of the operation as a string
 64        """
 65        service = RegistryService()
 66        
 67        # Determine which value to use
 68        value = None
 69        if value_dict is not None:
 70            value = value_dict
 71        elif value_str is not None:
 72            value = value_str
 73        
 74        if operation == "create_registry":
 75            if not registry_name:
 76                return "Error: registry_name is required for create_registry operation"
 77            
 78            result = service.create_registry(registry_name)
 79            return f"Registry '{registry_name}' created: {result}"
 80        
 81        elif operation == "list_registries":
 82            registries = service.list_registries()
 83            if not registries:
 84                return "No registries found"
 85            
 86            return f"Available registries: {', '.join(registries)}"
 87        
 88        # Operations that require a registry_name
 89        if not registry_name:
 90            return "Error: registry_name is required for this operation"
 91        
 92        if operation == "get_all":
 93            items = service.get_all(registry_name)
 94            if items is None:
 95                return f"Error: Registry '{registry_name}' not found"
 96            
 97            if not items:
 98                return f"Registry '{registry_name}' is empty"
 99            
100            return f"Items in registry '{registry_name}': {items}"
101        
102        elif operation == "list_keys":
103            keys = service.list_keys(registry_name)
104            if keys is None:
105                return f"Error: Registry '{registry_name}' not found"
106            
107            if not keys:
108                return f"Registry '{registry_name}' is empty"
109            
110            return f"Keys in registry '{registry_name}': {', '.join(keys)}"
111        
112        # Operations that require a key
113        if not key:
114            return "Error: key is required for this operation"
115        
116        if operation == "get":
117            item = service.get(registry_name, key)
118            if item is None:
119                return f"Item '{key}' not found in registry '{registry_name}'"
120            
121            return f"Item '{key}' in registry '{registry_name}': {item}"
122        
123        elif operation == "delete":
124            result = service.delete(registry_name, key)
125            return f"Item '{key}' deleted from registry '{registry_name}': {result}"
126        
127        # Operations that require a value
128        if operation in ["add", "update"]:
129            if value is None:
130                return "Error: either value_str or value_dict is required for this operation"
131            
132            if operation == "add":
133                result = service.add(registry_name, key, value)
134                return f"Item '{key}' added to registry '{registry_name}': {result}"
135            
136            elif operation == "update":
137                result = service.update(registry_name, key, value)
138                return f"Item '{key}' updated in registry '{registry_name}': {result}"
139        
140        return f"Error: Unknown operation '{operation}'"

Run the registry tool.

Returns: Result of the operation as a string

args_schema = <class 'RegistryToolArgsSchema'>

is_async = False

Inherited Members

heaven_base.baseheaventool.BaseHeavenTool: BaseHeavenTool; base_tool; get_spec; to_openai_function; get_openai_function; create_adk_tool; create