Repository: cpacker/MemGPT
File: letta/server/server.py (Python)

Code before:
            return FunctionReturn(
                id="null",
                function_call_id="null",
                date=get_utc_time(),
                status="error",
                function_return=error_msg,
                stdout=[""],
                stderr=[traceback.format_exc()],
            )

    def get_error_msg_for_func_return(self, tool_name, exception_message):
        # same as agent.py
        from letta.constants import MAX_ERROR_MESSAGE_CHAR_LIMIT

        error_msg = f"Error executing tool {tool_name}: {exception_message}"
        if len(error_msg) > MAX_ERROR_MESSAGE_CHAR_LIMIT:
            error_msg = error_msg[:MAX_ERROR_MESSAGE_CHAR_LIMIT]
        return error_msg

    # Composio wrappers
    def get_composio_client(self, api_key: Optional[str] = None):
        if api_key:
            return Composio(api_key=api_key)
        elif tool_settings.composio_api_key:
            return Composio(api_key=tool_settings.composio_api_key)
        else:
            return Composio()

    def get_composio_apps(self, api_key: Optional[str] = None) -> List["AppModel"]:
        """Get a list of all Composio apps with actions"""
        apps = self.get_composio_client(api_key=api_key).apps.get()
        apps_with_actions = []
        for app in apps:
            # A bit of hacky logic until composio patches this
            if app.meta["actionsCount"] > 0 and not app.name.lower().endswith("_beta"):
                apps_with_actions.append(app)

        return apps_with_actions

    def get_composio_actions_from_app_name(self, composio_app_name: str, api_key: Optional[str] = None) -> List["ActionModel"]:
        actions = self.get_composio_client(api_key=api_key).actions.get(apps=[composio_app_name])
        return actions


Review comment (suggestion):
maybe we should throw some kind of error or warning here if `len(llm_configs) > 1`? and similar for embedding config 

Code after:
        return letta_agent.get_context_window()

    def run_tool_from_source(
        self,
        user_id: str,
        tool_args: str,
        tool_source: str,
        tool_source_type: Optional[str] = None,
        tool_name: Optional[str] = None,
    ) -> FunctionReturn:
        """Run a tool from source code"""

        try:
            tool_args_dict = json.loads(tool_args)
        except json.JSONDecodeError:
            raise ValueError("Invalid JSON string for tool_args")

        if tool_source_type is not None and tool_source_type != "python":
            raise ValueError("Only Python source code is supported at this time")

        # NOTE: we're creating a floating Tool object and NOT persiting to DB
        tool = Tool(
            name=tool_name,
            source_code=tool_source,
        )
        assert tool.name is not None, "Failed to create tool object"

        # TODO eventually allow using agent state in tools
        agent_state = None

        # Next, attempt to run the tool with the sandbox
        try:
            sandbox_run_result = ToolExecutionSandbox(tool.name, tool_args_dict, user_id, tool_object=tool).run(agent_state=agent_state)
            function_response = str(sandbox_run_result.func_return)
            stdout = [s for s in sandbox_run_result.stdout if s.strip()]
            stderr = [s for s in sandbox_run_result.stderr if s.strip()]

            # expected error
            if stderr:
                error_msg = self.get_error_msg_for_func_return(tool.name, stderr[-1])
                return FunctionReturn(
                    id="null",
                    function_call_id="null",
                    date=get_utc_time(),
                    status="error",
                    function_return=error_msg,
                    stdout=stdout,
                    stderr=stderr,
                )

            return FunctionReturn(