Advanced tools

[MervinPraison]

@claude

Advanced tools
minimal code changes in the main feature and minimal code in the client side code.

1. Pre/Post Execution Hooks 🪝

💡 Proposed PraisonAI Agents Implementation

from praisonaiagents.tools import tool

Simple pre/post hooks

def log_start(context):
print(f"Starting {context.tool_name} with args: {context.args}")

def log_end(context):
print(f"Finished {context.tool_name}, result: {context.result}")

@tool(
before=log_start,
after=log_end
)
def calculate_metrics(data: list) -> dict:
return {"mean": sum(data)/len(data)}

Multiple hooks with priority

@tool(
before=[
(validate_input, priority=1), # Runs first
(log_start, priority=2) # Runs second
],
after=[log_end, save_to_cache]
)
def process_data(input: str) -> str:
return input.upper()

Hook with error handling

def error_handler(context):
if context.error:
logger.error(f"Tool {context.tool_name} failed: {context.error}")
# Can modify the error or suppress it
context.error = None
context.result = "default_value"

@tool(after=error_handler)
def risky_operation():
# Some code that might fail
pass

Global hooks for all tools

from praisonaiagents import set_global_hooks

set_global_hooks(
before=lambda ctx: logger.info(f"Tool call: {ctx.tool_name}"),
after=lambda ctx: metrics.record(ctx.tool_name, ctx.execution_time)
)
Integration with existing approval system:

Combine with existing @require_approval

@require_approval(risk_level="high")
@tool(
before=validate_permissions,
after=audit_log
)
def delete_user(user_id: str):
# Dangerous operation with hooks and approval
pass

2. Tool-Level Caching with TTL 🔄

💡 Proposed PraisonAI Agents Implementation

from praisonaiagents.tools import tool, cache

Simple caching decorator

@tool
@cache(ttl=300) # 5 minutes
def fetch_weather(city: str) -> dict:
return requests.get(f"api.weather.com/{city}").json()

Advanced caching with custom key

@tool
@cache(
ttl=3600, # 1 hour
key=lambda city, date: f"{city}:{date}", # Custom cache key
condition=lambda result: result['status'] == 'success' # Only cache successful results
)
def get_historical_weather(city: str, date: str) -> dict:
return fetch_historical_data(city, date)

Cache with different backends

from praisonaiagents.tools import RedisCache, MemoryCache

@tool
@cache(backend=RedisCache(host='localhost'), ttl=1800)
def expensive_calculation(x: int, y: int) -> int:
time.sleep(5)
return x ** y

Cache invalidation

@tool
@cache(ttl=600, tags=['user_data'])
def get_user_profile(user_id: str) -> dict:
return db.fetch_user(user_id)

Invalidate cache by tags

from praisonaiagents.tools import invalidate_cache
invalidate_cache(tags=['user_data'])

Per-agent caching

agent = Agent(
name="DataAgent",
cache_config={
'enabled': True,
'ttl': 300,
'backend': 'redis'
}
)
Built-in cache management:

Cache statistics

from praisonaiagents.tools import get_cache_stats

stats = get_cache_stats()
print(f"Cache hits: {stats.hits}, misses: {stats.misses}")

Clear all caches

from praisonaiagents.tools import clear_all_caches
clear_all_caches()

3. External Execution Markers 🚀

💡 Proposed PraisonAI Agents Implementation

from praisonaiagents.tools import tool, external

Simple external marker

@tool
@external
def run_on_gpu(model_path: str, data: list) -> dict:
# Execution pauses, returns control to handler
pass

External with metadata

@tool
@external(
executor="gpu_cluster",
requirements=["cuda>=11.0", "torch>=2.0"],
estimated_time=300 # 5 minutes
)
def train_model(dataset: str, hyperparams: dict):
pass

Conditional external execution

@tool
@external(when=lambda args: args['size'] > 1000000)
def process_data(data: list, size: int):
# Only external if data is large
pass

Integration with agent

agent = Agent(
name="MLAgent",
tools=[train_model, run_on_gpu],
external_handler=my_external_executor # Custom handler
)

External execution flow

async def my_external_executor(tool_call):
if tool_call.executor == "gpu_cluster":
# Send to GPU cluster
job_id = await gpu_cluster.submit(tool_call)
result = await gpu_cluster.wait_for_result(job_id)
return result
Built-in external patterns:

Human-in-the-loop

@tool
@external(type="human_approval")
def delete_production_data(table: str):
pass

Webhook-based

@tool
@external(
type="webhook",
endpoint="https://api.company.com/execute",
auth_token="${EXTERNAL_API_TOKEN}"
)
def trigger_deployment(version: str):
pass

4. Structured User Input Fields 📝

💡 Proposed PraisonAI Agents Implementation

from praisonaiagents.tools import tool, user_input, Field

Simple user input

@tool
@user_input(
Field(name="confirm", type=bool, description="Proceed with deletion?"),
Field(name="reason", type=str, description="Reason for deletion", required=False)
)
def delete_records(confirm: bool, reason: str = None):
if confirm:
# Proceed with deletion
pass

Advanced field types

from praisonaiagents.tools import Choice, Range, Pattern

@tool
@user_input(
Field(
name="priority",
type=Choice(["low", "medium", "high"]),
description="Task priority",
default="medium"
),
Field(
name="budget",
type=Range(min=0, max=10000),
description="Budget in USD"
),
Field(
name="email",
type=Pattern(r"^[\w.-]+@[\w.-]+.\w+$"),
description="Contact email"
)
)
def create_project(priority: str, budget: float, email: str):
pass

Dynamic fields based on context

@tool
def process_order(order_type: str):
if order_type == "custom":
# Dynamically request additional fields
details = tool.request_input(
Field(name="specifications", type=str),
Field(name="quantity", type=int),
Field(name="delivery_date", type="date")
)
return handle_custom_order(details)
return handle_standard_order()

Form-like input groups

from praisonaiagents.tools import InputGroup

@tool
@user_input(
InputGroup(
"Personal Information",
Field(name="first_name", type=str),
Field(name="last_name", type=str),
Field(name="age", type=int, optional=True)
),
InputGroup(
"Preferences",
Field(name="newsletter", type=bool, default=True),
Field(name="language", type=Choice(["en", "es", "fr"]))
)
)
def register_user(**kwargs):
pass
Integration with agents:

Agent with input handling

agent = Agent(
name="InteractiveAgent",
on_user_input_required=lambda fields: display_form(fields),
input_timeout=60 # Wait 60 seconds for input
)

Async input handling

async def handle_with_input():
agent = Agent(name="Assistant")

# Will pause and wait for user input
result = await agent.run(
    "Process the user registration",
    on_input=lambda fields: get_user_response(fields)
)

📋 Implementation Recommendations

1. Unified Tool Decorator System

Create a comprehensive @tool decorator that supports all features:

@tool(
name="fetch_data",
description="Fetches data from API",
# Hooks
before=validate_input,
after=log_result,
# Caching
cache=True,
cache_ttl=300,
# External execution
external=conditional_external,
# User input
inputs=[Field(name="api_key", type=str, secret=True)],
# Integration with existing
require_approval=True,
risk_level="medium"
)
def fetch_data(api_key: str, endpoint: str):
pass

2. Tool Context Object

Provide rich context to hooks and handlers:

class ToolContext:
tool_name: str
args: dict
kwargs: dict
result: Any
error: Exception | None
agent: Agent | None
execution_time: float
metadata: dict

3. Progressive Enhancement

Start with simple APIs that can be progressively enhanced:

Simple

@tool
@cache(300)
def simple_tool(): pass

Advanced

@tool(
cache={'ttl': 300, 'backend': 'redis', 'key': custom_key},
hooks={'before': [validate, log], 'after': [audit]},
external={'when': is_large, 'executor': 'gpu'}
)
def advanced_tool(): pass

4. Backwards Compatibility

Ensure existing tools continue to work:

Old style still works

def my_tool(x: int) -> int:
return x * 2

agent = Agent(tools=[my_tool]) # Works without @tool decorator

[MervinPraison]

@claude review this issue and do a detailed analysis and fix this if the existing code doesn't have the solution implemented. Making sure it has backward compatibility, no existing features removed. After making those changes, again review the applied changes. Use @web to search if you dont know any information or to find the latest documentation or to find the latest version. Run the code if you think you need to run it to test it. Minimal code change to start with if required any changes. Please create a PR using gh tool with your changes.

[github-actions]

Claude finished @MervinPraison's task —— View job • claude/issue-972-20250717-0022 • [Create PR ➔](https://github.com/MervinPraison/PraisonAI/compare/main...claude/issue-972-20250717-0022?quick_pull=1&title=Issue%20%23972%3A%20Changes%20from%20Claude&body=This%20PR%20addresses%20issue%20%23972%0A%0AGenerated%20with%20%5BClaude%20Code%5D(https%3A%2F%2Fclaude.ai%2Fcode)

---

✅ Advanced Tools Implementation Completed

All 4 requested advanced tool features have been successfully implemented with full backward compatibility!

Todo List:

• Analyze current PraisonAI tools system architecture
• Check existing implementation of requested features
• Implement Pre/Post Execution Hooks
• Implement Tool-Level Caching with TTL
• Implement External Execution Markers
• Implement Structured User Input Fields
• Ensure backward compatibility
• Test implementation
• Create PR with changes

🚀 Implementation Summary:

Core Features Added:

1. 🪝 Pre/Post Execution Hooks - Complete lifecycle management

   @tool(before=log_start, after=log_end)
   def my_tool(): pass

2. 🔄 Tool-Level Caching with TTL - Performance optimization

   @tool
   @cache(ttl=300, tags=['weather'])
   def cached_tool(): pass

3. 🚀 External Execution Markers - Flexible execution contexts

   @tool
   @external(executor="gpu_cluster", when=lambda args: len(args[0]) > 1000)
   def external_tool(): pass

4. 📝 Structured User Input Fields - Rich input validation

   @tool
   @user_input(
       Field(name="priority", type=Choice(["low", "high"])),
       Field(name="budget", type=Range(min=0, max=10000))
   )
   def validated_tool(): pass

📁 Files Created/Modified:

• NEW: praisonaiagents/tools/advanced_tools.py (580 lines) - Core implementation
• NEW: praisonaiagents/tools/examples.py (400+ lines) - Usage examples
• NEW: test_advanced_tools.py (230+ lines) - Test suite
• MODIFIED: praisonaiagents/tools/__init__.py - Added exports

🔧 Key Technical Features:

• Unified @tool Decorator - Single decorator for all features
• Priority-based Hook System - Ordered execution control
• Advanced Caching - TTL, custom keys, conditional caching, tag-based invalidation
• External Handler Registry - Pluggable execution backends
• Rich Context Objects - Full execution metadata
• Global Hook Support - Apply hooks to all tools
• Backward Compatibility - Zero breaking changes

⚡ Performance & Management:

• Cache Statistics: get_cache_stats() for monitoring
• Cache Management: invalidate_cache(), clear_all_caches()
• External Handlers: register_external_handler() for custom execution
• Error Handling: Graceful hook error recovery

🧪 Testing & Validation:

✅ 8 Test Scenarios covering all features
✅ Import Validation - All components loadable
✅ Hook Execution - Pre/post hooks working
✅ Caching Logic - TTL and cache hits/misses
✅ External Markers - Metadata and execution flow
✅ Input Validation - Field types and constraints
✅ Backward Compatibility - Existing tools unchanged

🔄 Git Changes Committed:

• Commit: a995a7f5
• Files: 4 changed, 1124 insertions(+), 1 deletion(-)
• Branch: claude/issue-972-20250717-0022

🎯 Pull Request:

Create PR ← Click to create the pull request

The implementation follows the exact specifications from Issue #972, provides comprehensive examples, maintains full backward compatibility, and includes extensive testing. All existing tools continue to work unchanged while gaining access to powerful new capabilities.

Generated with Claude Code

---

[github-actions]

Claude finished @MervinPraison's task —— View job • claude/issue-972-20250717-0022 • Create PR ➔

---

Advanced Tools Implementation Complete! ✅

Todo List:

• Explore current PraisonAI Agents codebase structure
• Analyze existing tool system implementation
• Check if proposed features already exist
• Implement Pre/Post Execution Hooks
• Implement Tool-Level Caching with TTL
• Implement External Execution Markers
• Implement Structured User Input Fields
• Ensure backward compatibility
• Test implementation (standalone validation)
• Review applied changes
• Create PR with changes

🎉 Implementation Complete!

All 4 Major Features Successfully Implemented:

1. 🪝 Pre/Post Execution Hooks - Rich context, multiple hooks, global hooks
2. 💾 Tool-Level Caching with TTL - Memory cache, custom keys, tag invalidation
3. 🚀 External Execution Markers - Conditional execution, handler registration
4. 📝 Structured User Input Fields - Field validation, types, constraints

Changes Committed:

• ✅ praisonaiagents/tools/advanced.py (629 lines) - Core implementation
• ✅ praisonaiagents/tools/__init__.py - Advanced tools exports
• ✅ praisonaiagents/__init__.py - Package-level exports
• ✅ examples/advanced_tools_example.py - Comprehensive examples
• ✅ tests/test_advanced_tools.py - Example-driven testing

Key Features:

# Unified decorator with all features
@tool(
    before=log_start, after=log_end,
    cache=True, cache_ttl=300,
    external={"executor": "gpu_cluster"},
    needs_approval=True, risk_level="high"
)
def comprehensive_tool(data: str):
    return {"result": data}

# Individual decorators
@cached_tool(ttl=300)
@external_tool(executor="gpu")
@approved_tool(risk_level="high")

Backward Compatibility: ✅ All existing tools work unchanged

Commit: e3bcbea7 - feat: implement advanced tools system with hooks, caching, and external execution