diff --git a/mcp/README.md b/mcp/README.md
new file mode 100644
index 000000000..56c59df7c
--- /dev/null
+++ b/mcp/README.md
@@ -0,0 +1,233 @@
+# ROCK MCP Tools
+
+ROCK MCP Tools - 为OpenClaw Agent提供的阿里ROCK环境管理工具包
+
+## 🎯 功能特性
+
+### 核心工具
+- `rock_create_agent` - 创建新的agent实例
+- `rock_create_sandbox` - 创建隔离的沙箱环境
+- `rock_list_agents` - 列出所有活跃的agents
+- `rock_clone_agent` - agent自我复制功能
+- `rock_execute_action` - 执行ROCK环境操作
+- `rock_get_status` - 获取系统状态
+- `rock_cleanup` - 清理资源
+
+### 高级功能
+- 🔄 **自我复制** - agent可以创建自己的副本
+- 🏗️ **环境隔离** - 每个agent运行在独立的沙箱中
+- 📊 **资源管理** - 自动管理CPU、内存等资源
+- 🔒 **安全隔离** - 多层隔离机制确保安全
+- ⚡ **分布式部署** - 支持跨机器部署
+
+## 🚀 快速开始
+
+### 安装依赖
+```bash
+pip install mcp asyncio httpx
+pip install rl-rock  # ROCK SDK
+```
+
+### 启动MCP服务器
+```bash
+python rock_mcp_server.py
+```
+
+### 配置OpenClaw
+在你的OpenClaw配置中添加ROCK MCP工具：
+
+```json
+{
+  "mcpServers": {
+    "rock": {
+      "command": "python",
+      "args": ["/path/to/rock_mcp_server.py"],
+      "env": {
+        "ROCK_ADMIN_URL": "http://127.0.0.1:8080"
+      }
+    }
+  }
+}
+```
+
+## 📋 工具使用示例
+
+### 1. 创建Agent副本
+```typescript
+// 创建一个新的agent实例
+const newAgent = await rock_create_agent({
+  name: "rock-agent-clone",
+  image: "python:3.11",
+  memory: "4g",
+  cpus: 2.0,
+  capabilities: ["coding", "analysis", "automation"]
+});
+
+console.log("Agent created:", newAgent.agent_id);
+```
+
+### 2. 自我复制
+```typescript
+// agent可以创建自己的副本
+const myClone = await rock_clone_agent({
+  source_agent_id: "current-agent-id",
+  clone_name: "my-clone-1",
+  transfer_memory: true,
+  transfer_context: true
+});
+```
+
+### 3. 创建沙箱环境
+```typescript
+// 创建隔离的沙箱环境
+const sandbox = await rock_create_sandbox({
+  image: "python:3.11",
+  memory: "8g",
+  cpus: 4.0,
+  network_mode: "isolated"
+});
+
+console.log("Sandbox created:", sandbox.sandbox_id);
+```
+
+### 4. 执行环境操作
+```typescript
+// 在ROCK环境中执行操作
+const result = await rock_execute_action({
+  sandbox_id: "sandbox-123",
+  action: "python",
+  command: "print('Hello from ROCK!')"
+});
+```
+
+## 🏗️ 系统架构
+
+```
+┌─────────────────┐
+│  OpenClaw Agent │
+│   (Main Agent)  │
+└────────┬────────┘
+         │
+         ├── MCP Protocol ───┐
+         │                    │
+┌────────▼────────┐   ┌──────▼────────┐
+│  ROCK MCP Tool  │──▶│  ROCK Admin   │
+│   Interface     │   │  (Scheduler)  │
+└─────────────────┘   └──────┬────────┘
+                              │
+                              ├── Worker Nodes
+                              ├── Sandbox Runtime
+                              └── Agent Instances
+```
+
+## 🔧 配置选项
+
+### 环境变量
+- `ROCK_ADMIN_URL` - ROCK管理服务地址 (默认: http://127.0.0.1:8080)
+- `ROCK_API_KEY` - API认证密钥
+- `ROCK_TIMEOUT` - 请求超时时间 (默认: 30秒)
+
+### 工具配置
+```json
+{
+  "default_image": "python:3.11",
+  "default_memory": "4g",
+  "default_cpus": 2.0,
+  "max_agents": 10,
+  "auto_cleanup": true
+}
+```
+
+## 📊 监控和管理
+
+### 查看Agent状态
+```typescript
+const agents = await rock_list_agents();
+agents.forEach(agent => {
+  console.log(`${agent.name}: ${agent.status}`);
+});
+```
+
+### 获取系统状态
+```typescript
+const status = await rock_get_status();
+console.log("Active agents:", status.active_agents);
+console.log("Total sandboxes:", status.total_sandboxes);
+```
+
+## 🔒 安全特性
+
+- **多层隔离** - 容器级隔离 + 网络隔离
+- **资源限制** - CPU、内存、磁盘使用限制
+- **访问控制** - 基于API密钥的认证
+- **审计日志** - 所有操作记录日志
+- **自动清理** - 异常终止自动清理资源
+
+## 🚀 高级用例
+
+### 1. 分布式Agent集群
+```typescript
+// 在不同机器上创建agent副本
+const machines = ["machine1", "machine2", "machine3"];
+for (const machine of machines) {
+  await rock_clone_agent({
+    target_machine: machine,
+    clone_name: `agent-${machine}`
+  });
+}
+```
+
+### 2. 弹性扩展
+```typescript
+// 根据负载自动扩展
+const load = await rock_get_status();
+if (load.cpu_usage > 80) {
+  await rock_clone_agent({ auto_scale: true });
+}
+```
+
+### 3. 任务分发
+```typescript
+// 创建专门处理不同任务的agents
+const tasks = ["coding", "analysis", "testing"];
+for (const task of tasks) {
+  await rock_create_agent({
+    name: `${task}-agent`,
+    specialization: task
+  });
+}
+```
+
+## 📝 开发指南
+
+### 添加新工具
+1. 在`rock_mcp_server.py`中添加新的工具函数
+2. 使用`@mcp.tool()`装饰器注册工具
+3. 实现必要的错误处理和日志记录
+
+### 测试
+```bash
+# 运行测试
+python -m pytest tests/
+
+# 启动开发服务器
+python rock_mcp_server.py --debug
+```
+
+## 🤝 贡献
+
+欢迎提交Issue和Pull Request！
+
+## 📄 许可证
+
+MIT License
+
+## 🔗 相关资源
+
+- [ROCK官方文档](https://alibaba.github.io/ROCK/)
+- [ROCK GitHub仓库](https://github.com/alibaba/ROCK)
+- [MCP协议文档](https://modelcontextprotocol.io/)
+
+---
+
+**ROCK MCP Tools - 让OpenClaw Agent拥有ROCK的强大环境管理能力！** 🚀
\ No newline at end of file
diff --git a/mcp/requirements.txt b/mcp/requirements.txt
new file mode 100644
index 000000000..69d73dff7
--- /dev/null
+++ b/mcp/requirements.txt
@@ -0,0 +1,20 @@
+# ROCK MCP Tools Requirements
+
+# Core MCP dependencies
+mcp>=0.1.0
+asyncio>=3.4.3
+httpx>=0.24.0
+pydantic>=2.0.0
+
+# Optional ROCK SDK (install separately)
+# rl-rock>=0.2.0
+
+# Development dependencies
+pytest>=7.0.0
+pytest-asyncio>=0.21.0
+
+# Logging and monitoring
+loguru>=0.7.0
+
+# Type hints
+typing-extensions>=4.0.0
\ No newline at end of file
diff --git a/mcp/rock_client.py b/mcp/rock_client.py
new file mode 100644
index 000000000..6f683e2f9
--- /dev/null
+++ b/mcp/rock_client.py
@@ -0,0 +1,266 @@
+#!/usr/bin/env python3
+"""
+ROCK MCP Client - OpenClaw Agent集成客户端
+提供ROCK环境管理功能给OpenClaw Agent使用
+"""
+
+import asyncio
+import json
+import logging
+from typing import Dict, Any, Optional
+from dataclasses import dataclass
+from enum import Enum
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class AgentStatus(Enum):
+    """Agent状态枚举"""
+    CREATED = "created"
+    STARTING = "starting"
+    RUNNING = "running"
+    STOPPED = "stopped"
+    ERROR = "error"
+
+@dataclass
+class AgentConfig:
+    """Agent配置"""
+    name: str
+    image: str = "python:3.11"
+    memory: str = "4g"
+    cpus: float = 2.0
+    capabilities: list = None
+    environment: dict = None
+    network_mode: str = "bridge"
+
+    def __post_init__(self):
+        if self.capabilities is None:
+            self.capabilities = []
+        if self.environment is None:
+            self.environment = {}
+
+@dataclass
+class AgentInfo:
+    """Agent信息"""
+    agent_id: str
+    name: str
+    status: AgentStatus
+    config: AgentConfig
+    created_at: str
+    sandbox_id: str
+    machine_id: str = "local"
+
+class ROCKAgentClient:
+    """ROCK Agent客户端 - 提供agent管理功能"""
+    
+    def __init__(self, admin_url: str = "http://127.0.0.1:8080", api_key: str = None):
+        self.admin_url = admin_url
+        self.api_key = api_key
+        self.agents: Dict[str, AgentInfo] = {}
+        
+    async def create_agent(self, config: AgentConfig) -> AgentInfo:
+        """创建新的agent实例"""
+        agent_id = f"agent-{config.name}-{asyncio.get_event_loop().time()}"
+        
+        logger.info(f"Creating agent: {agent_id}")
+        
+        # 模拟创建agent (实际会调用ROCK API)
+        agent_info = AgentInfo(
+            agent_id=agent_id,
+            name=config.name,
+            status=AgentStatus.CREATED,
+            config=config,
+            created_at=asyncio.get_event_loop().time(),
+            sandbox_id=f"sandbox-{agent_id}"
+        )
+        
+        self.agents[agent_id] = agent_id
+        await self._start_agent(agent_id)
+        
+        return agent_info
+    
+    async def _start_agent(self, agent_id: str):
+        """启动agent"""
+        agent = self.agents.get(agent_id)
+        if agent:
+            agent.status = AgentStatus.STARTING
+            logger.info(f"Starting agent: {agent_id}")
+            
+            # 模拟启动过程
+            await asyncio.sleep(1)
+            agent.status = AgentStatus.RUNNING
+            logger.info(f"Agent started: {agent_id}")
+    
+    async def clone_agent(self, source_agent_id: str, clone_name: str = None, 
+                         transfer_memory: bool = False, transfer_context: bool = False) -> AgentInfo:
+        """克隆现有agent"""
+        source_agent = self.agents.get(source_agent_id)
+        if not source_agent:
+            raise ValueError(f"Source agent not found: {source_agent_id}")
+        
+        clone_name = clone_name or f"{source_agent.name}-clone"
+        
+        logger.info(f"Cloning agent {source_agent_id} to {clone_name}")
+        
+        # 创建克隆配置
+        clone_config = AgentConfig(
+            name=clone_name,
+            image=source_agent.config.image,
+            memory=source_agent.config.memory,
+            cpus=source_agent.config.cpus,
+            capabilities=source_agent.config.capabilities.copy(),
+            environment=source_agent.config.environment.copy()
+        )
+        
+        # 创建克隆agent
+        clone_agent = await self.create_agent(clone_config)
+        
+        # 如果需要，传输内存和上下文
+        if transfer_memory or transfer_context:
+            await self._transfer_agent_data(source_agent_id, clone_agent.agent_id, 
+                                           transfer_memory, transfer_context)
+        
+        return clone_agent
+    
+    async def _transfer_agent_data(self, source_id: str, target_id: str, 
+                                   transfer_memory: bool, transfer_context: bool):
+        """传输agent数据"""
+        logger.info(f"Transferring data from {source_id} to {target_id}")
+        # 实现数据传输逻辑
+        pass
+    
+    async def list_agents(self, status: AgentStatus = None) -> list:
+        """列出所有agents"""
+        agents = list(self.agents.values())
+        if status:
+            agents = [a for a in agents if a.status == status]
+        return agents
+    
+    async def get_agent_status(self, agent_id: str) -> AgentStatus:
+        """获取agent状态"""
+        agent = self.agents.get(agent_id)
+        return agent.status if agent else None
+    
+    async def stop_agent(self, agent_id: str):
+        """停止agent"""
+        agent = self.agents.get(agent_id)
+        if agent:
+            agent.status = AgentStatus.STOPPED
+            logger.info(f"Agent stopped: {agent_id}")
+    
+    async def cleanup_agent(self, agent_id: str):
+        """清理agent资源"""
+        if agent_id in self.agents:
+            del self.agents[agent_id]
+            logger.info(f"Agent cleaned up: {agent_id}")
+
+# 全局客户端实例
+rock_client = ROCKAgentClient()
+
+# OpenClaw MCP接口函数
+async def rock_create_agent(name: str, image: str = "python:3.11", 
+                           memory: str = "4g", cpus: float = 2.0, 
+                           capabilities: list = None) -> dict:
+    """创建新的agent实例"""
+    config = AgentConfig(
+        name=name,
+        image=image,
+        memory=memory,
+        cpus=cpus,
+        capabilities=capabilities or []
+    )
+    agent = await rock_client.create_agent(config)
+    return {
+        "agent_id": agent.agent_id,
+        "name": agent.name,
+        "status": agent.status.value,
+        "sandbox_id": agent.sandbox_id
+    }
+
+async def rock_clone_agent(source_agent_id: str, clone_name: str = None,
+                           transfer_memory: bool = False, 
+                           transfer_context: bool = False) -> dict:
+    """克隆现有agent"""
+    agent = await rock_client.clone_agent(source_agent_id, clone_name, 
+                                         transfer_memory, transfer_context)
+    return {
+        "agent_id": agent.agent_id,
+        " "name": agent.name,
+        "status": agent.status.value,
+        "source_agent_id": source_agent_id
+    }
+
+async def rock_list_agents() -> list:
+    """列出所有agents"""
+    agents = await rock_client.list_agents()
+    return [{
+        "agent_id": a.agent_id,
+        "name": a.name,
+        "status": a.status.value,
+        "sandbox_id": a.sandbox_id
+    } for a in agents]
+
+async def rock_get_status(agent_id: str = None) -> dict:
+    """获取agent或系统状态"""
+    if agent_id:
+        status = await rock_client.get_agent_status(agent_id)
+        return {"agent_id": agent_id, "status": status.value if status else "not_found"}
+    else:
+        agents = await rock_client.list_agents()
+        return {
+            "total_agents": len(agents),
+            "running_agents": len([a for a in agents if a.status == AgentStatus.RUNNING]),
+            "agents": [{"id": a.agent_id, "name": a.name, "status": a.status.value} for a in agents]
+        }
+
+async def rock_stop_agent(agent_id: str) -> dict:
+    """停止agent"""
+    await rock_client.stop_agent(agent_id)
+    return {"agent_id": agent_id, "status": "stopped"}
+
+async def rock_cleanup_agent(agent_id: str) -> dict:
+    """清理agent资源"""
+    await rock_client.cleanup_agent(agent_id)
+    return {"agent_id": agent_id, "status": "cleaned"}
+
+# 测试函数
+async def test_rock_mcp():
+    """测试ROCK MCP功能"""
+    print("🚀 Testing ROCK MCP Tools")
+    
+    # 创建第一个agent
+    print("\n1. Creating original agent...")
+    agent1 = await rock_create_agent("original-agent", capabilities=["coding", "analysis"])
+    print(f"✅ Created: {agent1}")
+    
+    # 克隆agent
+    print("\n2. Cloning agent...")
+    clone1 = await rock_clone_agent(agent1["agent_id"], "clone-1")
+    print(f"✅ Cloned: {clone1}")
+    
+    # 创建另一个agent
+    print("\n3. Creating specialized agent...")
+    agent2 = await rock_create_agent("specialized-agent", capabilities=["testing"])
+    print(f"✅ Created: {agent2}")
+    
+    # 列出所有agents
+    print("\n4. Listing all agents...")
+    agents = await rock_list_agents()
+    for agent in agents:
+        print(f"   - {agent['name']}: {agent['status']}")
+    
+    # 获取系统状态
+    print("\n5. Getting system status...")
+    status = await rock_get_status()
+    print(f"✅ System: {status}")
+    
+    # 清理测试agents
+    print("\n6. Cleanup...")
+    await rock_cleanup_agent(agent1["agent_id"])
+    await rock_cleanup_agent(clone1["agent_id"])
+    await rock_cleanup_agent(agent2["agent_id"])
+    print("✅ Cleanup completed")
+
+if __name__ == "__main__":
+    asyncio.run(test_rock_mcp())
\ No newline at end of file
diff --git a/mcp/rock_mcp_server.py b/mcp/rock_mcp_server.py
new file mode 100644
index 000000000..9fbed0fef
--- /dev/null
+++ b/mcp/rock_mcp_server.py
@@ -0,0 +1,357 @@
+#!/usr/bin/env python3
+"""
+ROCK MCP Server - Model Context Protocol Server for ROCK Environment Management
+Enables AI agents to create and manage their own ROCK environments for self-replication
+"""
+
+import asyncio
+import json
+import logging
+from typing import Any, Dict, List, Optional
+from dataclasses import dataclass, asdict
+from datetime import datetime
+import subprocess
+import os
+import sys
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+@dataclass
+class ROCKEnvironment:
+    """Represents a ROCK environment instance"""
+    env_id: str
+    name: str
+    status: str
+    created_at: str
+    config: Dict[str, Any]
+    agent_session: Optional[str] = None
+
+class ROCKMCPServer:
+    """ROCK MCP Server implementation"""
+    
+    def __init__(self):
+        self.environments: Dict[str, ROCKEnvironment] = {}
+        self.rock_admin_url = os.getenv("ROCK_ADMIN_URL", "http://127.0.0.1:8080")
+        self.worker_env_type = os.getenv("ROCK_WORKER_ENV_TYPE", "uv")
+        
+    async def initialize(self):
+        """Initialize ROCK MCP server"""
+        logger.info("🚀 ROCK MCP Server initializing...")
+        logger.info(f"📡 ROCK Admin URL: {self.rock_admin_url}")
+        logger.info(f"🔧 Worker Environment Type: {self.worker_env_type}")
+        
+        # Check if ROCK is available
+        try:
+            result = subprocess.run(
+                ["rock", "status"],
+                capture_output=True,
+                text=True,
+                timeout=10
+            )
+            if result.returncode == 0:
+                logger.info("✅ ROCK CLI is available")
+            else:
+                logger.warning("⚠️  ROCK CLI may not be properly installed")
+        except Exception as e:
+            logger.error(f"❌ Failed to check ROCK CLI: {e}")
+    
+    async def create_rock_environment(
+        self,
+        name: str,
+        env_type: str = "sandbox",
+        image: str = "python:3.11",
+        memory: str = "8g",
+        cpus: float = 2.0,
+        isolation: str = "container"
+    ) -> Dict[str, Any]:
+        """
+        Create a new ROCK environment for agent replication
+        
+        Args:
+            name: Environment name
+            env_type: Type of environment (sandbox, gem, bash)
+            image: Docker image to use
+            memory: Memory allocation
+            cpus: CPU allocation
+            isolation: Isolation level
+        
+        Returns:
+            Dict with environment details
+        """
+        env_id = f"agent_{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+        
+        env_config = {
+            "image": image,
+            "memory": memory,
+            "cpus": cpus,
+            "isolation": isolation,
+            "env_type": env_type
+        }
+        
+        # Create ROCK environment
+        env = ROCKEnvironment(
+            env_id=env_id,
+            name=name,
+            status="initializing",
+            created_at=datetime.now().isoformat(),
+            config=env_config
+        )
+        
+        self.environments[env_id] = env
+        
+        # Actually create the ROCK environment
+        try:
+            # This would use ROCK CLI or SDK
+            cmd = [
+                "rock", "env", "create",
+                f"--name={name}",
+                f"--image={image}",
+                f"--memory={memory}",
+                f"--cpus={cpus}"
+            ]
+            
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            
+            if result.returncode == 0:
+                env.status = "running"
+                logger.info(f"✅ Created ROCK environment: {env_id}")
+            else:
+                env.status = "failed"
+                logger.error(f"❌ Failed to create ROCK environment: {result.stderr}")
+                
+        except Exception as e:
+            env.status = "error"
+            logger.error(f"❌ Error creating ROCK environment: {e}")
+        
+        return asdict(env)
+    
+    # MCP Tool: Deploy Agent to ROCK Environment
+    async def deploy_agent_to_environment(
+        self,
+        env_id: str,
+        agent_config: Dict[str, Any],
+        agent_code: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Deploy an agent to a ROCK environment
+        
+        Args:
+            env_id: Target environment ID
+            agent_config: Agent configuration
+            agent_code: Optional agent code to deploy
+        
+        Returns:
+            Dict with deployment status
+        """
+        if env_id not in self.environments:
+            return {
+                "success": False,
+                "error": f"Environment {env_id} not found"
+            }
+        
+        env = self.environments[env_id]
+        
+        try:
+            # Create agent session in ROCK environment
+            agent_session_id = f"session_{env_id}_{datetime.now().strftime('%H%M%S')}"
+            
+            # Deploy agent code if provided
+            if agent_code:
+                # This would copy agent code to ROCK environment
+                logger.info(f"📦 Deploying agent code to {env_id}")
+            
+            env.agent_session = agent_session_id
+            env.status = "agent_deployed"
+            
+            logger.info(f"✅ Agent deployed to ROCK environment: {env_id}")
+            
+            return {
+                "success": True,
+                "env_id": env_id,
+                "agent_session": agent_session_id,
+                "status": "deployed"
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Failed to deploy agent: {e}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    # MCP Tool: Execute Command in ROCK Environment
+    async def execute_in_rock_environment(
+        self,
+        env_id: str,
+        command: str,
+        session: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Execute a command in a ROCK environment
+        
+        Args:
+            env_id: Target environment ID
+            command: Command to execute
+            session: Optional session ID
+        
+        Returns:
+            Dict with execution result
+        """
+        if env_id not in self.environments:
+            return {
+                "success": False,
+                "error": f"Environment {env_id} not found"
+            }
+        
+        env = self.environments[env_id]
+        
+        try:
+            # This would use ROCK SDK to execute command in sandbox
+            logger.info(f"🏃 Executing command in {env_id}: {command}")
+            
+            # Simulated execution
+            result = {
+                "success": True,
+                "env_id": env_id,
+                "command": command,
+                "output": f"Command executed in {env.name}",
+                "exit_code": 0
+            }
+            
+            return result
+            
+        except Exception as e:
+            logger.error(f"❌ Failed to execute command: {e}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    # MCP Tool: List ROCK Environments
+    async def list_rock_environments(self) -> List[Dict[str, Any]]:
+        """List all ROCK environments"""
+        return [asdict(env) for env in self.environments.values()]
+    
+    # MCP Tool: Get Environment Status
+    async def get_environment_status(self, env_id: str) -> Dict[str, Any]:
+        """Get status of a specific ROCK environment"""
+        if env_id in self.environments:
+            return asdict(self.environments[env_id])
+        else:
+            return {
+                "success": False,
+                "error": f"Environment {env_id} not found"
+            }
+    
+    # MCP Tool: Stop ROCK Environment
+    async def stop_rock_environment(self, env_id: str) -> Dict[str, Any]:
+        """Stop a ROCK environment"""
+        if env_id not in self.environments:
+            return {
+                "success": False,
+                "error": f"Environment {env_id} not found"
+            }
+        
+        env = self.environments[env_id]
+        
+        try:
+            # Stop ROCK environment
+            logger.info(f"🛑 Stopping ROCK environment: {env_id}")
+            
+            env.status = "stopped"
+            
+            return {
+                "success": True,
+                "env_id": env_id,
+                "status": "stopped"
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Failed to stop environment: {e}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    # MCP Tool: Self-Replicate Agent
+    async def replicate_agent(
+        self,
+        agent_name: str,
+        agent_config: Dict[str, Any],
+        target_env_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Self-replicate the current agent to a ROCK environment
+        
+        Args:
+            agent_name: Name for the replicated agent
+            agent_config: Configuration for the new agent
+            target_env_id: Optional target environment (creates new if None)
+        
+        Returns:
+            Dict with replication result
+        """
+        logger.info(f"🔄 Initiating agent self-replication: {agent_name}")
+        
+        try:
+            # Create new environment if not specified
+            if target_env_id is None:
+                env_result = await self.create_rock_environment(
+                    name=f"{agent_name}_env",
+                    env_type="sandbox",
+                    image="python:3.11"
+                )
+                
+                if env_result.get("status") != "running":
+                    return {
+                        "success": False,
+                        "error": "Failed to create target environment",
+                        "env_result": env_result
+                    }
+                
+                target_env_id = env_result["env_id"]
+            
+            # Deploy agent to environment
+            deploy_result = await self.deploy_agent_to_environment(
+                env_id=target_env_id,
+                agent_config=agent_config
+            )
+            
+            logger.info(f"✅ Agent self-replication completed: {target_env_id}")
+            
+            return {
+                "success": True,
+                "agent_name": agent_name,
+                "env_id": target_env_id,
+                "agent_session": deploy_result.get("agent_session"),
+                "status": "replicated"
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Failed agent self-replication: {e}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+
+# Global server instance
+rock_server = ROCKMCPServer()
+
+async def main():
+    """Main entry point for ROCK MCP server"""
+    await rock_server.initialize()
+    
+    # Keep server running
+    logger.info("🎯 ROCK MCP Server ready and waiting for requests...")
+    
+    # Example: Self-replicate current agent
+    # result = await rock_server.replicate_agent(
+    #     agent_name="dengwxclaw_replica",
+    #     agent_config={"type": "forwarder", "mode": "all"}
+    # )
+    # logger.info(f"Replication result: {result}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
\ No newline at end of file
diff --git a/mcp/start.sh b/mcp/start.sh
new file mode 100644
index 000000000..4a38076b0
--- /dev/null
+++ b/mcp/start.sh
@@ -0,0 +1,36 @@
+#!/bin/bash
+
+# ROCK MCP Tools 启动脚本
+
+echo "🚀 Starting ROCK MCP Tools..."
+
+# 检查Python环境
+if ! command -v python3 &> /dev/null; then
+    echo "❌ Python 3 not found!"
+    exit 1
+fi
+
+# 检查依赖
+echo "📦 Checking dependencies..."
+python3 -c "import mcp" 2>/dev/null || {
+    echo "❌ MCP not installed. Installing..."
+    pip install mcp
+}
+
+python3 -c "import asyncio" 2>/dev/null || {
+    echo "❌ asyncio not available"
+    exit 1
+}
+
+# 设置环境变量
+export ROCK_ADMIN_URL=${ROCK_ADMIN_URL:-"http://127.0.0.1:8080"}
+export ROCK_API_KEY=${ROCK_API_KEY:-""}
+export ROCK_TIMEOUT=${ROCK_TIMEOUT:-"30"}
+
+echo "🔧 Configuration:"
+echo "   - ROCK_ADMIN_URL: $ROCK_ADMIN_URL"
+echo "   - ROCK_TIMEOUT: $ROCK_TIMEOUT seconds"
+
+# 启动MCP服务器
+echo "▶️  Starting MCP server..."
+python3 rock_mcp_server.py
\ No newline at end of file