feat: Enable basic sandboxed tool run functionality

letta/agent.py

@@ -5,6 +5,7 @@
 from abc import ABC, abstractmethod
 from typing import List, Literal, Optional, Tuple, Union
 
+from e2b_code_interpreter import Sandbox
 from tqdm import tqdm
 
 from letta.agent_store.storage import StorageConnector
@@ -651,6 +652,48 @@ def _handle_ai_response(
 
                 function_args["self"] = self  # need to attach self to arg since it's dynamically linked
 
+                # Execute tool in sandbox
+                if (function_name in ["print_hello_world", "print_message"]): # remove hardcode
+                    sbx = Sandbox(api_key="e2b_99c3fa2617e66233a57bf267e1e1f462745b0da5") # store api key in env var
+                    code = ""
+
+                    # 1. Import dependencies
+                    # package = ""
+                    # sbx.commands.run(f"pip3 install {package}")
+
+                    # 2. Set Params
+                    for param in function_args:
+                        if param != "self":
+                            code += param + ' = "' + function_args[param] + '"\n'
+
+                    # 3. Add function source code
+                    tool = [tool for tool in self.tools if tool.name == function_name][0]
+                    code += tool.source_code + "\n"
+
+                    # 4. Add function call
+                    code += function_name + "("
+
+                    # 5. Populate params for function call
+                    for param in function_args:
+                        if param != "self":
+                            code += param + ","
+
+                    code += ")"
+
+                    # 6. Execute code and extract result or throw
+                    execution = sbx.run_code(code)
+                    if execution.error is not None:
+                        raise execution.error
+                    elif len(execution.results) == 0:
+                        function_response = ""
+                    else:
+                        function_response = execution.results[0].text
+
+                    # 7. Kill sandbox
+                    sbx.kill()
+                else:
+                    function_response = function_to_call(**function_args)
+
                 function_response = function_to_call(**function_args)
                 if function_name in ["conversation_search", "conversation_search_date", "archival_memory_search"]:
                     # with certain functions we rely on the paging mechanism to handle overflow

tests/test_tools.py

@@ -162,3 +162,68 @@ def core_memory_clear(self: Agent):
 
 def test_custom_import_tool(client):
     pass
+
+
+def test_run_basic_tool_in_sandbox(client: Union[LocalClient, RESTClient]):
+    """Test creation of a simple tool with no input params"""
+
+    def print_hello_world():
+        """
+        Returns:
+            str: A static string "Hello world".
+
+        """
+        print("hello world")
+        return "hello world"
+
+    tools = client.list_tools()
+    print(f"Original tools {[t.name for t in tools]}")
+
+    tool = client.create_tool(print_hello_world, name="print_hello_world", tags=["extras"])
+
+    tools = client.list_tools()
+    assert tool in tools, f"Expected {tool.name} in {[t.name for t in tools]}"
+    print(f"Updated tools {[t.name for t in tools]}")
+
+    # check tool id
+    tool = client.get_tool(tool.id)
+    assert tool is not None, "Expected tool to be created"
+    assert tool.id == tool.id, f"Expected {tool.id} to be {tool.id}"
+
+    # create agent with tool
+    agent_state = client.create_agent(tools=[tool.name])
+    response = client.user_message(agent_id=agent_state.id, message="hi please use the tool called print_hello_world")
+
+
+def test_run_tool_with_str_params_in_sandbox(client: Union[LocalClient, RESTClient]):
+    """Test creation of a simple tool that relies on a provided string param"""
+
+    def print_message(message: str):
+        """
+        Args:
+            message (str): The message to print.
+
+        Returns:
+            str: A static string "Hello world".
+
+        """
+        print(message)
+        return message
+
+    tools = client.list_tools()
+    print(f"Original tools {[t.name for t in tools]}")
+
+    tool = client.create_tool(print_message, name="print_message", tags=["extras"])
+
+    tools = client.list_tools()
+    assert tool in tools, f"Expected {tool.name} in {[t.name for t in tools]}"
+    print(f"Updated tools {[t.name for t in tools]}")
+
+    # check tool id
+    tool = client.get_tool(tool.id)
+    assert tool is not None, "Expected tool to be created"
+    assert tool.id == tool.id, f"Expected {tool.id} to be {tool.id}"
+
+    # create agent with tool
+    agent_state = client.create_agent(tools=[tool.name])
+    response = client.user_message(agent_id=agent_state.id, message="hi please use the tool called print_message")