Merge main

nalgonda/agency_config_lock_manager.py

@@ -10,7 +10,7 @@ class AgencyConfigLockManager:
     """
 
     # Mapping from agency ID to its corresponding Lock.
-    _locks: "defaultdict[str, threading.Lock]" = defaultdict(threading.Lock)
+    _locks: dict[str, threading.Lock] = defaultdict(threading.Lock)
 
     @classmethod
     def get_lock(cls, agency_id: str) -> threading.Lock:

nalgonda/custom_tools/utils.py

@@ -1,4 +1,3 @@
-import os
 from pathlib import Path
 
 from agency_swarm.util import get_openai_client
@@ -26,7 +25,7 @@ def check_directory_traversal(directory: Path) -> Path:
     if ".." in directory.parts:
         raise ValueError("Directory traversal is not allowed.")
 
-    base_directory = Path(os.getcwd())
+    base_directory = Path.cwd()
 
     # Resolve the directory path against the base directory
     resolved_directory = (base_directory / directory).resolve()

nalgonda/custom_tools/write_and_save_program.py

@@ -1,4 +1,4 @@
-import os
+from pathlib import Path
 
 from agency_swarm import BaseTool
 from pydantic import Field
@@ -27,15 +27,15 @@ def run(self):
             return "Invalid file path. Directory traversal is not allowed."
 
         # Extract the directory path from the file name
-        directory_path = os.path.dirname(self.file_name)
+        directory_path = Path(self.file_name).parent
         agency_id = "test_agency_id"  # agency_id = self.context["agency_id"]  # TODO: pass agency_id to all tools
         directory = DATA_DIR / agency_id / directory_path
 
         # Ensure the directory exists
         directory.mkdir(parents=True, exist_ok=True)
 
         # Construct the full path using the directory and file name
-        full_path = directory / os.path.basename(self.file_name)
+        full_path = directory / Path(self.file_name).name
 
         # Write the file
         with open(full_path, "w") as f: