mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-09 17:31:52 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0434b23ada | ||
|
|
a9951e0694 | ||
|
|
073062cd96 |
No files matched your search
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.55" \
|
||||
"praisonai>=2.3.58" \
|
||||
"praisonai[chat]" \
|
||||
"embedchain[github,youtube]"
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.55" \
|
||||
"praisonai>=2.3.58" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[chat]" \
|
||||
"praisonai[realtime]" \
|
||||
|
||||
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.55" \
|
||||
"praisonai>=2.3.58" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[crewai]"
|
||||
|
||||
|
||||
@@ -1538,13 +1538,13 @@ Your Goal: {self.goal}"""
|
||||
logging.debug(f"{self.name} executing tool {function_name} with arguments: {arguments}")
|
||||
|
||||
# Check if approval is required for this tool
|
||||
from ..approval import is_approval_required, console_approval_callback, get_risk_level, mark_approved, ApprovalDecision
|
||||
from ..approval import is_approval_required, console_approval_callback, get_risk_level, mark_approved, ApprovalDecision, get_approval_callback
|
||||
if is_approval_required(function_name):
|
||||
risk_level = get_risk_level(function_name)
|
||||
logging.info(f"Tool {function_name} requires approval (risk level: {risk_level})")
|
||||
|
||||
# Use global approval callback or default console callback
|
||||
callback = approval_callback or console_approval_callback
|
||||
callback = get_approval_callback() or console_approval_callback
|
||||
|
||||
try:
|
||||
decision = callback(function_name, arguments, risk_level)
|
||||
|
||||
@@ -43,6 +43,15 @@ def set_approval_callback(callback_fn: Callable):
|
||||
global approval_callback
|
||||
approval_callback = callback_fn
|
||||
|
||||
def get_approval_callback() -> Optional[Callable]:
|
||||
"""Get the current approval callback function.
|
||||
|
||||
Returns the custom callback if set, otherwise None.
|
||||
This should be used instead of directly accessing approval_callback
|
||||
to ensure the latest callback is always used.
|
||||
"""
|
||||
return approval_callback
|
||||
|
||||
def mark_approved(tool_name: str):
|
||||
"""Mark a tool as approved in the current context."""
|
||||
approved = _approved_context.get(set())
|
||||
@@ -90,7 +99,7 @@ def require_approval(risk_level: RiskLevel = "high"):
|
||||
except Exception as e:
|
||||
# Fallback to sync approval if async fails
|
||||
logging.warning(f"Async approval failed, using sync fallback: {e}")
|
||||
callback = approval_callback or console_approval_callback
|
||||
callback = get_approval_callback() or console_approval_callback
|
||||
decision = callback(tool_name, kwargs, risk_level)
|
||||
|
||||
if not decision.approved:
|
||||
@@ -198,7 +207,7 @@ async def request_approval(function_name: str, arguments: Dict[str, Any]) -> App
|
||||
risk_level = TOOL_RISK_LEVELS.get(function_name, "medium")
|
||||
|
||||
# Use custom callback if set, otherwise use console callback
|
||||
callback = approval_callback or console_approval_callback
|
||||
callback = get_approval_callback() or console_approval_callback
|
||||
|
||||
try:
|
||||
# Handle async callbacks
|
||||
|
||||
@@ -19,6 +19,15 @@ from rich.console import Console
|
||||
from rich.live import Live
|
||||
import inspect
|
||||
|
||||
# Import display_tool_call for callback support (lazy import to avoid circular imports)
|
||||
_display_tool_call = None
|
||||
def _get_display_tool_call():
|
||||
global _display_tool_call
|
||||
if _display_tool_call is None:
|
||||
from ..main import display_tool_call
|
||||
_display_tool_call = display_tool_call
|
||||
return _display_tool_call
|
||||
|
||||
# Constants
|
||||
LOCAL_SERVER_API_KEY_PLACEHOLDER = "not-needed"
|
||||
|
||||
@@ -978,16 +987,19 @@ class OpenAIClient:
|
||||
function_name = tool_call.function.name
|
||||
arguments = json.loads(tool_call.function.arguments)
|
||||
|
||||
# Always trigger callback for tool call tracking (even when verbose=False)
|
||||
display_tool_call_fn = _get_display_tool_call()
|
||||
display_tool_call_fn(f"Calling function: {function_name}", console=console if verbose else None)
|
||||
|
||||
if verbose and console:
|
||||
console.print(f"[bold]Calling function:[/bold] {function_name}")
|
||||
console.print(f"[dim]Arguments:[/dim] {arguments}")
|
||||
|
||||
# Execute the tool
|
||||
tool_result = execute_tool_fn(function_name, arguments)
|
||||
results_str = json.dumps(tool_result) if tool_result else "Function returned an empty output"
|
||||
|
||||
if verbose and console:
|
||||
console.print(f"[dim]Result:[/dim] {results_str}")
|
||||
# Trigger callback with result
|
||||
display_tool_call_fn(f"Function {function_name} returned: {results_str[:200]}{'...' if len(results_str) > 200 else ''}", console=console if verbose else None)
|
||||
|
||||
messages.append({
|
||||
"role": "tool",
|
||||
@@ -1150,8 +1162,11 @@ class OpenAIClient:
|
||||
function_name = tool_call.function.name
|
||||
arguments = json.loads(tool_call.function.arguments)
|
||||
|
||||
# Always trigger callback for tool call tracking (even when verbose=False)
|
||||
display_tool_call_fn = _get_display_tool_call()
|
||||
display_tool_call_fn(f"Calling function: {function_name}", console=console if verbose else None)
|
||||
|
||||
if verbose and console:
|
||||
console.print(f"[bold]Calling function:[/bold] {function_name}")
|
||||
console.print(f"[dim]Arguments:[/dim] {arguments}")
|
||||
|
||||
# Execute the tool (async)
|
||||
@@ -1167,8 +1182,8 @@ class OpenAIClient:
|
||||
|
||||
results_str = json.dumps(tool_result) if tool_result else "Function returned an empty output"
|
||||
|
||||
if verbose and console:
|
||||
console.print(f"[dim]Result:[/dim] {results_str}")
|
||||
# Trigger callback with result
|
||||
display_tool_call_fn(f"Function {function_name} returned: {results_str[:200]}{'...' if len(results_str) > 200 else ''}", console=console if verbose else None)
|
||||
|
||||
messages.append({
|
||||
"role": "tool",
|
||||
@@ -1299,6 +1314,10 @@ class OpenAIClient:
|
||||
yield f"\n[Error parsing arguments for {function_name if 'function_name' in locals() else 'unknown function'}: {str(e)}]"
|
||||
continue
|
||||
|
||||
# Always trigger callback for tool call tracking (even when verbose=False)
|
||||
display_tool_call_fn = _get_display_tool_call()
|
||||
display_tool_call_fn(f"Calling function: {function_name}", console=None)
|
||||
|
||||
if verbose:
|
||||
yield f"\n[Calling function: {function_name}]"
|
||||
|
||||
@@ -1311,6 +1330,9 @@ class OpenAIClient:
|
||||
if verbose:
|
||||
yield f"\n[Function error: {str(e)}]"
|
||||
|
||||
# Trigger callback with result
|
||||
display_tool_call_fn(f"Function {function_name} returned: {results_str[:200]}{'...' if len(results_str) > 200 else ''}", console=None)
|
||||
|
||||
if verbose:
|
||||
yield f"\n[Function result: {results_str}]"
|
||||
|
||||
|
||||
@@ -231,15 +231,15 @@ def display_tool_call(message: str, console=None):
|
||||
if not message or not message.strip():
|
||||
logging.debug("Empty message in display_tool_call, returning early")
|
||||
return
|
||||
if console is None:
|
||||
console = Console()
|
||||
message = _clean_display_content(str(message))
|
||||
logging.debug(f"Cleaned message in display_tool_call: {repr(message)}")
|
||||
|
||||
# Execute synchronous callbacks
|
||||
# Execute synchronous callbacks (always, even when console is None)
|
||||
execute_sync_callback('tool_call', message=message)
|
||||
|
||||
console.print(Panel.fit(Text(message, style="bold cyan"), title="Tool Call", border_style="green"))
|
||||
# Only print panel if console is provided (verbose mode)
|
||||
if console is not None:
|
||||
console.print(Panel.fit(Text(message, style="bold cyan"), title="Tool Call", border_style="green"))
|
||||
|
||||
def display_error(message: str, console=None):
|
||||
if not message or not message.strip():
|
||||
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "praisonaiagents"
|
||||
version = "0.1.17"
|
||||
version = "0.1.19"
|
||||
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
|
||||
Generated
+1
-1
@@ -2422,7 +2422,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "praisonaiagents"
|
||||
version = "0.1.17"
|
||||
version = "0.1.19"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
|
||||
@@ -3,8 +3,8 @@ class Praisonai < Formula
|
||||
|
||||
desc "AI tools for various AI applications"
|
||||
homepage "https://github.com/MervinPraison/PraisonAI"
|
||||
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.55.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.55.tar.gz | shasum -a 256`.split.first
|
||||
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.58.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.58.tar.gz | shasum -a 256`.split.first
|
||||
license "MIT"
|
||||
|
||||
depends_on "[email protected]"
|
||||
|
||||
@@ -693,6 +693,11 @@ class PraisonAI:
|
||||
# Autonomy Mode - control AI action approval
|
||||
parser.add_argument("--autonomy", type=str, choices=["suggest", "auto_edit", "full_auto"], help="Set autonomy mode for AI actions")
|
||||
|
||||
# Tool Approval - control tool execution approval
|
||||
parser.add_argument("--trust", action="store_true", help="Auto-approve all tool executions (skip approval prompts)")
|
||||
parser.add_argument("--approve-level", type=str, choices=["low", "medium", "high", "critical"],
|
||||
help="Auto-approve tools up to this risk level (e.g., --approve-level high approves low/medium/high but prompts for critical)")
|
||||
|
||||
# Sandbox Execution - secure command execution
|
||||
parser.add_argument("--sandbox", type=str, choices=["off", "basic", "strict"], help="Enable sandboxed command execution")
|
||||
|
||||
@@ -2910,6 +2915,11 @@ Provide ONLY the commit message, no explanations."""
|
||||
else:
|
||||
agent_config["verbose"] = False
|
||||
|
||||
# Load default tools (same as interactive mode)
|
||||
default_tools = self._load_interactive_tools()
|
||||
if default_tools:
|
||||
agent_config["tools"] = default_tools
|
||||
|
||||
# Add llm if specified
|
||||
if hasattr(self, 'args') and self.args.llm:
|
||||
# Check if max_tokens is specified - pass as dict config
|
||||
@@ -2990,6 +3000,34 @@ Provide ONLY the commit message, no explanations."""
|
||||
|
||||
# ===== NEW CLI FEATURES INTEGRATION =====
|
||||
|
||||
# Tool Approval - Auto-approve tools based on --trust or --approve-level
|
||||
if getattr(self.args, 'trust', False) or getattr(self.args, 'approve_level', None):
|
||||
from praisonaiagents.approval import set_approval_callback, ApprovalDecision
|
||||
|
||||
if getattr(self.args, 'trust', False):
|
||||
# Auto-approve all tools
|
||||
def auto_approve_all(function_name, arguments, risk_level):
|
||||
return ApprovalDecision(approved=True, reason="Auto-approved via --trust flag")
|
||||
set_approval_callback(auto_approve_all)
|
||||
print("[bold yellow]⚠️ Trust mode enabled - all tool executions will be auto-approved[/bold yellow]")
|
||||
elif getattr(self.args, 'approve_level', None):
|
||||
# Auto-approve up to specified risk level
|
||||
max_level = self.args.approve_level
|
||||
risk_order = {"low": 1, "medium": 2, "high": 3, "critical": 4}
|
||||
max_level_value = risk_order.get(max_level, 3)
|
||||
|
||||
def level_based_approve(function_name, arguments, risk_level):
|
||||
tool_level_value = risk_order.get(risk_level, 4)
|
||||
if tool_level_value <= max_level_value:
|
||||
return ApprovalDecision(approved=True, reason=f"Auto-approved (level {risk_level} <= {max_level})")
|
||||
else:
|
||||
# Fall back to console approval for higher risk levels
|
||||
from praisonaiagents.approval import console_approval_callback
|
||||
return console_approval_callback(function_name, arguments, risk_level)
|
||||
|
||||
set_approval_callback(level_based_approve)
|
||||
print(f"[bold cyan]Auto-approve enabled for tools up to '{max_level}' risk level[/bold cyan]")
|
||||
|
||||
# Router - Smart model selection (must be before agent creation)
|
||||
if getattr(self.args, 'router', False):
|
||||
from .features.router import RouterHandler
|
||||
@@ -3228,7 +3266,7 @@ Now, {final_instruction.lower()}:"""
|
||||
|
||||
Shows:
|
||||
- "Generating..." with spinner while processing
|
||||
- Real-time tool call notifications
|
||||
- Real-time tool call notifications via registered callback
|
||||
- Agent handoff notifications
|
||||
"""
|
||||
import threading
|
||||
@@ -3237,7 +3275,20 @@ Now, {final_instruction.lower()}:"""
|
||||
from rich.live import Live
|
||||
from rich.text import Text
|
||||
|
||||
# Import callback registration from praisonaiagents
|
||||
from praisonaiagents import register_display_callback, sync_display_callbacks
|
||||
|
||||
console = Console()
|
||||
|
||||
# Get tool names for display
|
||||
tool_names = []
|
||||
if hasattr(agent, 'tools') and agent.tools:
|
||||
for tool in agent.tools:
|
||||
if hasattr(tool, '__name__'):
|
||||
tool_names.append(tool.__name__)
|
||||
elif hasattr(tool, 'name'):
|
||||
tool_names.append(tool.name)
|
||||
|
||||
status_info = {
|
||||
'status': 'Generating...',
|
||||
'tool_calls': [],
|
||||
@@ -3245,9 +3296,29 @@ Now, {final_instruction.lower()}:"""
|
||||
'done': False,
|
||||
'result': None,
|
||||
'error': None,
|
||||
'start_time': time.time()
|
||||
'start_time': time.time(),
|
||||
'available_tools': tool_names
|
||||
}
|
||||
|
||||
def tool_call_callback(message):
|
||||
"""Callback triggered when a tool is called."""
|
||||
# Extract tool name from message
|
||||
if "Calling function:" in message:
|
||||
# Format: "Calling function: function_name"
|
||||
parts = message.split("Calling function:")
|
||||
if len(parts) > 1:
|
||||
tool_name = parts[1].strip()
|
||||
if tool_name and tool_name not in status_info['tool_calls']:
|
||||
status_info['tool_calls'].append(tool_name)
|
||||
status_info['status'] = f"Using {tool_name}..."
|
||||
elif "Function " in message and " returned:" in message:
|
||||
# Format: "Function function_name returned: ..."
|
||||
# Tool execution completed, update status
|
||||
status_info['status'] = "Processing result..."
|
||||
|
||||
# Register callback for tool calls
|
||||
register_display_callback('tool_call', tool_call_callback)
|
||||
|
||||
def build_status_display():
|
||||
"""Build the status display text."""
|
||||
elapsed = time.time() - status_info['start_time']
|
||||
@@ -3258,6 +3329,13 @@ Now, {final_instruction.lower()}:"""
|
||||
text.append(f"{status_info['status']} ", style="bold")
|
||||
text.append(f"({elapsed:.1f}s)", style="dim")
|
||||
|
||||
# Show available tools on first line
|
||||
if status_info['available_tools'] and not status_info['tool_calls']:
|
||||
tools_str = ', '.join(status_info['available_tools'][:3])
|
||||
if len(status_info['available_tools']) > 3:
|
||||
tools_str += f" +{len(status_info['available_tools']) - 3} more"
|
||||
text.append(f"\n 🔧 Tools: {tools_str}", style="dim")
|
||||
|
||||
# Show recent tool calls
|
||||
if status_info['tool_calls']:
|
||||
text.append("\n")
|
||||
@@ -3276,14 +3354,6 @@ Now, {final_instruction.lower()}:"""
|
||||
def run_agent():
|
||||
"""Run the agent in background thread."""
|
||||
try:
|
||||
# Monkey-patch agent's execute_tool to capture tool calls
|
||||
original_execute_tool = agent.execute_tool
|
||||
def patched_execute_tool(tool_name, *args, **kwargs):
|
||||
status_info['tool_calls'].append(tool_name)
|
||||
status_info['status'] = f"Using {tool_name}..."
|
||||
return original_execute_tool(tool_name, *args, **kwargs)
|
||||
agent.execute_tool = patched_execute_tool
|
||||
|
||||
status_info['result'] = agent.start(prompt)
|
||||
except Exception as e:
|
||||
status_info['error'] = e
|
||||
@@ -3302,16 +3372,27 @@ Now, {final_instruction.lower()}:"""
|
||||
time.sleep(0.1)
|
||||
except KeyboardInterrupt:
|
||||
console.print("\n[dim]Interrupted[/dim]")
|
||||
# Unregister callback
|
||||
if 'tool_call' in sync_display_callbacks:
|
||||
del sync_display_callbacks['tool_call']
|
||||
return None
|
||||
|
||||
# Wait for thread to complete
|
||||
thread.join(timeout=1.0)
|
||||
|
||||
# Unregister callback to avoid memory leaks
|
||||
if 'tool_call' in sync_display_callbacks:
|
||||
del sync_display_callbacks['tool_call']
|
||||
|
||||
# Handle result
|
||||
if status_info['error']:
|
||||
console.print(f"[red]Error: {status_info['error']}[/red]")
|
||||
return None
|
||||
|
||||
# Show tool calls that were made (if any)
|
||||
if status_info['tool_calls']:
|
||||
console.print(f"[dim]Tools used: {', '.join(status_info['tool_calls'])}[/dim]")
|
||||
|
||||
return status_info['result']
|
||||
|
||||
def _handle_serve_command(self, args, unknown_args):
|
||||
|
||||
@@ -57,7 +57,7 @@ class CloudDeployer:
|
||||
file.write("FROM python:3.11-slim\n")
|
||||
file.write("WORKDIR /app\n")
|
||||
file.write("COPY . .\n")
|
||||
file.write("RUN pip install flask praisonai==2.3.55 gunicorn markdown\n")
|
||||
file.write("RUN pip install flask praisonai==2.3.58 gunicorn markdown\n")
|
||||
file.write("EXPOSE 8080\n")
|
||||
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
|
||||
|
||||
|
||||
@@ -1 +1 @@
|
||||
__version__ = "2.3.55"
|
||||
__version__ = "2.3.58"
|
||||
@@ -12,7 +12,7 @@ dependencies = [
|
||||
"rich>=13.7",
|
||||
"markdown>=3.5",
|
||||
"pyparsing>=3.0.0",
|
||||
"praisonaiagents>=0.1.17",
|
||||
"praisonaiagents>=0.1.19",
|
||||
"python-dotenv>=0.19.0",
|
||||
"instructor>=1.3.3",
|
||||
"PyYAML>=6.0",
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
"""
|
||||
Tests for the CLI approval system.
|
||||
|
||||
Tests:
|
||||
1. --trust flag auto-approves all tools
|
||||
2. --approve-level flag with different risk levels
|
||||
3. Approval callback integration
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add the src directory to the path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
|
||||
class TestApprovalFlags:
|
||||
"""Test CLI approval flags."""
|
||||
|
||||
def test_trust_flag_exists(self):
|
||||
"""Test that --trust flag is defined in argparse."""
|
||||
import argparse
|
||||
|
||||
# Check that the args object has the trust attribute after parsing
|
||||
# Parse with --help to avoid actually running anything
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--trust", action="store_true")
|
||||
parser.add_argument("--approve-level", type=str, choices=["low", "medium", "high", "critical"])
|
||||
|
||||
# Test that the arguments can be parsed
|
||||
args = parser.parse_args(['--trust'])
|
||||
assert args.trust is True
|
||||
|
||||
args = parser.parse_args([])
|
||||
assert args.trust is False
|
||||
|
||||
def test_approve_level_flag_exists(self):
|
||||
"""Test that --approve-level flag is defined in argparse."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--approve-level", type=str, choices=["low", "medium", "high", "critical"])
|
||||
|
||||
# Test that the arguments can be parsed
|
||||
args = parser.parse_args(['--approve-level', 'high'])
|
||||
assert args.approve_level == 'high'
|
||||
|
||||
args = parser.parse_args(['--approve-level', 'critical'])
|
||||
assert args.approve_level == 'critical'
|
||||
|
||||
def test_trust_flag_in_praisonai_parser(self):
|
||||
"""Test that --trust flag is in PraisonAI's actual parser."""
|
||||
from praisonai.cli.main import PraisonAI
|
||||
|
||||
# Create instance - parse_args is called in __init__
|
||||
praison = PraisonAI()
|
||||
|
||||
# Call parse_args to get the args
|
||||
args = praison.parse_args()
|
||||
|
||||
# The args should have trust attribute after parse_args
|
||||
assert hasattr(args, 'trust'), "PraisonAI.args should have 'trust' attribute"
|
||||
# Cleanup
|
||||
del praison
|
||||
|
||||
def test_approve_level_flag_in_praisonai_parser(self):
|
||||
"""Test that --approve-level flag is in PraisonAI's actual parser."""
|
||||
from praisonai.cli.main import PraisonAI
|
||||
|
||||
# Create instance - parse_args is called in __init__
|
||||
praison = PraisonAI()
|
||||
|
||||
# Call parse_args to get the args
|
||||
args = praison.parse_args()
|
||||
|
||||
# The args should have approve_level attribute after parse_args
|
||||
assert hasattr(args, 'approve_level'), "PraisonAI.args should have 'approve_level' attribute"
|
||||
|
||||
|
||||
class TestApprovalCallback:
|
||||
"""Test approval callback system."""
|
||||
|
||||
def test_set_approval_callback(self):
|
||||
"""Test that we can set a custom approval callback."""
|
||||
from praisonaiagents.approval import set_approval_callback, ApprovalDecision
|
||||
|
||||
callback_called = []
|
||||
|
||||
def custom_callback(function_name, arguments, risk_level):
|
||||
callback_called.append((function_name, arguments, risk_level))
|
||||
return ApprovalDecision(approved=True, reason="Test approved")
|
||||
|
||||
set_approval_callback(custom_callback)
|
||||
|
||||
# Verify callback was set
|
||||
from praisonaiagents.approval import approval_callback
|
||||
assert approval_callback == custom_callback
|
||||
|
||||
# Reset callback
|
||||
set_approval_callback(None)
|
||||
|
||||
def test_auto_approve_callback(self):
|
||||
"""Test auto-approve callback returns approved decision."""
|
||||
from praisonaiagents.approval import ApprovalDecision
|
||||
|
||||
def auto_approve_callback(function_name, arguments, risk_level):
|
||||
return ApprovalDecision(approved=True, reason="Auto-approved")
|
||||
|
||||
decision = auto_approve_callback("execute_command", {"command": "ls"}, "critical")
|
||||
assert decision.approved is True
|
||||
assert decision.reason == "Auto-approved"
|
||||
|
||||
def test_level_based_callback(self):
|
||||
"""Test level-based approval callback."""
|
||||
from praisonaiagents.approval import ApprovalDecision
|
||||
|
||||
def level_based_callback(function_name, arguments, risk_level, max_level="high"):
|
||||
"""Approve based on risk level threshold."""
|
||||
levels = {"low": 1, "medium": 2, "high": 3, "critical": 4}
|
||||
tool_level = levels.get(risk_level, 4)
|
||||
max_allowed = levels.get(max_level, 3)
|
||||
|
||||
if tool_level <= max_allowed:
|
||||
return ApprovalDecision(approved=True, reason=f"Auto-approved (level {risk_level} <= {max_level})")
|
||||
else:
|
||||
return ApprovalDecision(approved=False, reason=f"Denied (level {risk_level} > {max_level})")
|
||||
|
||||
# Test with max_level="high"
|
||||
# Should approve low, medium, high but deny critical
|
||||
assert level_based_callback("test", {}, "low", "high").approved is True
|
||||
assert level_based_callback("test", {}, "medium", "high").approved is True
|
||||
assert level_based_callback("test", {}, "high", "high").approved is True
|
||||
assert level_based_callback("test", {}, "critical", "high").approved is False
|
||||
|
||||
# Test with max_level="critical"
|
||||
# Should approve everything
|
||||
assert level_based_callback("test", {}, "critical", "critical").approved is True
|
||||
|
||||
|
||||
class TestApprovalRequirements:
|
||||
"""Test approval requirement configuration."""
|
||||
|
||||
def test_default_dangerous_tools(self):
|
||||
"""Test that default dangerous tools are configured."""
|
||||
from praisonaiagents.approval import APPROVAL_REQUIRED_TOOLS, TOOL_RISK_LEVELS
|
||||
|
||||
# Check critical tools
|
||||
assert "execute_command" in APPROVAL_REQUIRED_TOOLS
|
||||
assert TOOL_RISK_LEVELS.get("execute_command") == "critical"
|
||||
|
||||
# Check high risk tools
|
||||
assert "write_file" in APPROVAL_REQUIRED_TOOLS
|
||||
assert TOOL_RISK_LEVELS.get("write_file") == "high"
|
||||
|
||||
def test_remove_approval_requirement(self):
|
||||
"""Test removing approval requirement."""
|
||||
from praisonaiagents.approval import (
|
||||
add_approval_requirement,
|
||||
remove_approval_requirement,
|
||||
is_approval_required,
|
||||
APPROVAL_REQUIRED_TOOLS
|
||||
)
|
||||
|
||||
# Add a test tool
|
||||
add_approval_requirement("test_tool", "medium")
|
||||
assert is_approval_required("test_tool") is True
|
||||
|
||||
# Remove it
|
||||
remove_approval_requirement("test_tool")
|
||||
assert is_approval_required("test_tool") is False
|
||||
|
||||
def test_is_approval_required(self):
|
||||
"""Test checking if approval is required."""
|
||||
from praisonaiagents.approval import is_approval_required
|
||||
|
||||
# execute_command should require approval
|
||||
assert is_approval_required("execute_command") is True
|
||||
|
||||
# read_file should not require approval
|
||||
assert is_approval_required("read_file") is False
|
||||
|
||||
# list_files should not require approval
|
||||
assert is_approval_required("list_files") is False
|
||||
|
||||
|
||||
class TestApprovalDecision:
|
||||
"""Test ApprovalDecision class."""
|
||||
|
||||
def test_approval_decision_approved(self):
|
||||
"""Test approved decision."""
|
||||
from praisonaiagents.approval import ApprovalDecision
|
||||
|
||||
decision = ApprovalDecision(approved=True, reason="User approved")
|
||||
assert decision.approved is True
|
||||
assert decision.reason == "User approved"
|
||||
assert decision.modified_args == {}
|
||||
|
||||
def test_approval_decision_denied(self):
|
||||
"""Test denied decision."""
|
||||
from praisonaiagents.approval import ApprovalDecision
|
||||
|
||||
decision = ApprovalDecision(approved=False, reason="User denied")
|
||||
assert decision.approved is False
|
||||
assert decision.reason == "User denied"
|
||||
|
||||
def test_approval_decision_with_modified_args(self):
|
||||
"""Test decision with modified arguments."""
|
||||
from praisonaiagents.approval import ApprovalDecision
|
||||
|
||||
decision = ApprovalDecision(
|
||||
approved=True,
|
||||
reason="Approved with modifications",
|
||||
modified_args={"command": "ls -la"}
|
||||
)
|
||||
assert decision.approved is True
|
||||
assert decision.modified_args == {"command": "ls -la"}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Generated
+4
-4
@@ -4197,7 +4197,7 @@ requires-dist = [
|
||||
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonaiagents", specifier = ">=0.1.17" },
|
||||
{ name = "praisonaiagents", specifier = ">=0.1.19" },
|
||||
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
|
||||
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
|
||||
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
|
||||
@@ -4252,7 +4252,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "praisonaiagents"
|
||||
version = "0.1.17"
|
||||
version = "0.1.19"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
@@ -4261,9 +4261,9 @@ dependencies = [
|
||||
{ name = "pydantic" },
|
||||
{ name = "rich" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/3c/ed/fd26940dd2ee4506f2d6e178841b36d09f90f6a817ea052e0c1a9473f132/praisonaiagents-0.1.17.tar.gz", hash = "sha256:fa3bb9e77c6dbd16ed1fdb933d3669fac2208b1e58a2a5d5986fd630fef975c1", size = 647705, upload-time = "2025-12-18T21:52:30.244Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/1c/51/f80e2cded4b512bd79f23d240d8ed84dee61d040d15595520c7c7e74ff1f/praisonaiagents-0.1.19.tar.gz", hash = "sha256:075367e1de996b1ba288561254f601e41532bca402d567caec53c6bb602bd3c2", size = 658016, upload-time = "2025-12-20T03:05:04.051Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/94/40/261bbe4e0597442bd8d3aef9450c315063735970bb8f32f95cc4f0267b7b/praisonaiagents-0.1.17-py3-none-any.whl", hash = "sha256:61b98ede790e14fd7a21f4fc51729ba0a4b32ab1165f3ad9ee966757fa581c0f", size = 583244, upload-time = "2025-12-18T21:52:28.004Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/8b/0c/4c36dc5d8d3f945dadb6a7b44c65543405bedf238682f77d9f8f383c6b87/praisonaiagents-0.1.19-py3-none-any.whl", hash = "sha256:f5c2f04cc63a621e071fa316914cbb18a06ae62be05c541df4b1a38c9ef0db2b", size = 588948, upload-time = "2025-12-20T03:05:02.151Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
Reference in new issue
Block a user