Release v2.3.41

README.md

@@ -189,6 +189,7 @@ npm install praisonai
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
 | 💻 Code Interpreter Agents | [Example](examples/python/agents/code-agent.py) | [📖](https://docs.praison.ai/features/codeagent) |
+| ✏️ AI Code Editing Tools | [Example](#ai-code-editing) | [📖](https://docs.praison.ai/features/code) |
 | 📚 RAG Agents | [Example](examples/python/concepts/rag-agents.py) | [📖](https://docs.praison.ai/features/rag) |
 | 🤔 Async & Parallel Processing | [Example](examples/python/general/async_example.py) | [📖](https://docs.praison.ai/features/async) |
 | 🔢 Math Agents | [Example](examples/python/agents/math-agent.py) | [📖](https://docs.praison.ai/features/mathagent) |
@@ -2261,6 +2262,13 @@ PraisonAI provides zero-dependency persistent memory for agents. For detailed ex
 - 🔐 **MCP Security** - Origin validation, DNS rebinding prevention, secure sessions
 - 🔄 **MCP Resumability** - SSE stream recovery via Last-Event-ID
 
+### A2A & A2UI Protocols
+
+- 🔗 **A2A Protocol** - Agent-to-Agent communication for inter-agent collaboration
+- 🖼️ **A2UI Protocol** - Agent-to-User Interface for generating rich UIs from agents
+- 📋 **UI Templates** - ChatTemplate, ListTemplate, FormTemplate, DashboardTemplate
+- 🔧 **Surface Builder** - Fluent API for building declarative UIs
+
 ### Safety & Control
 
 - 🤝 **Agent Handoffs** - Transfer context between specialised agents

docker/Dockerfile.chat

@@ -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.39" \
+    "praisonai>=2.3.41" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -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.39" \
+    "praisonai>=2.3.41" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -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.39" \
+    "praisonai>=2.3.41" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

src/praisonai/README.md

@@ -189,6 +189,7 @@ npm install praisonai
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
 | 💻 Code Interpreter Agents | [Example](examples/python/agents/code-agent.py) | [📖](https://docs.praison.ai/features/codeagent) |
+| ✏️ AI Code Editing Tools | [Example](#ai-code-editing) | [📖](https://docs.praison.ai/features/code) |
 | 📚 RAG Agents | [Example](examples/python/concepts/rag-agents.py) | [📖](https://docs.praison.ai/features/rag) |
 | 🤔 Async & Parallel Processing | [Example](examples/python/general/async_example.py) | [📖](https://docs.praison.ai/features/async) |
 | 🔢 Math Agents | [Example](examples/python/agents/math-agent.py) | [📖](https://docs.praison.ai/features/mathagent) |
@@ -2261,6 +2262,13 @@ PraisonAI provides zero-dependency persistent memory for agents. For detailed ex
 - 🔐 **MCP Security** - Origin validation, DNS rebinding prevention, secure sessions
 - 🔄 **MCP Resumability** - SSE stream recovery via Last-Event-ID
 
+### A2A & A2UI Protocols
+
+- 🔗 **A2A Protocol** - Agent-to-Agent communication for inter-agent collaboration
+- 🖼️ **A2UI Protocol** - Agent-to-User Interface for generating rich UIs from agents
+- 📋 **UI Templates** - ChatTemplate, ListTemplate, FormTemplate, DashboardTemplate
+- 🔧 **Surface Builder** - Fluent API for building declarative UIs
+
 ### Safety & Control
 
 - 🤝 **Agent Handoffs** - Transfer context between specialised agents

src/praisonai/praisonai.rb

@@ -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.39.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.39.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.41.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.41.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/code/__init__.py

@@ -46,14 +46,40 @@ from .utils import (
     strip_line_numbers,
 )
 
+from .agent_tools import (
+    set_workspace,
+    get_workspace,
+    code_read_file,
+    code_write_file,
+    code_list_files,
+    code_apply_diff,
+    code_search_replace,
+    code_execute_command,
+    CODE_TOOLS,
+    create_diff_block,
+    create_multi_diff,
+)
+
 __all__ = [
-    # Main tools
+    # Main tools (low-level)
     "read_file",
     "write_file", 
     "list_files",
     "apply_diff",
     "search_replace",
     "execute_command",
+    # Agent tools (high-level, string returns)
+    "set_workspace",
+    "get_workspace",
+    "code_read_file",
+    "code_write_file",
+    "code_list_files",
+    "code_apply_diff",
+    "code_search_replace",
+    "code_execute_command",
+    "CODE_TOOLS",
+    "create_diff_block",
+    "create_multi_diff",
     # Diff utilities
     "DiffResult",
     "apply_search_replace_diff",

src/praisonai/praisonai/code/agent_tools.py

@@ -0,0 +1,344 @@
+"""
+Agent-compatible tool wrappers for PraisonAI Code.
+
+These functions are designed to be used as tools with PraisonAI agents.
+They return string results suitable for agent consumption.
+"""
+
+import os
+from typing import Optional
+
+from .tools import (
+    read_file as _read_file,
+    write_file as _write_file,
+    list_files as _list_files,
+    apply_diff as _apply_diff,
+    search_replace as _search_replace,
+    execute_command as _execute_command,
+)
+from .tools.apply_diff import create_diff_block, create_multi_diff
+
+
+# Global workspace setting for tools
+_workspace_root: Optional[str] = None
+
+
+def set_workspace(workspace_path: str) -> None:
+    """
+    Set the workspace root for all code tools.
+    
+    Args:
+        workspace_path: Absolute path to the workspace root
+    """
+    global _workspace_root
+    _workspace_root = os.path.abspath(workspace_path)
+
+
+def get_workspace() -> Optional[str]:
+    """Get the current workspace root."""
+    return _workspace_root
+
+
+def code_read_file(
+    path: str,
+    start_line: Optional[int] = None,
+    end_line: Optional[int] = None,
+) -> str:
+    """
+    Read the contents of a file.
+    
+    Use this tool to read source code files. The content is returned with
+    line numbers for easy reference when making edits.
+    
+    Args:
+        path: Path to the file (relative to workspace or absolute)
+        start_line: First line to read (1-indexed, optional)
+        end_line: Last line to read (1-indexed, optional)
+        
+    Returns:
+        File content with line numbers, or error message
+        
+    Example:
+        >>> content = code_read_file("src/main.py")
+        >>> content = code_read_file("src/main.py", start_line=10, end_line=50)
+    """
+    result = _read_file(
+        path=path,
+        start_line=start_line,
+        end_line=end_line,
+        add_line_nums=True,
+        workspace=_workspace_root,
+    )
+    
+    if result['success']:
+        header = f"File: {path}"
+        if start_line or end_line:
+            header += f" (lines {result['start_line']}-{result['end_line']} of {result['total_lines']})"
+        else:
+            header += f" ({result['total_lines']} lines)"
+        return f"{header}\n\n{result['content']}"
+    else:
+        return f"Error reading {path}: {result['error']}"
+
+
+def code_write_file(path: str, content: str) -> str:
+    """
+    Write content to a file, creating it if it doesn't exist.
+    
+    Use this tool to create new files or completely replace existing files.
+    For partial modifications, use code_apply_diff instead.
+    
+    Args:
+        path: Path to the file (relative to workspace or absolute)
+        content: Content to write to the file
+        
+    Returns:
+        Success message or error
+        
+    Example:
+        >>> result = code_write_file("src/new_module.py", "def hello():\\n    print('Hello!')")
+    """
+    result = _write_file(
+        path=path,
+        content=content,
+        workspace=_workspace_root,
+        create_directories=True,
+        strip_code_fences=True,
+    )
+    
+    if result['success']:
+        action = "Created" if result['created'] else "Updated"
+        return f"{action} file: {path} ({result['bytes_written']} bytes)"
+    else:
+        return f"Error writing {path}: {result['error']}"
+
+
+def code_list_files(
+    path: str = ".",
+    recursive: bool = False,
+    extensions: Optional[str] = None,
+) -> str:
+    """
+    List files in a directory.
+    
+    Use this tool to explore the codebase structure.
+    
+    Args:
+        path: Directory path (relative to workspace or absolute)
+        recursive: Whether to list recursively
+        extensions: Comma-separated list of extensions to filter (e.g., "py,js,ts")
+        
+    Returns:
+        Formatted list of files and directories
+        
+    Example:
+        >>> files = code_list_files("src", recursive=True, extensions="py")
+    """
+    ext_list = None
+    if extensions:
+        ext_list = [e.strip() for e in extensions.split(',')]
+    
+    result = _list_files(
+        path=path,
+        recursive=recursive,
+        workspace=_workspace_root,
+        extensions=ext_list,
+    )
+    
+    if result['success']:
+        lines = [f"Contents of {path}:"]
+        
+        # Add directories
+        for d in result['directories']:
+            lines.append(f"  📁 {d['path']}/")
+        
+        # Add files
+        for f in result['files']:
+            size = f.get('size', 0)
+            if size < 1024:
+                size_str = f"{size}B"
+            elif size < 1024 * 1024:
+                size_str = f"{size / 1024:.1f}KB"
+            else:
+                size_str = f"{size / (1024 * 1024):.1f}MB"
+            lines.append(f"  📄 {f['path']} ({size_str})")
+        
+        if result.get('truncated'):
+            lines.append(f"\n  ... (showing {result['total_count']} items, more available)")
+        
+        return '\n'.join(lines)
+    else:
+        return f"Error listing {path}: {result['error']}"
+
+
+def code_apply_diff(path: str, diff: str) -> str:
+    """
+    Apply a SEARCH/REPLACE diff to modify a file.
+    
+    Use this tool for precise, surgical edits to existing files.
+    The SEARCH section must match existing content exactly.
+    
+    Diff Format:
+        <<<<<<< SEARCH
+        :start_line:N
+        -------
+        [exact content to find]
+        =======
+        [new content to replace with]
+        >>>>>>> REPLACE
+    
+    Args:
+        path: Path to the file to modify
+        diff: The SEARCH/REPLACE diff content
+        
+    Returns:
+        Success message with number of changes, or error details
+        
+    Example:
+        >>> diff = '''<<<<<<< SEARCH
+        ... :start_line:10
+        ... -------
+        ... def old_function():
+        ...     pass
+        ... =======
+        ... def new_function():
+        ...     return True
+        ... >>>>>>> REPLACE'''
+        >>> result = code_apply_diff("src/main.py", diff)
+    """
+    result = _apply_diff(
+        path=path,
+        diff=diff,
+        workspace=_workspace_root,
+        fuzzy_threshold=0.95,  # Allow slight fuzzy matching
+    )
+    
+    if result['success']:
+        msg = f"Successfully applied {result['applied_count']} change(s) to {path}"
+        if result.get('failed_blocks'):
+            msg += f"\nWarning: {len(result['failed_blocks'])} block(s) failed to apply"
+        return msg
+    else:
+        error_msg = f"Error applying diff to {path}: {result['error']}"
+        if result.get('failed_blocks'):
+            for fb in result['failed_blocks'][:3]:  # Show first 3 failures
+                error_msg += f"\n- {fb.get('error', 'Unknown error')}"
+        return error_msg
+
+
+def code_search_replace(
+    path: str,
+    search: str,
+    replace: str,
+    is_regex: bool = False,
+) -> str:
+    """
+    Perform a search and replace operation on a file.
+    
+    Use this for simple text replacements. For complex multi-part edits,
+    use code_apply_diff instead.
+    
+    Args:
+        path: Path to the file
+        search: Text to search for (or regex pattern if is_regex=True)
+        replace: Text to replace with
+        is_regex: Whether to treat search as a regex pattern
+        
+    Returns:
+        Success message with replacement count, or error
+        
+    Example:
+        >>> result = code_search_replace("src/main.py", "old_name", "new_name")
+    """
+    result = _search_replace(
+        path=path,
+        operations=[{
+            'search': search,
+            'replace': replace,
+            'is_regex': is_regex,
+        }],
+        workspace=_workspace_root,
+    )
+    
+    if result['success']:
+        return f"Replaced {result['total_replacements']} occurrence(s) in {path}"
+    else:
+        return f"Error: {result['error']}"
+
+
+def code_execute_command(
+    command: str,
+    cwd: Optional[str] = None,
+) -> str:
+    """
+    Execute a shell command.
+    
+    Use this tool to run commands like tests, linters, or build tools.
+    
+    Args:
+        command: The command to execute
+        cwd: Working directory (relative to workspace or absolute)
+        
+    Returns:
+        Command output (stdout and stderr)
+        
+    Example:
+        >>> result = code_execute_command("python -m pytest tests/")
+        >>> result = code_execute_command("npm test", cwd="frontend")
+    """
+    work_dir = cwd
+    if work_dir and _workspace_root and not os.path.isabs(work_dir):
+        work_dir = os.path.join(_workspace_root, work_dir)
+    elif not work_dir and _workspace_root:
+        work_dir = _workspace_root
+    
+    result = _execute_command(
+        command=command,
+        cwd=work_dir,
+        timeout=120,
+    )
+    
+    output_parts = []
+    
+    if result['stdout']:
+        output_parts.append(f"stdout:\n{result['stdout']}")
+    
+    if result['stderr']:
+        output_parts.append(f"stderr:\n{result['stderr']}")
+    
+    if result['success']:
+        status = f"Command completed successfully (exit code: {result['exit_code']})"
+    else:
+        if result.get('error'):
+            status = f"Command failed: {result['error']}"
+        else:
+            status = f"Command failed (exit code: {result['exit_code']})"
+    
+    output_parts.insert(0, status)
+    
+    return '\n\n'.join(output_parts)
+
+
+# Export all agent tools
+CODE_TOOLS = [
+    code_read_file,
+    code_write_file,
+    code_list_files,
+    code_apply_diff,
+    code_search_replace,
+    code_execute_command,
+]
+
+__all__ = [
+    'set_workspace',
+    'get_workspace',
+    'code_read_file',
+    'code_write_file',
+    'code_list_files',
+    'code_apply_diff',
+    'code_search_replace',
+    'code_execute_command',
+    'CODE_TOOLS',
+    'create_diff_block',
+    'create_multi_diff',
+]

src/praisonai/praisonai/code/tools/__init__.py

@@ -0,0 +1,27 @@
+"""
+Code editing tools for PraisonAI agents.
+
+Provides tools for:
+- read_file: Read file contents with optional line ranges
+- write_file: Create or overwrite files
+- list_files: List directory contents
+- apply_diff: Apply SEARCH/REPLACE diffs
+- search_replace: Multiple search/replace operations
+- execute_command: Run shell commands
+"""
+
+from .read_file import read_file
+from .write_file import write_file
+from .list_files import list_files
+from .apply_diff import apply_diff
+from .search_replace import search_replace
+from .execute_command import execute_command
+
+__all__ = [
+    "read_file",
+    "write_file",
+    "list_files",
+    "apply_diff",
+    "search_replace",
+    "execute_command",
+]

src/praisonai/praisonai/code/tools/apply_diff.py

@@ -0,0 +1,221 @@
+"""
+Apply Diff Tool for PraisonAI Code.
+
+Provides functionality to apply SEARCH/REPLACE diffs to files
+with fuzzy matching support.
+"""
+
+import os
+from typing import Optional, Dict, Any
+
+from ..diff.diff_strategy import apply_search_replace_diff
+from ..utils.file_utils import (
+    file_exists,
+    is_path_within_directory,
+)
+
+
+def apply_diff(
+    path: str,
+    diff: str,
+    workspace: Optional[str] = None,
+    fuzzy_threshold: float = 1.0,
+    buffer_lines: int = 40,
+    backup: bool = False,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Apply a SEARCH/REPLACE diff to a file.
+    
+    This tool applies precise, targeted modifications to an existing file
+    by searching for specific sections of content and replacing them.
+    
+    Diff Format:
+        <<<<<<< SEARCH
+        :start_line:N
+        -------
+        [exact content to find]
+        =======
+        [new content to replace with]
+        >>>>>>> REPLACE
+    
+    Args:
+        path: Path to the file (absolute or relative to workspace)
+        diff: The SEARCH/REPLACE diff content
+        workspace: Workspace root directory (for relative paths)
+        fuzzy_threshold: Similarity threshold (0.0-1.0, 1.0 = exact match)
+        buffer_lines: Lines to search around start_line hint
+        backup: Whether to create a backup before modifying
+        encoding: File encoding (default: utf-8)
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - path: str
+        - applied_count: int (number of blocks applied)
+        - failed_blocks: list (details of failed blocks)
+        - error: str (if success is False)
+        
+    Example:
+        >>> diff = '''
+        ... <<<<<<< SEARCH
+        ... :start_line:1
+        ... -------
+        ... def old_function():
+        ...     pass
+        ... =======
+        ... def new_function():
+        ...     return True
+        ... >>>>>>> REPLACE
+        ... '''
+        >>> result = apply_diff("src/main.py", diff)
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Security check
+    if workspace:
+        if not is_path_within_directory(abs_path, workspace):
+            return {
+                'success': False,
+                'error': f"Path '{path}' is outside the workspace",
+                'path': path,
+                'applied_count': 0,
+            }
+    
+    # Check if file exists
+    if not file_exists(abs_path):
+        return {
+            'success': False,
+            'error': f"File not found: {path}",
+            'path': path,
+            'applied_count': 0,
+        }
+    
+    try:
+        # Read original content
+        with open(abs_path, 'r', encoding=encoding, errors='replace') as f:
+            original_content = f.read()
+        
+        # Apply the diff
+        result = apply_search_replace_diff(
+            original_content=original_content,
+            diff_content=diff,
+            fuzzy_threshold=fuzzy_threshold,
+            buffer_lines=buffer_lines,
+        )
+        
+        if not result.success:
+            return {
+                'success': False,
+                'error': result.error or "Failed to apply diff",
+                'path': path,
+                'applied_count': result.applied_count,
+                'failed_blocks': result.failed_blocks,
+            }
+        
+        # Create backup if requested
+        backup_path = None
+        if backup:
+            import time
+            timestamp = int(time.time())
+            backup_path = f"{abs_path}.backup.{timestamp}"
+            with open(backup_path, 'w', encoding=encoding) as f:
+                f.write(original_content)
+        
+        # Write the modified content
+        with open(abs_path, 'w', encoding=encoding) as f:
+            f.write(result.content)
+        
+        return {
+            'success': True,
+            'path': path,
+            'absolute_path': abs_path,
+            'applied_count': result.applied_count,
+            'failed_blocks': result.failed_blocks if result.failed_blocks else None,
+            'backup_path': backup_path,
+        }
+        
+    except PermissionError:
+        return {
+            'success': False,
+            'error': f"Permission denied: {path}",
+            'path': path,
+            'applied_count': 0,
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error applying diff to {path}: {str(e)}",
+            'path': path,
+            'applied_count': 0,
+        }
+
+
+def create_diff_block(
+    search_content: str,
+    replace_content: str,
+    start_line: Optional[int] = None,
+) -> str:
+    """
+    Create a properly formatted SEARCH/REPLACE diff block.
+    
+    Args:
+        search_content: The content to search for
+        replace_content: The content to replace with
+        start_line: Optional line number hint
+        
+    Returns:
+        Formatted diff block string
+        
+    Example:
+        >>> diff = create_diff_block(
+        ...     "def old():\\n    pass",
+        ...     "def new():\\n    return True",
+        ...     start_line=10
+        ... )
+    """
+    lines = ["<<<<<<< SEARCH"]
+    
+    if start_line:
+        lines.append(f":start_line:{start_line}")
+    
+    lines.append("-------")
+    lines.append(search_content)
+    lines.append("=======")
+    lines.append(replace_content)
+    lines.append(">>>>>>> REPLACE")
+    
+    return '\n'.join(lines)
+
+
+def create_multi_diff(blocks: list) -> str:
+    """
+    Create a diff with multiple SEARCH/REPLACE blocks.
+    
+    Args:
+        blocks: List of dicts with 'search', 'replace', and optional 'start_line'
+        
+    Returns:
+        Combined diff string with all blocks
+        
+    Example:
+        >>> diff = create_multi_diff([
+        ...     {'search': 'old1', 'replace': 'new1', 'start_line': 5},
+        ...     {'search': 'old2', 'replace': 'new2', 'start_line': 20},
+        ... ])
+    """
+    diff_blocks = []
+    
+    for block in blocks:
+        diff_block = create_diff_block(
+            search_content=block.get('search', ''),
+            replace_content=block.get('replace', ''),
+            start_line=block.get('start_line'),
+        )
+        diff_blocks.append(diff_block)
+    
+    return '\n\n'.join(diff_blocks)

src/praisonai/praisonai/code/tools/execute_command.py

@@ -0,0 +1,275 @@
+"""
+Execute Command Tool for PraisonAI Code.
+
+Provides functionality to execute shell commands safely.
+"""
+
+import os
+import subprocess
+import shlex
+from typing import Optional, Dict, Any
+
+
+# Commands that are generally safe to auto-run
+SAFE_COMMANDS = {
+    'ls', 'dir', 'pwd', 'echo', 'cat', 'head', 'tail', 'wc',
+    'grep', 'find', 'which', 'whereis', 'type', 'file',
+    'date', 'whoami', 'hostname', 'uname',
+    'python', 'python3', 'node', 'npm', 'npx', 'pip', 'pip3',
+    'git', 'cargo', 'go', 'ruby', 'java', 'javac',
+}
+
+# Commands that should never be auto-run
+DANGEROUS_COMMANDS = {
+    'rm', 'rmdir', 'del', 'format', 'mkfs',
+    'dd', 'shred', 'chmod', 'chown', 'chgrp',
+    'kill', 'killall', 'pkill',
+    'shutdown', 'reboot', 'halt', 'poweroff',
+    'sudo', 'su', 'doas',
+    'curl', 'wget', 'ssh', 'scp', 'rsync',
+    'mv', 'cp',  # Can be destructive
+}
+
+
+def execute_command(
+    command: str,
+    cwd: Optional[str] = None,
+    workspace: Optional[str] = None,
+    timeout: int = 120,
+    capture_output: bool = True,
+    shell: bool = True,
+    env: Optional[Dict[str, str]] = None,
+) -> Dict[str, Any]:
+    """
+    Execute a shell command and return the result.
+    
+    This tool executes shell commands with safety checks and output capture.
+    
+    Args:
+        command: The command to execute
+        cwd: Working directory for the command
+        workspace: Workspace root (for security validation)
+        timeout: Command timeout in seconds
+        capture_output: Whether to capture stdout/stderr
+        shell: Whether to run through shell
+        env: Additional environment variables
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - exit_code: int
+        - stdout: str
+        - stderr: str
+        - command: str
+        - error: str (if success is False)
+        
+    Example:
+        >>> result = execute_command("python --version")
+        >>> print(result['stdout'])
+    """
+    if not command or not command.strip():
+        return {
+            'success': False,
+            'error': "Empty command",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+    
+    # Resolve working directory
+    if cwd:
+        if workspace and not os.path.isabs(cwd):
+            work_dir = os.path.abspath(os.path.join(workspace, cwd))
+        else:
+            work_dir = os.path.abspath(cwd)
+    elif workspace:
+        work_dir = workspace
+    else:
+        work_dir = os.getcwd()
+    
+    # Validate working directory exists
+    if not os.path.isdir(work_dir):
+        return {
+            'success': False,
+            'error': f"Working directory not found: {cwd}",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+    
+    # Prepare environment
+    cmd_env = os.environ.copy()
+    if env:
+        cmd_env.update(env)
+    
+    # Set PAGER to cat to avoid interactive pagers
+    cmd_env['PAGER'] = 'cat'
+    
+    try:
+        # Execute command
+        if shell:
+            result = subprocess.run(
+                command,
+                shell=True,
+                cwd=work_dir,
+                capture_output=capture_output,
+                timeout=timeout,
+                env=cmd_env,
+                text=True,
+            )
+        else:
+            # Parse command for non-shell execution
+            args = shlex.split(command)
+            result = subprocess.run(
+                args,
+                cwd=work_dir,
+                capture_output=capture_output,
+                timeout=timeout,
+                env=cmd_env,
+                text=True,
+            )
+        
+        return {
+            'success': result.returncode == 0,
+            'exit_code': result.returncode,
+            'stdout': result.stdout if capture_output else '',
+            'stderr': result.stderr if capture_output else '',
+            'command': command,
+            'cwd': work_dir,
+        }
+        
+    except subprocess.TimeoutExpired:
+        return {
+            'success': False,
+            'error': f"Command timed out after {timeout} seconds",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+    except FileNotFoundError as e:
+        return {
+            'success': False,
+            'error': f"Command not found: {str(e)}",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+    except PermissionError:
+        return {
+            'success': False,
+            'error': "Permission denied",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error executing command: {str(e)}",
+            'command': command,
+            'exit_code': -1,
+            'stdout': '',
+            'stderr': '',
+        }
+
+
+def is_safe_command(command: str) -> bool:
+    """
+    Check if a command is considered safe to auto-run.
+    
+    Args:
+        command: The command to check
+        
+    Returns:
+        True if the command is considered safe
+    """
+    if not command:
+        return False
+    
+    # Get the base command (first word)
+    parts = shlex.split(command)
+    if not parts:
+        return False
+    
+    base_cmd = os.path.basename(parts[0]).lower()
+    
+    # Check against dangerous commands
+    if base_cmd in DANGEROUS_COMMANDS:
+        return False
+    
+    # Check against safe commands
+    if base_cmd in SAFE_COMMANDS:
+        return True
+    
+    # Default to unsafe
+    return False
+
+
+def get_command_info(command: str) -> Dict[str, Any]:
+    """
+    Get information about a command without executing it.
+    
+    Args:
+        command: The command to analyze
+        
+    Returns:
+        Dictionary with command analysis
+    """
+    if not command:
+        return {
+            'valid': False,
+            'error': "Empty command",
+        }
+    
+    try:
+        parts = shlex.split(command)
+        base_cmd = parts[0] if parts else ''
+        
+        return {
+            'valid': True,
+            'base_command': base_cmd,
+            'arguments': parts[1:] if len(parts) > 1 else [],
+            'is_safe': is_safe_command(command),
+            'is_dangerous': os.path.basename(base_cmd).lower() in DANGEROUS_COMMANDS,
+        }
+    except ValueError as e:
+        return {
+            'valid': False,
+            'error': f"Invalid command syntax: {str(e)}",
+        }
+
+
+def run_python(
+    code: str,
+    cwd: Optional[str] = None,
+    timeout: int = 60,
+) -> Dict[str, Any]:
+    """
+    Execute Python code and return the result.
+    
+    Args:
+        code: Python code to execute
+        cwd: Working directory
+        timeout: Execution timeout
+        
+    Returns:
+        Dictionary with execution result
+    """
+    # Create a temporary command to run the code
+    import sys
+    python_cmd = sys.executable
+    
+    # Escape the code for command line
+    escaped_code = code.replace('\\', '\\\\').replace('"', '\\"')
+    command = f'{python_cmd} -c "{escaped_code}"'
+    
+    return execute_command(
+        command=command,
+        cwd=cwd,
+        timeout=timeout,
+    )

src/praisonai/praisonai/code/tools/list_files.py

@@ -0,0 +1,274 @@
+"""
+List Files Tool for PraisonAI Code.
+
+Provides functionality to list directory contents with filtering options.
+"""
+
+import os
+from typing import Optional, List, Dict, Any
+
+from ..utils.ignore_utils import (
+    should_ignore_path,
+    load_gitignore_patterns,
+)
+
+
+def list_files(
+    path: str,
+    recursive: bool = False,
+    workspace: Optional[str] = None,
+    max_files: int = 200,
+    include_hidden: bool = False,
+    extensions: Optional[List[str]] = None,
+    respect_gitignore: bool = True,
+) -> Dict[str, Any]:
+    """
+    List files and directories in a given path.
+    
+    This tool lists the contents of a directory, optionally recursively,
+    with support for filtering and gitignore patterns.
+    
+    Args:
+        path: Path to the directory (absolute or relative to workspace)
+        recursive: Whether to list recursively
+        workspace: Workspace root directory (for relative paths)
+        max_files: Maximum number of files to return
+        include_hidden: Whether to include hidden files (starting with .)
+        extensions: List of file extensions to include (e.g., ['py', 'js'])
+        respect_gitignore: Whether to respect .gitignore patterns
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - files: List of file entries with path, type, size
+        - directories: List of directory entries
+        - total_count: int (total items found)
+        - truncated: bool (True if max_files was reached)
+        - error: str (if success is False)
+        
+    Example:
+        >>> result = list_files("src", recursive=True, extensions=['py'])
+        >>> for f in result['files']:
+        ...     print(f['path'])
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Check if directory exists
+    if not os.path.isdir(abs_path):
+        return {
+            'success': False,
+            'error': f"Directory not found: {path}",
+            'files': [],
+            'directories': [],
+            'total_count': 0,
+        }
+    
+    # Load gitignore patterns
+    ignore_patterns = []
+    if respect_gitignore:
+        # Load from workspace root if available
+        if workspace:
+            ignore_patterns = load_gitignore_patterns(workspace)
+        # Also load from the target directory
+        ignore_patterns.extend(load_gitignore_patterns(abs_path))
+    
+    files = []
+    directories = []
+    total_count = 0
+    truncated = False
+    
+    try:
+        if recursive:
+            # Walk directory tree
+            for root, dirs, filenames in os.walk(abs_path):
+                # Filter hidden directories
+                if not include_hidden:
+                    dirs[:] = [d for d in dirs if not d.startswith('.')]
+                
+                # Filter ignored directories
+                if respect_gitignore:
+                    base_dir = workspace or abs_path
+                    dirs[:] = [
+                        d for d in dirs 
+                        if not should_ignore_path(
+                            os.path.join(root, d), 
+                            ignore_patterns, 
+                            base_dir, 
+                            is_directory=True
+                        )
+                    ]
+                
+                # Process files
+                for filename in filenames:
+                    if total_count >= max_files:
+                        truncated = True
+                        break
+                    
+                    # Skip hidden files
+                    if not include_hidden and filename.startswith('.'):
+                        continue
+                    
+                    file_path = os.path.join(root, filename)
+                    rel_path = os.path.relpath(file_path, abs_path)
+                    
+                    # Check gitignore
+                    if respect_gitignore:
+                        base_dir = workspace or abs_path
+                        if should_ignore_path(file_path, ignore_patterns, base_dir):
+                            continue
+                    
+                    # Filter by extension
+                    if extensions:
+                        ext = os.path.splitext(filename)[1].lstrip('.')
+                        if ext.lower() not in [e.lower() for e in extensions]:
+                            continue
+                    
+                    # Get file info
+                    try:
+                        stat = os.stat(file_path)
+                        files.append({
+                            'path': rel_path,
+                            'name': filename,
+                            'size': stat.st_size,
+                            'modified': stat.st_mtime,
+                        })
+                        total_count += 1
+                    except OSError:
+                        continue
+                
+                if truncated:
+                    break
+                
+                # Add directories
+                for dirname in dirs:
+                    if total_count >= max_files:
+                        truncated = True
+                        break
+                    
+                    dir_path = os.path.join(root, dirname)
+                    rel_path = os.path.relpath(dir_path, abs_path)
+                    
+                    directories.append({
+                        'path': rel_path,
+                        'name': dirname,
+                    })
+                    total_count += 1
+        else:
+            # List only top level
+            for entry in os.listdir(abs_path):
+                if total_count >= max_files:
+                    truncated = True
+                    break
+                
+                # Skip hidden files
+                if not include_hidden and entry.startswith('.'):
+                    continue
+                
+                entry_path = os.path.join(abs_path, entry)
+                
+                # Check gitignore
+                if respect_gitignore:
+                    base_dir = workspace or abs_path
+                    is_dir = os.path.isdir(entry_path)
+                    if should_ignore_path(entry_path, ignore_patterns, base_dir, is_dir):
+                        continue
+                
+                if os.path.isfile(entry_path):
+                    # Filter by extension
+                    if extensions:
+                        ext = os.path.splitext(entry)[1].lstrip('.')
+                        if ext.lower() not in [e.lower() for e in extensions]:
+                            continue
+                    
+                    try:
+                        stat = os.stat(entry_path)
+                        files.append({
+                            'path': entry,
+                            'name': entry,
+                            'size': stat.st_size,
+                            'modified': stat.st_mtime,
+                        })
+                        total_count += 1
+                    except OSError:
+                        continue
+                        
+                elif os.path.isdir(entry_path):
+                    directories.append({
+                        'path': entry,
+                        'name': entry,
+                    })
+                    total_count += 1
+        
+        # Sort results
+        files.sort(key=lambda x: x['path'])
+        directories.sort(key=lambda x: x['path'])
+        
+        return {
+            'success': True,
+            'files': files,
+            'directories': directories,
+            'total_count': total_count,
+            'truncated': truncated,
+            'path': path,
+        }
+        
+    except PermissionError:
+        return {
+            'success': False,
+            'error': f"Permission denied: {path}",
+            'files': [],
+            'directories': [],
+            'total_count': 0,
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error listing {path}: {str(e)}",
+            'files': [],
+            'directories': [],
+            'total_count': 0,
+        }
+
+
+def format_file_list(result: Dict[str, Any], show_size: bool = True) -> str:
+    """
+    Format a list_files result as a readable string.
+    
+    Args:
+        result: Result from list_files()
+        show_size: Whether to show file sizes
+        
+    Returns:
+        Formatted string representation
+    """
+    if not result['success']:
+        return f"Error: {result.get('error', 'Unknown error')}"
+    
+    lines = []
+    
+    # Add directories first
+    for d in result['directories']:
+        lines.append(f"📁 {d['path']}/")
+    
+    # Add files
+    for f in result['files']:
+        if show_size:
+            size = f.get('size', 0)
+            if size < 1024:
+                size_str = f"{size}B"
+            elif size < 1024 * 1024:
+                size_str = f"{size / 1024:.1f}KB"
+            else:
+                size_str = f"{size / (1024 * 1024):.1f}MB"
+            lines.append(f"📄 {f['path']} ({size_str})")
+        else:
+            lines.append(f"📄 {f['path']}")
+    
+    if result.get('truncated'):
+        lines.append(f"\n... (truncated, showing {result['total_count']} of more items)")
+    
+    return '\n'.join(lines)

src/praisonai/praisonai/code/tools/read_file.py

@@ -0,0 +1,206 @@
+"""
+Read File Tool for PraisonAI Code.
+
+Provides functionality to read file contents with optional line ranges
+and line number annotations.
+"""
+
+import os
+from typing import Optional, List, Dict, Any
+
+from ..utils.file_utils import (
+    add_line_numbers,
+    is_binary_file,
+    file_exists,
+    is_path_within_directory,
+)
+
+
+def read_file(
+    path: str,
+    start_line: Optional[int] = None,
+    end_line: Optional[int] = None,
+    add_line_nums: bool = True,
+    workspace: Optional[str] = None,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Read file contents with optional line range and line numbers.
+    
+    This tool reads the contents of a file, optionally limiting to specific
+    line ranges and adding line number annotations for AI context.
+    
+    Args:
+        path: Path to the file (absolute or relative to workspace)
+        start_line: First line to read (1-indexed, inclusive)
+        end_line: Last line to read (1-indexed, inclusive)
+        add_line_nums: Whether to add line numbers to output
+        workspace: Workspace root directory (for relative paths)
+        encoding: File encoding (default: utf-8)
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - content: str (file content, possibly with line numbers)
+        - total_lines: int (total lines in file)
+        - start_line: int (actual start line read)
+        - end_line: int (actual end line read)
+        - error: str (if success is False)
+        
+    Example:
+        >>> result = read_file("src/main.py", start_line=1, end_line=50)
+        >>> if result['success']:
+        ...     print(result['content'])
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Security check - ensure path is within workspace if specified
+    if workspace:
+        if not is_path_within_directory(abs_path, workspace):
+            return {
+                'success': False,
+                'error': f"Path '{path}' is outside the workspace",
+                'content': None,
+                'total_lines': 0,
+            }
+    
+    # Check if file exists
+    if not file_exists(abs_path):
+        return {
+            'success': False,
+            'error': f"File not found: {path}",
+            'content': None,
+            'total_lines': 0,
+        }
+    
+    # Check if binary
+    if is_binary_file(abs_path):
+        return {
+            'success': False,
+            'error': f"Cannot read binary file: {path}",
+            'content': None,
+            'total_lines': 0,
+        }
+    
+    try:
+        # Read file content
+        with open(abs_path, 'r', encoding=encoding, errors='replace') as f:
+            lines = f.readlines()
+        
+        total_lines = len(lines)
+        
+        # Determine line range
+        actual_start = start_line if start_line and start_line > 0 else 1
+        actual_end = end_line if end_line and end_line > 0 else total_lines
+        
+        # Clamp to valid range
+        actual_start = max(1, min(actual_start, total_lines))
+        actual_end = max(actual_start, min(actual_end, total_lines))
+        
+        # Extract lines (convert to 0-indexed)
+        selected_lines = lines[actual_start - 1:actual_end]
+        content = ''.join(selected_lines)
+        
+        # Remove trailing newline if present
+        if content.endswith('\n'):
+            content = content[:-1]
+        
+        # Add line numbers if requested
+        if add_line_nums:
+            content = add_line_numbers(content, actual_start)
+        
+        return {
+            'success': True,
+            'content': content,
+            'total_lines': total_lines,
+            'start_line': actual_start,
+            'end_line': actual_end,
+            'path': path,
+        }
+        
+    except PermissionError:
+        return {
+            'success': False,
+            'error': f"Permission denied: {path}",
+            'content': None,
+            'total_lines': 0,
+        }
+    except UnicodeDecodeError as e:
+        return {
+            'success': False,
+            'error': f"Encoding error reading {path}: {e}",
+            'content': None,
+            'total_lines': 0,
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error reading {path}: {str(e)}",
+            'content': None,
+            'total_lines': 0,
+        }
+
+
+def read_multiple_files(
+    files: List[Dict[str, Any]],
+    workspace: Optional[str] = None,
+    add_line_nums: bool = True,
+) -> Dict[str, Any]:
+    """
+    Read multiple files in a single operation.
+    
+    Args:
+        files: List of file specifications, each with:
+            - path: str (required)
+            - start_line: int (optional)
+            - end_line: int (optional)
+        workspace: Workspace root directory
+        add_line_nums: Whether to add line numbers
+        
+    Returns:
+        Dictionary with:
+        - success: bool (True if all files read successfully)
+        - results: List of individual file results
+        - errors: List of error messages
+        
+    Example:
+        >>> files = [
+        ...     {'path': 'src/main.py', 'start_line': 1, 'end_line': 50},
+        ...     {'path': 'src/utils.py'},
+        ... ]
+        >>> result = read_multiple_files(files, workspace='/project')
+    """
+    results = []
+    errors = []
+    all_success = True
+    
+    for file_spec in files:
+        path = file_spec.get('path')
+        if not path:
+            errors.append("Missing 'path' in file specification")
+            all_success = False
+            continue
+        
+        result = read_file(
+            path=path,
+            start_line=file_spec.get('start_line'),
+            end_line=file_spec.get('end_line'),
+            add_line_nums=add_line_nums,
+            workspace=workspace,
+        )
+        
+        results.append(result)
+        
+        if not result['success']:
+            all_success = False
+            errors.append(result.get('error', f"Failed to read {path}"))
+    
+    return {
+        'success': all_success,
+        'results': results,
+        'errors': errors if errors else None,
+    }

src/praisonai/praisonai/code/tools/search_replace.py

@@ -0,0 +1,248 @@
+"""
+Search and Replace Tool for PraisonAI Code.
+
+Provides functionality to perform multiple search/replace operations
+on a file in a single operation.
+"""
+
+import os
+import re
+from typing import Optional, List, Dict, Any
+
+from ..utils.file_utils import (
+    file_exists,
+    is_path_within_directory,
+)
+
+
+def search_replace(
+    path: str,
+    operations: List[Dict[str, Any]],
+    workspace: Optional[str] = None,
+    backup: bool = False,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Perform multiple search/replace operations on a file.
+    
+    This tool allows precise modifications to file content by performing
+    multiple search and replace operations in sequence.
+    
+    Args:
+        path: Path to the file (absolute or relative to workspace)
+        operations: List of operations, each with:
+            - search: str (text to find)
+            - replace: str (text to replace with)
+            - is_regex: bool (optional, treat search as regex)
+            - count: int (optional, max replacements, -1 for all)
+        workspace: Workspace root directory
+        backup: Whether to create a backup before modifying
+        encoding: File encoding
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - path: str
+        - operations_applied: int
+        - total_replacements: int
+        - failed_operations: list
+        - error: str (if success is False)
+        
+    Example:
+        >>> operations = [
+        ...     {'search': 'old_name', 'replace': 'new_name'},
+        ...     {'search': r'def (\\w+)\\(', 'replace': r'def renamed_\\\\1(', 'is_regex': True},
+        ... ]
+        >>> result = search_replace("src/main.py", operations)
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Security check
+    if workspace:
+        if not is_path_within_directory(abs_path, workspace):
+            return {
+                'success': False,
+                'error': f"Path '{path}' is outside the workspace",
+                'path': path,
+                'operations_applied': 0,
+                'total_replacements': 0,
+            }
+    
+    # Check if file exists
+    if not file_exists(abs_path):
+        return {
+            'success': False,
+            'error': f"File not found: {path}",
+            'path': path,
+            'operations_applied': 0,
+            'total_replacements': 0,
+        }
+    
+    if not operations:
+        return {
+            'success': False,
+            'error': "No operations provided",
+            'path': path,
+            'operations_applied': 0,
+            'total_replacements': 0,
+        }
+    
+    try:
+        # Read original content
+        with open(abs_path, 'r', encoding=encoding, errors='replace') as f:
+            content = f.read()
+        
+        original_content = content
+        operations_applied = 0
+        total_replacements = 0
+        failed_operations = []
+        
+        # Apply each operation
+        for i, op in enumerate(operations):
+            search = op.get('search')
+            replace = op.get('replace', '')
+            is_regex = op.get('is_regex', False)
+            count = op.get('count', -1)  # -1 means replace all
+            
+            if not search:
+                failed_operations.append({
+                    'index': i,
+                    'error': "Missing 'search' field",
+                })
+                continue
+            
+            try:
+                if is_regex:
+                    # Regex replacement
+                    if count == -1:
+                        new_content, num_subs = re.subn(search, replace, content)
+                    else:
+                        new_content, num_subs = re.subn(search, replace, content, count=count)
+                else:
+                    # Literal replacement
+                    if count == -1:
+                        num_subs = content.count(search)
+                        new_content = content.replace(search, replace)
+                    else:
+                        num_subs = min(content.count(search), count)
+                        new_content = content
+                        for _ in range(count):
+                            if search in new_content:
+                                new_content = new_content.replace(search, replace, 1)
+                            else:
+                                break
+                
+                if num_subs > 0:
+                    content = new_content
+                    operations_applied += 1
+                    total_replacements += num_subs
+                else:
+                    failed_operations.append({
+                        'index': i,
+                        'search': search[:50],
+                        'error': "No matches found",
+                    })
+                    
+            except re.error as e:
+                failed_operations.append({
+                    'index': i,
+                    'search': search[:50],
+                    'error': f"Invalid regex: {str(e)}",
+                })
+        
+        # Check if any changes were made
+        if content == original_content:
+            return {
+                'success': False,
+                'error': "No changes were made - no matches found",
+                'path': path,
+                'operations_applied': 0,
+                'total_replacements': 0,
+                'failed_operations': failed_operations,
+            }
+        
+        # Create backup if requested
+        backup_path = None
+        if backup:
+            import time
+            timestamp = int(time.time())
+            backup_path = f"{abs_path}.backup.{timestamp}"
+            with open(backup_path, 'w', encoding=encoding) as f:
+                f.write(original_content)
+        
+        # Write the modified content
+        with open(abs_path, 'w', encoding=encoding) as f:
+            f.write(content)
+        
+        return {
+            'success': True,
+            'path': path,
+            'absolute_path': abs_path,
+            'operations_applied': operations_applied,
+            'total_replacements': total_replacements,
+            'failed_operations': failed_operations if failed_operations else None,
+            'backup_path': backup_path,
+        }
+        
+    except PermissionError:
+        return {
+            'success': False,
+            'error': f"Permission denied: {path}",
+            'path': path,
+            'operations_applied': 0,
+            'total_replacements': 0,
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error performing search/replace on {path}: {str(e)}",
+            'path': path,
+            'operations_applied': 0,
+            'total_replacements': 0,
+        }
+
+
+def simple_replace(
+    path: str,
+    search: str,
+    replace: str,
+    workspace: Optional[str] = None,
+    is_regex: bool = False,
+    count: int = -1,
+    backup: bool = False,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Perform a single search/replace operation on a file.
+    
+    Convenience wrapper around search_replace for single operations.
+    
+    Args:
+        path: Path to the file
+        search: Text to find
+        replace: Text to replace with
+        workspace: Workspace root directory
+        is_regex: Whether search is a regex pattern
+        count: Max replacements (-1 for all)
+        backup: Whether to create a backup
+        encoding: File encoding
+        
+    Returns:
+        Dictionary with success status and details
+    """
+    return search_replace(
+        path=path,
+        operations=[{
+            'search': search,
+            'replace': replace,
+            'is_regex': is_regex,
+            'count': count,
+        }],
+        workspace=workspace,
+        backup=backup,
+        encoding=encoding,
+    )

src/praisonai/praisonai/code/tools/write_file.py

@@ -0,0 +1,217 @@
+"""
+Write File Tool for PraisonAI Code.
+
+Provides functionality to create or overwrite files with content.
+"""
+
+import os
+from typing import Optional, Dict, Any
+
+from ..utils.file_utils import (
+    file_exists,
+    create_directories_for_file,
+    is_path_within_directory,
+    detect_line_ending,
+    normalize_line_endings,
+)
+from ..utils.text_utils import (
+    unescape_html_entities,
+    strip_markdown_code_fences,
+)
+
+
+def write_file(
+    path: str,
+    content: str,
+    workspace: Optional[str] = None,
+    create_directories: bool = True,
+    backup: bool = False,
+    strip_code_fences: bool = True,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Write content to a file, creating it if it doesn't exist.
+    
+    This tool writes content to a file, optionally creating parent
+    directories and handling common AI output artifacts like code fences.
+    
+    Args:
+        path: Path to the file (absolute or relative to workspace)
+        content: Content to write to the file
+        workspace: Workspace root directory (for relative paths)
+        create_directories: Whether to create parent directories
+        backup: Whether to create a backup of existing files
+        strip_code_fences: Whether to strip markdown code fences
+        encoding: File encoding (default: utf-8)
+        
+    Returns:
+        Dictionary with:
+        - success: bool
+        - path: str (the path written to)
+        - created: bool (True if file was created, False if overwritten)
+        - backup_path: str (if backup was created)
+        - error: str (if success is False)
+        
+    Example:
+        >>> result = write_file("src/new_file.py", "print('hello')")
+        >>> if result['success']:
+        ...     print(f"Wrote to {result['path']}")
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Security check - ensure path is within workspace if specified
+    if workspace:
+        if not is_path_within_directory(abs_path, workspace):
+            return {
+                'success': False,
+                'error': f"Path '{path}' is outside the workspace",
+                'path': path,
+            }
+    
+    # Process content
+    processed_content = content
+    
+    # Strip markdown code fences if requested
+    if strip_code_fences:
+        processed_content = strip_markdown_code_fences(processed_content)
+    
+    # Unescape HTML entities (common in AI output)
+    processed_content = unescape_html_entities(processed_content)
+    
+    # Check if file exists
+    file_existed = file_exists(abs_path)
+    backup_path = None
+    
+    try:
+        # Create backup if requested and file exists
+        if backup and file_existed:
+            import time
+            timestamp = int(time.time())
+            backup_path = f"{abs_path}.backup.{timestamp}"
+            with open(abs_path, 'r', encoding=encoding, errors='replace') as src:
+                with open(backup_path, 'w', encoding=encoding) as dst:
+                    dst.write(src.read())
+        
+        # Create parent directories if needed
+        if create_directories:
+            if not create_directories_for_file(abs_path):
+                return {
+                    'success': False,
+                    'error': f"Failed to create directories for {path}",
+                    'path': path,
+                }
+        
+        # Preserve line endings if file exists
+        if file_existed:
+            try:
+                with open(abs_path, 'r', encoding=encoding, errors='replace') as f:
+                    original_content = f.read()
+                original_line_ending = detect_line_ending(original_content)
+                processed_content = normalize_line_endings(processed_content, original_line_ending)
+            except Exception:
+                pass  # If we can't read, just use the content as-is
+        
+        # Write the file
+        with open(abs_path, 'w', encoding=encoding) as f:
+            f.write(processed_content)
+        
+        return {
+            'success': True,
+            'path': path,
+            'absolute_path': abs_path,
+            'created': not file_existed,
+            'backup_path': backup_path,
+            'bytes_written': len(processed_content.encode(encoding)),
+        }
+        
+    except PermissionError:
+        return {
+            'success': False,
+            'error': f"Permission denied: {path}",
+            'path': path,
+        }
+    except OSError as e:
+        return {
+            'success': False,
+            'error': f"OS error writing {path}: {str(e)}",
+            'path': path,
+        }
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error writing {path}: {str(e)}",
+            'path': path,
+        }
+
+
+def append_to_file(
+    path: str,
+    content: str,
+    workspace: Optional[str] = None,
+    create_if_missing: bool = True,
+    encoding: str = 'utf-8',
+) -> Dict[str, Any]:
+    """
+    Append content to an existing file.
+    
+    Args:
+        path: Path to the file
+        content: Content to append
+        workspace: Workspace root directory
+        create_if_missing: Create file if it doesn't exist
+        encoding: File encoding
+        
+    Returns:
+        Dictionary with success status and details
+    """
+    # Resolve path
+    if workspace and not os.path.isabs(path):
+        abs_path = os.path.abspath(os.path.join(workspace, path))
+    else:
+        abs_path = os.path.abspath(path)
+    
+    # Security check
+    if workspace:
+        if not is_path_within_directory(abs_path, workspace):
+            return {
+                'success': False,
+                'error': f"Path '{path}' is outside the workspace",
+                'path': path,
+            }
+    
+    file_existed = file_exists(abs_path)
+    
+    if not file_existed and not create_if_missing:
+        return {
+            'success': False,
+            'error': f"File not found: {path}",
+            'path': path,
+        }
+    
+    try:
+        # Create directories if needed
+        if not file_existed:
+            create_directories_for_file(abs_path)
+        
+        # Append to file
+        with open(abs_path, 'a', encoding=encoding) as f:
+            f.write(content)
+        
+        return {
+            'success': True,
+            'path': path,
+            'absolute_path': abs_path,
+            'created': not file_existed,
+            'bytes_appended': len(content.encode(encoding)),
+        }
+        
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Error appending to {path}: {str(e)}",
+            'path': path,
+        }

src/praisonai/praisonai/deploy.py

@@ -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.39 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.41 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.3.39"
+__version__ = "2.3.41"

src/praisonai/server.json

@@ -0,0 +1,21 @@
+{
+  "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json",
+  "name": "io.github.MervinPraison/praisonai",
+  "title": "PraisonAI",
+  "description": "AI Agents Framework with Self Reflection and MCP support",
+  "repository": {
+    "url": "https://github.com/MervinPraison/PraisonAI",
+    "source": "github"
+  },
+  "version": "2.3.41",
+  "packages": [
+    {
+      "registryType": "pypi",
+      "identifier": "praisonai",
+      "version": "2.3.41",
+      "transport": {
+        "type": "stdio"
+      }
+    }
+  ]
+}

src/praisonai/tests/unit/code/__init__.py

@@ -0,0 +1 @@
+"""Tests for praisonai.code module."""

src/praisonai/tests/unit/code/test_agent_integration.py

@@ -0,0 +1,235 @@
+"""
+Integration tests for praisonai.code agent tools.
+"""
+
+import os
+import tempfile
+
+
+class TestAgentTools:
+    """Tests for agent-compatible tool wrappers."""
+    
+    def test_set_and_get_workspace(self):
+        """Test setting and getting workspace."""
+        from praisonai.code import set_workspace, get_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            assert get_workspace() == temp_dir
+    
+    def test_code_read_file(self):
+        """Test code_read_file agent tool."""
+        from praisonai.code import code_read_file, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            # Create test file
+            file_path = os.path.join(temp_dir, "test.py")
+            with open(file_path, 'w') as f:
+                f.write("def hello():\n    print('hello')\n")
+            
+            result = code_read_file("test.py")
+            
+            assert "File: test.py" in result
+            assert "def hello():" in result
+            assert "1 |" in result  # Line numbers
+    
+    def test_code_write_file(self):
+        """Test code_write_file agent tool."""
+        from praisonai.code import code_write_file, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            result = code_write_file("new_file.py", "print('hello')")
+            
+            assert "Created" in result
+            assert os.path.exists(os.path.join(temp_dir, "new_file.py"))
+    
+    def test_code_list_files(self):
+        """Test code_list_files agent tool."""
+        from praisonai.code import code_list_files, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            # Create some files
+            open(os.path.join(temp_dir, "file1.py"), 'w').close()
+            open(os.path.join(temp_dir, "file2.py"), 'w').close()
+            os.makedirs(os.path.join(temp_dir, "subdir"))
+            
+            result = code_list_files(".")
+            
+            assert "file1.py" in result
+            assert "file2.py" in result
+            assert "subdir" in result
+    
+    def test_code_apply_diff(self):
+        """Test code_apply_diff agent tool."""
+        from praisonai.code import code_apply_diff, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            # Create test file
+            file_path = os.path.join(temp_dir, "test.py")
+            with open(file_path, 'w') as f:
+                f.write("def old_function():\n    return 'old'\n")
+            
+            diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def old_function():
+    return 'old'
+=======
+def new_function():
+    return 'new'
+>>>>>>> REPLACE"""
+            
+            result = code_apply_diff("test.py", diff)
+            
+            assert "Successfully applied" in result
+            
+            # Verify file was modified
+            with open(file_path, 'r') as f:
+                content = f.read()
+            assert "def new_function():" in content
+    
+    def test_code_search_replace(self):
+        """Test code_search_replace agent tool."""
+        from praisonai.code import code_search_replace, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            # Create test file
+            file_path = os.path.join(temp_dir, "test.py")
+            with open(file_path, 'w') as f:
+                f.write("old_name = 'value'\nprint(old_name)\n")
+            
+            result = code_search_replace("test.py", "old_name", "new_name")
+            
+            assert "Replaced" in result
+            assert "2" in result  # 2 occurrences
+    
+    def test_code_execute_command(self):
+        """Test code_execute_command agent tool."""
+        from praisonai.code import code_execute_command, set_workspace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            result = code_execute_command("echo 'Hello from test'")
+            
+            assert "completed successfully" in result
+            assert "Hello" in result
+    
+    def test_code_tools_list(self):
+        """Test CODE_TOOLS list contains all tools."""
+        from praisonai.code import CODE_TOOLS
+        
+        assert len(CODE_TOOLS) == 6
+        
+        tool_names = [t.__name__ for t in CODE_TOOLS]
+        assert "code_read_file" in tool_names
+        assert "code_write_file" in tool_names
+        assert "code_list_files" in tool_names
+        assert "code_apply_diff" in tool_names
+        assert "code_search_replace" in tool_names
+        assert "code_execute_command" in tool_names
+
+
+class TestDiffHelpers:
+    """Tests for diff helper functions."""
+    
+    def test_create_diff_block(self):
+        """Test creating a diff block."""
+        from praisonai.code import create_diff_block
+        
+        diff = create_diff_block(
+            search_content="old content",
+            replace_content="new content",
+            start_line=10,
+        )
+        
+        assert "<<<<<<< SEARCH" in diff
+        assert ":start_line:10" in diff
+        assert "old content" in diff
+        assert "=======" in diff
+        assert "new content" in diff
+        assert ">>>>>>> REPLACE" in diff
+    
+    def test_create_multi_diff(self):
+        """Test creating multiple diff blocks."""
+        from praisonai.code import create_multi_diff
+        
+        blocks = [
+            {'search': 'old1', 'replace': 'new1', 'start_line': 5},
+            {'search': 'old2', 'replace': 'new2', 'start_line': 20},
+        ]
+        
+        diff = create_multi_diff(blocks)
+        
+        # Should contain two SEARCH/REPLACE blocks
+        assert diff.count("<<<<<<< SEARCH") == 2
+        assert diff.count(">>>>>>> REPLACE") == 2
+        assert "old1" in diff
+        assert "new1" in diff
+        assert "old2" in diff
+        assert "new2" in diff
+
+
+class TestEndToEndWorkflow:
+    """End-to-end workflow tests."""
+    
+    def test_read_modify_write_workflow(self):
+        """Test a complete read-modify-write workflow."""
+        from praisonai.code import (
+            set_workspace, code_read_file, code_apply_diff, code_execute_command
+        )
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            set_workspace(temp_dir)
+            
+            # Create a Python file
+            file_path = os.path.join(temp_dir, "calculator.py")
+            with open(file_path, 'w') as f:
+                f.write('''def add(a, b):
+    return a + b
+
+def subtract(a, b):
+    return a - b
+
+if __name__ == "__main__":
+    print(add(2, 3))
+''')
+            
+            # Read the file
+            content = code_read_file("calculator.py")
+            assert "def add(a, b):" in content
+            
+            # Apply a diff to add a multiply function
+            diff = '''<<<<<<< SEARCH
+:start_line:4
+-------
+def subtract(a, b):
+    return a - b
+=======
+def subtract(a, b):
+    return a - b
+
+def multiply(a, b):
+    return a * b
+>>>>>>> REPLACE'''
+            
+            result = code_apply_diff("calculator.py", diff)
+            assert "Successfully applied" in result
+            
+            # Verify the change
+            content = code_read_file("calculator.py")
+            assert "def multiply(a, b):" in content
+            
+            # Run the file to verify it's valid Python
+            result = code_execute_command("python calculator.py")
+            assert "5" in result  # add(2, 3) = 5

src/praisonai/tests/unit/code/test_diff_strategy.py

@@ -0,0 +1,341 @@
+"""
+Tests for praisonai.code.diff.diff_strategy module.
+"""
+
+import os
+import tempfile
+
+
+class TestValidateDiffFormat:
+    """Tests for validate_diff_format function."""
+    
+    def test_valid_single_block(self):
+        """Test valid single SEARCH/REPLACE block."""
+        from praisonai.code.diff.diff_strategy import validate_diff_format
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+old content
+=======
+new content
+>>>>>>> REPLACE"""
+        
+        is_valid, error = validate_diff_format(diff)
+        assert is_valid is True
+        assert error is None
+    
+    def test_valid_multiple_blocks(self):
+        """Test valid multiple SEARCH/REPLACE blocks."""
+        from praisonai.code.diff.diff_strategy import validate_diff_format
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+old1
+=======
+new1
+>>>>>>> REPLACE
+
+<<<<<<< SEARCH
+:start_line:10
+-------
+old2
+=======
+new2
+>>>>>>> REPLACE"""
+        
+        is_valid, error = validate_diff_format(diff)
+        assert is_valid is True
+    
+    def test_missing_replace_marker(self):
+        """Test missing REPLACE marker."""
+        from praisonai.code.diff.diff_strategy import validate_diff_format
+        
+        diff = """<<<<<<< SEARCH
+old content
+=======
+new content"""
+        
+        is_valid, error = validate_diff_format(diff)
+        assert is_valid is False
+        assert ">>>>>>> REPLACE" in error
+    
+    def test_missing_separator(self):
+        """Test missing separator."""
+        from praisonai.code.diff.diff_strategy import validate_diff_format
+        
+        diff = """<<<<<<< SEARCH
+old content
+>>>>>>> REPLACE"""
+        
+        is_valid, error = validate_diff_format(diff)
+        assert is_valid is False
+
+
+class TestParseDiffBlocks:
+    """Tests for parse_diff_blocks function."""
+    
+    def test_parse_single_block(self):
+        """Test parsing single block."""
+        from praisonai.code.diff.diff_strategy import parse_diff_blocks
+        
+        diff = """<<<<<<< SEARCH
+:start_line:5
+-------
+def old():
+    pass
+=======
+def new():
+    return True
+>>>>>>> REPLACE"""
+        
+        blocks, error = parse_diff_blocks(diff)
+        
+        assert error is None
+        assert len(blocks) == 1
+        assert blocks[0].start_line == 5
+        assert "def old():" in blocks[0].search_content
+        assert "def new():" in blocks[0].replace_content
+    
+    def test_parse_multiple_blocks(self):
+        """Test parsing multiple blocks."""
+        from praisonai.code.diff.diff_strategy import parse_diff_blocks
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+first
+=======
+first_new
+>>>>>>> REPLACE
+
+<<<<<<< SEARCH
+:start_line:10
+-------
+second
+=======
+second_new
+>>>>>>> REPLACE"""
+        
+        blocks, error = parse_diff_blocks(diff)
+        
+        assert error is None
+        assert len(blocks) == 2
+        # Should be sorted by start_line
+        assert blocks[0].start_line == 1
+        assert blocks[1].start_line == 10
+    
+    def test_parse_block_without_start_line(self):
+        """Test parsing block without start_line hint."""
+        from praisonai.code.diff.diff_strategy import parse_diff_blocks
+        
+        diff = """<<<<<<< SEARCH
+-------
+old content
+=======
+new content
+>>>>>>> REPLACE"""
+        
+        blocks, error = parse_diff_blocks(diff)
+        
+        assert error is None
+        assert len(blocks) == 1
+        assert blocks[0].start_line is None
+
+
+class TestApplySearchReplaceDiff:
+    """Tests for apply_search_replace_diff function."""
+    
+    def test_exact_match_replacement(self):
+        """Test exact match replacement."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """def hello():
+    print("hello")
+
+def world():
+    print("world")"""
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def hello():
+    print("hello")
+=======
+def hello():
+    print("Hello, World!")
+>>>>>>> REPLACE"""
+        
+        result = apply_search_replace_diff(original, diff)
+        
+        assert result.success is True
+        assert result.applied_count == 1
+        assert 'print("Hello, World!")' in result.content
+    
+    def test_multiple_replacements(self):
+        """Test multiple replacements in one diff."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """line1
+line2
+line3
+line4
+line5"""
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+line1
+=======
+LINE1
+>>>>>>> REPLACE
+
+<<<<<<< SEARCH
+:start_line:5
+-------
+line5
+=======
+LINE5
+>>>>>>> REPLACE"""
+        
+        result = apply_search_replace_diff(original, diff)
+        
+        assert result.success is True
+        assert result.applied_count == 2
+        assert "LINE1" in result.content
+        assert "LINE5" in result.content
+    
+    def test_fuzzy_matching(self):
+        """Test fuzzy matching with threshold."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """def hello():
+    print("hello")"""
+        
+        # Slightly different content (extra space)
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def hello():
+    print("hello")
+=======
+def hello():
+    print("goodbye")
+>>>>>>> REPLACE"""
+        
+        # With exact matching, this should work
+        result = apply_search_replace_diff(original, diff, fuzzy_threshold=1.0)
+        
+        assert result.success is True
+        assert 'print("goodbye")' in result.content
+    
+    def test_no_match_found(self):
+        """Test when no match is found."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """def hello():
+    print("hello")"""
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def nonexistent():
+    pass
+=======
+def new():
+    pass
+>>>>>>> REPLACE"""
+        
+        result = apply_search_replace_diff(original, diff)
+        
+        assert result.success is False
+        assert len(result.failed_blocks) > 0
+    
+    def test_indentation_preservation(self):
+        """Test that indentation is preserved."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """class MyClass:
+    def method(self):
+        pass"""
+        
+        diff = """<<<<<<< SEARCH
+:start_line:2
+-------
+    def method(self):
+        pass
+=======
+    def method(self):
+        return True
+>>>>>>> REPLACE"""
+        
+        result = apply_search_replace_diff(original, diff)
+        
+        assert result.success is True
+        # Check indentation is preserved
+        lines = result.content.split('\n')
+        assert lines[1].startswith('    ')  # Method should be indented
+        assert lines[2].startswith('        ')  # Body should be double indented
+    
+    def test_identical_search_replace_error(self):
+        """Test that identical search/replace is rejected."""
+        from praisonai.code.diff.diff_strategy import apply_search_replace_diff
+        
+        original = """def hello():
+    pass"""
+        
+        diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def hello():
+    pass
+=======
+def hello():
+    pass
+>>>>>>> REPLACE"""
+        
+        result = apply_search_replace_diff(original, diff)
+        
+        # Should fail because search and replace are identical
+        assert result.success is False
+
+
+class TestDiffIntegration:
+    """Integration tests for diff operations."""
+    
+    def test_apply_diff_to_file(self):
+        """Test applying diff to an actual file."""
+        from praisonai.code.tools.apply_diff import apply_diff
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create test file
+            file_path = os.path.join(temp_dir, "test.py")
+            with open(file_path, 'w') as f:
+                f.write("""def old_function():
+    return "old"
+
+def another():
+    pass""")
+            
+            diff = """<<<<<<< SEARCH
+:start_line:1
+-------
+def old_function():
+    return "old"
+=======
+def new_function():
+    return "new"
+>>>>>>> REPLACE"""
+            
+            result = apply_diff(file_path, diff)
+            
+            assert result['success'] is True
+            assert result['applied_count'] == 1
+            
+            # Verify file was modified
+            with open(file_path, 'r') as f:
+                content = f.read()
+            
+            assert "def new_function():" in content
+            assert 'return "new"' in content

src/praisonai/tests/unit/code/test_file_utils.py

@@ -0,0 +1,228 @@
+"""
+Tests for praisonai.code.utils.file_utils module.
+"""
+
+import os
+import tempfile
+
+
+class TestAddLineNumbers:
+    """Tests for add_line_numbers function."""
+    
+    def test_add_line_numbers_basic(self):
+        """Test basic line number addition."""
+        from praisonai.code.utils.file_utils import add_line_numbers
+        
+        content = "line1\nline2\nline3"
+        result = add_line_numbers(content)
+        
+        assert "1 | line1" in result
+        assert "2 | line2" in result
+        assert "3 | line3" in result
+    
+    def test_add_line_numbers_custom_start(self):
+        """Test line numbers with custom start."""
+        from praisonai.code.utils.file_utils import add_line_numbers
+        
+        content = "line1\nline2"
+        result = add_line_numbers(content, start_line=10)
+        
+        assert "10 | line1" in result
+        assert "11 | line2" in result
+    
+    def test_add_line_numbers_empty(self):
+        """Test with empty content."""
+        from praisonai.code.utils.file_utils import add_line_numbers
+        
+        result = add_line_numbers("")
+        assert result == ""
+    
+    def test_add_line_numbers_single_line(self):
+        """Test with single line."""
+        from praisonai.code.utils.file_utils import add_line_numbers
+        
+        result = add_line_numbers("hello")
+        assert "1 | hello" in result
+
+
+class TestStripLineNumbers:
+    """Tests for strip_line_numbers function."""
+    
+    def test_strip_line_numbers_basic(self):
+        """Test basic line number stripping."""
+        from praisonai.code.utils.file_utils import strip_line_numbers
+        
+        content = "  1 | line1\n  2 | line2"
+        result = strip_line_numbers(content)
+        
+        assert result == "line1\nline2"
+    
+    def test_strip_line_numbers_no_numbers(self):
+        """Test with content that has no line numbers."""
+        from praisonai.code.utils.file_utils import strip_line_numbers
+        
+        content = "line1\nline2"
+        result = strip_line_numbers(content)
+        
+        assert result == "line1\nline2"
+    
+    def test_strip_line_numbers_tab_separator(self):
+        """Test with tab separator."""
+        from praisonai.code.utils.file_utils import strip_line_numbers
+        
+        content = "1\tline1\n2\tline2"
+        result = strip_line_numbers(content)
+        
+        assert result == "line1\nline2"
+
+
+class TestEveryLineHasLineNumbers:
+    """Tests for every_line_has_line_numbers function."""
+    
+    def test_all_lines_have_numbers(self):
+        """Test when all lines have numbers."""
+        from praisonai.code.utils.file_utils import every_line_has_line_numbers
+        
+        content = "  1 | line1\n  2 | line2"
+        assert every_line_has_line_numbers(content) is True
+    
+    def test_no_lines_have_numbers(self):
+        """Test when no lines have numbers."""
+        from praisonai.code.utils.file_utils import every_line_has_line_numbers
+        
+        content = "line1\nline2"
+        assert every_line_has_line_numbers(content) is False
+    
+    def test_mixed_lines(self):
+        """Test when some lines have numbers."""
+        from praisonai.code.utils.file_utils import every_line_has_line_numbers
+        
+        content = "  1 | line1\nline2"
+        assert every_line_has_line_numbers(content) is False
+    
+    def test_empty_content(self):
+        """Test with empty content."""
+        from praisonai.code.utils.file_utils import every_line_has_line_numbers
+        
+        assert every_line_has_line_numbers("") is False
+
+
+class TestNormalizeLineEndings:
+    """Tests for normalize_line_endings function."""
+    
+    def test_normalize_crlf_to_lf(self):
+        """Test converting CRLF to LF."""
+        from praisonai.code.utils.file_utils import normalize_line_endings
+        
+        content = "line1\r\nline2\r\n"
+        result = normalize_line_endings(content, '\n')
+        
+        assert result == "line1\nline2\n"
+        assert '\r' not in result
+    
+    def test_normalize_lf_to_crlf(self):
+        """Test converting LF to CRLF."""
+        from praisonai.code.utils.file_utils import normalize_line_endings
+        
+        content = "line1\nline2\n"
+        result = normalize_line_endings(content, '\r\n')
+        
+        assert result == "line1\r\nline2\r\n"
+
+
+class TestIsBinaryFile:
+    """Tests for is_binary_file function."""
+    
+    def test_text_file(self):
+        """Test with a text file."""
+        from praisonai.code.utils.file_utils import is_binary_file
+        
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
+            f.write("Hello, world!")
+            temp_path = f.name
+        
+        try:
+            assert is_binary_file(temp_path) is False
+        finally:
+            os.unlink(temp_path)
+    
+    def test_binary_file(self):
+        """Test with a binary file."""
+        from praisonai.code.utils.file_utils import is_binary_file
+        
+        with tempfile.NamedTemporaryFile(mode='wb', suffix='.bin', delete=False) as f:
+            f.write(b'\x00\x01\x02\x03\x04\x05')
+            temp_path = f.name
+        
+        try:
+            assert is_binary_file(temp_path) is True
+        finally:
+            os.unlink(temp_path)
+
+
+class TestFileExists:
+    """Tests for file_exists function."""
+    
+    def test_existing_file(self):
+        """Test with existing file."""
+        from praisonai.code.utils.file_utils import file_exists
+        
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            temp_path = f.name
+        
+        try:
+            assert file_exists(temp_path) is True
+        finally:
+            os.unlink(temp_path)
+    
+    def test_nonexistent_file(self):
+        """Test with nonexistent file."""
+        from praisonai.code.utils.file_utils import file_exists
+        
+        assert file_exists("/nonexistent/path/file.txt") is False
+    
+    def test_directory(self):
+        """Test with directory (should return False)."""
+        from praisonai.code.utils.file_utils import file_exists
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            assert file_exists(temp_dir) is False
+
+
+class TestIsPathWithinDirectory:
+    """Tests for is_path_within_directory function."""
+    
+    def test_path_within(self):
+        """Test path within directory."""
+        from praisonai.code.utils.file_utils import is_path_within_directory
+        
+        assert is_path_within_directory("/home/user/project/file.py", "/home/user/project") is True
+    
+    def test_path_outside(self):
+        """Test path outside directory."""
+        from praisonai.code.utils.file_utils import is_path_within_directory
+        
+        assert is_path_within_directory("/home/other/file.py", "/home/user/project") is False
+    
+    def test_path_traversal(self):
+        """Test path traversal attempt."""
+        from praisonai.code.utils.file_utils import is_path_within_directory
+        
+        # This should resolve to /home/other, which is outside /home/user/project
+        assert is_path_within_directory("/home/user/project/../other/file.py", "/home/user/project") is False
+
+
+class TestCreateDirectoriesForFile:
+    """Tests for create_directories_for_file function."""
+    
+    def test_create_directories(self):
+        """Test creating directories for a file path."""
+        from praisonai.code.utils.file_utils import create_directories_for_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "subdir1", "subdir2", "file.txt")
+            
+            result = create_directories_for_file(file_path)
+            
+            assert result is True
+            assert os.path.isdir(os.path.dirname(file_path))

src/praisonai/tests/unit/code/test_tools.py

@@ -0,0 +1,379 @@
+"""
+Tests for praisonai.code.tools module.
+"""
+
+import os
+import tempfile
+
+
+class TestReadFile:
+    """Tests for read_file tool."""
+    
+    def test_read_entire_file(self):
+        """Test reading entire file."""
+        from praisonai.code.tools.read_file import read_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("line1\nline2\nline3")
+            
+            result = read_file(file_path, add_line_nums=False)
+            
+            assert result['success'] is True
+            assert result['content'] == "line1\nline2\nline3"
+            assert result['total_lines'] == 3
+    
+    def test_read_with_line_numbers(self):
+        """Test reading with line numbers."""
+        from praisonai.code.tools.read_file import read_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("line1\nline2\nline3")
+            
+            result = read_file(file_path, add_line_nums=True)
+            
+            assert result['success'] is True
+            assert "1 | line1" in result['content']
+            assert "2 | line2" in result['content']
+    
+    def test_read_line_range(self):
+        """Test reading specific line range."""
+        from praisonai.code.tools.read_file import read_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("line1\nline2\nline3\nline4\nline5")
+            
+            result = read_file(file_path, start_line=2, end_line=4, add_line_nums=False)
+            
+            assert result['success'] is True
+            assert result['start_line'] == 2
+            assert result['end_line'] == 4
+            assert "line2" in result['content']
+            assert "line4" in result['content']
+            assert "line1" not in result['content']
+            assert "line5" not in result['content']
+    
+    def test_read_nonexistent_file(self):
+        """Test reading nonexistent file."""
+        from praisonai.code.tools.read_file import read_file
+        
+        result = read_file("/nonexistent/path/file.txt")
+        
+        assert result['success'] is False
+        assert "not found" in result['error'].lower()
+    
+    def test_read_with_workspace(self):
+        """Test reading with workspace path."""
+        from praisonai.code.tools.read_file import read_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "subdir", "test.txt")
+            os.makedirs(os.path.dirname(file_path))
+            with open(file_path, 'w') as f:
+                f.write("content")
+            
+            result = read_file("subdir/test.txt", workspace=temp_dir, add_line_nums=False)
+            
+            assert result['success'] is True
+            assert result['content'] == "content"
+    
+    def test_read_outside_workspace(self):
+        """Test reading file outside workspace is blocked."""
+        from praisonai.code.tools.read_file import read_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            result = read_file("../../../etc/passwd", workspace=temp_dir)
+            
+            assert result['success'] is False
+            assert "outside" in result['error'].lower()
+
+
+class TestWriteFile:
+    """Tests for write_file tool."""
+    
+    def test_write_new_file(self):
+        """Test writing a new file."""
+        from praisonai.code.tools.write_file import write_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "new_file.txt")
+            
+            result = write_file(file_path, "Hello, World!")
+            
+            assert result['success'] is True
+            assert result['created'] is True
+            
+            with open(file_path, 'r') as f:
+                assert f.read() == "Hello, World!"
+    
+    def test_overwrite_existing_file(self):
+        """Test overwriting an existing file."""
+        from praisonai.code.tools.write_file import write_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "existing.txt")
+            with open(file_path, 'w') as f:
+                f.write("old content")
+            
+            result = write_file(file_path, "new content")
+            
+            assert result['success'] is True
+            assert result['created'] is False
+            
+            with open(file_path, 'r') as f:
+                assert f.read() == "new content"
+    
+    def test_create_directories(self):
+        """Test creating parent directories."""
+        from praisonai.code.tools.write_file import write_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "a", "b", "c", "file.txt")
+            
+            result = write_file(file_path, "content", create_directories=True)
+            
+            assert result['success'] is True
+            assert os.path.exists(file_path)
+    
+    def test_strip_code_fences(self):
+        """Test stripping markdown code fences."""
+        from praisonai.code.tools.write_file import write_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.py")
+            
+            content = """```python
+def hello():
+    pass
+```"""
+            
+            result = write_file(file_path, content, strip_code_fences=True)
+            
+            assert result['success'] is True
+            
+            with open(file_path, 'r') as f:
+                saved_content = f.read()
+            
+            assert "```" not in saved_content
+            assert "def hello():" in saved_content
+    
+    def test_write_with_backup(self):
+        """Test writing with backup."""
+        from praisonai.code.tools.write_file import write_file
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("original")
+            
+            result = write_file(file_path, "modified", backup=True)
+            
+            assert result['success'] is True
+            assert result['backup_path'] is not None
+            assert os.path.exists(result['backup_path'])
+            
+            with open(result['backup_path'], 'r') as f:
+                assert f.read() == "original"
+
+
+class TestListFiles:
+    """Tests for list_files tool."""
+    
+    def test_list_files_basic(self):
+        """Test basic file listing."""
+        from praisonai.code.tools.list_files import list_files
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create some files
+            open(os.path.join(temp_dir, "file1.txt"), 'w').close()
+            open(os.path.join(temp_dir, "file2.py"), 'w').close()
+            os.makedirs(os.path.join(temp_dir, "subdir"))
+            
+            result = list_files(temp_dir)
+            
+            assert result['success'] is True
+            assert len(result['files']) == 2
+            assert len(result['directories']) == 1
+    
+    def test_list_files_recursive(self):
+        """Test recursive file listing."""
+        from praisonai.code.tools.list_files import list_files
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create nested structure
+            os.makedirs(os.path.join(temp_dir, "a", "b"))
+            open(os.path.join(temp_dir, "file1.txt"), 'w').close()
+            open(os.path.join(temp_dir, "a", "file2.txt"), 'w').close()
+            open(os.path.join(temp_dir, "a", "b", "file3.txt"), 'w').close()
+            
+            result = list_files(temp_dir, recursive=True)
+            
+            assert result['success'] is True
+            assert len(result['files']) == 3
+    
+    def test_list_files_with_extension_filter(self):
+        """Test filtering by extension."""
+        from praisonai.code.tools.list_files import list_files
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            open(os.path.join(temp_dir, "file1.txt"), 'w').close()
+            open(os.path.join(temp_dir, "file2.py"), 'w').close()
+            open(os.path.join(temp_dir, "file3.py"), 'w').close()
+            
+            result = list_files(temp_dir, extensions=['py'])
+            
+            assert result['success'] is True
+            assert len(result['files']) == 2
+            assert all(f['name'].endswith('.py') for f in result['files'])
+    
+    def test_list_files_exclude_hidden(self):
+        """Test excluding hidden files."""
+        from praisonai.code.tools.list_files import list_files
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            open(os.path.join(temp_dir, "visible.txt"), 'w').close()
+            open(os.path.join(temp_dir, ".hidden"), 'w').close()
+            
+            result = list_files(temp_dir, include_hidden=False)
+            
+            assert result['success'] is True
+            assert len(result['files']) == 1
+            assert result['files'][0]['name'] == "visible.txt"
+    
+    def test_list_files_max_limit(self):
+        """Test max files limit."""
+        from praisonai.code.tools.list_files import list_files
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create many files
+            for i in range(10):
+                open(os.path.join(temp_dir, f"file{i}.txt"), 'w').close()
+            
+            result = list_files(temp_dir, max_files=5)
+            
+            assert result['success'] is True
+            assert result['total_count'] == 5
+            assert result['truncated'] is True
+
+
+class TestSearchReplace:
+    """Tests for search_replace tool."""
+    
+    def test_simple_replace(self):
+        """Test simple string replacement."""
+        from praisonai.code.tools.search_replace import search_replace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("Hello, old world!")
+            
+            result = search_replace(file_path, [
+                {'search': 'old', 'replace': 'new'}
+            ])
+            
+            assert result['success'] is True
+            assert result['total_replacements'] == 1
+            
+            with open(file_path, 'r') as f:
+                assert f.read() == "Hello, new world!"
+    
+    def test_multiple_operations(self):
+        """Test multiple search/replace operations."""
+        from praisonai.code.tools.search_replace import search_replace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("foo bar foo baz")
+            
+            result = search_replace(file_path, [
+                {'search': 'foo', 'replace': 'FOO'},
+                {'search': 'bar', 'replace': 'BAR'},
+            ])
+            
+            assert result['success'] is True
+            assert result['operations_applied'] == 2
+            
+            with open(file_path, 'r') as f:
+                content = f.read()
+            
+            assert "FOO" in content
+            assert "BAR" in content
+    
+    def test_regex_replace(self):
+        """Test regex replacement."""
+        from praisonai.code.tools.search_replace import search_replace
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("def func1(): pass\ndef func2(): pass")
+            
+            result = search_replace(file_path, [
+                {'search': r'def (\w+)\(\)', 'replace': r'def renamed_\1()', 'is_regex': True}
+            ])
+            
+            assert result['success'] is True
+            
+            with open(file_path, 'r') as f:
+                content = f.read()
+            
+            assert "renamed_func1" in content
+            assert "renamed_func2" in content
+
+
+class TestExecuteCommand:
+    """Tests for execute_command tool."""
+    
+    def test_simple_command(self):
+        """Test simple command execution."""
+        from praisonai.code.tools.execute_command import execute_command
+        
+        result = execute_command("echo 'Hello, World!'")
+        
+        assert result['success'] is True
+        assert result['exit_code'] == 0
+        assert "Hello" in result['stdout']
+    
+    def test_command_with_cwd(self):
+        """Test command with working directory."""
+        from praisonai.code.tools.execute_command import execute_command
+        
+        with tempfile.TemporaryDirectory() as temp_dir:
+            result = execute_command("pwd", cwd=temp_dir)
+            
+            assert result['success'] is True
+            assert temp_dir in result['stdout'] or os.path.basename(temp_dir) in result['stdout']
+    
+    def test_command_failure(self):
+        """Test command that fails."""
+        from praisonai.code.tools.execute_command import execute_command
+        
+        result = execute_command("exit 1", shell=True)
+        
+        assert result['success'] is False
+        assert result['exit_code'] == 1
+    
+    def test_command_timeout(self):
+        """Test command timeout."""
+        from praisonai.code.tools.execute_command import execute_command
+        
+        result = execute_command("sleep 10", timeout=1)
+        
+        assert result['success'] is False
+        assert "timed out" in result['error'].lower()
+    
+    def test_is_safe_command(self):
+        """Test safe command detection."""
+        from praisonai.code.tools.execute_command import is_safe_command
+        
+        assert is_safe_command("ls -la") is True
+        assert is_safe_command("echo hello") is True
+        assert is_safe_command("rm -rf /") is False
+        assert is_safe_command("sudo apt install") is False