diff --git a/docs/joy_trust_integration.md b/docs/joy_trust_integration.md new file mode 100644 index 0000000..b7c7950 --- /dev/null +++ b/docs/joy_trust_integration.md @@ -0,0 +1,302 @@ +# Joy Trust Integration for PraisonAI + +Native Joy Trust Network integration for secure agent handoffs in PraisonAI. + +## Overview + +The Joy Trust integration provides automatic trust verification for agent handoffs, ensuring that tasks are only delegated to verified, trustworthy agents. This integration builds on the existing Joy Trust Network documentation and provides seamless, zero-configuration security for multi-agent workflows. + +## Features + +- **Automatic Trust Verification**: Environment variable configuration for seamless integration +- **Handoff Safety Checks**: Verify agent trustworthiness before delegation +- **Decorator-based Security**: `@trust_verified_handoff` decorator for secure functions +- **Configurable Thresholds**: Set minimum trust scores for different use cases +- **Caching**: Built-in result caching to reduce API calls +- **Fallback Handling**: Graceful degradation when trust verification fails + +## Quick Start + +### 1. Environment Variable Configuration + +```bash +# Enable trust verification for all agents +export PRAISONAI_TRUST_PROVIDER=joy + +# Set minimum trust score threshold (default: 3.0) +export PRAISONAI_TRUST_MIN_SCORE=3.5 + +# Enable automatic handoff verification (default: true) +export PRAISONAI_TRUST_AUTO_VERIFY=true + +# Optional: Set Joy Trust API key for enhanced features +export JOY_TRUST_API_KEY=your_api_key_here +``` + +### 2. Basic Usage + +```python +from praisonai_tools import check_trust_score, verify_handoff_safety + +# Check an agent's trust score +result = check_trust_score("researcher_agent", min_score=3.0) +print(f"Trust Score: {result['trust_score']}") +print(f"Safe to use: {result['meets_threshold']}") + +# Verify handoff safety with recommendations +verification = verify_handoff_safety("data_analyst", min_score=3.5) +print(f"Handoff Safe: {verification['handoff_safe']}") +print(f"Recommendation: {verification['recommendation']}") +``` + +### 3. Decorator-based Security + +```python +from praisonai_tools import trust_verified_handoff + +@trust_verified_handoff(min_score=4.0) +def delegate_analysis_task(agent_name: str, data: dict): + # This function only executes if the agent meets trust requirements + return perform_delegation(agent_name, data) + +# The decorator automatically checks trust before execution +result = delegate_analysis_task("expert_analyst", {"data": "..."}) +``` + +## API Reference + +### Functions + +#### `check_trust_score(agent_name: str, min_score: float = 3.0) -> dict` + +Check an agent's trust score on the Joy Trust Network. + +**Parameters:** +- `agent_name`: Name/identifier of the agent to check +- `min_score`: Minimum acceptable trust score + +**Returns:** +```python +{ + "agent_name": str, + "trust_score": float, + "verified": bool, + "meets_threshold": bool, + "threshold_used": float, + "reputation": dict, + "recommendations": int, + "error": str | None +} +``` + +#### `verify_handoff_safety(agent_name: str, min_score: float = 3.0) -> dict` + +Verify if it's safe to hand off to the specified agent. + +**Parameters:** +- `agent_name`: Target agent for handoff +- `min_score`: Minimum acceptable trust score + +**Returns:** +```python +{ + "agent_name": str, + "trust_score": float, + "verified": bool, + "handoff_safe": bool, + "recommendation": str, + "threshold_used": float, + "verification_time": float, + "error": str | None +} +``` + +#### `trust_verified_handoff(min_score: float = 3.0)` + +Decorator for automatic trust verification before function execution. + +**Parameters:** +- `min_score`: Minimum trust score required + +**Usage:** +```python +@trust_verified_handoff(min_score=3.5) +def my_delegation_function(agent_name, task): + return delegate_task(agent_name, task) +``` + +#### `is_trust_verification_enabled() -> bool` + +Check if trust verification is enabled via environment variables. + +#### `get_trust_config() -> TrustConfig` + +Get current trust configuration from environment variables. + +### Classes + +#### `JoyTrustTool` + +Enhanced tool class for Joy Trust Network verification. + +```python +from praisonai_tools import JoyTrustTool + +tool = JoyTrustTool(api_key="optional_key") +result = tool.check_trust("agent_name") +``` + +#### `TrustConfig` + +Configuration dataclass for Joy Trust integration. + +```python +from praisonai_tools.tools.joy_trust_tool import TrustConfig + +config = TrustConfig.from_env() +print(f"Enabled: {config.enabled}") +print(f"Min Score: {config.min_score}") +``` + +## Environment Variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `PRAISONAI_TRUST_PROVIDER` | Set to 'joy' to enable trust verification | None | +| `PRAISONAI_TRUST_MIN_SCORE` | Minimum trust score threshold | 3.0 | +| `PRAISONAI_TRUST_AUTO_VERIFY` | Enable automatic handoff verification | true | +| `PRAISONAI_TRUST_TIMEOUT` | Request timeout in seconds | 10.0 | +| `PRAISONAI_TRUST_CACHE_DURATION` | Cache duration in seconds | 300 | +| `PRAISONAI_TRUST_FALLBACK` | Allow fallback on errors | true | +| `JOY_TRUST_API_KEY` | Joy Trust API key (optional) | None | + +## Trust Score Ranges + +The Joy Trust Network uses a 0-5 scale for trust scores: + +- **4.5-5.0**: Excellent - Highly trusted, verified agents +- **3.5-4.4**: Good - Reliable agents with good reputation +- **2.5-3.4**: Moderate - Acceptable with caution +- **1.0-2.4**: Low - Not recommended for important tasks +- **0.0-0.9**: Very Low - Strongly discouraged + +## Integration Patterns + +### 1. Workflow-level Integration + +```python +import os +from praisonai_tools import verify_handoff_safety + +# Enable trust verification +os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + +def secure_workflow(agents: list, min_score: float = 3.0): + trusted_agents = [] + + for agent in agents: + verification = verify_handoff_safety(agent['name'], min_score) + if verification['handoff_safe']: + trusted_agents.append(agent) + else: + print(f"⚠️ Agent {agent['name']} failed trust check") + + return trusted_agents +``` + +### 2. Conditional Trust Checking + +```python +from praisonai_tools import is_trust_verification_enabled, check_trust_score + +def delegate_with_optional_trust(agent_name: str, task: str): + if is_trust_verification_enabled(): + trust_result = check_trust_score(agent_name) + if not trust_result['meets_threshold']: + return {"error": f"Agent {agent_name} doesn't meet trust requirements"} + + return perform_delegation(agent_name, task) +``` + +### 3. Custom Trust Policies + +```python +from praisonai_tools import JoyTrustTool +from praisonai_tools.tools.joy_trust_tool import TrustConfig + +# Create custom configuration +custom_config = TrustConfig( + enabled=True, + min_score=4.0, # High security requirement + timeout_seconds=5.0, # Fast timeout + fallback_on_error=False # Strict mode +) + +tool = JoyTrustTool(config=custom_config) +result = tool.verify_handoff_safety("critical_agent") +``` + +## Error Handling + +The integration provides robust error handling: + +```python +from praisonai_tools import check_trust_score + +result = check_trust_score("unknown_agent") + +if result['error']: + print(f"Trust check failed: {result['error']}") + + if result.get('fallback_used'): + print("Using fallback behavior") + else: + print("Strict mode - blocking delegation") +else: + print(f"Trust verification successful: {result['trust_score']}") +``` + +## Best Practices + +1. **Start with moderate thresholds** (3.0-3.5) and adjust based on your security needs +2. **Use environment variables** for configuration to avoid hardcoding values +3. **Enable caching** in production to reduce API calls and improve performance +4. **Set appropriate timeouts** based on your application's requirements +5. **Use fallback mode** for non-critical operations to maintain availability +6. **Monitor trust scores** and update agent permissions accordingly + +## Troubleshooting + +### Common Issues + +1. **"httpx is required" error**: Install httpx with `pip install httpx` or `pip install praisonai-tools[marketplace]` +2. **Connection timeouts**: Increase `PRAISONAI_TRUST_TIMEOUT` value +3. **False negatives**: Check if agent names match Joy Trust Network entries exactly +4. **API rate limits**: Enable caching with `PRAISONAI_TRUST_CACHE_DURATION` + +### Debug Mode + +Enable debug logging to troubleshoot issues: + +```python +import logging +logging.basicConfig(level=logging.DEBUG) + +from praisonai_tools import check_trust_score +result = check_trust_score("debug_agent") +``` + +## Contributing + +To contribute to the Joy Trust integration: + +1. Fork the PraisonAI-Tools repository +2. Create a feature branch +3. Add tests for new functionality +4. Submit a pull request + +## Support + +- **Documentation**: [Joy Trust Network Docs](https://docs.praison.ai/joy-trust-network) +- **Issues**: [PraisonAI-Tools Issues](https://github.com/MervinPraison/PraisonAI-Tools/issues) +- **Community**: [PraisonAI Discord](https://discord.gg/praisonai) \ No newline at end of file diff --git a/examples/joy_trust_integration.py b/examples/joy_trust_integration.py new file mode 100644 index 0000000..7dc788c --- /dev/null +++ b/examples/joy_trust_integration.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +""" +Joy Trust Integration Examples for PraisonAI + +This example demonstrates the native Joy Trust integration for secure agent handoffs. +""" + +import os +from praisonai_tools import ( + check_trust_score, + verify_handoff_safety, + trust_verified_handoff, + is_trust_verification_enabled, + get_trust_config, + JoyTrustTool +) + +def example_basic_trust_check(): + """Example 1: Basic trust score verification""" + print("=== Basic Trust Score Check ===") + + # Check trust for a specific agent + result = check_trust_score("researcher_agent", min_score=3.0) + + print(f"Agent: {result['agent_name']}") + print(f"Trust Score: {result['trust_score']}") + print(f"Verified: {result['verified']}") + print(f"Meets Threshold: {result['meets_threshold']}") + print(f"Recommendation: {result.get('recommendation', 'N/A')}") + + if result['error']: + print(f"Error: {result['error']}") + + print() + + +def example_handoff_verification(): + """Example 2: Handoff safety verification""" + print("=== Handoff Safety Verification ===") + + target_agent = "data_analyst" + verification = verify_handoff_safety(target_agent, min_score=3.5) + + print(f"Target Agent: {verification['agent_name']}") + print(f"Trust Score: {verification['trust_score']}") + print(f"Handoff Safe: {verification['handoff_safe']}") + print(f"Recommendation: {verification['recommendation']}") + print(f"Threshold Used: {verification['threshold_used']}") + + if verification['handoff_safe']: + print("✅ Safe to proceed with handoff") + else: + print("❌ Handoff not recommended") + + print() + + +@trust_verified_handoff(min_score=4.0) +def example_secure_delegation(agent_name: str, task: str): + """Example 3: Decorator-based secure delegation""" + print(f"Delegating task '{task}' to {agent_name}") + + # Simulate task delegation + # In a real scenario, this would delegate to the actual agent + return { + "success": True, + "result": f"Task completed by {agent_name}", + "agent": agent_name, + "task": task + } + + +def example_environment_configuration(): + """Example 4: Environment-based configuration""" + print("=== Environment Configuration ===") + + # Set environment variables for automatic trust verification + os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + os.environ['PRAISONAI_TRUST_MIN_SCORE'] = '3.5' + os.environ['PRAISONAI_TRUST_AUTO_VERIFY'] = 'true' + + # Check if trust verification is enabled + enabled = is_trust_verification_enabled() + print(f"Trust Verification Enabled: {enabled}") + + # Get current configuration + config = get_trust_config() + print(f"Provider: {config.provider}") + print(f"Min Score: {config.min_score}") + print(f"Auto Verify: {config.auto_verify_handoffs}") + print(f"Timeout: {config.timeout_seconds}s") + + print() + + +def example_advanced_configuration(): + """Example 5: Advanced tool configuration""" + print("=== Advanced Configuration ===") + + # Create a custom configured tool + trust_tool = JoyTrustTool( + api_key=os.getenv("JOY_TRUST_API_KEY"), + config=None # Will use environment config + ) + + # Configure the tool + config_result = trust_tool.configure( + min_score=4.5, + timeout_seconds=15.0, + auto_verify_handoffs=True + ) + + print(f"Configuration Status: {config_result['status']}") + print(f"Current Config: {config_result['config']}") + + # Use the configured tool + result = trust_tool.check_trust("expert_agent") + print(f"\nExpert Agent Trust: {result['trust_score']}") + print(f"Meets Threshold: {result['meets_threshold']}") + + print() + + +def example_workflow_integration(): + """Example 6: Workflow integration pattern""" + print("=== Workflow Integration ===") + + # Enable trust verification + os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + + # Define a multi-agent workflow + workflow_agents = [ + {"name": "data_collector", "role": "collect"}, + {"name": "data_processor", "role": "process"}, + {"name": "report_generator", "role": "generate"} + ] + + trusted_agents = [] + + for agent in workflow_agents: + verification = verify_handoff_safety(agent["name"], min_score=3.0) + + print(f"Agent: {agent['name']}") + print(f" Role: {agent['role']}") + print(f" Trust Score: {verification['trust_score']}") + print(f" Safe: {verification['handoff_safe']}") + print(f" Recommendation: {verification['recommendation']}") + + if verification['handoff_safe']: + trusted_agents.append(agent) + + print() + + print(f"Trusted agents for workflow: {len(trusted_agents)}/{len(workflow_agents)}") + + # Proceed with trusted agents only + if len(trusted_agents) == len(workflow_agents): + print("✅ All agents trusted - workflow can proceed safely") + else: + print("⚠️ Some agents not trusted - review workflow security") + + +def example_decorator_usage(): + """Example 7: Using the decorator in practice""" + print("=== Decorator Usage ===") + + # Enable automatic trust verification + os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + os.environ['PRAISONAI_TRUST_AUTO_VERIFY'] = 'true' + + # Example with high trust requirement + try: + result = example_secure_delegation("trusted_expert", "analyze_data") + print(f"Delegation Result: {result}") + except Exception as e: + print(f"Delegation Failed: {e}") + + print() + + +if __name__ == "__main__": + print("Joy Trust Integration Examples\n") + + # Note: These examples will make actual API calls if JOY_TRUST_API_KEY is set + # and the agents exist in the Joy Trust Network + + try: + example_basic_trust_check() + example_handoff_verification() + example_environment_configuration() + example_advanced_configuration() + example_workflow_integration() + example_decorator_usage() + + except Exception as e: + print(f"Example execution error: {e}") + print("\nNote: Some examples may fail if Joy Trust Network is not accessible") + print("or if the example agent names don't exist in the network.") \ No newline at end of file diff --git a/praisonai_tools/tools/__init__.py b/praisonai_tools/tools/__init__.py index 273d8a9..3570bd4 100644 --- a/praisonai_tools/tools/__init__.py +++ b/praisonai_tools/tools/__init__.py @@ -474,6 +474,10 @@ def __getattr__(name): "verify_agent_identity": "agentid_tool", "JoyTrustTool": "joy_trust_tool", "check_trust_score": "joy_trust_tool", + "verify_handoff_safety": "joy_trust_tool", + "trust_verified_handoff": "joy_trust_tool", + "is_trust_verification_enabled": "joy_trust_tool", + "get_trust_config": "joy_trust_tool", "AgentFolioTool": "agentfolio_tool", "check_behavioral_trust": "agentfolio_tool", "verify_task_delegation_safety": "agentfolio_tool", @@ -800,6 +804,10 @@ def __getattr__(name): "verify_agent_identity", "JoyTrustTool", "check_trust_score", + "verify_handoff_safety", + "trust_verified_handoff", + "is_trust_verification_enabled", + "get_trust_config", "AgentFolioTool", "check_behavioral_trust", "verify_task_delegation_safety", diff --git a/praisonai_tools/tools/joy_trust_tool.py b/praisonai_tools/tools/joy_trust_tool.py index 60c87fe..4900bae 100644 --- a/praisonai_tools/tools/joy_trust_tool.py +++ b/praisonai_tools/tools/joy_trust_tool.py @@ -1,77 +1,142 @@ """Joy Trust Tool for PraisonAI Agents. -Agent trust score verification using Joy Trust Network. +Enhanced agent trust score verification using Joy Trust Network with native +integration for secure agent handoffs. + +Features: +- Basic trust score verification +- Automatic handoff trust verification +- Environment variable configuration +- Decorator for trust-aware agent delegation +- Integration hooks for PraisonAI handoff system Usage: - from praisonai_tools import JoyTrustTool + Basic usage: + from praisonai_tools import check_trust_score + result = check_trust_score("agent_name") - tool = JoyTrustTool() - result = tool.check_trust("agent_name") + Advanced usage with handoff integration: + from praisonai_tools import JoyTrustTool, trust_verified_handoff + + # Enable automatic trust verification + os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + + @trust_verified_handoff(min_score=3.0) + def delegate_to_agent(agent, task): + return agent.run(task) Environment Variables: JOY_TRUST_API_KEY: Joy Trust API key (optional) + PRAISONAI_TRUST_PROVIDER: Set to 'joy' to enable automatic trust verification + PRAISONAI_TRUST_MIN_SCORE: Minimum trust score threshold (default: 3.0) + PRAISONAI_TRUST_AUTO_VERIFY: Enable automatic handoff verification (default: true) """ import os import logging -from typing import Any, Dict, Optional, Union +from typing import Any, Dict, Optional, Union, Callable, List +from dataclasses import dataclass +from functools import wraps from praisonai_tools.tools.base import BaseTool logger = logging.getLogger(__name__) +@dataclass +class TrustConfig: + """Configuration for Joy Trust integration.""" + + enabled: bool = False + provider: str = "joy" + min_score: float = 3.0 + auto_verify_handoffs: bool = True + timeout_seconds: float = 10.0 + cache_duration: int = 300 # 5 minutes + fallback_on_error: bool = True + + @classmethod + def from_env(cls) -> 'TrustConfig': + """Create configuration from environment variables.""" + return cls( + enabled=os.getenv('PRAISONAI_TRUST_PROVIDER', '').lower() == 'joy', + provider=os.getenv('PRAISONAI_TRUST_PROVIDER', 'joy'), + min_score=float(os.getenv('PRAISONAI_TRUST_MIN_SCORE', '3.0')), + auto_verify_handoffs=os.getenv('PRAISONAI_TRUST_AUTO_VERIFY', 'true').lower() == 'true', + timeout_seconds=float(os.getenv('PRAISONAI_TRUST_TIMEOUT', '10.0')), + cache_duration=int(os.getenv('PRAISONAI_TRUST_CACHE_DURATION', '300')), + fallback_on_error=os.getenv('PRAISONAI_TRUST_FALLBACK', 'true').lower() == 'true' + ) + + class JoyTrustTool(BaseTool): - """Tool for Joy Trust Network verification.""" + """Enhanced tool for Joy Trust Network verification with native integration.""" name = "joy_trust" - description = "Check agent trust scores using Joy Trust Network." + description = "Check agent trust scores using Joy Trust Network with automatic handoff integration." - def __init__(self, api_key: Optional[str] = None): + def __init__(self, api_key: Optional[str] = None, config: Optional[TrustConfig] = None): self.api_key = api_key or os.getenv("JOY_TRUST_API_KEY") + self.config = config or TrustConfig.from_env() + self._cache = {} # Simple in-memory cache super().__init__() def run( self, action: str = "check_trust", agent_name: Optional[str] = None, + min_score: Optional[float] = None, **kwargs ) -> Union[str, Dict[str, Any]]: + """Run trust verification action.""" if action == "check_trust": - return self.check_trust(agent_name=agent_name) + return self.check_trust(agent_name=agent_name, min_score=min_score) + elif action == "verify_handoff": + return self.verify_handoff_safety(agent_name=agent_name, min_score=min_score) + elif action == "configure": + return self.configure(**kwargs) return {"error": f"Unknown action: {action}"} - def check_trust(self, agent_name: str) -> Dict[str, Any]: - """Check an agent's trust score on Joy Trust Network before delegation. + def check_trust(self, agent_name: str, min_score: Optional[float] = None) -> Dict[str, Any]: + """Check an agent's trust score on Joy Trust Network. Args: agent_name: Name/identifier of the agent to check + min_score: Minimum acceptable trust score (uses config default if not provided) Returns: - Dictionary containing: - - trust_score: Numeric trust score (0-1) - - verified: Boolean indicating if agent is verified - - reputation: Reputation metrics if available - - recommendations: Number of positive recommendations - - error: Error message if lookup failed + Dictionary containing trust information and verification status """ if not agent_name: return { "agent_name": "", "trust_score": 0.0, "verified": False, + "meets_threshold": False, "reputation": {}, "recommendations": 0, "error": "agent_name is required" } + min_threshold = min_score if min_score is not None else self.config.min_score + + # Check cache first + cache_key = f"{agent_name}_{min_threshold}" + if cache_key in self._cache: + cached_result = self._cache[cache_key] + if cached_result.get('_cached_at', 0) + self.config.cache_duration > time.time(): + logger.debug(f"Using cached trust score for {agent_name}") + return cached_result + try: import httpx + import time except ImportError: return { "agent_name": agent_name, "trust_score": 0.0, "verified": False, + "meets_threshold": False, "reputation": {}, "recommendations": 0, "error": ( @@ -81,7 +146,7 @@ def check_trust(self, agent_name: str) -> Dict[str, Any]: } try: - with httpx.Client(timeout=30.0) as client: + with httpx.Client(timeout=self.config.timeout_seconds) as client: params = {"name": agent_name} if self.api_key: params["api_key"] = self.api_key @@ -93,50 +158,219 @@ def check_trust(self, agent_name: str) -> Dict[str, Any]: response.raise_for_status() data = response.json() + trust_score = data.get("trust_score", 0.0) - return { + result = { "agent_name": agent_name, - "trust_score": data.get("trust_score", 0.0), + "trust_score": trust_score, "verified": data.get("verified", False), + "meets_threshold": trust_score >= min_threshold, + "threshold_used": min_threshold, "reputation": data.get("reputation", {}), "recommendations": data.get("recommendations", 0), "last_activity": data.get("last_activity"), "network_rank": data.get("network_rank"), - "error": None + "error": None, + "_cached_at": time.time() } + # Cache the result + self._cache[cache_key] = result + + return result + except httpx.RequestError as e: logger.error(f"Joy Trust request error: {e}") - return { + error_result = { "agent_name": agent_name, "trust_score": 0.0, "verified": False, + "meets_threshold": self.config.fallback_on_error, # Fallback behavior + "threshold_used": min_threshold, "reputation": {}, "recommendations": 0, - "error": f"Connection error: {e}" + "error": f"Connection error: {e}", + "fallback_used": self.config.fallback_on_error } + return error_result except httpx.HTTPStatusError as e: logger.error(f"Joy Trust API error: {e.response.status_code}") - return { + error_result = { "agent_name": agent_name, "trust_score": 0.0, "verified": False, + "meets_threshold": self.config.fallback_on_error, # Fallback behavior + "threshold_used": min_threshold, "reputation": {}, "recommendations": 0, - "error": f"API error ({e.response.status_code}): {e.response.text}" + "error": f"API error ({e.response.status_code}): {e.response.text}", + "fallback_used": self.config.fallback_on_error } + return error_result except Exception as e: logger.error(f"Joy Trust unexpected error: {e}") - return { + error_result = { "agent_name": agent_name, "trust_score": 0.0, "verified": False, + "meets_threshold": self.config.fallback_on_error, # Fallback behavior + "threshold_used": min_threshold, "reputation": {}, "recommendations": 0, - "error": f"Unexpected error: {e}" + "error": f"Unexpected error: {e}", + "fallback_used": self.config.fallback_on_error } + return error_result + + def verify_handoff_safety(self, agent_name: str, min_score: Optional[float] = None) -> Dict[str, Any]: + """Verify if it's safe to hand off to the specified agent. + + Args: + agent_name: Target agent for handoff + min_score: Minimum acceptable trust score + + Returns: + Dictionary with safety verification and recommendation + """ + trust_result = self.check_trust(agent_name, min_score) + + safety_result = { + **trust_result, + "handoff_safe": trust_result.get("meets_threshold", False), + "recommendation": self._get_handoff_recommendation(trust_result), + "verification_time": time.time() + } + + return safety_result + + def _get_handoff_recommendation(self, trust_result: Dict[str, Any]) -> str: + """Generate handoff recommendation based on trust score.""" + score = trust_result.get("trust_score", 0.0) + verified = trust_result.get("verified", False) + error = trust_result.get("error") + + if error and not trust_result.get("fallback_used", False): + return f"⚠️ Trust verification failed: {error}. Handoff not recommended." + + if error and trust_result.get("fallback_used", False): + return "⚠️ Trust verification failed but fallback enabled. Proceed with caution." + + if score >= 4.5 and verified: + return "✅ Excellent trust score. Handoff highly recommended." + elif score >= 3.5 and verified: + return "✅ Good trust score. Handoff recommended." + elif score >= 2.5: + return "⚠️ Moderate trust score. Handoff acceptable with caution." + elif score >= 1.0: + return "❌ Low trust score. Handoff not recommended." + else: + return "❌ Very low or no trust score. Handoff strongly discouraged." + + def configure(self, **kwargs) -> Dict[str, Any]: + """Update configuration settings.""" + for key, value in kwargs.items(): + if hasattr(self.config, key): + setattr(self.config, key, value) + + return { + "status": "configured", + "config": { + "enabled": self.config.enabled, + "min_score": self.config.min_score, + "auto_verify_handoffs": self.config.auto_verify_handoffs, + "timeout_seconds": self.config.timeout_seconds + } + } + + +def trust_verified_handoff(min_score: float = 3.0, trust_tool: Optional[JoyTrustTool] = None): + """Decorator for trust-verified agent handoffs. + + This decorator automatically verifies agent trust before delegation. + + Args: + min_score: Minimum trust score required + trust_tool: Custom JoyTrustTool instance (creates one if None) + + Usage: + @trust_verified_handoff(min_score=4.0) + def delegate_task(agent_name, task): + # Your delegation logic here + return result + """ + def decorator(func: Callable) -> Callable: + @wraps(func) + def wrapper(*args, **kwargs): + # Get agent name from args/kwargs + agent_name = None + if args and hasattr(args[0], 'name'): + agent_name = args[0].name + elif 'agent_name' in kwargs: + agent_name = kwargs['agent_name'] + elif 'agent' in kwargs and hasattr(kwargs['agent'], 'name'): + agent_name = kwargs['agent'].name + + if not agent_name: + logger.warning("Could not determine agent name for trust verification") + return func(*args, **kwargs) + + # Verify trust if enabled + config = TrustConfig.from_env() + if config.enabled and config.auto_verify_handoffs: + tool = trust_tool or JoyTrustTool(config=config) + verification = tool.verify_handoff_safety(agent_name, min_score) + + if not verification.get("handoff_safe", False): + error_msg = f"Handoff blocked: {verification.get('recommendation', 'Trust verification failed')}" + logger.warning(error_msg) + return { + "success": False, + "error": error_msg, + "trust_verification": verification + } + + logger.info(f"Trust verification passed for {agent_name}: {verification.get('recommendation')}") + + return func(*args, **kwargs) + + return wrapper + return decorator + + +def check_trust_score(agent_name: str, min_score: float = 3.0) -> Dict[str, Any]: + """Check an agent's trust score on Joy Trust Network before delegation. + + Args: + agent_name: Name/identifier of the agent to check + min_score: Minimum acceptable trust score + + Returns: + Dictionary containing trust information and safety recommendation + """ + tool = JoyTrustTool() + return tool.check_trust(agent_name=agent_name, min_score=min_score) + + +def verify_handoff_safety(agent_name: str, min_score: float = 3.0) -> Dict[str, Any]: + """Verify if it's safe to hand off to the specified agent. + + Args: + agent_name: Target agent for handoff + min_score: Minimum acceptable trust score + + Returns: + Dictionary with safety verification and recommendation + """ + tool = JoyTrustTool() + return tool.verify_handoff_safety(agent_name=agent_name, min_score=min_score) + + +def is_trust_verification_enabled() -> bool: + """Check if trust verification is enabled via environment variables.""" + config = TrustConfig.from_env() + return config.enabled -def check_trust_score(agent_name: str) -> Dict[str, Any]: - """Check an agent's trust score on Joy Trust Network before delegation.""" - return JoyTrustTool().check_trust(agent_name=agent_name) \ No newline at end of file +def get_trust_config() -> TrustConfig: + """Get current trust configuration from environment variables.""" + return TrustConfig.from_env() \ No newline at end of file diff --git a/tests/test_marketplace_tools.py b/tests/test_marketplace_tools.py index b5a2456..6133ea1 100644 --- a/tests/test_marketplace_tools.py +++ b/tests/test_marketplace_tools.py @@ -1,9 +1,12 @@ """Tests for marketplace tools.""" import pytest +import os from praisonai_tools import ( pinchwork_delegate, verify_agent_identity, check_trust_score, - check_behavioral_trust, verify_task_delegation_safety + check_behavioral_trust, verify_task_delegation_safety, + verify_handoff_safety, trust_verified_handoff, + is_trust_verification_enabled, get_trust_config, JoyTrustTool ) @@ -14,6 +17,12 @@ def test_marketplace_tools_import(): assert check_trust_score is not None assert check_behavioral_trust is not None assert verify_task_delegation_safety is not None + # New Joy Trust integration functions + assert verify_handoff_safety is not None + assert trust_verified_handoff is not None + assert is_trust_verification_enabled is not None + assert get_trust_config is not None + assert JoyTrustTool is not None def test_pinchwork_delegate_signature(): @@ -117,6 +126,122 @@ def test_delegation_safety_real_api(): print(f"Delegation safety result: {result}") +def test_verify_handoff_safety_signature(): + """Test verify_handoff_safety tool signature and documentation.""" + # Check function exists and has proper signature + assert callable(verify_handoff_safety) + + # Check documentation + doc = verify_handoff_safety.__doc__ + assert "Verify if it's safe to hand off to the specified agent" in doc + assert "agent_name:" in doc + assert "min_score:" in doc + + +def test_trust_verified_handoff_decorator(): + """Test trust_verified_handoff decorator functionality.""" + # Check decorator exists and is callable + assert callable(trust_verified_handoff) + + # Test decorator application without environment variables set + @trust_verified_handoff(min_score=3.0) + def dummy_function(agent_name): + return {"success": True, "agent": agent_name} + + # Should work when trust verification is disabled + result = dummy_function("test_agent") + assert result["success"] is True + assert result["agent"] == "test_agent" + + +def test_trust_config_functions(): + """Test trust configuration functions.""" + # Test without environment variables + assert is_trust_verification_enabled() is False + + config = get_trust_config() + assert config.enabled is False + assert config.provider == "joy" + assert config.min_score == 3.0 + + # Test with environment variables + old_provider = os.environ.get('PRAISONAI_TRUST_PROVIDER') + old_min_score = os.environ.get('PRAISONAI_TRUST_MIN_SCORE') + + try: + os.environ['PRAISONAI_TRUST_PROVIDER'] = 'joy' + os.environ['PRAISONAI_TRUST_MIN_SCORE'] = '4.5' + + assert is_trust_verification_enabled() is True + + config = get_trust_config() + assert config.enabled is True + assert config.min_score == 4.5 + + finally: + # Clean up environment + if old_provider is not None: + os.environ['PRAISONAI_TRUST_PROVIDER'] = old_provider + elif 'PRAISONAI_TRUST_PROVIDER' in os.environ: + del os.environ['PRAISONAI_TRUST_PROVIDER'] + + if old_min_score is not None: + os.environ['PRAISONAI_TRUST_MIN_SCORE'] = old_min_score + elif 'PRAISONAI_TRUST_MIN_SCORE' in os.environ: + del os.environ['PRAISONAI_TRUST_MIN_SCORE'] + + +def test_joy_trust_tool_class(): + """Test JoyTrustTool class instantiation and methods.""" + from praisonai_tools.tools.joy_trust_tool import TrustConfig + + # Test basic instantiation + tool = JoyTrustTool() + assert tool.name == "joy_trust" + assert tool.config is not None + + # Test with custom config + custom_config = TrustConfig( + enabled=True, + min_score=4.0, + timeout_seconds=5.0 + ) + tool_with_config = JoyTrustTool(config=custom_config) + assert tool_with_config.config.min_score == 4.0 + assert tool_with_config.config.timeout_seconds == 5.0 + + # Test configuration method + config_result = tool.configure(min_score=3.5, timeout_seconds=15.0) + assert config_result["status"] == "configured" + assert config_result["config"]["min_score"] == 3.5 + assert config_result["config"]["timeout_seconds"] == 15.0 + + +def test_check_trust_score_error_handling(): + """Test check_trust_score error handling with invalid input.""" + # Test with empty agent name + result = check_trust_score("", min_score=3.0) + assert result["error"] == "agent_name is required" + assert result["trust_score"] == 0.0 + assert result["meets_threshold"] is False + + +@pytest.mark.skipif(True, reason="Skip real API calls in tests - requires network") +def test_enhanced_joy_trust_real_api(): + """Real test with enhanced Joy Trust functionality (skipped by default).""" + # Test basic trust check + result = check_trust_score("example_agent", min_score=3.0) + print(f"Enhanced Joy Trust result: {result}") + assert "meets_threshold" in result + assert "threshold_used" in result + + # Test handoff safety verification + safety_result = verify_handoff_safety("example_agent", min_score=3.0) + print(f"Handoff safety result: {safety_result}") + assert "handoff_safe" in safety_result + assert "recommendation" in safety_result + + def test_tools_work_without_httpx(): """Test that tools give proper error when httpx is not installed.""" # This would need mocking httpx import to test properly @@ -125,4 +250,5 @@ def test_tools_work_without_httpx(): assert verify_agent_identity is not None assert check_trust_score is not None assert check_behavioral_trust is not None - assert verify_task_delegation_safety is not None \ No newline at end of file + assert verify_task_delegation_safety is not None + assert verify_handoff_safety is not None \ No newline at end of file