Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 19 additions & 3 deletions AgentCrew/modules/chat/console_ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,7 @@ def display_tool_confirmation_request(self, tool_info):
# Use Rich to print the prompt but still need to use input() for user interaction
self.console.print(
Text(
"\nAllow this tool to run? [y]es/[n]o/[all] future calls: ",
"\nAllow this tool to run? [y]es/[n]o/[a]lways in this session/[f]orever (this and future sessions): ",
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should be [a]ll (in this session) instead of [a]lways in this session

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed

style=RICH_STYLE_YELLOW,
),
end="",
Expand All @@ -471,7 +471,7 @@ def display_tool_confirmation_request(self, tool_info):
confirmation_id, {"action": "deny"}
)
break
elif response in ["all", "a"]:
elif response in ["a", "all"]:
self.message_handler.resolve_tool_confirmation(
confirmation_id, {"action": "approve_all"}
)
Expand All @@ -482,9 +482,25 @@ def display_tool_confirmation_request(self, tool_info):
approved_text.append("' for this session.")
self.console.print(approved_text)
break
elif response in ["f", "forever"]:
# Add tool to persistent auto-approval list
from AgentCrew.modules.config import ConfigManagement
config_manager = ConfigManagement()
config_manager.write_auto_approval_tools(tool_use["name"], add=True)

self.message_handler.resolve_tool_confirmation(
confirmation_id, {"action": "approve"}
)
saved_text = Text(
"✓ Tool '", style=RICH_STYLE_YELLOW
)
saved_text.append(tool_use["name"])
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

merge to 1

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed

saved_text.append("' will be auto-approved forever.")
self.console.print(saved_text)
break
else:
self.console.print(
Text("Please enter 'y', 'n', or 'all'.", style=RICH_STYLE_YELLOW)
Text("Please enter 'y', 'n', 'a', or 'f'.", style=RICH_STYLE_YELLOW)
)
self._start_input_thread()

Expand Down
14 changes: 11 additions & 3 deletions AgentCrew/modules/chat/message/tool_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,26 @@
import asyncio

from AgentCrew.modules import logger
from AgentCrew.modules.config import ConfigManagement
from AgentCrew.modules.llm.message import MessageTransformer


class ToolManager:
"""Manages tool execution and confirmation."""

def _load_persistent_auto_approved_tools(self):
"""Load persistent auto-approved tools from config."""
config_manager = ConfigManagement()
return set(config_manager.get_auto_approval_tools())

def __init__(self, message_handler):
from AgentCrew.modules.chat.message import MessageHandler

if isinstance(message_handler, MessageHandler):
self.message_handler = message_handler
self._auto_approved_tools = set() # Track tools approved for all future calls

self._auto_approved_tools = self._load_persistent_auto_approved_tools()

self._pending_confirmations = {} # Store futures for confirmation requests
self._next_confirmation_id = 0 # ID counter for confirmation requests
self.yolo_mode = False # Enable/disable auto-approval mode
Expand Down Expand Up @@ -62,7 +70,6 @@ async def execute_tool(self, tool_use: Dict[str, Any]):
)
return

# For all other tools, check if confirmation is needed
if not self.yolo_mode and tool_name not in self._auto_approved_tools:
# Request confirmation from the user
confirmation = await self._wait_for_tool_confirmation(tool_use)
Expand Down Expand Up @@ -242,4 +249,5 @@ def _post_tool_transfer(self, tool_use, tool_result):

def reset_approved_tools(self):
"""Reset approved tools for a new conversation."""
self._auto_approved_tools = set()
# Reload persistent tools from config and reset session-based approvals
self._auto_approved_tools = self._load_persistent_auto_approved_tools()
43 changes: 39 additions & 4 deletions AgentCrew/modules/config/config_management.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,28 +232,30 @@ def read_global_config_data(self) -> Dict[str, Any]:
config_path = self._get_global_config_file_path()
try:
if not os.path.exists(config_path):
return {"api_keys": {}} # Default structure if file doesn't exist
return {"api_keys": {}, "auto_approval_tools": []} # Default structure if file doesn't exist
with open(config_path, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
logger.warning(
f"Warning: Global config file {config_path} does not contain a valid JSON object. Returning default."
)
return {"api_keys": {}}
return {"api_keys": {}, "auto_approval_tools": []}
# Ensure api_keys key exists and is a dict
if "api_keys" not in data or not isinstance(data.get("api_keys"), dict):
data["api_keys"] = {}
if "auto_approval_tools" not in data or not isinstance(data.get("auto_approval_tools"), list):
data["auto_approval_tools"] = []
return data
except json.JSONDecodeError:
logger.warning(
f"Warning: Error decoding global config file {config_path}. Returning default config."
)
return {"api_keys": {}}
return {"api_keys": {}, "auto_approval_tools": []}
except Exception as e:
logger.warning(
f"Warning: Could not read global config file {config_path}: {e}. Returning default config."
)
return {"api_keys": {}}
return {"api_keys": {}, "auto_approval_tools": []}

def write_global_config_data(self, config_data: Dict[str, Any]) -> None:
"""Writes data to the global config.json file."""
Expand Down Expand Up @@ -561,3 +563,36 @@ def get_last_used_agent(self) -> Optional[str]:
"""
last_used = self.get_last_used_settings()
return last_used.get("agent")

def get_auto_approval_tools(self) -> List[str]:
"""
Get the list of auto-approved tools from global config.

Returns:
List of tool names that are auto-approved.
"""
global_config = self.read_global_config_data()
return global_config.get("auto_approval_tools", [])

def write_auto_approval_tools(self, tool_name: str, add: bool = True) -> None:
"""
Add or remove a tool from the auto-approval list in global config.

Args:
tool_name: Name of the tool to add/remove from auto-approval list.
add: True to add the tool, False to remove it.
"""
try:
global_config = self.read_global_config_data()
auto_approval_tools = global_config.get("auto_approval_tools", [])

if add and tool_name not in auto_approval_tools:
auto_approval_tools.append(tool_name)
elif not add and tool_name in auto_approval_tools:
auto_approval_tools.remove(tool_name)

global_config["auto_approval_tools"] = auto_approval_tools
self.write_global_config_data(global_config)
except Exception as e:
action = "add" if add else "remove"
logger.warning(f"Warning: Failed to {action} tool {tool_name} from auto-approval list: {e}")