Release v2.8.9

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.8.7" \
+    "praisonai>=2.8.9" \
     "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.8.7" \
+    "praisonai>=2.8.9" \
     "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.8.7" \
+    "praisonai>=2.8.9" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/acp/README.md

@@ -0,0 +1,137 @@
+# ACP (Agent Client Protocol) Examples
+
+This directory contains examples for using PraisonAI with the Agent Client Protocol (ACP).
+
+## What is ACP?
+
+The Agent Client Protocol (ACP) is a standardized JSON-RPC 2.0 protocol that allows code editors and IDEs to communicate with AI coding agents. It enables seamless integration between PraisonAI and editors like:
+
+- **Zed**
+- **JetBrains IDEs** (IntelliJ, PyCharm, WebStorm, etc.)
+- **VSCode** (via extensions)
+- **Toad**
+
+## Quick Start
+
+### CLI Usage
+
+```bash
+# Start ACP server with defaults
+praisonai acp
+
+# Enable debug logging
+praisonai acp --debug
+
+# Allow file writes
+praisonai acp --allow-write
+
+# Use a specific model
+praisonai acp -m gpt-4o
+
+# Resume last session
+praisonai acp --resume --last
+```
+
+### Python API
+
+```python
+from praisonai.acp import serve
+
+serve(
+    workspace=".",
+    debug=True,
+    allow_write=True,
+)
+```
+
+## Editor Configuration
+
+### Zed
+
+Add to `~/.config/zed/settings.json`:
+
+```json
+{
+  "agent_servers": {
+    "PraisonAI": {
+      "command": "praisonai",
+      "args": ["acp"],
+      "env": {}
+    }
+  }
+}
+```
+
+### JetBrains
+
+Add to `~/.jetbrains/acp.json`:
+
+```json
+{
+  "agent_servers": {
+    "PraisonAI": {
+      "command": "praisonai",
+      "args": ["acp"],
+      "env": {}
+    }
+  }
+}
+```
+
+### Toad
+
+```bash
+toad acp "praisonai acp"
+```
+
+## Examples
+
+- `basic_acp_server.py` - Simple ACP server with default settings
+- `custom_agent_acp.py` - ACP server with a custom agent
+
+## CLI Options
+
+| Option | Description |
+|--------|-------------|
+| `-w, --workspace` | Workspace root directory |
+| `-a, --agent` | Agent name or config file |
+| `-m, --model` | LLM model to use |
+| `-r, --resume` | Resume session by ID |
+| `--last` | Resume the last session |
+| `--approve` | Approval mode: manual, auto, scoped |
+| `--read-only` | Read-only mode (default) |
+| `--allow-write` | Allow file writes |
+| `--allow-shell` | Allow shell commands |
+| `--debug` | Enable debug logging |
+
+## Installation
+
+```bash
+pip install praisonai[acp]
+```
+
+## Protocol Details
+
+ACP uses JSON-RPC 2.0 over stdio:
+- **stdin**: Receives JSON-RPC requests from the client
+- **stdout**: Sends JSON-RPC responses to the client
+- **stderr**: Debug/log output (never pollutes stdout)
+
+### Supported Methods
+
+**Agent Methods (client → agent):**
+- `initialize` - Negotiate protocol version and capabilities
+- `authenticate` - Optional authentication
+- `session/new` - Create a new session
+- `session/load` - Load an existing session
+- `session/prompt` - Send user message
+- `session/cancel` - Cancel ongoing operations
+- `session/set_mode` - Change operating mode
+
+**Client Methods (agent → client):**
+- `session/update` - Send progress updates
+- `session/request_permission` - Request user approval
+- `fs/read_text_file` - Read file contents
+- `fs/write_text_file` - Write file contents
+- `terminal/create` - Create terminal
+- `terminal/output` - Get terminal output

examples/acp/basic_acp_server.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+"""
+Basic ACP Server Example
+
+This example demonstrates how to run PraisonAI as an ACP server
+that can be connected to by IDEs like Zed, JetBrains, VSCode, or Toad.
+
+Usage:
+    # Run from command line
+    praisonai acp
+    
+    # Or run this script directly
+    python basic_acp_server.py
+"""
+
+from praisonai.acp import serve
+
+if __name__ == "__main__":
+    # Start ACP server with default settings
+    # This will listen on stdin/stdout for JSON-RPC messages
+    serve(
+        workspace=".",           # Current directory as workspace
+        agent="default",         # Use default agent
+        debug=False,             # Set to True for debug logging to stderr
+        read_only=True,          # Safe by default
+        approval_mode="manual",  # Require approval for actions
+    )

examples/acp/custom_agent_acp.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""
+Custom Agent ACP Server Example
+
+This example shows how to run a custom PraisonAI agent as an ACP server.
+
+Usage:
+    python custom_agent_acp.py
+"""
+
+from praisonai.acp import serve, ACPServer, ACPConfig
+from praisonaiagents import Agent
+
+# Create a custom agent
+agent = Agent(
+    name="CodeAssistant",
+    instructions="""You are an expert coding assistant. You help users:
+    - Write clean, efficient code
+    - Debug issues
+    - Explain complex concepts
+    - Review code for best practices
+    
+    Always provide clear explanations with your code suggestions.""",
+    model="gpt-4o-mini",
+)
+
+if __name__ == "__main__":
+    # Create config
+    config = ACPConfig(
+        workspace=".",
+        debug=True,
+        allow_write=True,  # Allow file modifications
+        approval_mode="manual",
+    )
+    
+    # Create server with custom agent
+    server = ACPServer(config=config, agent=agent)
+    
+    # Run the server
+    serve(
+        workspace=".",
+        debug=True,
+        allow_write=True,
+        approval_mode="manual",
+    )

src/praisonai/examples/acp/README.md

@@ -0,0 +1,137 @@
+# ACP (Agent Client Protocol) Examples
+
+This directory contains examples for using PraisonAI with the Agent Client Protocol (ACP).
+
+## What is ACP?
+
+The Agent Client Protocol (ACP) is a standardized JSON-RPC 2.0 protocol that allows code editors and IDEs to communicate with AI coding agents. It enables seamless integration between PraisonAI and editors like:
+
+- **Zed**
+- **JetBrains IDEs** (IntelliJ, PyCharm, WebStorm, etc.)
+- **VSCode** (via extensions)
+- **Toad**
+
+## Quick Start
+
+### CLI Usage
+
+```bash
+# Start ACP server with defaults
+praisonai acp
+
+# Enable debug logging
+praisonai acp --debug
+
+# Allow file writes
+praisonai acp --allow-write
+
+# Use a specific model
+praisonai acp -m gpt-4o
+
+# Resume last session
+praisonai acp --resume --last
+```
+
+### Python API
+
+```python
+from praisonai.acp import serve
+
+serve(
+    workspace=".",
+    debug=True,
+    allow_write=True,
+)
+```
+
+## Editor Configuration
+
+### Zed
+
+Add to `~/.config/zed/settings.json`:
+
+```json
+{
+  "agent_servers": {
+    "PraisonAI": {
+      "command": "praisonai",
+      "args": ["acp"],
+      "env": {}
+    }
+  }
+}
+```
+
+### JetBrains
+
+Add to `~/.jetbrains/acp.json`:
+
+```json
+{
+  "agent_servers": {
+    "PraisonAI": {
+      "command": "praisonai",
+      "args": ["acp"],
+      "env": {}
+    }
+  }
+}
+```
+
+### Toad
+
+```bash
+toad acp "praisonai acp"
+```
+
+## Examples
+
+- `basic_acp_server.py` - Simple ACP server with default settings
+- `custom_agent_acp.py` - ACP server with a custom agent
+
+## CLI Options
+
+| Option | Description |
+|--------|-------------|
+| `-w, --workspace` | Workspace root directory |
+| `-a, --agent` | Agent name or config file |
+| `-m, --model` | LLM model to use |
+| `-r, --resume` | Resume session by ID |
+| `--last` | Resume the last session |
+| `--approve` | Approval mode: manual, auto, scoped |
+| `--read-only` | Read-only mode (default) |
+| `--allow-write` | Allow file writes |
+| `--allow-shell` | Allow shell commands |
+| `--debug` | Enable debug logging |
+
+## Installation
+
+```bash
+pip install praisonai[acp]
+```
+
+## Protocol Details
+
+ACP uses JSON-RPC 2.0 over stdio:
+- **stdin**: Receives JSON-RPC requests from the client
+- **stdout**: Sends JSON-RPC responses to the client
+- **stderr**: Debug/log output (never pollutes stdout)
+
+### Supported Methods
+
+**Agent Methods (client → agent):**
+- `initialize` - Negotiate protocol version and capabilities
+- `authenticate` - Optional authentication
+- `session/new` - Create a new session
+- `session/load` - Load an existing session
+- `session/prompt` - Send user message
+- `session/cancel` - Cancel ongoing operations
+- `session/set_mode` - Change operating mode
+
+**Client Methods (agent → client):**
+- `session/update` - Send progress updates
+- `session/request_permission` - Request user approval
+- `fs/read_text_file` - Read file contents
+- `fs/write_text_file` - Write file contents
+- `terminal/create` - Create terminal
+- `terminal/output` - Get terminal output

src/praisonai/examples/acp/basic_acp_server.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+"""
+Basic ACP Server Example
+
+This example demonstrates how to run PraisonAI as an ACP server
+that can be connected to by IDEs like Zed, JetBrains, VSCode, or Toad.
+
+Usage:
+    # Run from command line
+    praisonai acp
+    
+    # Or run this script directly
+    python basic_acp_server.py
+"""
+
+from praisonai.acp import serve
+
+if __name__ == "__main__":
+    # Start ACP server with default settings
+    # This will listen on stdin/stdout for JSON-RPC messages
+    serve(
+        workspace=".",           # Current directory as workspace
+        agent="default",         # Use default agent
+        debug=False,             # Set to True for debug logging to stderr
+        read_only=True,          # Safe by default
+        approval_mode="manual",  # Require approval for actions
+    )

src/praisonai/examples/acp/custom_agent_acp.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""
+Custom Agent ACP Server Example
+
+This example shows how to run a custom PraisonAI agent as an ACP server.
+
+Usage:
+    python custom_agent_acp.py
+"""
+
+from praisonai.acp import serve, ACPServer, ACPConfig
+from praisonaiagents import Agent
+
+# Create a custom agent
+agent = Agent(
+    name="CodeAssistant",
+    instructions="""You are an expert coding assistant. You help users:
+    - Write clean, efficient code
+    - Debug issues
+    - Explain complex concepts
+    - Review code for best practices
+    
+    Always provide clear explanations with your code suggestions.""",
+    model="gpt-4o-mini",
+)
+
+if __name__ == "__main__":
+    # Create config
+    config = ACPConfig(
+        workspace=".",
+        debug=True,
+        allow_write=True,  # Allow file modifications
+        approval_mode="manual",
+    )
+    
+    # Create server with custom agent
+    server = ACPServer(config=config, agent=agent)
+    
+    # Run the server
+    serve(
+        workspace=".",
+        debug=True,
+        allow_write=True,
+        approval_mode="manual",
+    )

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

src/praisonai/praisonai/acp/__init__.py

@@ -0,0 +1,54 @@
+"""
+Agent Client Protocol (ACP) support for PraisonAI.
+
+This module provides ACP server functionality that allows IDEs/editors
+(Zed, JetBrains, VSCode, Toad) to connect to PraisonAI agents.
+
+Usage:
+    # CLI
+    praisonai acp [OPTIONS]
+    
+    # Python API
+    from praisonai.acp import serve
+    serve(workspace=".", agent="default")
+"""
+
+import importlib.util
+
+__all__ = [
+    "serve",
+    "ACPServer",
+    "ACPSession",
+    "ACPConfig",
+]
+
+# Lazy imports for performance
+def __getattr__(name: str):
+    """Lazy import ACP components."""
+    if name == "serve":
+        from .server import serve
+        return serve
+    if name == "ACPServer":
+        from .server import ACPServer
+        return ACPServer
+    if name == "ACPSession":
+        from .session import ACPSession
+        return ACPSession
+    if name == "ACPConfig":
+        from .config import ACPConfig
+        return ACPConfig
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def _check_acp_available() -> bool:
+    """Check if agent-client-protocol package is installed."""
+    return importlib.util.find_spec("acp") is not None
+
+
+def _get_install_instructions() -> str:
+    """Get installation instructions for ACP support."""
+    return (
+        "ACP support requires the agent-client-protocol package.\n"
+        "Install with: pip install praisonai[acp]\n"
+        "Or: pip install agent-client-protocol"
+    )

src/praisonai/praisonai/acp/config.py

@@ -0,0 +1,159 @@
+"""
+ACP configuration management.
+
+Handles configuration precedence:
+1. CLI flags (highest)
+2. Environment variables
+3. Config file (~/.praison/config.yaml)
+4. Defaults (lowest)
+"""
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class ACPConfig:
+    """Configuration for ACP server."""
+    
+    # Workspace settings
+    workspace: Path = field(default_factory=Path.cwd)
+    
+    # Agent settings
+    agent: str = "default"
+    agents_config: Optional[Path] = None
+    router_enabled: bool = False
+    model: Optional[str] = None
+    
+    # Session settings
+    resume_session: Optional[str] = None
+    resume_last: bool = False
+    
+    # Permission settings
+    read_only: bool = True
+    allow_write: bool = False
+    allow_shell: bool = False
+    allow_network: bool = False
+    approval_mode: str = "manual"  # manual, auto, scoped
+    
+    # Workspace boundaries
+    allowed_paths: List[Path] = field(default_factory=list)
+    
+    # Debug settings
+    debug: bool = False
+    
+    # Profile
+    profile: Optional[str] = None
+    
+    def __post_init__(self):
+        """Validate and normalize configuration."""
+        if isinstance(self.workspace, str):
+            self.workspace = Path(self.workspace).resolve()
+        
+        if self.agents_config and isinstance(self.agents_config, str):
+            self.agents_config = Path(self.agents_config)
+        
+        # Set up default allowed paths
+        if not self.allowed_paths:
+            self.allowed_paths = [
+                self.workspace,
+                Path.home() / ".praison",
+            ]
+        
+        # Normalize allowed paths
+        self.allowed_paths = [
+            Path(p).resolve() if isinstance(p, str) else p.resolve()
+            for p in self.allowed_paths
+        ]
+    
+    @classmethod
+    def from_env(cls) -> "ACPConfig":
+        """Create config from environment variables."""
+        return cls(
+            workspace=Path(os.environ.get("PRAISONAI_WORKSPACE", ".")).resolve(),
+            agent=os.environ.get("PRAISONAI_AGENT", "default"),
+            model=os.environ.get("PRAISONAI_MODEL") or os.environ.get("OPENAI_MODEL"),
+            debug=os.environ.get("PRAISONAI_DEBUG", "").lower() in ("1", "true", "yes"),
+            read_only=os.environ.get("PRAISONAI_READ_ONLY", "true").lower() in ("1", "true", "yes"),
+            approval_mode=os.environ.get("PRAISONAI_APPROVAL_MODE", "manual"),
+        )
+    
+    @classmethod
+    def from_file(cls, path: Optional[Path] = None) -> "ACPConfig":
+        """Load config from YAML file."""
+        if path is None:
+            path = Path.home() / ".praison" / "config.yaml"
+        
+        if not path.exists():
+            return cls()
+        
+        try:
+            import yaml
+            with open(path) as f:
+                data = yaml.safe_load(f) or {}
+            
+            acp_config = data.get("acp", {})
+            return cls(
+                workspace=Path(acp_config.get("workspace", ".")).resolve(),
+                agent=acp_config.get("agent", "default"),
+                model=acp_config.get("model"),
+                debug=acp_config.get("debug", False),
+                read_only=acp_config.get("read_only", True),
+                approval_mode=acp_config.get("approval_mode", "manual"),
+            )
+        except Exception:
+            return cls()
+    
+    @classmethod
+    def merge(cls, *configs: "ACPConfig") -> "ACPConfig":
+        """Merge multiple configs, later configs override earlier ones."""
+        result = cls()
+        for config in configs:
+            for field_name in [
+                "workspace", "agent", "agents_config", "router_enabled",
+                "model", "resume_session", "resume_last", "read_only",
+                "allow_write", "allow_shell", "allow_network", "approval_mode",
+                "debug", "profile"
+            ]:
+                value = getattr(config, field_name)
+                # Only override if value is not default/None
+                if value is not None and value != getattr(cls(), field_name, None):
+                    setattr(result, field_name, value)
+        return result
+    
+    def is_path_allowed(self, path: Path) -> bool:
+        """Check if a path is within allowed boundaries."""
+        path = Path(path).resolve()
+        return any(
+            path == allowed or allowed in path.parents
+            for allowed in self.allowed_paths
+        )
+    
+    def can_write(self) -> bool:
+        """Check if write operations are allowed."""
+        return not self.read_only or self.allow_write
+    
+    def can_execute_shell(self) -> bool:
+        """Check if shell commands are allowed."""
+        return self.allow_shell
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert config to dictionary."""
+        return {
+            "workspace": str(self.workspace),
+            "agent": self.agent,
+            "agents_config": str(self.agents_config) if self.agents_config else None,
+            "router_enabled": self.router_enabled,
+            "model": self.model,
+            "resume_session": self.resume_session,
+            "resume_last": self.resume_last,
+            "read_only": self.read_only,
+            "allow_write": self.allow_write,
+            "allow_shell": self.allow_shell,
+            "allow_network": self.allow_network,
+            "approval_mode": self.approval_mode,
+            "debug": self.debug,
+            "profile": self.profile,
+        }

src/praisonai/praisonai/acp/server.py

@@ -0,0 +1,587 @@
+"""
+ACP Server implementation for PraisonAI.
+
+This module implements the Agent Client Protocol server that allows
+IDEs/editors to communicate with PraisonAI agents.
+"""
+
+import asyncio
+import logging
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+from .config import ACPConfig
+from .session import ACPSession, SessionStore
+
+# Configure logging to stderr only (stdout reserved for JSON-RPC)
+logger = logging.getLogger(__name__)
+
+
+def _setup_stderr_logging(debug: bool = False) -> None:
+    """Configure logging to stderr only."""
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(logging.Formatter(
+        "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+    ))
+    
+    root_logger = logging.getLogger("praisonai.acp")
+    root_logger.handlers.clear()
+    root_logger.addHandler(handler)
+    root_logger.setLevel(logging.DEBUG if debug else logging.INFO)
+
+
+class ACPServer:
+    """
+    ACP Server that exposes PraisonAI agents to IDE clients.
+    
+    Implements the Agent Client Protocol (JSON-RPC 2.0 over stdio).
+    """
+    
+    def __init__(
+        self,
+        config: Optional[ACPConfig] = None,
+        agent: Optional[Any] = None,
+        agents: Optional[Any] = None,
+    ):
+        """
+        Initialize ACP server.
+        
+        Args:
+            config: ACP configuration
+            agent: Optional pre-configured Agent instance
+            agents: Optional pre-configured PraisonAIAgents instance
+        """
+        self.config = config or ACPConfig()
+        self._agent = agent
+        self._agents = agents
+        self._session_store = SessionStore()
+        self._sessions: Dict[str, ACPSession] = {}
+        self._client = None  # ACP client connection
+        self._cancelled_sessions: set = set()
+        
+        _setup_stderr_logging(self.config.debug)
+    
+    def _get_agent(self):
+        """Lazy load or create agent."""
+        if self._agent is not None:
+            return self._agent
+        
+        if self._agents is not None:
+            return self._agents
+        
+        # Create default agent
+        try:
+            from praisonaiagents import Agent
+            self._agent = Agent(
+                name="PraisonAI",
+                instructions="You are a helpful AI coding assistant.",
+                model=self.config.model or "gpt-4o-mini",
+            )
+            return self._agent
+        except ImportError:
+            logger.error("praisonaiagents not installed")
+            raise RuntimeError("praisonaiagents package required")
+    
+    def on_connect(self, client) -> None:
+        """Called when client connects."""
+        self._client = client
+        logger.info("Client connected")
+    
+    async def initialize(
+        self,
+        protocol_version: int,
+        client_capabilities: Optional[Dict[str, Any]] = None,
+        client_info: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Handle initialize request.
+        
+        Negotiate protocol version and exchange capabilities.
+        """
+        logger.info(f"Initialize: protocol_version={protocol_version}, client_info={client_info}")
+        
+        # Import here to avoid circular imports and check availability
+        try:
+            from acp import PROTOCOL_VERSION
+        except ImportError:
+            PROTOCOL_VERSION = 1
+        
+        return {
+            "protocolVersion": min(protocol_version, PROTOCOL_VERSION),
+            "agentCapabilities": {
+                "loadSession": True,
+                "promptCapabilities": {
+                    "image": False,
+                    "audio": False,
+                    "embeddedContext": True,
+                },
+                "mcpCapabilities": {
+                    "http": False,
+                    "sse": False,
+                },
+            },
+            "agentInfo": {
+                "name": "praisonai",
+                "title": "PraisonAI",
+                "version": self._get_version(),
+            },
+            "authMethods": [],
+        }
+    
+    async def authenticate(self, method_id: str, **kwargs: Any) -> Optional[Dict[str, Any]]:
+        """Handle authenticate request (optional)."""
+        logger.info(f"Authenticate: method_id={method_id}")
+        return {}
+    
+    async def new_session(
+        self,
+        cwd: str,
+        mcp_servers: List[Dict[str, Any]],
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Create a new conversation session.
+        
+        Args:
+            cwd: Working directory for the session
+            mcp_servers: List of MCP server configurations
+        """
+        logger.info(f"New session: cwd={cwd}")
+        
+        workspace = Path(cwd).resolve()
+        session = ACPSession.create(
+            workspace=workspace,
+            agent_id=self.config.agent,
+        )
+        session.mcp_servers = mcp_servers
+        
+        self._sessions[session.session_id] = session
+        self._session_store.save(session)
+        
+        logger.info(f"Created session: {session.session_id}")
+        
+        return {
+            "sessionId": session.session_id,
+            "modes": self._get_available_modes(),
+        }
+    
+    async def load_session(
+        self,
+        cwd: str,
+        mcp_servers: List[Dict[str, Any]],
+        session_id: str,
+        **kwargs: Any,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Load an existing session.
+        
+        Replays conversation history via session/update notifications.
+        """
+        logger.info(f"Load session: session_id={session_id}")
+        
+        # Try to load from store
+        session = self._session_store.load(session_id)
+        if session is None:
+            logger.warning(f"Session not found: {session_id}")
+            return None
+        
+        # Update workspace and MCP servers
+        session.workspace = Path(cwd).resolve()
+        session.mcp_servers = mcp_servers
+        session.update_activity()
+        
+        self._sessions[session_id] = session
+        
+        # Replay conversation history
+        await self._replay_session_history(session)
+        
+        return {}
+    
+    async def list_sessions(
+        self,
+        cursor: Optional[str] = None,
+        cwd: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """List available sessions."""
+        sessions = self._session_store.list_sessions()
+        
+        # Filter by cwd if provided
+        if cwd:
+            cwd_path = Path(cwd).resolve()
+            sessions = [s for s in sessions if s.workspace == cwd_path]
+        
+        return {
+            "sessions": [
+                {
+                    "sessionId": s.session_id,
+                    "createdAt": s.created_at,
+                    "lastActivity": s.last_activity,
+                }
+                for s in sessions
+            ],
+        }
+    
+    async def set_session_mode(
+        self,
+        mode_id: str,
+        session_id: str,
+        **kwargs: Any,
+    ) -> Optional[Dict[str, Any]]:
+        """Set the operating mode for a session."""
+        logger.info(f"Set session mode: session_id={session_id}, mode_id={mode_id}")
+        
+        session = self._sessions.get(session_id)
+        if session:
+            session.mode = mode_id
+            self._session_store.save(session)
+        
+        return {}
+    
+    async def set_session_model(
+        self,
+        model_id: str,
+        session_id: str,
+        **kwargs: Any,
+    ) -> Optional[Dict[str, Any]]:
+        """Set the LLM model for a session."""
+        logger.info(f"Set session model: session_id={session_id}, model_id={model_id}")
+        
+        session = self._sessions.get(session_id)
+        if session:
+            session.model = model_id
+            self._session_store.save(session)
+        
+        return {}
+    
+    async def prompt(
+        self,
+        prompt: List[Dict[str, Any]],
+        session_id: str,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Handle user prompt.
+        
+        This is the main interaction method where the user sends a message
+        and the agent processes it.
+        """
+        logger.info(f"Prompt: session_id={session_id}")
+        
+        session = self._sessions.get(session_id)
+        if session is None:
+            logger.error(f"Session not found: {session_id}")
+            return {"stopReason": "refusal"}
+        
+        # Check if cancelled
+        if session_id in self._cancelled_sessions:
+            self._cancelled_sessions.discard(session_id)
+            return {"stopReason": "cancelled"}
+        
+        # Extract text from prompt blocks
+        user_message = self._extract_prompt_text(prompt)
+        session.add_message("user", user_message)
+        
+        try:
+            # Get agent response
+            agent = self._get_agent()
+            
+            # Send thinking update
+            await self._send_thought(session_id, "Processing your request...")
+            
+            # Generate response
+            response = await self._generate_response(agent, user_message, session)
+            
+            # Send response as agent message chunks
+            await self._send_agent_message(session_id, response)
+            
+            # Save session
+            session.add_message("assistant", response)
+            self._session_store.save(session)
+            
+            return {"stopReason": "end_turn"}
+            
+        except asyncio.CancelledError:
+            return {"stopReason": "cancelled"}
+        except Exception as e:
+            logger.exception(f"Error processing prompt: {e}")
+            await self._send_agent_message(session_id, f"Error: {str(e)}")
+            return {"stopReason": "refusal"}
+    
+    async def fork_session(
+        self,
+        cwd: str,
+        session_id: str,
+        mcp_servers: Optional[List[Dict[str, Any]]] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """Fork an existing session."""
+        logger.info(f"Fork session: session_id={session_id}")
+        
+        original = self._sessions.get(session_id)
+        if original is None:
+            original = self._session_store.load(session_id)
+        
+        if original is None:
+            # Create new session if original not found
+            return await self.new_session(cwd, mcp_servers or [])
+        
+        # Create forked session
+        forked = ACPSession.create(
+            workspace=Path(cwd).resolve(),
+            agent_id=original.agent_id,
+        )
+        forked.messages = original.messages.copy()
+        forked.mcp_servers = mcp_servers or original.mcp_servers
+        
+        self._sessions[forked.session_id] = forked
+        self._session_store.save(forked)
+        
+        return {"sessionId": forked.session_id}
+    
+    async def resume_session(
+        self,
+        cwd: str,
+        session_id: str,
+        mcp_servers: Optional[List[Dict[str, Any]]] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """Resume an existing session."""
+        return await self.load_session(cwd, mcp_servers or [], session_id)
+    
+    async def cancel(self, session_id: str, **kwargs: Any) -> None:
+        """Cancel ongoing operations for a session."""
+        logger.info(f"Cancel: session_id={session_id}")
+        self._cancelled_sessions.add(session_id)
+    
+    async def ext_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Handle extension method calls."""
+        logger.info(f"Extension method: {method}")
+        return {"error": f"Unknown extension method: {method}"}
+    
+    async def ext_notification(self, method: str, params: Dict[str, Any]) -> None:
+        """Handle extension notifications."""
+        logger.info(f"Extension notification: {method}")
+    
+    # Helper methods
+    
+    def _get_version(self) -> str:
+        """Get PraisonAI version."""
+        try:
+            from praisonai.version import __version__
+            return __version__
+        except ImportError:
+            return "0.0.0"
+    
+    def _get_available_modes(self) -> Optional[List[Dict[str, Any]]]:
+        """Get available operating modes."""
+        return [
+            {
+                "id": "manual",
+                "name": "Manual",
+                "description": "Requires approval for all actions",
+            },
+            {
+                "id": "auto",
+                "name": "Auto",
+                "description": "Auto-approve within workspace",
+            },
+        ]
+    
+    def _extract_prompt_text(self, prompt: List[Dict[str, Any]]) -> str:
+        """Extract text content from prompt blocks."""
+        texts = []
+        for block in prompt:
+            if isinstance(block, dict):
+                block_type = block.get("type", "")
+                if block_type == "text":
+                    texts.append(block.get("text", ""))
+                elif block_type == "resource":
+                    resource = block.get("resource", {})
+                    if "text" in resource:
+                        texts.append(f"[File: {resource.get('uri', 'unknown')}]\n{resource['text']}")
+        return "\n".join(texts)
+    
+    async def _generate_response(
+        self,
+        agent: Any,
+        message: str,
+        session: ACPSession,
+    ) -> str:
+        """Generate agent response."""
+        try:
+            # Try async chat first
+            if hasattr(agent, "achat"):
+                response = await agent.achat(message)
+            elif hasattr(agent, "chat"):
+                # Run sync chat in executor
+                loop = asyncio.get_event_loop()
+                response = await loop.run_in_executor(None, agent.chat, message)
+            else:
+                response = "Agent does not support chat interface"
+            
+            return str(response) if response else ""
+        except Exception as e:
+            logger.exception(f"Error generating response: {e}")
+            return f"Error generating response: {str(e)}"
+    
+    async def _send_agent_message(self, session_id: str, text: str) -> None:
+        """Send agent message chunk to client."""
+        if self._client is None:
+            return
+        
+        try:
+            from acp import update_agent_message_text
+            update = update_agent_message_text(text)
+            await self._client.session_update(session_id=session_id, update=update)
+        except ImportError:
+            # Fallback without ACP SDK
+            logger.debug(f"Agent message: {text[:100]}...")
+    
+    async def _send_thought(self, session_id: str, text: str) -> None:
+        """Send agent thought to client."""
+        if self._client is None:
+            return
+        
+        try:
+            from acp import update_agent_thought_text
+            update = update_agent_thought_text(text)
+            await self._client.session_update(session_id=session_id, update=update)
+        except ImportError:
+            logger.debug(f"Agent thought: {text}")
+    
+    async def _replay_session_history(self, session: ACPSession) -> None:
+        """Replay session history to client."""
+        if self._client is None:
+            return
+        
+        try:
+            from acp import update_user_message_text, update_agent_message_text
+            
+            for msg in session.messages:
+                role = msg.get("role", "")
+                content = msg.get("content", "")
+                
+                if role == "user":
+                    update = update_user_message_text(str(content))
+                elif role == "assistant":
+                    update = update_agent_message_text(str(content))
+                else:
+                    continue
+                
+                await self._client.session_update(
+                    session_id=session.session_id,
+                    update=update,
+                )
+        except ImportError:
+            logger.debug("Skipping history replay (ACP SDK not available)")
+
+
+async def _run_server(config: ACPConfig) -> None:
+    """Run the ACP server."""
+    try:
+        from acp import Agent as ACPAgent, run_agent
+    except ImportError:
+        logger.error(
+            "agent-client-protocol package not installed.\n"
+            "Install with: pip install praisonai[acp]\n"
+            "Or: pip install agent-client-protocol"
+        )
+        sys.exit(1)
+    
+    server = ACPServer(config=config)
+    
+    # Create ACP-compatible agent wrapper
+    class PraisonACPAgent(ACPAgent):
+        def on_connect(self, conn):
+            server.on_connect(conn)
+        
+        async def initialize(self, protocol_version, client_capabilities=None, client_info=None, **kwargs):
+            return await server.initialize(protocol_version, client_capabilities, client_info, **kwargs)
+        
+        async def authenticate(self, method_id, **kwargs):
+            return await server.authenticate(method_id, **kwargs)
+        
+        async def new_session(self, cwd, mcp_servers, **kwargs):
+            return await server.new_session(cwd, mcp_servers, **kwargs)
+        
+        async def load_session(self, cwd, mcp_servers, session_id, **kwargs):
+            return await server.load_session(cwd, mcp_servers, session_id, **kwargs)
+        
+        async def list_sessions(self, cursor=None, cwd=None, **kwargs):
+            return await server.list_sessions(cursor, cwd, **kwargs)
+        
+        async def set_session_mode(self, mode_id, session_id, **kwargs):
+            return await server.set_session_mode(mode_id, session_id, **kwargs)
+        
+        async def set_session_model(self, model_id, session_id, **kwargs):
+            return await server.set_session_model(model_id, session_id, **kwargs)
+        
+        async def prompt(self, prompt, session_id, **kwargs):
+            return await server.prompt(prompt, session_id, **kwargs)
+        
+        async def fork_session(self, cwd, session_id, mcp_servers=None, **kwargs):
+            return await server.fork_session(cwd, session_id, mcp_servers, **kwargs)
+        
+        async def resume_session(self, cwd, session_id, mcp_servers=None, **kwargs):
+            return await server.resume_session(cwd, session_id, mcp_servers, **kwargs)
+        
+        async def cancel(self, session_id, **kwargs):
+            return await server.cancel(session_id, **kwargs)
+        
+        async def ext_method(self, method, params):
+            return await server.ext_method(method, params)
+        
+        async def ext_notification(self, method, params):
+            return await server.ext_notification(method, params)
+    
+    logger.info("Starting PraisonAI ACP server...")
+    await run_agent(PraisonACPAgent(), use_unstable_protocol=True)
+
+
+def serve(
+    workspace: Union[str, Path] = ".",
+    agent: str = "default",
+    model: Optional[str] = None,
+    resume: Optional[str] = None,
+    resume_last: bool = False,
+    debug: bool = False,
+    read_only: bool = True,
+    allow_write: bool = False,
+    allow_shell: bool = False,
+    approval_mode: str = "manual",
+    **kwargs: Any,
+) -> None:
+    """
+    Start the ACP server.
+    
+    This is the main entry point for running PraisonAI as an ACP agent.
+    
+    Args:
+        workspace: Working directory for the session
+        agent: Agent name or configuration
+        model: LLM model to use
+        resume: Session ID to resume
+        resume_last: Resume the last session
+        debug: Enable debug logging
+        read_only: Read-only mode (default True)
+        allow_write: Allow file writes
+        allow_shell: Allow shell commands
+        approval_mode: Approval mode (manual, auto, scoped)
+    """
+    config = ACPConfig(
+        workspace=Path(workspace).resolve(),
+        agent=agent,
+        model=model,
+        resume_session=resume,
+        resume_last=resume_last,
+        debug=debug,
+        read_only=read_only,
+        allow_write=allow_write,
+        allow_shell=allow_shell,
+        approval_mode=approval_mode,
+    )
+    
+    asyncio.run(_run_server(config))

src/praisonai/praisonai/acp/session.py

@@ -0,0 +1,219 @@
+"""
+ACP session management.
+
+Handles session creation, persistence, and resume functionality.
+"""
+
+import json
+import logging
+import time
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ACPSession:
+    """Represents an ACP conversation session."""
+    
+    session_id: str
+    workspace: Path
+    created_at: float = field(default_factory=time.time)
+    last_activity: float = field(default_factory=time.time)
+    
+    # Agent attribution
+    agent_id: Optional[str] = None
+    run_id: Optional[str] = None
+    trace_id: Optional[str] = None
+    
+    # Session state
+    mode: str = "manual"  # manual, auto, full_auto
+    model: Optional[str] = None
+    
+    # Conversation history (for resume)
+    messages: List[Dict[str, Any]] = field(default_factory=list)
+    tool_calls: List[Dict[str, Any]] = field(default_factory=list)
+    
+    # MCP servers
+    mcp_servers: List[Dict[str, Any]] = field(default_factory=list)
+    
+    @classmethod
+    def create(cls, workspace: Path, agent_id: Optional[str] = None) -> "ACPSession":
+        """Create a new session."""
+        session_id = f"sess_{uuid.uuid4().hex[:16]}"
+        return cls(
+            session_id=session_id,
+            workspace=workspace,
+            agent_id=agent_id,
+            run_id=f"run_{uuid.uuid4().hex[:8]}",
+            trace_id=f"trace_{uuid.uuid4().hex[:8]}",
+        )
+    
+    def update_activity(self) -> None:
+        """Update last activity timestamp."""
+        self.last_activity = time.time()
+    
+    def add_message(self, role: str, content: Any) -> None:
+        """Add a message to the conversation history."""
+        self.messages.append({
+            "role": role,
+            "content": content,
+            "timestamp": time.time(),
+        })
+        self.update_activity()
+    
+    def add_tool_call(self, tool_call_id: str, title: str, status: str, **kwargs) -> None:
+        """Add a tool call to the history."""
+        self.tool_calls.append({
+            "tool_call_id": tool_call_id,
+            "title": title,
+            "status": status,
+            "timestamp": time.time(),
+            **kwargs,
+        })
+        self.update_activity()
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize session to dictionary."""
+        return {
+            "session_id": self.session_id,
+            "workspace": str(self.workspace),
+            "created_at": self.created_at,
+            "last_activity": self.last_activity,
+            "agent_id": self.agent_id,
+            "run_id": self.run_id,
+            "trace_id": self.trace_id,
+            "mode": self.mode,
+            "model": self.model,
+            "messages": self.messages,
+            "tool_calls": self.tool_calls,
+            "mcp_servers": self.mcp_servers,
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ACPSession":
+        """Deserialize session from dictionary."""
+        return cls(
+            session_id=data["session_id"],
+            workspace=Path(data["workspace"]),
+            created_at=data.get("created_at", time.time()),
+            last_activity=data.get("last_activity", time.time()),
+            agent_id=data.get("agent_id"),
+            run_id=data.get("run_id"),
+            trace_id=data.get("trace_id"),
+            mode=data.get("mode", "manual"),
+            model=data.get("model"),
+            messages=data.get("messages", []),
+            tool_calls=data.get("tool_calls", []),
+            mcp_servers=data.get("mcp_servers", []),
+        )
+
+
+class SessionStore:
+    """Persistent storage for ACP sessions."""
+    
+    def __init__(self, storage_dir: Optional[Path] = None):
+        """Initialize session store."""
+        if storage_dir is None:
+            storage_dir = Path.home() / ".praison" / "acp" / "sessions"
+        self.storage_dir = storage_dir
+        self.storage_dir.mkdir(parents=True, exist_ok=True)
+        self._last_session_file = self.storage_dir / ".last_session"
+    
+    def _session_path(self, session_id: str) -> Path:
+        """Get path for session file."""
+        return self.storage_dir / f"{session_id}.json"
+    
+    def save(self, session: ACPSession) -> None:
+        """Save session to disk."""
+        try:
+            path = self._session_path(session.session_id)
+            with open(path, "w") as f:
+                json.dump(session.to_dict(), f, indent=2)
+            
+            # Update last session pointer
+            with open(self._last_session_file, "w") as f:
+                f.write(session.session_id)
+            
+            logger.debug(f"Saved session {session.session_id}")
+        except Exception as e:
+            logger.error(f"Failed to save session: {e}")
+    
+    def load(self, session_id: str) -> Optional[ACPSession]:
+        """Load session from disk."""
+        try:
+            path = self._session_path(session_id)
+            if not path.exists():
+                logger.warning(f"Session not found: {session_id}")
+                return None
+            
+            with open(path) as f:
+                data = json.load(f)
+            
+            return ACPSession.from_dict(data)
+        except Exception as e:
+            logger.error(f"Failed to load session: {e}")
+            return None
+    
+    def load_last(self) -> Optional[ACPSession]:
+        """Load the most recent session."""
+        try:
+            if not self._last_session_file.exists():
+                return None
+            
+            with open(self._last_session_file) as f:
+                session_id = f.read().strip()
+            
+            if session_id:
+                return self.load(session_id)
+            return None
+        except Exception as e:
+            logger.error(f"Failed to load last session: {e}")
+            return None
+    
+    def delete(self, session_id: str) -> bool:
+        """Delete a session."""
+        try:
+            path = self._session_path(session_id)
+            if path.exists():
+                path.unlink()
+                logger.debug(f"Deleted session {session_id}")
+                return True
+            return False
+        except Exception as e:
+            logger.error(f"Failed to delete session: {e}")
+            return False
+    
+    def list_sessions(self, limit: int = 50) -> List[ACPSession]:
+        """List all sessions, sorted by last activity."""
+        sessions = []
+        try:
+            for path in self.storage_dir.glob("sess_*.json"):
+                try:
+                    with open(path) as f:
+                        data = json.load(f)
+                    sessions.append(ACPSession.from_dict(data))
+                except Exception:
+                    continue
+            
+            # Sort by last activity, most recent first
+            sessions.sort(key=lambda s: s.last_activity, reverse=True)
+            return sessions[:limit]
+        except Exception as e:
+            logger.error(f"Failed to list sessions: {e}")
+            return []
+    
+    def cleanup_old_sessions(self, max_age_days: int = 30) -> int:
+        """Remove sessions older than max_age_days."""
+        cutoff = time.time() - (max_age_days * 24 * 60 * 60)
+        removed = 0
+        
+        for session in self.list_sessions(limit=1000):
+            if session.last_activity < cutoff:
+                if self.delete(session.session_id):
+                    removed += 1
+        
+        return removed

src/praisonai/praisonai/cli/features/acp.py

@@ -0,0 +1,236 @@
+"""
+ACP CLI handler for PraisonAI.
+
+Provides the `praisonai acp` command for IDE/editor integration.
+"""
+
+import argparse
+import logging
+import sys
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+class ACPHandler:
+    """Handler for ACP CLI commands."""
+    
+    @staticmethod
+    def add_arguments(parser: argparse.ArgumentParser) -> None:
+        """Add ACP-specific arguments to parser."""
+        parser.add_argument(
+            "-w", "--workspace",
+            type=str,
+            default=".",
+            help="Workspace root directory (default: current directory)"
+        )
+        parser.add_argument(
+            "-a", "--agent",
+            type=str,
+            default="default",
+            help="Agent name or configuration file"
+        )
+        parser.add_argument(
+            "--agents",
+            type=str,
+            help="Multi-agent configuration YAML file"
+        )
+        parser.add_argument(
+            "--router",
+            action="store_true",
+            help="Enable router agent for task delegation"
+        )
+        parser.add_argument(
+            "-m", "--model",
+            type=str,
+            help="LLM model to use"
+        )
+        parser.add_argument(
+            "-r", "--resume",
+            type=str,
+            nargs="?",
+            const="__last__",
+            help="Resume session by ID (or --resume --last for most recent)"
+        )
+        parser.add_argument(
+            "--last",
+            action="store_true",
+            help="Resume the last session (use with --resume)"
+        )
+        parser.add_argument(
+            "--approve",
+            type=str,
+            choices=["manual", "auto", "scoped"],
+            default="manual",
+            help="Approval mode for actions (default: manual)"
+        )
+        parser.add_argument(
+            "--read-only",
+            action="store_true",
+            default=True,
+            help="Read-only mode (default: enabled)"
+        )
+        parser.add_argument(
+            "--allow-write",
+            action="store_true",
+            help="Allow file write operations"
+        )
+        parser.add_argument(
+            "--allow-shell",
+            action="store_true",
+            help="Allow shell command execution"
+        )
+        parser.add_argument(
+            "--allow-network",
+            action="store_true",
+            help="Allow network requests"
+        )
+        parser.add_argument(
+            "--debug",
+            action="store_true",
+            help="Enable debug logging to stderr"
+        )
+        parser.add_argument(
+            "--profile",
+            type=str,
+            help="Use named profile from config"
+        )
+    
+    @staticmethod
+    def handle(args: argparse.Namespace) -> int:
+        """
+        Handle ACP command execution.
+        
+        Returns:
+            Exit code (0 for success, non-zero for error)
+        """
+        # Check if ACP SDK is available
+        try:
+            import importlib.util
+            if importlib.util.find_spec("acp") is None:
+                print(
+                    "Error: agent-client-protocol package not installed.\n"
+                    "Install with: pip install praisonai[acp]\n"
+                    "Or: pip install agent-client-protocol",
+                    file=sys.stderr
+                )
+                return 1
+        except Exception as e:
+            print(f"Error checking ACP availability: {e}", file=sys.stderr)
+            return 1
+        
+        # Import ACP module
+        try:
+            from praisonai.acp import serve
+        except ImportError as e:
+            print(f"Error importing ACP module: {e}", file=sys.stderr)
+            return 1
+        
+        # Determine resume session
+        resume_session: Optional[str] = None
+        resume_last = False
+        
+        if args.resume:
+            if args.resume == "__last__" or args.last:
+                resume_last = True
+            else:
+                resume_session = args.resume
+        elif args.last:
+            resume_last = True
+        
+        # Determine read-only mode
+        read_only = args.read_only and not args.allow_write
+        
+        try:
+            # Start ACP server
+            serve(
+                workspace=args.workspace,
+                agent=args.agent,
+                model=args.model,
+                resume=resume_session,
+                resume_last=resume_last,
+                debug=args.debug,
+                read_only=read_only,
+                allow_write=args.allow_write,
+                allow_shell=args.allow_shell,
+                approval_mode=args.approve,
+            )
+            return 0
+        except KeyboardInterrupt:
+            logger.info("ACP server stopped by user")
+            return 0
+        except Exception as e:
+            logger.exception(f"ACP server error: {e}")
+            print(f"Error: {e}", file=sys.stderr)
+            return 1
+
+
+def setup_acp_subparser(subparsers) -> None:
+    """Set up the ACP subcommand parser."""
+    acp_parser = subparsers.add_parser(
+        "acp",
+        help="Start ACP server for IDE/editor integration",
+        description=(
+            "Start the Agent Client Protocol (ACP) server.\n\n"
+            "This allows IDEs and editors (Zed, JetBrains, VSCode, Toad) "
+            "to connect to PraisonAI agents via JSON-RPC over stdio."
+        ),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Examples:
+  praisonai acp                    # Start with defaults
+  praisonai acp --debug            # Enable debug logging
+  praisonai acp --resume --last    # Resume last session
+  praisonai acp --allow-write      # Enable file writes
+  praisonai acp -m gpt-4o          # Use specific model
+
+Editor Configuration:
+
+  Zed (~/.config/zed/settings.json):
+    {
+      "agent_servers": {
+        "PraisonAI": {
+          "command": "praisonai",
+          "args": ["acp"],
+          "env": {}
+        }
+      }
+    }
+
+  JetBrains (~/.jetbrains/acp.json):
+    {
+      "agent_servers": {
+        "PraisonAI": {
+          "command": "praisonai",
+          "args": ["acp"],
+          "env": {}
+        }
+      }
+    }
+
+  Toad:
+    toad acp "praisonai acp"
+"""
+    )
+    ACPHandler.add_arguments(acp_parser)
+    acp_parser.set_defaults(func=lambda args: ACPHandler.handle(args))
+
+
+def run_acp_command(args: Optional[list] = None) -> int:
+    """
+    Run ACP command directly.
+    
+    This can be called from the main CLI or as a standalone entry point.
+    """
+    parser = argparse.ArgumentParser(
+        prog="praisonai acp",
+        description="Start ACP server for IDE/editor integration"
+    )
+    ACPHandler.add_arguments(parser)
+    
+    parsed_args = parser.parse_args(args)
+    return ACPHandler.handle(parsed_args)
+
+
+if __name__ == "__main__":
+    sys.exit(run_acp_command())

src/praisonai/praisonai/cli/main.py

@@ -698,7 +698,7 @@ class PraisonAI:
             return default_args
         
         # Define special commands
-        special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall']
+        special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp']
         
         parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
         parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -898,6 +898,12 @@ class PraisonAI:
             self._handle_serve_command(args, unknown_args)
             sys.exit(0)
 
+        # Handle ACP command - Agent Client Protocol server for IDE integration
+        if args.command == 'acp':
+            from praisonai.cli.features.acp import run_acp_command
+            exit_code = run_acp_command(unknown_args)
+            sys.exit(exit_code)
+
         # Handle both command and flag versions for call
         if args.command == 'call' or args.call:
             if not CALL_MODULE_AVAILABLE:

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

src/praisonai/praisonai/ui/chat.py

@@ -319,7 +319,6 @@ User Question: {message.content}
 
     message_history.append({"role": "user", "content": user_message})
     msg = cl.Message(content="")
-    await msg.send()
 
     completion_params = {
         "model": model_name,
@@ -349,6 +348,7 @@ User Question: {message.content}
     full_response = ""
     tool_calls = []
     current_tool_call = None
+    msg_sent = False
 
     async for part in response:
         if 'choices' in part and len(part['choices']) > 0:
@@ -356,6 +356,10 @@ User Question: {message.content}
 
             if 'content' in delta and delta['content'] is not None:
                 token = delta['content']
+                # Send message on first token to avoid delay
+                if not msg_sent:
+                    await msg.send()
+                    msg_sent = True
                 await msg.stream_token(token)
                 full_response += token
 
@@ -382,6 +386,10 @@ User Question: {message.content}
     if current_tool_call:
         tool_calls.append(current_tool_call)
 
+    # Ensure message is sent even if no content (e.g., tool calls only)
+    if not msg_sent:
+        await msg.send()
+
     logger.debug(f"Full response: {full_response}")
     logger.debug(f"Tool calls: {tool_calls}")
     message_history.append({"role": "assistant", "content": full_response})

src/praisonai/praisonai/ui/code.py

@@ -398,7 +398,6 @@ Context:
     message_history.append({"role": "user", "content": user_message})
 
     msg = cl.Message(content="")
-    await msg.send()
 
     # Use PraisonAI Agents if available, otherwise fallback to litellm
     if PRAISONAI_AGENTS_AVAILABLE:
@@ -433,41 +432,47 @@ Then use the Claude Code tool to handle those requests.
 
 For informational questions, explanations, or general conversations, respond normally without using Claude Code."""
 
-        # Create agent
+        # Create agent with streaming enabled
         agent = Agent(
             name="PraisonAI Assistant",
             instructions=instructions,
             llm=model_name,
-            tools=available_tools if available_tools else None
+            tools=available_tools if available_tools else None,
+            stream=True
         )
         
         # Execute agent with streaming
         full_response = ""
+        msg_sent = False
         
-        # Use agent's streaming capabilities if available
         try:
-            # For now, use synchronous execution and stream the result
-            # TODO: Implement proper streaming when PraisonAI agents support it
-            result = agent.start(user_message)
+            # Use async chat for proper streaming
+            result = await agent.achat(user_message)
             
-            # Stream the response character by character for better UX
+            # Get the response text
             if hasattr(result, 'raw'):
                 response_text = result.raw
             else:
                 response_text = str(result)
             
-            for char in response_text:
-                await msg.stream_token(char)
-                full_response += char
-                # Small delay to make streaming visible
-                await asyncio.sleep(0.01)
+            # Send message on first content
+            if not msg_sent:
+                await msg.send()
+                msg_sent = True
+            
+            # Stream in word chunks for better UX (not char-by-char which is too slow)
+            words = response_text.split(' ')
+            for i, word in enumerate(words):
+                token = word + (' ' if i < len(words) - 1 else '')
+                await msg.stream_token(token)
+                full_response += token
             
         except Exception as e:
             error_response = f"Error executing agent: {str(e)}"
-            for char in error_response:
-                await msg.stream_token(char)
-                full_response += char
-                await asyncio.sleep(0.01)
+            if not msg_sent:
+                await msg.send()
+            await msg.stream_token(error_response)
+            full_response = error_response
         
         msg.content = full_response
         await msg.update()
@@ -475,6 +480,7 @@ For informational questions, explanations, or general conversations, respond nor
     except Exception as e:
         error_msg = f"Failed to use PraisonAI Agents: {str(e)}"
         logger.error(error_msg)
+        await msg.send()
         await msg.stream_token(error_msg)
         msg.content = error_msg
         await msg.update()
@@ -515,6 +521,7 @@ async def handle_with_litellm(user_message, model_name, message_history, msg, im
     full_response = ""
     tool_calls = []
     current_tool_call = None
+    msg_sent = False
 
     async for part in response:
         logger.debug(f"LLM part: {part}")
@@ -523,6 +530,10 @@ async def handle_with_litellm(user_message, model_name, message_history, msg, im
             
             if 'content' in delta and delta['content'] is not None:
                 token = delta['content']
+                # Send message on first token
+                if not msg_sent:
+                    await msg.send()
+                    msg_sent = True
                 await msg.stream_token(token)
                 full_response += token
             
@@ -549,6 +560,10 @@ async def handle_with_litellm(user_message, model_name, message_history, msg, im
     if current_tool_call:
         tool_calls.append(current_tool_call)
 
+    # Ensure message is sent even if no content (tool calls only)
+    if not msg_sent:
+        await msg.send()
+
     logger.debug(f"Full response: {full_response}")
     logger.debug(f"Tool calls: {tool_calls}")
     message_history.append({"role": "assistant", "content": full_response})

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.8.7"
+__version__ = "2.8.9"

src/praisonai/pyproject.toml

@@ -100,6 +100,9 @@ autogen-v4 = [
     "praisonai-tools>=0.1.0", 
     "crewai"
 ]
+acp = [
+    "agent-client-protocol>=0.7.0",
+]
 
 [project.urls]
 Homepage = "https://docs.praison.ai"

src/praisonai/tests/integration/acp/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for ACP module."""

src/praisonai/tests/integration/acp/test_acp_real_llm.py

@@ -0,0 +1,206 @@
+"""
+Real LLM integration tests for ACP.
+
+These tests use actual API keys to verify end-to-end functionality.
+API keys should be set via environment variables.
+"""
+
+import os
+import tempfile
+from pathlib import Path
+
+import pytest
+
+# Skip if no API key
+pytestmark = pytest.mark.skipif(
+    not os.environ.get("OPENAI_API_KEY"),
+    reason="OPENAI_API_KEY not set"
+)
+
+
+class TestACPRealLLM:
+    """Real LLM tests for ACP server."""
+    
+    @pytest.fixture
+    def acp_server(self):
+        """Create ACP server with real agent."""
+        from praisonai.acp.server import ACPServer
+        from praisonai.acp.config import ACPConfig
+        
+        config = ACPConfig(
+            workspace=Path(tempfile.mkdtemp()),
+            debug=True,
+            model="gpt-4o-mini",
+        )
+        return ACPServer(config=config)
+    
+    @pytest.mark.asyncio
+    async def test_initialize_real(self, acp_server):
+        """Test real initialize handshake."""
+        result = await acp_server.initialize(
+            protocol_version=1,
+            client_capabilities={"fs": {"readTextFile": True}},
+            client_info={"name": "test-harness", "version": "1.0"},
+        )
+        
+        assert result["protocolVersion"] == 1
+        assert result["agentInfo"]["name"] == "praisonai"
+        assert "agentCapabilities" in result
+        
+        # Log for verification (masked)
+        print(f"[TEST] Initialize successful: protocol={result['protocolVersion']}")
+    
+    @pytest.mark.asyncio
+    async def test_session_lifecycle(self, acp_server):
+        """Test complete session lifecycle."""
+        # Initialize
+        await acp_server.initialize(protocol_version=1)
+        
+        # Create session
+        session_result = await acp_server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        )
+        
+        session_id = session_result["sessionId"]
+        assert session_id.startswith("sess_")
+        
+        # Log for verification
+        print(f"[TEST] Session created: {session_id[:20]}...")
+        
+        # Set mode
+        await acp_server.set_session_mode(
+            mode_id="manual",
+            session_id=session_id,
+        )
+        
+        # Cancel
+        await acp_server.cancel(session_id=session_id)
+        
+        print("[TEST] Session lifecycle complete")
+    
+    @pytest.mark.asyncio
+    async def test_prompt_with_real_llm(self, acp_server):
+        """Test prompt with real LLM call."""
+        # Initialize
+        await acp_server.initialize(protocol_version=1)
+        
+        # Create session
+        session_result = await acp_server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        )
+        session_id = session_result["sessionId"]
+        
+        # Send a simple prompt
+        prompt_result = await acp_server.prompt(
+            prompt=[{"type": "text", "text": "What is 2+2? Reply with just the number."}],
+            session_id=session_id,
+        )
+        
+        # Verify response
+        assert "stopReason" in prompt_result
+        assert prompt_result["stopReason"] in ["end_turn", "cancelled", "refusal"]
+        
+        # Log for verification (no secrets)
+        print(f"[TEST] Prompt completed: stopReason={prompt_result['stopReason']}")
+        print(f"[TEST] Session: {session_id[:20]}...")
+    
+    @pytest.mark.asyncio
+    async def test_session_resume(self, acp_server):
+        """Test session resume functionality."""
+        # Initialize
+        await acp_server.initialize(protocol_version=1)
+        
+        # Create session
+        session_result = await acp_server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        )
+        session_id = session_result["sessionId"]
+        
+        # Add a message to the session
+        acp_server._sessions[session_id].add_message("user", "Test message")
+        acp_server._session_store.save(acp_server._sessions[session_id])
+        
+        # Create new server instance (simulating restart)
+        from praisonai.acp.config import ACPConfig
+        config = ACPConfig(workspace=Path("/tmp/test"), debug=True)
+        from praisonai.acp.server import ACPServer as ACPServerClass
+        new_server = ACPServerClass(config=config)
+        
+        # Resume session
+        resume_result = await new_server.load_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+            session_id=session_id,
+        )
+        
+        assert resume_result is not None
+        assert session_id in new_server._sessions
+        
+        # Verify message was preserved
+        loaded_session = new_server._sessions[session_id]
+        assert len(loaded_session.messages) == 1
+        assert loaded_session.messages[0]["content"] == "Test message"
+        
+        print(f"[TEST] Session resume successful: {session_id[:20]}...")
+
+
+class TestACPPermissionGating:
+    """Tests for permission gating."""
+    
+    def test_read_only_default(self):
+        """Verify read-only is default."""
+        from praisonai.acp.config import ACPConfig
+        
+        config = ACPConfig()
+        assert config.read_only is True
+        assert config.can_write() is False
+        assert config.can_execute_shell() is False
+    
+    def test_explicit_write_permission(self):
+        """Test explicit write permission."""
+        from praisonai.acp.config import ACPConfig
+        
+        config = ACPConfig(allow_write=True)
+        assert config.can_write() is True
+    
+    def test_workspace_boundary(self):
+        """Test workspace boundary enforcement."""
+        from praisonai.acp.config import ACPConfig
+        
+        config = ACPConfig(workspace="/tmp/test_workspace")
+        
+        # Within workspace - allowed
+        assert config.is_path_allowed(Path("/tmp/test_workspace/file.txt"))
+        assert config.is_path_allowed(Path("/tmp/test_workspace/subdir/file.txt"))
+        
+        # Outside workspace - not allowed
+        assert not config.is_path_allowed(Path("/etc/passwd"))
+        assert not config.is_path_allowed(Path("/var/log/system.log"))
+
+
+class TestACPTracing:
+    """Tests for tracing and attribution."""
+    
+    def test_session_has_trace_ids(self):
+        """Verify sessions have trace IDs."""
+        from praisonai.acp.session import ACPSession
+        
+        session = ACPSession.create(
+            workspace=Path("/tmp/test"),
+            agent_id="test_agent",
+        )
+        
+        assert session.session_id is not None
+        assert session.run_id is not None
+        assert session.trace_id is not None
+        assert session.agent_id == "test_agent"
+        
+        # IDs should have proper prefixes
+        assert session.session_id.startswith("sess_")
+        assert session.run_id.startswith("run_")
+        assert session.trace_id.startswith("trace_")
+        
+        print(f"[TEST] Trace IDs: session={session.session_id[:15]}, run={session.run_id}, trace={session.trace_id}")

src/praisonai/tests/integration/acp/test_acp_stdio.py

@@ -0,0 +1,324 @@
+"""
+Integration tests for ACP stdio communication.
+
+Tests the ACP server by spawning it as a subprocess and communicating
+via JSON-RPC over stdio.
+"""
+
+import json
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import pytest
+
+
+class TestACPStdioIntegration:
+    """Integration tests for ACP stdio communication."""
+    
+    @pytest.fixture
+    def acp_process(self):
+        """Start ACP server as subprocess."""
+        # Use the CLI module directly to avoid crewai import issues
+        proc = subprocess.Popen(
+            [
+                sys.executable, "-c",
+                """
+import sys
+import asyncio
+from praisonai.acp.server import ACPServer
+from praisonai.acp.config import ACPConfig
+
+async def run():
+    config = ACPConfig(debug=True)
+    server = ACPServer(config=config)
+    
+    # Simple JSON-RPC handler for testing
+    import json
+    
+    while True:
+        try:
+            line = sys.stdin.readline()
+            if not line:
+                break
+            
+            request = json.loads(line.strip())
+            method = request.get("method", "")
+            params = request.get("params", {})
+            req_id = request.get("id")
+            
+            if method == "initialize":
+                result = await server.initialize(
+                    protocol_version=params.get("protocolVersion", 1),
+                    client_capabilities=params.get("clientCapabilities"),
+                    client_info=params.get("clientInfo"),
+                )
+            elif method == "session/new":
+                result = await server.new_session(
+                    cwd=params.get("cwd", "."),
+                    mcp_servers=params.get("mcpServers", []),
+                )
+            elif method == "session/cancel":
+                await server.cancel(session_id=params.get("sessionId", ""))
+                result = None
+            else:
+                result = {"error": f"Unknown method: {method}"}
+            
+            response = {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "result": result,
+            }
+            print(json.dumps(response), flush=True)
+        except Exception as e:
+            print(json.dumps({"jsonrpc": "2.0", "id": None, "error": {"code": -32603, "message": str(e)}}), flush=True)
+
+asyncio.run(run())
+"""
+            ],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+        )
+        yield proc
+        proc.terminate()
+        proc.wait(timeout=5)
+    
+    def send_request(self, proc, method: str, params: dict, req_id: int = 1) -> dict:
+        """Send JSON-RPC request and get response."""
+        request = {
+            "jsonrpc": "2.0",
+            "id": req_id,
+            "method": method,
+            "params": params,
+        }
+        proc.stdin.write(json.dumps(request) + "\n")
+        proc.stdin.flush()
+        
+        # Read response with timeout
+        response_line = proc.stdout.readline()
+        if response_line:
+            return json.loads(response_line.strip())
+        return None
+    
+    def test_initialize(self, acp_process):
+        """Test initialize handshake."""
+        response = self.send_request(
+            acp_process,
+            "initialize",
+            {
+                "protocolVersion": 1,
+                "clientCapabilities": {"fs": {"readTextFile": True}},
+                "clientInfo": {"name": "test-client", "version": "1.0"},
+            },
+        )
+        
+        assert response is not None
+        assert response["jsonrpc"] == "2.0"
+        assert response["id"] == 1
+        assert "result" in response
+        
+        result = response["result"]
+        assert result["protocolVersion"] == 1
+        assert "agentCapabilities" in result
+        assert result["agentInfo"]["name"] == "praisonai"
+    
+    def test_new_session(self, acp_process):
+        """Test session creation."""
+        # First initialize
+        self.send_request(
+            acp_process,
+            "initialize",
+            {"protocolVersion": 1},
+            req_id=1,
+        )
+        
+        # Then create session
+        response = self.send_request(
+            acp_process,
+            "session/new",
+            {
+                "cwd": "/tmp/test",
+                "mcpServers": [],
+            },
+            req_id=2,
+        )
+        
+        assert response is not None
+        assert response["id"] == 2
+        assert "result" in response
+        
+        result = response["result"]
+        assert "sessionId" in result
+        assert result["sessionId"].startswith("sess_")
+    
+    def test_cancel_notification(self, acp_process):
+        """Test cancel notification."""
+        # Initialize
+        self.send_request(acp_process, "initialize", {"protocolVersion": 1}, req_id=1)
+        
+        # Create session
+        new_resp = self.send_request(
+            acp_process,
+            "session/new",
+            {"cwd": "/tmp", "mcpServers": []},
+            req_id=2,
+        )
+        session_id = new_resp["result"]["sessionId"]
+        
+        # Send cancel (notification, but we're treating it as request for testing)
+        response = self.send_request(
+            acp_process,
+            "session/cancel",
+            {"sessionId": session_id},
+            req_id=3,
+        )
+        
+        assert response is not None
+        assert response["id"] == 3
+
+
+class TestACPNoStdoutPollution:
+    """Tests to ensure ACP doesn't pollute stdout."""
+    
+    def test_no_print_statements(self):
+        """Verify ACP modules don't have print statements that would pollute stdout."""
+        import ast
+        from pathlib import Path
+        
+        acp_dir = Path(__file__).parent.parent.parent.parent / "praisonai" / "acp"
+        
+        for py_file in acp_dir.glob("*.py"):
+            content = py_file.read_text()
+            try:
+                tree = ast.parse(content)
+                for node in ast.walk(tree):
+                    if isinstance(node, ast.Call):
+                        if isinstance(node.func, ast.Name) and node.func.id == "print":
+                            # Check if it's inside a function that's not for debugging
+                            pytest.fail(f"Found print() in {py_file.name} - stdout must be reserved for JSON-RPC")
+            except SyntaxError:
+                pass  # Skip files with syntax errors
+    
+    def test_logging_to_stderr(self):
+        """Verify logging is configured for stderr."""
+        from praisonai.acp.server import _setup_stderr_logging
+        import logging
+        import sys
+        
+        _setup_stderr_logging(debug=True)
+        
+        logger = logging.getLogger("praisonai.acp")
+        
+        # Check that handlers write to stderr
+        for handler in logger.handlers:
+            if hasattr(handler, 'stream'):
+                assert handler.stream == sys.stderr, "Logger must write to stderr, not stdout"
+
+
+class TestACPInteractiveModeCompatibility:
+    """Tests for ACP and interactive mode compatibility."""
+    
+    def test_acp_import_does_not_break_interactive(self):
+        """Verify importing ACP doesn't break interactive mode imports."""
+        # This should not raise any errors
+        from praisonai.acp import ACPConfig
+        
+        # Interactive mode imports should still work
+        from praisonai.cli.features.interactive_tui import InteractiveConfig
+        
+        # Both should be usable
+        acp_config = ACPConfig()
+        interactive_config = InteractiveConfig()
+        
+        assert acp_config is not None
+        assert interactive_config is not None
+    
+    def test_acp_lazy_import(self):
+        """Verify ACP uses lazy imports."""
+        import time
+        
+        start = time.time()
+        import praisonai.acp as acp_module  # noqa: F401
+        import_time = time.time() - start
+        
+        # Import should be fast (< 100ms)
+        assert import_time < 0.1, f"ACP import too slow: {import_time}s"
+    
+    def test_interactive_mode_without_acp_sdk(self):
+        """Verify interactive mode works even if ACP SDK is not installed."""
+        # Mock the ACP SDK not being available
+        import sys
+        
+        # Save original
+        original_modules = sys.modules.copy()
+        
+        try:
+            # Remove acp from modules if present
+            if 'acp' in sys.modules:
+                del sys.modules['acp']
+            
+            # Interactive mode should still work
+            from praisonai.cli.features.interactive_tui import InteractiveSession
+            
+            session = InteractiveSession()
+            assert session is not None
+        finally:
+            # Restore
+            sys.modules.update(original_modules)
+
+
+class TestACPSessionPersistence:
+    """Tests for ACP session persistence."""
+    
+    def test_session_save_and_load(self):
+        """Test session persistence across server restarts."""
+        from praisonai.acp.session import ACPSession, SessionStore
+        
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create and save session
+            session = ACPSession.create(workspace=Path("/tmp/test"))
+            session.add_message("user", "Hello")
+            session.add_message("assistant", "Hi there!")
+            store.save(session)
+            
+            # Create new store instance (simulating restart)
+            store2 = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Load session
+            loaded = store2.load(session.session_id)
+            
+            assert loaded is not None
+            assert loaded.session_id == session.session_id
+            assert len(loaded.messages) == 2
+            assert loaded.messages[0]["content"] == "Hello"
+    
+    def test_resume_last_session(self):
+        """Test --resume --last functionality."""
+        from praisonai.acp.session import ACPSession, SessionStore
+        
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create multiple sessions
+            session1 = ACPSession.create(workspace=Path("/tmp/test1"))
+            store.save(session1)
+            
+            time.sleep(0.01)  # Ensure different timestamps
+            
+            session2 = ACPSession.create(workspace=Path("/tmp/test2"))
+            session2.add_message("user", "Last session message")
+            store.save(session2)
+            
+            # Load last session
+            last = store.load_last()
+            
+            assert last is not None
+            assert last.session_id == session2.session_id
+            assert len(last.messages) == 1

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

@@ -0,0 +1 @@
+"""Unit tests for ACP module."""

src/praisonai/tests/unit/acp/test_config.py

@@ -0,0 +1,166 @@
+"""Unit tests for ACP configuration."""
+
+import os
+import tempfile
+from pathlib import Path
+
+from praisonai.acp.config import ACPConfig
+
+
+class TestACPConfig:
+    """Tests for ACPConfig class."""
+    
+    def test_default_config(self):
+        """Test default configuration values."""
+        config = ACPConfig()
+        
+        assert config.workspace == Path.cwd()
+        assert config.agent == "default"
+        assert config.agents_config is None
+        assert config.router_enabled is False
+        assert config.model is None
+        assert config.resume_session is None
+        assert config.resume_last is False
+        assert config.read_only is True
+        assert config.allow_write is False
+        assert config.allow_shell is False
+        assert config.allow_network is False
+        assert config.approval_mode == "manual"
+        assert config.debug is False
+        assert config.profile is None
+    
+    def test_workspace_normalization(self):
+        """Test workspace path normalization."""
+        config = ACPConfig(workspace=".")
+        assert config.workspace == Path.cwd()
+        
+        config = ACPConfig(workspace="/tmp")
+        # Use resolve() to handle macOS symlinks (/tmp -> /private/tmp)
+        assert config.workspace == Path("/tmp").resolve()
+    
+    def test_allowed_paths_default(self):
+        """Test default allowed paths."""
+        config = ACPConfig()
+        
+        assert config.workspace in config.allowed_paths
+        assert Path.home() / ".praison" in config.allowed_paths
+    
+    def test_is_path_allowed(self):
+        """Test path allowlist checking."""
+        config = ACPConfig(workspace="/tmp/test_workspace")
+        
+        # Workspace itself should be allowed
+        assert config.is_path_allowed(Path("/tmp/test_workspace"))
+        
+        # Subdirectories should be allowed
+        assert config.is_path_allowed(Path("/tmp/test_workspace/subdir"))
+        
+        # Parent directories should NOT be allowed
+        assert not config.is_path_allowed(Path("/tmp"))
+        
+        # Unrelated paths should NOT be allowed
+        assert not config.is_path_allowed(Path("/var/log"))
+    
+    def test_can_write(self):
+        """Test write permission checking."""
+        # Read-only mode
+        config = ACPConfig(read_only=True, allow_write=False)
+        assert not config.can_write()
+        
+        # Allow write overrides read-only
+        config = ACPConfig(read_only=True, allow_write=True)
+        assert config.can_write()
+        
+        # Not read-only
+        config = ACPConfig(read_only=False)
+        assert config.can_write()
+    
+    def test_can_execute_shell(self):
+        """Test shell execution permission checking."""
+        config = ACPConfig(allow_shell=False)
+        assert not config.can_execute_shell()
+        
+        config = ACPConfig(allow_shell=True)
+        assert config.can_execute_shell()
+    
+    def test_from_env(self, monkeypatch):
+        """Test configuration from environment variables."""
+        monkeypatch.setenv("PRAISONAI_WORKSPACE", "/tmp/env_workspace")
+        monkeypatch.setenv("PRAISONAI_AGENT", "test_agent")
+        monkeypatch.setenv("PRAISONAI_MODEL", "gpt-4")
+        monkeypatch.setenv("PRAISONAI_DEBUG", "true")
+        monkeypatch.setenv("PRAISONAI_READ_ONLY", "false")
+        monkeypatch.setenv("PRAISONAI_APPROVAL_MODE", "auto")
+        
+        config = ACPConfig.from_env()
+        
+        # Use resolve() to handle macOS symlinks
+        assert config.workspace == Path("/tmp/env_workspace").resolve()
+        assert config.agent == "test_agent"
+        assert config.model == "gpt-4"
+        assert config.debug is True
+        assert config.read_only is False
+        assert config.approval_mode == "auto"
+    
+    def test_from_file(self):
+        """Test configuration from YAML file."""
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
+            f.write("""
+acp:
+  workspace: /tmp/file_workspace
+  agent: file_agent
+  model: gpt-3.5-turbo
+  debug: true
+  read_only: false
+  approval_mode: scoped
+""")
+            f.flush()
+            
+            config = ACPConfig.from_file(Path(f.name))
+            
+            # Use resolve() to handle macOS symlinks
+            assert config.workspace == Path("/tmp/file_workspace").resolve()
+            assert config.agent == "file_agent"
+            assert config.model == "gpt-3.5-turbo"
+            assert config.debug is True
+            assert config.read_only is False
+            assert config.approval_mode == "scoped"
+        
+        os.unlink(f.name)
+    
+    def test_from_file_missing(self):
+        """Test configuration from non-existent file."""
+        config = ACPConfig.from_file(Path("/nonexistent/config.yaml"))
+        
+        # Should return default config
+        assert config.agent == "default"
+        assert config.debug is False
+    
+    def test_to_dict(self):
+        """Test configuration serialization."""
+        config = ACPConfig(
+            workspace="/tmp/test",
+            agent="test_agent",
+            model="gpt-4",
+            debug=True,
+        )
+        
+        data = config.to_dict()
+        
+        # Use resolve() to handle macOS symlinks
+        assert data["workspace"] == str(Path("/tmp/test").resolve())
+        assert data["agent"] == "test_agent"
+        assert data["model"] == "gpt-4"
+        assert data["debug"] is True
+    
+    def test_merge(self):
+        """Test configuration merging."""
+        config1 = ACPConfig(agent="agent1", model="model1")
+        config2 = ACPConfig(agent="agent2", debug=True)
+        
+        merged = ACPConfig.merge(config1, config2)
+        
+        # Later config overrides
+        assert merged.agent == "agent2"
+        assert merged.model == "model1"  # Not overridden
+        assert merged.debug is True

src/praisonai/tests/unit/acp/test_server.py

@@ -0,0 +1,244 @@
+"""Unit tests for ACP server."""
+
+import asyncio
+from pathlib import Path
+from praisonai.acp.config import ACPConfig
+from praisonai.acp.server import ACPServer
+
+
+class TestACPServer:
+    """Tests for ACPServer class."""
+    
+    def test_server_init(self):
+        """Test server initialization."""
+        config = ACPConfig(workspace="/tmp/test", debug=True)
+        server = ACPServer(config=config)
+        
+        assert server.config == config
+        assert server._sessions == {}
+        assert server._cancelled_sessions == set()
+    
+    def test_server_init_default_config(self):
+        """Test server initialization with default config."""
+        server = ACPServer()
+        
+        assert server.config is not None
+        assert server.config.workspace == Path.cwd()
+    
+    def test_initialize(self):
+        """Test initialize handler."""
+        server = ACPServer()
+        
+        result = asyncio.run(server.initialize(
+            protocol_version=1,
+            client_capabilities={"fs": {"readTextFile": True}},
+            client_info={"name": "test-client", "version": "1.0"},
+        ))
+        
+        assert result["protocolVersion"] == 1
+        assert "agentCapabilities" in result
+        assert result["agentCapabilities"]["loadSession"] is True
+        assert result["agentInfo"]["name"] == "praisonai"
+    
+    def test_new_session(self):
+        """Test new_session handler."""
+        server = ACPServer()
+        
+        result = asyncio.run(server.new_session(
+            cwd="/tmp/test_workspace",
+            mcp_servers=[],
+        ))
+        
+        assert "sessionId" in result
+        assert result["sessionId"].startswith("sess_")
+        assert result["sessionId"] in server._sessions
+    
+    def test_new_session_with_mcp_servers(self):
+        """Test new_session with MCP servers."""
+        server = ACPServer()
+        
+        mcp_servers = [
+            {"name": "test", "command": "/usr/bin/test", "args": []}
+        ]
+        
+        result = asyncio.run(server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=mcp_servers,
+        ))
+        
+        session = server._sessions[result["sessionId"]]
+        assert session.mcp_servers == mcp_servers
+    
+    def test_set_session_mode(self):
+        """Test set_session_mode handler."""
+        server = ACPServer()
+        
+        # Create a session first
+        new_result = asyncio.run(server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        ))
+        session_id = new_result["sessionId"]
+        
+        # Set mode
+        result = asyncio.run(server.set_session_mode(
+            mode_id="auto",
+            session_id=session_id,
+        ))
+        
+        assert result == {}
+        assert server._sessions[session_id].mode == "auto"
+    
+    def test_cancel(self):
+        """Test cancel handler."""
+        server = ACPServer()
+        
+        # Create a session first
+        new_result = asyncio.run(server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        ))
+        session_id = new_result["sessionId"]
+        
+        # Cancel it
+        asyncio.run(server.cancel(session_id=session_id))
+        
+        assert session_id in server._cancelled_sessions
+    
+    def test_authenticate(self):
+        """Test authenticate handler."""
+        server = ACPServer()
+        
+        result = asyncio.run(server.authenticate(method_id="test"))
+        
+        assert result == {}
+    
+    def test_list_sessions(self):
+        """Test list_sessions handler."""
+        import tempfile
+        with tempfile.TemporaryDirectory() as tmpdir:
+            from praisonai.acp.session import SessionStore
+            server = ACPServer()
+            # Use isolated session store
+            server._session_store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create some sessions
+            asyncio.run(server.new_session(cwd="/tmp/test1", mcp_servers=[]))
+            asyncio.run(server.new_session(cwd="/tmp/test2", mcp_servers=[]))
+            
+            result = asyncio.run(server.list_sessions())
+            
+            assert "sessions" in result
+            assert len(result["sessions"]) == 2
+    
+    def test_fork_session(self):
+        """Test fork_session handler."""
+        server = ACPServer()
+        
+        # Create original session
+        original = asyncio.run(server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        ))
+        original_id = original["sessionId"]
+        
+        # Add a message to original
+        server._sessions[original_id].add_message("user", "Hello")
+        
+        # Fork it
+        forked = asyncio.run(server.fork_session(
+            cwd="/tmp/test2",
+            session_id=original_id,
+        ))
+        
+        assert forked["sessionId"] != original_id
+        assert forked["sessionId"] in server._sessions
+        
+        # Forked session should have the message
+        forked_session = server._sessions[forked["sessionId"]]
+        assert len(forked_session.messages) == 1
+    
+    def test_extract_prompt_text(self):
+        """Test prompt text extraction."""
+        server = ACPServer()
+        
+        prompt = [
+            {"type": "text", "text": "Hello world"},
+            {"type": "text", "text": "How are you?"},
+        ]
+        
+        result = server._extract_prompt_text(prompt)
+        
+        assert "Hello world" in result
+        assert "How are you?" in result
+    
+    def test_extract_prompt_text_with_resource(self):
+        """Test prompt text extraction with resource."""
+        server = ACPServer()
+        
+        prompt = [
+            {"type": "text", "text": "Check this file:"},
+            {
+                "type": "resource",
+                "resource": {
+                    "uri": "file:///tmp/test.py",
+                    "text": "print('hello')",
+                }
+            },
+        ]
+        
+        result = server._extract_prompt_text(prompt)
+        
+        assert "Check this file:" in result
+        assert "print('hello')" in result
+        assert "file:///tmp/test.py" in result
+    
+    def test_get_available_modes(self):
+        """Test available modes."""
+        server = ACPServer()
+        
+        modes = server._get_available_modes()
+        
+        assert modes is not None
+        assert len(modes) >= 2
+        
+        mode_ids = [m["id"] for m in modes]
+        assert "manual" in mode_ids
+        assert "auto" in mode_ids
+
+
+class TestACPServerPrompt:
+    """Tests for ACPServer prompt handling."""
+    
+    def test_prompt_cancelled(self):
+        """Test prompt with cancelled session."""
+        server = ACPServer()
+        
+        # Create session
+        new_result = asyncio.run(server.new_session(
+            cwd="/tmp/test",
+            mcp_servers=[],
+        ))
+        session_id = new_result["sessionId"]
+        
+        # Mark as cancelled
+        server._cancelled_sessions.add(session_id)
+        
+        # Try to prompt
+        result = asyncio.run(server.prompt(
+            prompt=[{"type": "text", "text": "Hello"}],
+            session_id=session_id,
+        ))
+        
+        assert result["stopReason"] == "cancelled"
+    
+    def test_prompt_nonexistent_session(self):
+        """Test prompt with non-existent session."""
+        server = ACPServer()
+        
+        result = asyncio.run(server.prompt(
+            prompt=[{"type": "text", "text": "Hello"}],
+            session_id="nonexistent",
+        ))
+        
+        assert result["stopReason"] == "refusal"

src/praisonai/tests/unit/acp/test_session.py

@@ -0,0 +1,203 @@
+"""Unit tests for ACP session management."""
+
+import tempfile
+import time
+from pathlib import Path
+
+from praisonai.acp.session import ACPSession, SessionStore
+
+
+class TestACPSession:
+    """Tests for ACPSession class."""
+    
+    def test_create_session(self):
+        """Test session creation."""
+        session = ACPSession.create(
+            workspace=Path("/tmp/test"),
+            agent_id="test_agent",
+        )
+        
+        assert session.session_id.startswith("sess_")
+        assert session.workspace == Path("/tmp/test")
+        assert session.agent_id == "test_agent"
+        assert session.run_id is not None
+        assert session.trace_id is not None
+        assert session.mode == "manual"
+    
+    def test_update_activity(self):
+        """Test activity timestamp update."""
+        session = ACPSession.create(workspace=Path("/tmp"))
+        original_time = session.last_activity
+        
+        time.sleep(0.01)
+        session.update_activity()
+        
+        assert session.last_activity > original_time
+    
+    def test_add_message(self):
+        """Test adding messages to session."""
+        session = ACPSession.create(workspace=Path("/tmp"))
+        
+        session.add_message("user", "Hello")
+        session.add_message("assistant", "Hi there!")
+        
+        assert len(session.messages) == 2
+        assert session.messages[0]["role"] == "user"
+        assert session.messages[0]["content"] == "Hello"
+        assert session.messages[1]["role"] == "assistant"
+        assert session.messages[1]["content"] == "Hi there!"
+    
+    def test_add_tool_call(self):
+        """Test adding tool calls to session."""
+        session = ACPSession.create(workspace=Path("/tmp"))
+        
+        session.add_tool_call(
+            tool_call_id="call_001",
+            title="Read file",
+            status="completed",
+            path="/tmp/test.txt",
+        )
+        
+        assert len(session.tool_calls) == 1
+        assert session.tool_calls[0]["tool_call_id"] == "call_001"
+        assert session.tool_calls[0]["title"] == "Read file"
+        assert session.tool_calls[0]["status"] == "completed"
+        assert session.tool_calls[0]["path"] == "/tmp/test.txt"
+    
+    def test_to_dict(self):
+        """Test session serialization."""
+        session = ACPSession.create(
+            workspace=Path("/tmp/test"),
+            agent_id="test_agent",
+        )
+        session.add_message("user", "Test message")
+        
+        data = session.to_dict()
+        
+        assert data["session_id"] == session.session_id
+        assert data["workspace"] == "/tmp/test"
+        assert data["agent_id"] == "test_agent"
+        assert len(data["messages"]) == 1
+    
+    def test_from_dict(self):
+        """Test session deserialization."""
+        data = {
+            "session_id": "sess_test123",
+            "workspace": "/tmp/test",
+            "created_at": 1234567890.0,
+            "last_activity": 1234567891.0,
+            "agent_id": "test_agent",
+            "run_id": "run_abc",
+            "trace_id": "trace_xyz",
+            "mode": "auto",
+            "model": "gpt-4",
+            "messages": [{"role": "user", "content": "Hello"}],
+            "tool_calls": [],
+            "mcp_servers": [],
+        }
+        
+        session = ACPSession.from_dict(data)
+        
+        assert session.session_id == "sess_test123"
+        assert session.workspace == Path("/tmp/test")
+        assert session.agent_id == "test_agent"
+        assert session.mode == "auto"
+        assert session.model == "gpt-4"
+        assert len(session.messages) == 1
+
+
+class TestSessionStore:
+    """Tests for SessionStore class."""
+    
+    def test_save_and_load(self):
+        """Test saving and loading sessions."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            session = ACPSession.create(workspace=Path("/tmp/test"))
+            session.add_message("user", "Test")
+            
+            store.save(session)
+            
+            loaded = store.load(session.session_id)
+            
+            assert loaded is not None
+            assert loaded.session_id == session.session_id
+            assert len(loaded.messages) == 1
+    
+    def test_load_nonexistent(self):
+        """Test loading non-existent session."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            loaded = store.load("nonexistent_session")
+            
+            assert loaded is None
+    
+    def test_load_last(self):
+        """Test loading the last session."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create and save multiple sessions
+            session1 = ACPSession.create(workspace=Path("/tmp/test1"))
+            store.save(session1)
+            
+            session2 = ACPSession.create(workspace=Path("/tmp/test2"))
+            store.save(session2)
+            
+            # Last saved should be session2
+            last = store.load_last()
+            
+            assert last is not None
+            assert last.session_id == session2.session_id
+    
+    def test_delete(self):
+        """Test deleting a session."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            session = ACPSession.create(workspace=Path("/tmp/test"))
+            store.save(session)
+            
+            # Verify it exists
+            assert store.load(session.session_id) is not None
+            
+            # Delete it
+            result = store.delete(session.session_id)
+            assert result is True
+            
+            # Verify it's gone
+            assert store.load(session.session_id) is None
+    
+    def test_list_sessions(self):
+        """Test listing sessions."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create multiple sessions
+            for i in range(3):
+                session = ACPSession.create(workspace=Path(f"/tmp/test{i}"))
+                store.save(session)
+                time.sleep(0.01)  # Ensure different timestamps
+            
+            sessions = store.list_sessions()
+            
+            assert len(sessions) == 3
+            # Should be sorted by last_activity, most recent first
+            assert sessions[0].last_activity >= sessions[1].last_activity
+            assert sessions[1].last_activity >= sessions[2].last_activity
+    
+    def test_list_sessions_limit(self):
+        """Test listing sessions with limit."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            store = SessionStore(storage_dir=Path(tmpdir))
+            
+            # Create multiple sessions
+            for i in range(5):
+                session = ACPSession.create(workspace=Path(f"/tmp/test{i}"))
+                store.save(session)
+            
+            sessions = store.list_sessions(limit=2)
+            
+            assert len(sessions) == 2

src/praisonai/uv.lock

@@ -9,6 +9,18 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 
+[[package]]
+name = "agent-client-protocol"
+version = "0.7.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/db/7c/12da39be4f73026fd9b02144df5f64d803488cf1439aa221b0edb7c305e3/agent_client_protocol-0.7.1.tar.gz", hash = "sha256:8d7031209e14c3f2f987e3b95e7d9c3286158e7b2af1bf43d6aae5b8a429249f", size = 66226, upload-time = "2025-12-28T13:58:57.012Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/03/48/48d2fb454f911147432cd779f548e188274e1700f1cbe0a258e78158331a/agent_client_protocol-0.7.1-py3-none-any.whl", hash = "sha256:4ffe999488f2b23db26f09becdfaa2aaae6529f0847a52bca61bc2c628001c0f", size = 53771, upload-time = "2025-12-28T13:58:55.967Z" },
+]
+
 [[package]]
 name = "agentops"
 version = "0.3.16"
@@ -3979,6 +3991,9 @@ dependencies = [
 ]
 
 [package.optional-dependencies]
+acp = [
+    { name = "agent-client-protocol" },
+]
 agentops = [
     { name = "agentops" },
 ]
@@ -4075,6 +4090,7 @@ ui = [
 
 [package.metadata]
 requires-dist = [
+    { name = "agent-client-protocol", marker = "extra == 'acp'", specifier = ">=0.7.0" },
     { name = "agentops", marker = "extra == 'agentops'", specifier = ">=0.3.12" },
     { name = "aiosqlite", marker = "extra == 'chat'", specifier = ">=0.20.0" },
     { name = "aiosqlite", marker = "extra == 'code'", specifier = ">=0.20.0" },
@@ -4147,7 +4163,7 @@ requires-dist = [
     { name = "websockets", marker = "extra == 'realtime'", specifier = ">=12.0" },
     { name = "yfinance", marker = "extra == 'realtime'", specifier = ">=0.2.44" },
 ]
-provides-extras = ["ui", "gradio", "api", "agentops", "google", "openai", "anthropic", "cohere", "chat", "code", "realtime", "call", "train", "crewai", "autogen", "autogen-v4"]
+provides-extras = ["ui", "gradio", "api", "agentops", "google", "openai", "anthropic", "cohere", "chat", "code", "realtime", "call", "train", "crewai", "autogen", "autogen-v4", "acp"]
 
 [[package]]
 name = "praisonai-tools"