feat: enhance Joy Trust integration for secure agent handoffs (fixes MervinPraison/PraisonAI#1324) (#16)

Enhanced Joy Trust tool with native integration features:

- Added automatic handoff trust verification
- Environment variable configuration (PRAISONAI_TRUST_PROVIDER=joy)
- @trust_verified_handoff decorator for secure delegation
- Handoff safety verification with recommendations
- Configurable trust score thresholds and caching
- Fallback handling for robust operation
- Comprehensive examples and documentation
- Updated tests for new functionality

Addresses feature request for native Joy Trust integration to make
trust verification automatic and free for PraisonAI users.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: MervinPraison <[email protected]>
Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-04-09 12:40:09 +01:00
1 parent 324d3458d9
commit 1275ca4792
5 files changed
+900 -32

No files matched your search

+302
View File
@@ -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)
+198
View File
@@ -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.")
+8
View File
@@ -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 @@ __all__ = [
"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",
+264 -30
View File
@@ -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 @@ class JoyTrustTool(BaseTool):
}
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 @@ class JoyTrustTool(BaseTool):
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 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)
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 get_trust_config() -> TrustConfig:
"""Get current trust configuration from environment variables."""
return TrustConfig.from_env()
+128 -2
View File
@@ -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
assert verify_task_delegation_safety is not None
assert verify_handoff_safety is not None