Compare commits

...

47 Commits

Author SHA1 Message Date
Jiayi Zhang 52a13f2a57 Update README.md
Pre-commit checks / pre-commit-check (push) Failing after 0s
2026-01-04 11:11:56 +08:00
Jiayi Zhang 211a245c95 Merge pull request #1266 from cnJasonZ/feat/jiekouAI
Add Jiekou.AI as new LLM provider
2025-11-14 18:50:00 +08:00
cnJasonZ 872d5bb59e feat: add JiekouAI as new provider 2025-11-07 13:48:49 +08:00
Jiayi Zhang 37e7f9c8f9 Merge pull request #1256 from FoundationAgents/chart-visualization
Merge pull request #861 from 666haiwen/feat/chart-visualization
2025-10-28 17:52:27 +08:00
Jiayi Zhang 67d6c1c00d Merge pull request #1206 from XYDT-AI/sandbox
Feat: support 4 sandbox tools by using daytona
2025-09-19 15:29:19 +08:00
GhostC efccc9fcc9 Merge branch 'sandbox' of https://github.com/XYDT-AI/OpenManus-improve into sandbox 2025-09-14 18:19:22 +08:00
GhostC c26bc9262e put base64 in the currect position for sb_vision 2025-09-14 18:15:01 +08:00
GhostC 52cf406467 Merge branch 'FoundationAgents:main' into sandbox 2025-09-08 15:53:40 +08:00
Jiayi Zhang 20c7207bda Update README.md 2025-09-04 18:39:07 +08:00
GhostC decd25ab34 Merge branch 'FoundationAgents:main' into sandbox 2025-08-16 17:47:07 +08:00
GhostC 5e7aba3e4b modify sb_browser_use tool's result 2025-08-16 17:46:02 +08:00
Jiayi Zhang 3e604acb93 Update Group Link. 2025-08-13 10:38:39 +08:00
GhostC 23f2e85ae9 modify requirements.txt 2025-07-30 23:36:44 +08:00
GhostC 9cb3bd16c4 remove sb_*.py to tool/sandbox&modify requirements.txt 2025-07-30 18:59:33 +08:00
GhostC 802e212ef2 Apply pre-commit formatting fixes 2025-07-27 22:31:39 +08:00
GhostC f08ed23914 Merge branch 'sandbox' of https://github.com/XYDT-AI/OpenManus-sandbox into sandbox 2025-07-27 22:27:35 +08:00
GhostC a010fe7048 Apply pre-commit formatting fixes 2025-07-27 22:26:50 +08:00
GhostC 386aeb46b6 Merge branch 'FoundationAgents:main' into sandbox 2025-07-27 22:00:14 +08:00
GhostC 66f4ae046e feat:support 4 sandbox tools by using daytona 2025-07-27 21:59:18 +08:00
GhostC f99dfac5bb add README and delete sandbox_id 2025-07-27 18:38:53 +08:00
Jiayi Zhang ac66e8fd44 Merge pull request #1197 from SNHuan/tool/Crawl4aiTool
add Crawl4aiTool
2025-07-23 17:47:52 +08:00
SNHuan 08cd0443b6 Format crawl4ai.py with pre-commit hooks 2025-07-23 17:36:53 +08:00
white-rm 50f56f5708 eliminate sandox redundant information 2025-07-22 23:50:13 +08:00
SNHuan 4a28a7912c add Crawl4aiTool 2025-07-22 14:29:16 +08:00
cyb1278588254 0246513495 debug sb_files_tool & sb_vision_tool 2025-07-20 17:26:15 +08:00
GhostC c94422748a Merge branch 'sandbox' of https://github.com/XYDT-AI/OpenManus-sandbox into sandbox 2025-07-20 16:02:58 +08:00
GhostC 3301d54625 support use existing sandbox by id 2025-07-20 16:02:18 +08:00
GhostC 0a6207d26f Merge branch 'FoundationAgents:main' into sandbox 2025-07-20 14:39:36 +08:00
GhostC 398ae434ed add main for sandbox_agent 2025-07-20 02:17:08 +08:00
cyb1278588254 5731b63984 init sb_files_tool & sb_vision_tool 2025-07-19 23:48:23 +08:00
white-rm 6fb48f4dbe update browser sandbox tool 2 2025-07-19 23:37:01 +08:00
white-rm c016d5627c update browser sandbox tool message 2025-07-16 00:24:56 +08:00
white-rm e4c4d4a00f update browser sandbox tool 2025-07-13 21:09:58 +08:00
GhostC cc283546c0 Merge branch 'sandbox' of https://github.com/XYDT-AI/OpenManus-sandbox into sandbox 2025-07-10 23:03:36 +08:00
GhostC 7ce31d1c34 init sandbox agent 2025-07-10 23:02:50 +08:00
white-rm 904dec240b update shell sandox 2025-07-08 22:40:59 +08:00
GhostC 17e1becb9f test sb_browser_use_tool 2025-07-08 22:24:02 +08:00
GhostC 41e12eb1de Merge branch 'sandbox' of https://github.com/XYDT-AI/OpenManus-sandbox into sandbox 2025-07-07 00:57:04 +08:00
GhostC fb30c432f3 add daytona config 2025-07-06 23:57:00 +08:00
white-rm 686ace2b09 add shell sandox 2025-07-06 22:42:09 +08:00
GhostC d7e3c14b60 debug computer use tool 2025-07-06 15:13:41 +08:00
GhostC 2a6287422d add sandbox tool 2025-07-01 21:49:22 +08:00
GhostC 9513686155 modify BaseTool 2025-06-30 23:12:44 +08:00
white-rm a6a45d7e02 add sandox utils 2025-06-30 00:49:18 +08:00
GhostC 527fd6483b add utils 2025-06-29 10:12:23 +08:00
GhostC fd5b070cb4 sandbox init 2025-06-27 19:38:41 +08:00
mannaandpoem f21c7a2e55 Merge pull request #861 from 666haiwen/feat/chart-visualization
feat: add chart visualization tools which support png/html output
2025-03-22 13:54:27 +08:00
31 changed files with 3241 additions and 68 deletions
+13 -15
View File
@@ -16,24 +16,22 @@ repos:
rev: v2.0.1
hooks:
- id: autoflake
args: [
--remove-all-unused-imports,
--ignore-init-module-imports,
--expand-star-imports,
--remove-duplicate-keys,
--remove-unused-variables,
--recursive,
--in-place,
--exclude=__init__.py,
]
args:
[
--remove-all-unused-imports,
--ignore-init-module-imports,
--expand-star-imports,
--remove-duplicate-keys,
--remove-unused-variables,
--recursive,
--in-place,
--exclude=__init__.py,
]
files: \.py$
- repo: https://github.com/pycqa/isort
rev: 5.12.0
hooks:
- id: isort
args: [
"--profile", "black",
"--filter-files",
"--lines-after-imports=2",
]
args:
["--profile", "black", "--filter-files", "--lines-after-imports=2"]
+2 -3
View File
@@ -174,8 +174,7 @@ Thanks to [PPIO](https://ppinfra.com/user/register?invited_by=OCPKCN&utm_source=
## Acknowledgement
Thanks to [anthropic-computer-use](https://github.com/anthropics/anthropic-quickstarts/tree/main/computer-use-demo)
and [browser-use](https://github.com/browser-use/browser-use) for providing basic support for this project!
Thanks to [anthropic-computer-use](https://github.com/anthropics/anthropic-quickstarts/tree/main/computer-use-demo), [browser-use](https://github.com/browser-use/browser-use) and [crawl4ai](https://github.com/unclecode/crawl4ai) for providing basic support for this project!
Additionally, we are grateful to [AAAJ](https://github.com/metauto-ai/agent-as-a-judge), [MetaGPT](https://github.com/geekan/MetaGPT), [OpenHands](https://github.com/All-Hands-AI/OpenHands) and [SWE-agent](https://github.com/SWE-agent/SWE-agent).
@@ -186,7 +185,7 @@ OpenManus is built by contributors from MetaGPT. Huge thanks to this agent commu
## Cite
```bibtex
@misc{openmanus2025,
author = {Xinbin Liang and Jinyu Xiang and Zhaoyang Yu and Jiayi Zhang and Sirui Hong and Sheng Fan and Xiao Tang},
author = {Xinbin Liang and Jinyu Xiang and Zhaoyang Yu and Jiayi Zhang and Sirui Hong and Sheng Fan and Xiao Tang and Bang Liu and Yuyu Luo and Chenglin Wu},
title = {OpenManus: An open-source framework for building general AI agents},
year = {2025},
publisher = {Zenodo},
+5
View File
@@ -8,6 +8,7 @@ from app.logger import logger
from app.prompt.browser import NEXT_STEP_PROMPT, SYSTEM_PROMPT
from app.schema import Message, ToolChoice
from app.tool import BrowserUseTool, Terminate, ToolCollection
from app.tool.sandbox.sb_browser_tool import SandboxBrowserTool
# Avoid circular import if BrowserAgent needs BrowserContextHelper
@@ -22,6 +23,10 @@ class BrowserContextHelper:
async def get_browser_state(self) -> Optional[dict]:
browser_tool = self.agent.available_tools.get_tool(BrowserUseTool().name)
if not browser_tool:
browser_tool = self.agent.available_tools.get_tool(
SandboxBrowserTool().name
)
if not browser_tool or not hasattr(browser_tool, "get_current_state"):
logger.warning("BrowserUseTool not found or doesn't have get_current_state")
return None
+223
View File
@@ -0,0 +1,223 @@
from typing import Dict, List, Optional
from pydantic import Field, model_validator
from app.agent.browser import BrowserContextHelper
from app.agent.toolcall import ToolCallAgent
from app.config import config
from app.daytona.sandbox import create_sandbox, delete_sandbox
from app.daytona.tool_base import SandboxToolsBase
from app.logger import logger
from app.prompt.manus import NEXT_STEP_PROMPT, SYSTEM_PROMPT
from app.tool import Terminate, ToolCollection
from app.tool.ask_human import AskHuman
from app.tool.mcp import MCPClients, MCPClientTool
from app.tool.sandbox.sb_browser_tool import SandboxBrowserTool
from app.tool.sandbox.sb_files_tool import SandboxFilesTool
from app.tool.sandbox.sb_shell_tool import SandboxShellTool
from app.tool.sandbox.sb_vision_tool import SandboxVisionTool
class SandboxManus(ToolCallAgent):
"""A versatile general-purpose agent with support for both local and MCP tools."""
name: str = "SandboxManus"
description: str = "A versatile agent that can solve various tasks using multiple sandbox-tools including MCP-based tools"
system_prompt: str = SYSTEM_PROMPT.format(directory=config.workspace_root)
next_step_prompt: str = NEXT_STEP_PROMPT
max_observe: int = 10000
max_steps: int = 20
# MCP clients for remote tool access
mcp_clients: MCPClients = Field(default_factory=MCPClients)
# Add general-purpose tools to the tool collection
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(
# PythonExecute(),
# BrowserUseTool(),
# StrReplaceEditor(),
AskHuman(),
Terminate(),
)
)
special_tool_names: list[str] = Field(default_factory=lambda: [Terminate().name])
browser_context_helper: Optional[BrowserContextHelper] = None
# Track connected MCP servers
connected_servers: Dict[str, str] = Field(
default_factory=dict
) # server_id -> url/command
_initialized: bool = False
sandbox_link: Optional[dict[str, dict[str, str]]] = Field(default_factory=dict)
@model_validator(mode="after")
def initialize_helper(self) -> "SandboxManus":
"""Initialize basic components synchronously."""
self.browser_context_helper = BrowserContextHelper(self)
return self
@classmethod
async def create(cls, **kwargs) -> "SandboxManus":
"""Factory method to create and properly initialize a Manus instance."""
instance = cls(**kwargs)
await instance.initialize_mcp_servers()
await instance.initialize_sandbox_tools()
instance._initialized = True
return instance
async def initialize_sandbox_tools(
self,
password: str = config.daytona.VNC_password,
) -> None:
try:
# 创建新沙箱
if password:
sandbox = create_sandbox(password=password)
self.sandbox = sandbox
else:
raise ValueError("password must be provided")
vnc_link = sandbox.get_preview_link(6080)
website_link = sandbox.get_preview_link(8080)
vnc_url = vnc_link.url if hasattr(vnc_link, "url") else str(vnc_link)
website_url = (
website_link.url if hasattr(website_link, "url") else str(website_link)
)
# Get the actual sandbox_id from the created sandbox
actual_sandbox_id = sandbox.id if hasattr(sandbox, "id") else "new_sandbox"
if not self.sandbox_link:
self.sandbox_link = {}
self.sandbox_link[actual_sandbox_id] = {
"vnc": vnc_url,
"website": website_url,
}
logger.info(f"VNC URL: {vnc_url}")
logger.info(f"Website URL: {website_url}")
SandboxToolsBase._urls_printed = True
sb_tools = [
SandboxBrowserTool(sandbox),
SandboxFilesTool(sandbox),
SandboxShellTool(sandbox),
SandboxVisionTool(sandbox),
]
self.available_tools.add_tools(*sb_tools)
except Exception as e:
logger.error(f"Error initializing sandbox tools: {e}")
raise
async def initialize_mcp_servers(self) -> None:
"""Initialize connections to configured MCP servers."""
for server_id, server_config in config.mcp_config.servers.items():
try:
if server_config.type == "sse":
if server_config.url:
await self.connect_mcp_server(server_config.url, server_id)
logger.info(
f"Connected to MCP server {server_id} at {server_config.url}"
)
elif server_config.type == "stdio":
if server_config.command:
await self.connect_mcp_server(
server_config.command,
server_id,
use_stdio=True,
stdio_args=server_config.args,
)
logger.info(
f"Connected to MCP server {server_id} using command {server_config.command}"
)
except Exception as e:
logger.error(f"Failed to connect to MCP server {server_id}: {e}")
async def connect_mcp_server(
self,
server_url: str,
server_id: str = "",
use_stdio: bool = False,
stdio_args: List[str] = None,
) -> None:
"""Connect to an MCP server and add its tools."""
if use_stdio:
await self.mcp_clients.connect_stdio(
server_url, stdio_args or [], server_id
)
self.connected_servers[server_id or server_url] = server_url
else:
await self.mcp_clients.connect_sse(server_url, server_id)
self.connected_servers[server_id or server_url] = server_url
# Update available tools with only the new tools from this server
new_tools = [
tool for tool in self.mcp_clients.tools if tool.server_id == server_id
]
self.available_tools.add_tools(*new_tools)
async def disconnect_mcp_server(self, server_id: str = "") -> None:
"""Disconnect from an MCP server and remove its tools."""
await self.mcp_clients.disconnect(server_id)
if server_id:
self.connected_servers.pop(server_id, None)
else:
self.connected_servers.clear()
# Rebuild available tools without the disconnected server's tools
base_tools = [
tool
for tool in self.available_tools.tools
if not isinstance(tool, MCPClientTool)
]
self.available_tools = ToolCollection(*base_tools)
self.available_tools.add_tools(*self.mcp_clients.tools)
async def delete_sandbox(self, sandbox_id: str) -> None:
"""Delete a sandbox by ID."""
try:
await delete_sandbox(sandbox_id)
logger.info(f"Sandbox {sandbox_id} deleted successfully")
if sandbox_id in self.sandbox_link:
del self.sandbox_link[sandbox_id]
except Exception as e:
logger.error(f"Error deleting sandbox {sandbox_id}: {e}")
raise e
async def cleanup(self):
"""Clean up Manus agent resources."""
if self.browser_context_helper:
await self.browser_context_helper.cleanup_browser()
# Disconnect from all MCP servers only if we were initialized
if self._initialized:
await self.disconnect_mcp_server()
await self.delete_sandbox(self.sandbox.id if self.sandbox else "unknown")
self._initialized = False
async def think(self) -> bool:
"""Process current state and decide next actions with appropriate context."""
if not self._initialized:
await self.initialize_mcp_servers()
self._initialized = True
original_prompt = self.next_step_prompt
recent_messages = self.memory.messages[-3:] if self.memory.messages else []
browser_in_use = any(
tc.function.name == SandboxBrowserTool().name
for msg in recent_messages
if msg.tool_calls
for tc in msg.tool_calls
)
if browser_in_use:
self.next_step_prompt = (
await self.browser_context_helper.format_next_step_prompt()
)
result = await super().think()
# Restore original prompt
self.next_step_prompt = original_prompt
return result
+32
View File
@@ -105,6 +105,25 @@ class SandboxSettings(BaseModel):
)
class DaytonaSettings(BaseModel):
daytona_api_key: str
daytona_server_url: Optional[str] = Field(
"https://app.daytona.io/api", description=""
)
daytona_target: Optional[str] = Field("us", description="enum ['eu', 'us']")
sandbox_image_name: Optional[str] = Field("whitezxj/sandbox:0.1.0", description="")
sandbox_entrypoint: Optional[str] = Field(
"/usr/bin/supervisord -n -c /etc/supervisor/conf.d/supervisord.conf",
description="",
)
# sandbox_id: Optional[str] = Field(
# None, description="ID of the daytona sandbox to use, if any"
# )
VNC_password: Optional[str] = Field(
"123456", description="VNC password for the vnc service in sandbox"
)
class MCPServerConfig(BaseModel):
"""Configuration for a single MCP server"""
@@ -167,6 +186,9 @@ class AppConfig(BaseModel):
run_flow_config: Optional[RunflowSettings] = Field(
None, description="Run flow configuration"
)
daytona_config: Optional[DaytonaSettings] = Field(
None, description="Daytona configuration"
)
class Config:
arbitrary_types_allowed = True
@@ -268,6 +290,11 @@ class Config:
sandbox_settings = SandboxSettings(**sandbox_config)
else:
sandbox_settings = SandboxSettings()
daytona_config = raw_config.get("daytona", {})
if daytona_config:
daytona_settings = DaytonaSettings(**daytona_config)
else:
daytona_settings = DaytonaSettings()
mcp_config = raw_config.get("mcp", {})
mcp_settings = None
@@ -296,6 +323,7 @@ class Config:
"search_config": search_settings,
"mcp_config": mcp_settings,
"run_flow_config": run_flow_settings,
"daytona_config": daytona_settings,
}
self._config = AppConfig(**config_dict)
@@ -308,6 +336,10 @@ class Config:
def sandbox(self) -> SandboxSettings:
return self._config.sandbox
@property
def daytona(self) -> DaytonaSettings:
return self._config.daytona_config
@property
def browser_config(self) -> Optional[BrowserSettings]:
return self._config.browser_config
+57
View File
@@ -0,0 +1,57 @@
# Agent with Daytona sandbox
## Prerequisites
- conda activate 'Your OpenManus python env'
- pip install daytona==0.21.8 structlog==25.4.0
## Setup & Running
1. daytona config :
```bash
cd OpenManus
cp config/config.example-daytona.toml config/config.toml
```
2. get daytona apikey :
goto https://app.daytona.io/dashboard/keys and create your apikey
3. set your apikey in config.toml
```toml
# daytona config
[daytona]
daytona_api_key = ""
#daytona_server_url = "https://app.daytona.io/api"
#daytona_target = "us" #Daytona is currently available in the following regions:United States (us)、Europe (eu)
#sandbox_image_name = "whitezxj/sandbox:0.1.0" #If you don't use this default image,sandbox tools may be useless
#sandbox_entrypoint = "/usr/bin/supervisord -n -c /etc/supervisor/conf.d/supervisord.conf" #If you change this entrypoint,server in sandbox may be useless
#VNC_password = #The password you set to log in sandbox by VNC,it will be 123456 if you don't set
```
2. Run :
```bash
cd OpenManus
python sandbox_main.py
```
3. Send tasks to Agent
You can sent tasks to Agent by terminate,agent will use sandbox tools to handle your tasks.
4. See results
If agent use sb_browser_use tool, you can see the operations by VNC link, The VNC link will print in the termination,e.g.:https://6080-sandbox-123456.h7890.daytona.work.
If agent use sb_shell tool, you can see the results by terminate of sandbox in https://app.daytona.io/dashboard/sandboxes.
Agent can use sb_files tool to operate files to sandbox.
## Example
You can send task e.g.:"帮我在https://hk.trip.com/travel-guide/guidebook/nanjing-9696/?ishideheader=true&isHideNavBar=YES&disableFontScaling=1&catalogId=514634&locale=zh-HK查询相关信息上制定一份南京旅游攻略,并在工作区保存为index.html"
Then you can see the agent's browser action in VNC link(https://6080-sandbox-123456.h7890.proxy.daytona.work) and you can see the html made by agent in Website URL(https://8080-sandbox-123456.h7890.proxy.daytona.work).
## Learn More
- [Daytona Documentation](https://www.daytona.io/docs/)
+165
View File
@@ -0,0 +1,165 @@
import time
from daytona import (
CreateSandboxFromImageParams,
Daytona,
DaytonaConfig,
Resources,
Sandbox,
SandboxState,
SessionExecuteRequest,
)
from app.config import config
from app.utils.logger import logger
# load_dotenv()
daytona_settings = config.daytona
logger.info("Initializing Daytona sandbox configuration")
daytona_config = DaytonaConfig(
api_key=daytona_settings.daytona_api_key,
server_url=daytona_settings.daytona_server_url,
target=daytona_settings.daytona_target,
)
if daytona_config.api_key:
logger.info("Daytona API key configured successfully")
else:
logger.warning("No Daytona API key found in environment variables")
if daytona_config.server_url:
logger.info(f"Daytona server URL set to: {daytona_config.server_url}")
else:
logger.warning("No Daytona server URL found in environment variables")
if daytona_config.target:
logger.info(f"Daytona target set to: {daytona_config.target}")
else:
logger.warning("No Daytona target found in environment variables")
daytona = Daytona(daytona_config)
logger.info("Daytona client initialized")
async def get_or_start_sandbox(sandbox_id: str):
"""Retrieve a sandbox by ID, check its state, and start it if needed."""
logger.info(f"Getting or starting sandbox with ID: {sandbox_id}")
try:
sandbox = daytona.get(sandbox_id)
# Check if sandbox needs to be started
if (
sandbox.state == SandboxState.ARCHIVED
or sandbox.state == SandboxState.STOPPED
):
logger.info(f"Sandbox is in {sandbox.state} state. Starting...")
try:
daytona.start(sandbox)
# Wait a moment for the sandbox to initialize
# sleep(5)
# Refresh sandbox state after starting
sandbox = daytona.get(sandbox_id)
# Start supervisord in a session when restarting
start_supervisord_session(sandbox)
except Exception as e:
logger.error(f"Error starting sandbox: {e}")
raise e
logger.info(f"Sandbox {sandbox_id} is ready")
return sandbox
except Exception as e:
logger.error(f"Error retrieving or starting sandbox: {str(e)}")
raise e
def start_supervisord_session(sandbox: Sandbox):
"""Start supervisord in a session."""
session_id = "supervisord-session"
try:
logger.info(f"Creating session {session_id} for supervisord")
sandbox.process.create_session(session_id)
# Execute supervisord command
sandbox.process.execute_session_command(
session_id,
SessionExecuteRequest(
command="exec /usr/bin/supervisord -n -c /etc/supervisor/conf.d/supervisord.conf",
var_async=True,
),
)
time.sleep(25) # Wait a bit to ensure supervisord starts properly
logger.info(f"Supervisord started in session {session_id}")
except Exception as e:
logger.error(f"Error starting supervisord session: {str(e)}")
raise e
def create_sandbox(password: str, project_id: str = None):
"""Create a new sandbox with all required services configured and running."""
logger.info("Creating new Daytona sandbox environment")
logger.info("Configuring sandbox with browser-use image and environment variables")
labels = None
if project_id:
logger.info(f"Using sandbox_id as label: {project_id}")
labels = {"id": project_id}
params = CreateSandboxFromImageParams(
image=daytona_settings.sandbox_image_name,
public=True,
labels=labels,
env_vars={
"CHROME_PERSISTENT_SESSION": "true",
"RESOLUTION": "1024x768x24",
"RESOLUTION_WIDTH": "1024",
"RESOLUTION_HEIGHT": "768",
"VNC_PASSWORD": password,
"ANONYMIZED_TELEMETRY": "false",
"CHROME_PATH": "",
"CHROME_USER_DATA": "",
"CHROME_DEBUGGING_PORT": "9222",
"CHROME_DEBUGGING_HOST": "localhost",
"CHROME_CDP": "",
},
resources=Resources(
cpu=2,
memory=4,
disk=5,
),
auto_stop_interval=15,
auto_archive_interval=24 * 60,
)
# Create the sandbox
sandbox = daytona.create(params)
logger.info(f"Sandbox created with ID: {sandbox.id}")
# Start supervisord in a session for new sandbox
start_supervisord_session(sandbox)
logger.info(f"Sandbox environment successfully initialized")
return sandbox
async def delete_sandbox(sandbox_id: str):
"""Delete a sandbox by its ID."""
logger.info(f"Deleting sandbox with ID: {sandbox_id}")
try:
# Get the sandbox
sandbox = daytona.get(sandbox_id)
# Delete the sandbox
daytona.delete(sandbox)
logger.info(f"Successfully deleted sandbox {sandbox_id}")
return True
except Exception as e:
logger.error(f"Error deleting sandbox {sandbox_id}: {str(e)}")
raise e
+138
View File
@@ -0,0 +1,138 @@
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, ClassVar, Dict, Optional
from daytona import Daytona, DaytonaConfig, Sandbox, SandboxState
from pydantic import Field
from app.config import config
from app.daytona.sandbox import create_sandbox, start_supervisord_session
from app.tool.base import BaseTool
from app.utils.files_utils import clean_path
from app.utils.logger import logger
# load_dotenv()
daytona_settings = config.daytona
daytona_config = DaytonaConfig(
api_key=daytona_settings.daytona_api_key,
server_url=daytona_settings.daytona_server_url,
target=daytona_settings.daytona_target,
)
daytona = Daytona(daytona_config)
@dataclass
class ThreadMessage:
"""
Represents a message to be added to a thread.
"""
type: str
content: Dict[str, Any]
is_llm_message: bool = False
metadata: Optional[Dict[str, Any]] = None
timestamp: Optional[float] = field(
default_factory=lambda: datetime.now().timestamp()
)
def to_dict(self) -> Dict[str, Any]:
"""Convert the message to a dictionary for API calls"""
return {
"type": self.type,
"content": self.content,
"is_llm_message": self.is_llm_message,
"metadata": self.metadata or {},
"timestamp": self.timestamp,
}
class SandboxToolsBase(BaseTool):
"""Base class for all sandbox tools that provides project-based sandbox access."""
# Class variable to track if sandbox URLs have been printed
_urls_printed: ClassVar[bool] = False
# Required fields
project_id: Optional[str] = None
# thread_manager: Optional[ThreadManager] = None
# Private fields (not part of the model schema)
_sandbox: Optional[Sandbox] = None
_sandbox_id: Optional[str] = None
_sandbox_pass: Optional[str] = None
workspace_path: str = Field(default="/workspace", exclude=True)
_sessions: dict[str, str] = {}
class Config:
arbitrary_types_allowed = True # Allow non-pydantic types like ThreadManager
underscore_attrs_are_private = True
async def _ensure_sandbox(self) -> Sandbox:
"""Ensure we have a valid sandbox instance, retrieving it from the project if needed."""
if self._sandbox is None:
# Get or start the sandbox
try:
self._sandbox = create_sandbox(password=config.daytona.VNC_password)
# Log URLs if not already printed
if not SandboxToolsBase._urls_printed:
vnc_link = self._sandbox.get_preview_link(6080)
website_link = self._sandbox.get_preview_link(8080)
vnc_url = (
vnc_link.url if hasattr(vnc_link, "url") else str(vnc_link)
)
website_url = (
website_link.url
if hasattr(website_link, "url")
else str(website_link)
)
print("\033[95m***")
print(f"VNC URL: {vnc_url}")
print(f"Website URL: {website_url}")
print("***\033[0m")
SandboxToolsBase._urls_printed = True
except Exception as e:
logger.error(f"Error retrieving or starting sandbox: {str(e)}")
raise e
else:
if (
self._sandbox.state == SandboxState.ARCHIVED
or self._sandbox.state == SandboxState.STOPPED
):
logger.info(f"Sandbox is in {self._sandbox.state} state. Starting...")
try:
daytona.start(self._sandbox)
# Wait a moment for the sandbox to initialize
# sleep(5)
# Refresh sandbox state after starting
# Start supervisord in a session when restarting
start_supervisord_session(self._sandbox)
except Exception as e:
logger.error(f"Error starting sandbox: {e}")
raise e
return self._sandbox
@property
def sandbox(self) -> Sandbox:
"""Get the sandbox instance, ensuring it exists."""
if self._sandbox is None:
raise RuntimeError("Sandbox not initialized. Call _ensure_sandbox() first.")
return self._sandbox
@property
def sandbox_id(self) -> str:
"""Get the sandbox ID, ensuring it exists."""
if self._sandbox_id is None:
raise RuntimeError(
"Sandbox ID not initialized. Call _ensure_sandbox() first."
)
return self._sandbox_id
def clean_path(self, path: str) -> str:
"""Clean and normalize a path to be relative to /workspace."""
cleaned_path = clean_path(path, self.workspace_path)
logger.debug(f"Cleaned path: {path} -> {cleaned_path}")
return cleaned_path
+2
View File
@@ -1,6 +1,7 @@
from app.tool.base import BaseTool
from app.tool.bash import Bash
from app.tool.browser_use_tool import BrowserUseTool
from app.tool.crawl4ai import Crawl4aiTool
from app.tool.create_chat_completion import CreateChatCompletion
from app.tool.planning import PlanningTool
from app.tool.str_replace_editor import StrReplaceEditor
@@ -19,4 +20,5 @@ __all__ = [
"ToolCollection",
"CreateChatCompletion",
"PlanningTool",
"Crawl4aiTool",
]
+124 -23
View File
@@ -1,35 +1,38 @@
import json
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, Union
from pydantic import BaseModel, Field
from app.utils.logger import logger
class BaseTool(ABC, BaseModel):
name: str
description: str
parameters: Optional[dict] = None
class Config:
arbitrary_types_allowed = True
# class BaseTool(ABC, BaseModel):
# name: str
# description: str
# parameters: Optional[dict] = None
async def __call__(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
return await self.execute(**kwargs)
# class Config:
# arbitrary_types_allowed = True
@abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
# async def __call__(self, **kwargs) -> Any:
# """Execute the tool with given parameters."""
# return await self.execute(**kwargs)
def to_param(self) -> Dict:
"""Convert tool to function call format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
# @abstractmethod
# async def execute(self, **kwargs) -> Any:
# """Execute the tool with given parameters."""
# def to_param(self) -> Dict:
# """Convert tool to function call format."""
# return {
# "type": "function",
# "function": {
# "name": self.name,
# "description": self.description,
# "parameters": self.parameters,
# },
# }
class ToolResult(BaseModel):
@@ -72,6 +75,104 @@ class ToolResult(BaseModel):
return type(self)(**{**self.dict(), **kwargs})
class BaseTool(ABC, BaseModel):
"""Consolidated base class for all tools combining BaseModel and Tool functionality.
Provides:
- Pydantic model validation
- Schema registration
- Standardized result handling
- Abstract execution interface
Attributes:
name (str): Tool name
description (str): Tool description
parameters (dict): Tool parameters schema
_schemas (Dict[str, List[ToolSchema]]): Registered method schemas
"""
name: str
description: str
parameters: Optional[dict] = None
# _schemas: Dict[str, List[ToolSchema]] = {}
class Config:
arbitrary_types_allowed = True
underscore_attrs_are_private = False
# def __init__(self, **data):
# """Initialize tool with model validation and schema registration."""
# super().__init__(**data)
# logger.debug(f"Initializing tool class: {self.__class__.__name__}")
# self._register_schemas()
# def _register_schemas(self):
# """Register schemas from all decorated methods."""
# for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
# if hasattr(method, 'tool_schemas'):
# self._schemas[name] = method.tool_schemas
# logger.debug(f"Registered schemas for method '{name}' in {self.__class__.__name__}")
async def __call__(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
return await self.execute(**kwargs)
@abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
def to_param(self) -> Dict:
"""Convert tool to function call format.
Returns:
Dictionary with tool metadata in OpenAI function calling format
"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
# def get_schemas(self) -> Dict[str, List[ToolSchema]]:
# """Get all registered tool schemas.
# Returns:
# Dict mapping method names to their schema definitions
# """
# return self._schemas
def success_response(self, data: Union[Dict[str, Any], str]) -> ToolResult:
"""Create a successful tool result.
Args:
data: Result data (dictionary or string)
Returns:
ToolResult with success=True and formatted output
"""
if isinstance(data, str):
text = data
else:
text = json.dumps(data, indent=2)
logger.debug(f"Created success response for {self.__class__.__name__}")
return ToolResult(output=text)
def fail_response(self, msg: str) -> ToolResult:
"""Create a failed tool result.
Args:
msg: Error message describing the failure
Returns:
ToolResult with success=False and error message
"""
logger.debug(f"Tool {self.__class__.__name__} returned failed result: {msg}")
return ToolResult(error=msg)
class CLIResult(ToolResult):
"""A ToolResult that can be rendered as a CLI output."""
+487
View File
@@ -0,0 +1,487 @@
import asyncio
import base64
import logging
import os
import time
from typing import Dict, Literal, Optional
import aiohttp
from pydantic import Field
from app.daytona.tool_base import Sandbox, SandboxToolsBase
from app.tool.base import ToolResult
KEYBOARD_KEYS = [
"a",
"b",
"c",
"d",
"e",
"f",
"g",
"h",
"i",
"j",
"k",
"l",
"m",
"n",
"o",
"p",
"q",
"r",
"s",
"t",
"u",
"v",
"w",
"x",
"y",
"z",
"0",
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
"enter",
"esc",
"backspace",
"tab",
"space",
"delete",
"ctrl",
"alt",
"shift",
"win",
"up",
"down",
"left",
"right",
"f1",
"f2",
"f3",
"f4",
"f5",
"f6",
"f7",
"f8",
"f9",
"f10",
"f11",
"f12",
"ctrl+c",
"ctrl+v",
"ctrl+x",
"ctrl+z",
"ctrl+a",
"ctrl+s",
"alt+tab",
"alt+f4",
"ctrl+alt+delete",
]
MOUSE_BUTTONS = ["left", "right", "middle"]
_COMPUTER_USE_DESCRIPTION = """\
A comprehensive computer automation tool that allows interaction with the desktop environment.
* This tool provides commands for controlling mouse, keyboard, and taking screenshots
* It maintains state including current mouse position
* Use this when you need to automate desktop applications, fill forms, or perform GUI interactions
Key capabilities include:
* Mouse Control: Move, click, drag, scroll
* Keyboard Input: Type text, press keys or key combinations
* Screenshots: Capture and save screen images
* Waiting: Pause execution for specified duration
"""
class ComputerUseTool(SandboxToolsBase):
"""Computer automation tool for controlling the desktop environment."""
name: str = "computer_use"
description: str = _COMPUTER_USE_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"move_to",
"click",
"scroll",
"typing",
"press",
"wait",
"mouse_down",
"mouse_up",
"drag_to",
"hotkey",
"screenshot",
],
"description": "The computer action to perform",
},
"x": {"type": "number", "description": "X coordinate for mouse actions"},
"y": {"type": "number", "description": "Y coordinate for mouse actions"},
"button": {
"type": "string",
"enum": MOUSE_BUTTONS,
"description": "Mouse button for click/drag actions",
"default": "left",
},
"num_clicks": {
"type": "integer",
"description": "Number of clicks",
"enum": [1, 2, 3],
"default": 1,
},
"amount": {
"type": "integer",
"description": "Scroll amount (positive for up, negative for down)",
"minimum": -10,
"maximum": 10,
},
"text": {"type": "string", "description": "Text to type"},
"key": {
"type": "string",
"enum": KEYBOARD_KEYS,
"description": "Key to press",
},
"keys": {
"type": "string",
"enum": KEYBOARD_KEYS,
"description": "Key combination to press",
},
"duration": {
"type": "number",
"description": "Duration in seconds to wait",
"default": 0.5,
},
},
"required": ["action"],
"dependencies": {
"move_to": ["x", "y"],
"click": [],
"scroll": ["amount"],
"typing": ["text"],
"press": ["key"],
"wait": [],
"mouse_down": [],
"mouse_up": [],
"drag_to": ["x", "y"],
"hotkey": ["keys"],
"screenshot": [],
},
}
session: Optional[aiohttp.ClientSession] = Field(default=None, exclude=True)
mouse_x: int = Field(default=0, exclude=True)
mouse_y: int = Field(default=0, exclude=True)
api_base_url: Optional[str] = Field(default=None, exclude=True)
def __init__(self, sandbox: Optional[Sandbox] = None, **data):
"""Initialize with optional sandbox."""
super().__init__(**data)
if sandbox is not None:
self._sandbox = sandbox # 直接操作基类的私有属性
self.api_base_url = sandbox.get_preview_link(8000).url
logging.info(
f"Initialized ComputerUseTool with API URL: {self.api_base_url}"
)
@classmethod
def create_with_sandbox(cls, sandbox: Sandbox) -> "ComputerUseTool":
"""Factory method to create a tool with sandbox."""
return cls(sandbox=sandbox) # 通过构造函数初始化
async def _get_session(self) -> aiohttp.ClientSession:
"""Get or create aiohttp session for API requests."""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
return self.session
async def _api_request(
self, method: str, endpoint: str, data: Optional[Dict] = None
) -> Dict:
"""Send request to automation service API."""
try:
session = await self._get_session()
url = f"{self.api_base_url}/api{endpoint}"
logging.debug(f"API request: {method} {url} {data}")
if method.upper() == "GET":
async with session.get(url) as response:
result = await response.json()
else: # POST
async with session.post(url, json=data) as response:
result = await response.json()
logging.debug(f"API response: {result}")
return result
except Exception as e:
logging.error(f"API request failed: {str(e)}")
return {"success": False, "error": str(e)}
async def execute(
self,
action: Literal[
"move_to",
"click",
"scroll",
"typing",
"press",
"wait",
"mouse_down",
"mouse_up",
"drag_to",
"hotkey",
"screenshot",
],
x: Optional[float] = None,
y: Optional[float] = None,
button: str = "left",
num_clicks: int = 1,
amount: Optional[int] = None,
text: Optional[str] = None,
key: Optional[str] = None,
keys: Optional[str] = None,
duration: float = 0.5,
**kwargs,
) -> ToolResult:
"""
Execute a specified computer automation action.
Args:
action: The action to perform
x: X coordinate for mouse actions
y: Y coordinate for mouse actions
button: Mouse button for click/drag actions
num_clicks: Number of clicks to perform
amount: Scroll amount (positive for up, negative for down)
text: Text to type
key: Key to press
keys: Key combination to press
duration: Duration in seconds to wait
**kwargs: Additional arguments
Returns:
ToolResult with the action's output or error
"""
try:
if action == "move_to":
if x is None or y is None:
return ToolResult(error="x and y coordinates are required")
x_int = int(round(float(x)))
y_int = int(round(float(y)))
result = await self._api_request(
"POST", "/automation/mouse/move", {"x": x_int, "y": y_int}
)
if result.get("success", False):
self.mouse_x = x_int
self.mouse_y = y_int
return ToolResult(output=f"Moved to ({x_int}, {y_int})")
else:
return ToolResult(
error=f"Failed to move: {result.get('error', 'Unknown error')}"
)
elif action == "click":
x_val = x if x is not None else self.mouse_x
y_val = y if y is not None else self.mouse_y
x_int = int(round(float(x_val)))
y_int = int(round(float(y_val)))
num_clicks = int(num_clicks)
result = await self._api_request(
"POST",
"/automation/mouse/click",
{
"x": x_int,
"y": y_int,
"clicks": num_clicks,
"button": button.lower(),
},
)
if result.get("success", False):
self.mouse_x = x_int
self.mouse_y = y_int
return ToolResult(
output=f"{num_clicks} {button} click(s) performed at ({x_int}, {y_int})"
)
else:
return ToolResult(
error=f"Failed to click: {result.get('error', 'Unknown error')}"
)
elif action == "scroll":
if amount is None:
return ToolResult(error="Scroll amount is required")
amount = int(float(amount))
amount = max(-10, min(10, amount))
result = await self._api_request(
"POST",
"/automation/mouse/scroll",
{"clicks": amount, "x": self.mouse_x, "y": self.mouse_y},
)
if result.get("success", False):
direction = "up" if amount > 0 else "down"
steps = abs(amount)
return ToolResult(
output=f"Scrolled {direction} {steps} step(s) at position ({self.mouse_x}, {self.mouse_y})"
)
else:
return ToolResult(
error=f"Failed to scroll: {result.get('error', 'Unknown error')}"
)
elif action == "typing":
if text is None:
return ToolResult(error="Text is required for typing")
text = str(text)
result = await self._api_request(
"POST",
"/automation/keyboard/write",
{"message": text, "interval": 0.01},
)
if result.get("success", False):
return ToolResult(output=f"Typed: {text}")
else:
return ToolResult(
error=f"Failed to type: {result.get('error', 'Unknown error')}"
)
elif action == "press":
if key is None:
return ToolResult(error="Key is required for press action")
key = str(key).lower()
result = await self._api_request(
"POST", "/automation/keyboard/press", {"keys": key, "presses": 1}
)
if result.get("success", False):
return ToolResult(output=f"Pressed key: {key}")
else:
return ToolResult(
error=f"Failed to press key: {result.get('error', 'Unknown error')}"
)
elif action == "wait":
duration = float(duration)
duration = max(0, min(10, duration))
await asyncio.sleep(duration)
return ToolResult(output=f"Waited {duration} seconds")
elif action == "mouse_down":
x_val = x if x is not None else self.mouse_x
y_val = y if y is not None else self.mouse_y
x_int = int(round(float(x_val)))
y_int = int(round(float(y_val)))
result = await self._api_request(
"POST",
"/automation/mouse/down",
{"x": x_int, "y": y_int, "button": button.lower()},
)
if result.get("success", False):
self.mouse_x = x_int
self.mouse_y = y_int
return ToolResult(
output=f"{button} button pressed at ({x_int}, {y_int})"
)
else:
return ToolResult(
error=f"Failed to press button: {result.get('error', 'Unknown error')}"
)
elif action == "mouse_up":
x_val = x if x is not None else self.mouse_x
y_val = y if y is not None else self.mouse_y
x_int = int(round(float(x_val)))
y_int = int(round(float(y_val)))
result = await self._api_request(
"POST",
"/automation/mouse/up",
{"x": x_int, "y": y_int, "button": button.lower()},
)
if result.get("success", False):
self.mouse_x = x_int
self.mouse_y = y_int
return ToolResult(
output=f"{button} button released at ({x_int}, {y_int})"
)
else:
return ToolResult(
error=f"Failed to release button: {result.get('error', 'Unknown error')}"
)
elif action == "drag_to":
if x is None or y is None:
return ToolResult(error="x and y coordinates are required")
target_x = int(round(float(x)))
target_y = int(round(float(y)))
start_x = self.mouse_x
start_y = self.mouse_y
result = await self._api_request(
"POST",
"/automation/mouse/drag",
{"x": target_x, "y": target_y, "duration": 0.3, "button": "left"},
)
if result.get("success", False):
self.mouse_x = target_x
self.mouse_y = target_y
return ToolResult(
output=f"Dragged from ({start_x}, {start_y}) to ({target_x}, {target_y})"
)
else:
return ToolResult(
error=f"Failed to drag: {result.get('error', 'Unknown error')}"
)
elif action == "hotkey":
if keys is None:
return ToolResult(error="Keys are required for hotkey action")
keys = str(keys).lower().strip()
key_sequence = keys.split("+")
result = await self._api_request(
"POST",
"/automation/keyboard/hotkey",
{"keys": key_sequence, "interval": 0.01},
)
if result.get("success", False):
return ToolResult(output=f"Pressed key combination: {keys}")
else:
return ToolResult(
error=f"Failed to press keys: {result.get('error', 'Unknown error')}"
)
elif action == "screenshot":
result = await self._api_request("POST", "/automation/screenshot")
if "image" in result:
base64_str = result["image"]
timestamp = time.strftime("%Y%m%d_%H%M%S")
# Save screenshot to file
screenshots_dir = "screenshots"
if not os.path.exists(screenshots_dir):
os.makedirs(screenshots_dir)
timestamped_filename = os.path.join(
screenshots_dir, f"screenshot_{timestamp}.png"
)
latest_filename = "latest_screenshot.png"
# Decode base64 string and save to file
img_data = base64.b64decode(base64_str)
with open(timestamped_filename, "wb") as f:
f.write(img_data)
# Save a copy as the latest screenshot
with open(latest_filename, "wb") as f:
f.write(img_data)
return ToolResult(
output=f"Screenshot saved as {timestamped_filename}",
base64_image=base64_str,
)
else:
return ToolResult(error="Failed to capture screenshot")
else:
return ToolResult(error=f"Unknown action: {action}")
except Exception as e:
return ToolResult(error=f"Computer action failed: {str(e)}")
async def cleanup(self):
"""Clean up resources."""
if self.session and not self.session.closed:
await self.session.close()
self.session = None
def __del__(self):
"""Ensure cleanup on destruction."""
if hasattr(self, "session") and self.session is not None:
try:
asyncio.run(self.cleanup())
except RuntimeError:
loop = asyncio.new_event_loop()
loop.run_until_complete(self.cleanup())
loop.close()
+269
View File
@@ -0,0 +1,269 @@
"""
Crawl4AI Web Crawler Tool for OpenManus
This tool integrates Crawl4AI, a high-performance web crawler designed for LLMs and AI agents,
providing fast, precise, and AI-ready data extraction with clean Markdown generation.
"""
import asyncio
from typing import List, Union
from urllib.parse import urlparse
from app.logger import logger
from app.tool.base import BaseTool, ToolResult
class Crawl4aiTool(BaseTool):
"""
Web crawler tool powered by Crawl4AI.
Provides clean markdown extraction optimized for AI processing.
"""
name: str = "crawl4ai"
description: str = """Web crawler that extracts clean, AI-ready content from web pages.
Features:
- Extracts clean markdown content optimized for LLMs
- Handles JavaScript-heavy sites and dynamic content
- Supports multiple URLs in a single request
- Fast and reliable with built-in error handling
Perfect for content analysis, research, and feeding web content to AI models."""
parameters: dict = {
"type": "object",
"properties": {
"urls": {
"type": "array",
"items": {"type": "string"},
"description": "(required) List of URLs to crawl. Can be a single URL or multiple URLs.",
"minItems": 1,
},
"timeout": {
"type": "integer",
"description": "(optional) Timeout in seconds for each URL. Default is 30.",
"default": 30,
"minimum": 5,
"maximum": 120,
},
"bypass_cache": {
"type": "boolean",
"description": "(optional) Whether to bypass cache and fetch fresh content. Default is false.",
"default": False,
},
"word_count_threshold": {
"type": "integer",
"description": "(optional) Minimum word count for content blocks. Default is 10.",
"default": 10,
"minimum": 1,
},
},
"required": ["urls"],
}
async def execute(
self,
urls: Union[str, List[str]],
timeout: int = 30,
bypass_cache: bool = False,
word_count_threshold: int = 10,
) -> ToolResult:
"""
Execute web crawling for the specified URLs.
Args:
urls: Single URL string or list of URLs to crawl
timeout: Timeout in seconds for each URL
bypass_cache: Whether to bypass cache
word_count_threshold: Minimum word count for content blocks
Returns:
ToolResult with crawl results
"""
# Normalize URLs to list
if isinstance(urls, str):
url_list = [urls]
else:
url_list = urls
# Validate URLs
valid_urls = []
for url in url_list:
if self._is_valid_url(url):
valid_urls.append(url)
else:
logger.warning(f"Invalid URL skipped: {url}")
if not valid_urls:
return ToolResult(error="No valid URLs provided")
try:
# Import crawl4ai components
from crawl4ai import (
AsyncWebCrawler,
BrowserConfig,
CacheMode,
CrawlerRunConfig,
)
# Configure browser settings
browser_config = BrowserConfig(
headless=True,
verbose=False,
browser_type="chromium",
ignore_https_errors=True,
java_script_enabled=True,
)
# Configure crawler settings
run_config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS if bypass_cache else CacheMode.ENABLED,
word_count_threshold=word_count_threshold,
process_iframes=True,
remove_overlay_elements=True,
excluded_tags=["script", "style"],
page_timeout=timeout * 1000, # Convert to milliseconds
verbose=False,
wait_until="domcontentloaded",
)
results = []
successful_count = 0
failed_count = 0
# Process each URL
async with AsyncWebCrawler(config=browser_config) as crawler:
for url in valid_urls:
try:
logger.info(f"🕷️ Crawling URL: {url}")
start_time = asyncio.get_event_loop().time()
result = await crawler.arun(url=url, config=run_config)
end_time = asyncio.get_event_loop().time()
execution_time = end_time - start_time
if result.success:
# Count words in markdown
word_count = 0
if hasattr(result, "markdown") and result.markdown:
word_count = len(result.markdown.split())
# Count links
links_count = 0
if hasattr(result, "links") and result.links:
internal_links = result.links.get("internal", [])
external_links = result.links.get("external", [])
links_count = len(internal_links) + len(external_links)
# Count images
images_count = 0
if hasattr(result, "media") and result.media:
images = result.media.get("images", [])
images_count = len(images)
results.append(
{
"url": url,
"success": True,
"status_code": getattr(result, "status_code", 200),
"title": result.metadata.get("title")
if result.metadata
else None,
"markdown": result.markdown
if hasattr(result, "markdown")
else None,
"word_count": word_count,
"links_count": links_count,
"images_count": images_count,
"execution_time": execution_time,
}
)
successful_count += 1
logger.info(
f"✅ Successfully crawled {url} in {execution_time:.2f}s"
)
else:
results.append(
{
"url": url,
"success": False,
"error_message": getattr(
result, "error_message", "Unknown error"
),
"execution_time": execution_time,
}
)
failed_count += 1
logger.warning(f"❌ Failed to crawl {url}")
except Exception as e:
error_msg = f"Error crawling {url}: {str(e)}"
logger.error(error_msg)
results.append(
{"url": url, "success": False, "error_message": error_msg}
)
failed_count += 1
# Format output
output_lines = [f"🕷️ Crawl4AI Results Summary:"]
output_lines.append(f"📊 Total URLs: {len(valid_urls)}")
output_lines.append(f"✅ Successful: {successful_count}")
output_lines.append(f"❌ Failed: {failed_count}")
output_lines.append("")
for i, result in enumerate(results, 1):
output_lines.append(f"{i}. {result['url']}")
if result["success"]:
output_lines.append(
f" ✅ Status: Success (HTTP {result.get('status_code', 'N/A')})"
)
if result.get("title"):
output_lines.append(f" 📄 Title: {result['title']}")
if result.get("markdown"):
# Show first 300 characters of markdown content
content_preview = result["markdown"]
if len(result["markdown"]) > 300:
content_preview += "..."
output_lines.append(f" 📝 Content: {content_preview}")
output_lines.append(
f" 📊 Stats: {result.get('word_count', 0)} words, {result.get('links_count', 0)} links, {result.get('images_count', 0)} images"
)
if result.get("execution_time"):
output_lines.append(
f" ⏱️ Time: {result['execution_time']:.2f}s"
)
else:
output_lines.append(f" ❌ Status: Failed")
if result.get("error_message"):
output_lines.append(f" 🚫 Error: {result['error_message']}")
output_lines.append("")
return ToolResult(output="\n".join(output_lines))
except ImportError:
error_msg = "Crawl4AI is not installed. Please install it with: pip install crawl4ai"
logger.error(error_msg)
return ToolResult(error=error_msg)
except Exception as e:
error_msg = f"Crawl4AI execution failed: {str(e)}"
logger.error(error_msg)
return ToolResult(error=error_msg)
def _is_valid_url(self, url: str) -> bool:
"""Validate if a URL is properly formatted."""
try:
result = urlparse(url)
return all([result.scheme, result.netloc]) and result.scheme in [
"http",
"https",
]
except Exception:
return False
+450
View File
@@ -0,0 +1,450 @@
import base64
import io
import json
import traceback
from typing import Optional # Add this import for Optional
from PIL import Image
from pydantic import Field
from app.daytona.tool_base import ( # Ensure Sandbox is imported correctly
Sandbox,
SandboxToolsBase,
ThreadMessage,
)
from app.tool.base import ToolResult
from app.utils.logger import logger
# Context = TypeVar("Context")
_BROWSER_DESCRIPTION = """\
A sandbox-based browser automation tool that allows interaction with web pages through various actions.
* This tool provides commands for controlling a browser session in a sandboxed environment
* It maintains state across calls, keeping the browser session alive until explicitly closed
* Use this when you need to browse websites, fill forms, click buttons, or extract content in a secure sandbox
* Each action requires specific parameters as defined in the tool's dependencies
Key capabilities include:
* Navigation: Go to specific URLs, go back in history
* Interaction: Click elements by index, input text, send keyboard commands
* Scrolling: Scroll up/down by pixel amount or scroll to specific text
* Tab management: Switch between tabs or close tabs
* Content extraction: Get dropdown options or select dropdown options
"""
# noinspection PyArgumentList
class SandboxBrowserTool(SandboxToolsBase):
"""Tool for executing tasks in a Daytona sandbox with browser-use capabilities."""
name: str = "sandbox_browser"
description: str = _BROWSER_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"navigate_to",
"go_back",
"wait",
"click_element",
"input_text",
"send_keys",
"switch_tab",
"close_tab",
"scroll_down",
"scroll_up",
"scroll_to_text",
"get_dropdown_options",
"select_dropdown_option",
"click_coordinates",
"drag_drop",
],
"description": "The browser action to perform",
},
"url": {
"type": "string",
"description": "URL for 'navigate_to' action",
},
"index": {
"type": "integer",
"description": "Element index for interaction actions",
},
"text": {
"type": "string",
"description": "Text for input or scroll actions",
},
"amount": {
"type": "integer",
"description": "Pixel amount to scroll",
},
"page_id": {
"type": "integer",
"description": "Tab ID for tab management actions",
},
"keys": {
"type": "string",
"description": "Keys to send for keyboard actions",
},
"seconds": {
"type": "integer",
"description": "Seconds to wait",
},
"x": {
"type": "integer",
"description": "X coordinate for click or drag actions",
},
"y": {
"type": "integer",
"description": "Y coordinate for click or drag actions",
},
"element_source": {
"type": "string",
"description": "Source element for drag and drop",
},
"element_target": {
"type": "string",
"description": "Target element for drag and drop",
},
},
"required": ["action"],
"dependencies": {
"navigate_to": ["url"],
"click_element": ["index"],
"input_text": ["index", "text"],
"send_keys": ["keys"],
"switch_tab": ["page_id"],
"close_tab": ["page_id"],
"scroll_down": ["amount"],
"scroll_up": ["amount"],
"scroll_to_text": ["text"],
"get_dropdown_options": ["index"],
"select_dropdown_option": ["index", "text"],
"click_coordinates": ["x", "y"],
"drag_drop": ["element_source", "element_target"],
"wait": ["seconds"],
},
}
browser_message: Optional[ThreadMessage] = Field(default=None, exclude=True)
def __init__(
self, sandbox: Optional[Sandbox] = None, thread_id: Optional[str] = None, **data
):
"""Initialize with optional sandbox and thread_id."""
super().__init__(**data)
if sandbox is not None:
self._sandbox = sandbox # Directly set the base class private attribute
def _validate_base64_image(
self, base64_string: str, max_size_mb: int = 10
) -> tuple[bool, str]:
"""
Validate base64 image data.
Args:
base64_string: The base64 encoded image data
max_size_mb: Maximum allowed image size in megabytes
Returns:
Tuple of (is_valid, error_message)
"""
try:
if not base64_string or len(base64_string) < 10:
return False, "Base64 string is empty or too short"
if base64_string.startswith("data:"):
try:
base64_string = base64_string.split(",", 1)[1]
except (IndexError, ValueError):
return False, "Invalid data URL format"
import re
if not re.match(r"^[A-Za-z0-9+/]*={0,2}$", base64_string):
return False, "Invalid base64 characters detected"
if len(base64_string) % 4 != 0:
return False, "Invalid base64 string length"
try:
image_data = base64.b64decode(base64_string, validate=True)
except Exception as e:
return False, f"Base64 decoding failed: {str(e)}"
max_size_bytes = max_size_mb * 1024 * 1024
if len(image_data) > max_size_bytes:
return False, f"Image size exceeds limit ({max_size_bytes} bytes)"
try:
image_stream = io.BytesIO(image_data)
with Image.open(image_stream) as img:
img.verify()
supported_formats = {"JPEG", "PNG", "GIF", "BMP", "WEBP", "TIFF"}
if img.format not in supported_formats:
return False, f"Unsupported image format: {img.format}"
image_stream.seek(0)
with Image.open(image_stream) as img_check:
width, height = img_check.size
max_dimension = 8192
if width > max_dimension or height > max_dimension:
return (
False,
f"Image dimensions exceed limit ({max_dimension}x{max_dimension})",
)
if width < 1 or height < 1:
return False, f"Invalid image dimensions: {width}x{height}"
except Exception as e:
return False, f"Invalid image data: {str(e)}"
return True, "Valid image"
except Exception as e:
logger.error(f"Unexpected error during base64 image validation: {e}")
return False, f"Validation error: {str(e)}"
async def _execute_browser_action(
self, endpoint: str, params: dict = None, method: str = "POST"
) -> ToolResult:
"""Execute a browser automation action through the sandbox API."""
try:
await self._ensure_sandbox()
url = f"http://localhost:8003/api/automation/{endpoint}"
if method == "GET" and params:
query_params = "&".join([f"{k}={v}" for k, v in params.items()])
url = f"{url}?{query_params}"
curl_cmd = (
f"curl -s -X {method} '{url}' -H 'Content-Type: application/json'"
)
else:
curl_cmd = (
f"curl -s -X {method} '{url}' -H 'Content-Type: application/json'"
)
if params:
json_data = json.dumps(params)
curl_cmd += f" -d '{json_data}'"
logger.debug(f"Executing curl command: {curl_cmd}")
response = self.sandbox.process.exec(curl_cmd, timeout=30)
if response.exit_code == 0:
try:
result = json.loads(response.result)
result.setdefault("content", "")
result.setdefault("role", "assistant")
if "screenshot_base64" in result:
screenshot_data = result["screenshot_base64"]
is_valid, validation_message = self._validate_base64_image(
screenshot_data
)
if not is_valid:
logger.warning(
f"Screenshot validation failed: {validation_message}"
)
result["image_validation_error"] = validation_message
del result["screenshot_base64"]
# added_message = await self.thread_manager.add_message(
# thread_id=self.thread_id,
# type="browser_state",
# content=result,
# is_llm_message=False
# )
message = ThreadMessage(
type="browser_state", content=result, is_llm_message=False
)
self.browser_message = message
success_response = {
"success": result.get("success", False),
"message": result.get("message", "Browser action completed"),
}
# if added_message and 'message_id' in added_message:
# success_response['message_id'] = added_message['message_id']
for field in [
"url",
"title",
"element_count",
"pixels_below",
"ocr_text",
"image_url",
]:
if field in result:
success_response[field] = result[field]
return (
self.success_response(success_response)
if success_response["success"]
else self.fail_response(success_response)
)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse response JSON: {e}")
return self.fail_response(f"Failed to parse response JSON: {e}")
else:
logger.error(f"Browser automation request failed: {response}")
return self.fail_response(
f"Browser automation request failed: {response}"
)
except Exception as e:
logger.error(f"Error executing browser action: {e}")
logger.debug(traceback.format_exc())
return self.fail_response(f"Error executing browser action: {e}")
async def execute(
self,
action: str,
url: Optional[str] = None,
index: Optional[int] = None,
text: Optional[str] = None,
amount: Optional[int] = None,
page_id: Optional[int] = None,
keys: Optional[str] = None,
seconds: Optional[int] = None,
x: Optional[int] = None,
y: Optional[int] = None,
element_source: Optional[str] = None,
element_target: Optional[str] = None,
**kwargs,
) -> ToolResult:
"""
Execute a browser action in the sandbox environment.
Args:
action: The browser action to perform
url: URL for navigation
index: Element index for interaction
text: Text for input or scroll actions
amount: Pixel amount to scroll
page_id: Tab ID for tab management
keys: Keys to send for keyboard actions
seconds: Seconds to wait
x: X coordinate for click/drag
y: Y coordinate for click/drag
element_source: Source element for drag and drop
element_target: Target element for drag and drop
Returns:
ToolResult with the action's output or error
"""
# async with self.lock:
try:
# Navigation actions
if action == "navigate_to":
if not url:
return self.fail_response("URL is required for navigation")
return await self._execute_browser_action("navigate_to", {"url": url})
elif action == "go_back":
return await self._execute_browser_action("go_back", {})
# Interaction actions
elif action == "click_element":
if index is None:
return self.fail_response("Index is required for click_element")
return await self._execute_browser_action(
"click_element", {"index": index}
)
elif action == "input_text":
if index is None or not text:
return self.fail_response(
"Index and text are required for input_text"
)
return await self._execute_browser_action(
"input_text", {"index": index, "text": text}
)
elif action == "send_keys":
if not keys:
return self.fail_response("Keys are required for send_keys")
return await self._execute_browser_action("send_keys", {"keys": keys})
# Tab management
elif action == "switch_tab":
if page_id is None:
return self.fail_response("Page ID is required for switch_tab")
return await self._execute_browser_action(
"switch_tab", {"page_id": page_id}
)
elif action == "close_tab":
if page_id is None:
return self.fail_response("Page ID is required for close_tab")
return await self._execute_browser_action(
"close_tab", {"page_id": page_id}
)
# Scrolling actions
elif action == "scroll_down":
params = {"amount": amount} if amount is not None else {}
return await self._execute_browser_action("scroll_down", params)
elif action == "scroll_up":
params = {"amount": amount} if amount is not None else {}
return await self._execute_browser_action("scroll_up", params)
elif action == "scroll_to_text":
if not text:
return self.fail_response("Text is required for scroll_to_text")
return await self._execute_browser_action(
"scroll_to_text", {"text": text}
)
# Dropdown actions
elif action == "get_dropdown_options":
if index is None:
return self.fail_response(
"Index is required for get_dropdown_options"
)
return await self._execute_browser_action(
"get_dropdown_options", {"index": index}
)
elif action == "select_dropdown_option":
if index is None or not text:
return self.fail_response(
"Index and text are required for select_dropdown_option"
)
return await self._execute_browser_action(
"select_dropdown_option", {"index": index, "text": text}
)
# Coordinate-based actions
elif action == "click_coordinates":
if x is None or y is None:
return self.fail_response(
"X and Y coordinates are required for click_coordinates"
)
return await self._execute_browser_action(
"click_coordinates", {"x": x, "y": y}
)
elif action == "drag_drop":
if not element_source or not element_target:
return self.fail_response(
"Source and target elements are required for drag_drop"
)
return await self._execute_browser_action(
"drag_drop",
{
"element_source": element_source,
"element_target": element_target,
},
)
# Utility actions
elif action == "wait":
seconds_to_wait = seconds if seconds is not None else 3
return await self._execute_browser_action(
"wait", {"seconds": seconds_to_wait}
)
else:
return self.fail_response(f"Unknown action: {action}")
except Exception as e:
logger.error(f"Error executing browser action: {e}")
return self.fail_response(f"Error executing browser action: {e}")
async def get_current_state(
self, message: Optional[ThreadMessage] = None
) -> ToolResult:
"""
Get the current browser state as a ToolResult.
If context is not provided, uses self.context.
"""
try:
# Use provided context or fall back to self.context
message = message or self.browser_message
if not message:
return ToolResult(error="Browser context not initialized")
state = message.content
screenshot = state.get("screenshot_base64")
# Build the state info with all required fields
state_info = {
"url": state.get("url", ""),
"title": state.get("title", ""),
"tabs": [tab.model_dump() for tab in state.get("tabs", [])],
"pixels_above": getattr(state, "pixels_above", 0),
"pixels_below": getattr(state, "pixels_below", 0),
"help": "[0], [1], [2], etc., represent clickable indices corresponding to the elements listed. Clicking on these indices will navigate to or interact with the respective content behind them.",
}
return ToolResult(
output=json.dumps(state_info, indent=4, ensure_ascii=False),
base64_image=screenshot,
)
except Exception as e:
return ToolResult(error=f"Failed to get browser state: {str(e)}")
@classmethod
def create_with_sandbox(cls, sandbox: Sandbox) -> "SandboxBrowserTool":
"""Factory method to create a tool with sandbox."""
return cls(sandbox=sandbox)
+361
View File
@@ -0,0 +1,361 @@
import asyncio
from typing import Optional, TypeVar
from pydantic import Field
from app.daytona.tool_base import Sandbox, SandboxToolsBase
from app.tool.base import ToolResult
from app.utils.files_utils import clean_path, should_exclude_file
from app.utils.logger import logger
Context = TypeVar("Context")
_FILES_DESCRIPTION = """\
A sandbox-based file system tool that allows file operations in a secure sandboxed environment.
* This tool provides commands for creating, reading, updating, and deleting files in the workspace
* All operations are performed relative to the /workspace directory for security
* Use this when you need to manage files, edit code, or manipulate file contents in a sandbox
* Each action requires specific parameters as defined in the tool's dependencies
Key capabilities include:
* File creation: Create new files with specified content and permissions
* File modification: Replace specific strings or completely rewrite files
* File deletion: Remove files from the workspace
* File reading: Read file contents with optional line range specification
"""
class SandboxFilesTool(SandboxToolsBase):
name: str = "sandbox_files"
description: str = _FILES_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"create_file",
"str_replace",
"full_file_rewrite",
"delete_file",
],
"description": "The file operation to perform",
},
"file_path": {
"type": "string",
"description": "Path to the file, relative to /workspace (e.g., 'src/main.py')",
},
"file_contents": {
"type": "string",
"description": "Content to write to the file",
},
"old_str": {
"type": "string",
"description": "Text to be replaced (must appear exactly once)",
},
"new_str": {
"type": "string",
"description": "Replacement text",
},
"permissions": {
"type": "string",
"description": "File permissions in octal format (e.g., '644')",
"default": "644",
},
},
"required": ["action"],
"dependencies": {
"create_file": ["file_path", "file_contents"],
"str_replace": ["file_path", "old_str", "new_str"],
"full_file_rewrite": ["file_path", "file_contents"],
"delete_file": ["file_path"],
},
}
SNIPPET_LINES: int = Field(default=4, exclude=True)
# workspace_path: str = Field(default="/workspace", exclude=True)
# sandbox: Optional[Sandbox] = Field(default=None, exclude=True)
def __init__(
self, sandbox: Optional[Sandbox] = None, thread_id: Optional[str] = None, **data
):
"""Initialize with optional sandbox and thread_id."""
super().__init__(**data)
if sandbox is not None:
self._sandbox = sandbox
def clean_path(self, path: str) -> str:
"""Clean and normalize a path to be relative to /workspace"""
return clean_path(path, self.workspace_path)
def _should_exclude_file(self, rel_path: str) -> bool:
"""Check if a file should be excluded based on path, name, or extension"""
return should_exclude_file(rel_path)
def _file_exists(self, path: str) -> bool:
"""Check if a file exists in the sandbox"""
try:
self.sandbox.fs.get_file_info(path)
return True
except Exception:
return False
async def get_workspace_state(self) -> dict:
"""Get the current workspace state by reading all files"""
files_state = {}
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
files = self.sandbox.fs.list_files(self.workspace_path)
for file_info in files:
rel_path = file_info.name
# Skip excluded files and directories
if self._should_exclude_file(rel_path) or file_info.is_dir:
continue
try:
full_path = f"{self.workspace_path}/{rel_path}"
content = self.sandbox.fs.download_file(full_path).decode()
files_state[rel_path] = {
"content": content,
"is_dir": file_info.is_dir,
"size": file_info.size,
"modified": file_info.mod_time,
}
except Exception as e:
print(f"Error reading file {rel_path}: {e}")
except UnicodeDecodeError:
print(f"Skipping binary file: {rel_path}")
return files_state
except Exception as e:
print(f"Error getting workspace state: {str(e)}")
return {}
async def execute(
self,
action: str,
file_path: Optional[str] = None,
file_contents: Optional[str] = None,
old_str: Optional[str] = None,
new_str: Optional[str] = None,
permissions: Optional[str] = "644",
**kwargs,
) -> ToolResult:
"""
Execute a file operation in the sandbox environment.
Args:
action: The file operation to perform
file_path: Path to the file relative to /workspace
file_contents: Content to write to the file
old_str: Text to be replaced (for str_replace)
new_str: Replacement text (for str_replace)
permissions: File permissions in octal format
Returns:
ToolResult with the operation's output or error
"""
async with asyncio.Lock():
try:
# File creation
if action == "create_file":
if not file_path or not file_contents:
return self.fail_response(
"file_path and file_contents are required for create_file"
)
return await self._create_file(
file_path, file_contents, permissions
)
# String replacement
elif action == "str_replace":
if not file_path or not old_str or not new_str:
return self.fail_response(
"file_path, old_str, and new_str are required for str_replace"
)
return await self._str_replace(file_path, old_str, new_str)
# Full file rewrite
elif action == "full_file_rewrite":
if not file_path or not file_contents:
return self.fail_response(
"file_path and file_contents are required for full_file_rewrite"
)
return await self._full_file_rewrite(
file_path, file_contents, permissions
)
# File deletion
elif action == "delete_file":
if not file_path:
return self.fail_response(
"file_path is required for delete_file"
)
return await self._delete_file(file_path)
else:
return self.fail_response(f"Unknown action: {action}")
except Exception as e:
logger.error(f"Error executing file action: {e}")
return self.fail_response(f"Error executing file action: {e}")
async def _create_file(
self, file_path: str, file_contents: str, permissions: str = "644"
) -> ToolResult:
"""Create a new file with the provided contents"""
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
file_path = self.clean_path(file_path)
full_path = f"{self.workspace_path}/{file_path}"
if self._file_exists(full_path):
return self.fail_response(
f"File '{file_path}' already exists. Use full_file_rewrite to modify existing files."
)
# Create parent directories if needed
parent_dir = "/".join(full_path.split("/")[:-1])
if parent_dir:
self.sandbox.fs.create_folder(parent_dir, "755")
# Write the file content
self.sandbox.fs.upload_file(file_contents.encode(), full_path)
self.sandbox.fs.set_file_permissions(full_path, permissions)
message = f"File '{file_path}' created successfully."
# Check if index.html was created and add 8080 server info (only in root workspace)
if file_path.lower() == "index.html":
try:
website_link = self.sandbox.get_preview_link(8080)
website_url = (
website_link.url
if hasattr(website_link, "url")
else str(website_link).split("url='")[1].split("'")[0]
)
message += f"\n\n[Auto-detected index.html - HTTP server available at: {website_url}]"
message += "\n[Note: Use the provided HTTP server URL above instead of starting a new server]"
except Exception as e:
logger.warning(
f"Failed to get website URL for index.html: {str(e)}"
)
return self.success_response(message)
except Exception as e:
return self.fail_response(f"Error creating file: {str(e)}")
async def _str_replace(
self, file_path: str, old_str: str, new_str: str
) -> ToolResult:
"""Replace specific text in a file"""
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
file_path = self.clean_path(file_path)
full_path = f"{self.workspace_path}/{file_path}"
if not self._file_exists(full_path):
return self.fail_response(f"File '{file_path}' does not exist")
content = self.sandbox.fs.download_file(full_path).decode()
old_str = old_str.expandtabs()
new_str = new_str.expandtabs()
occurrences = content.count(old_str)
if occurrences == 0:
return self.fail_response(f"String '{old_str}' not found in file")
if occurrences > 1:
lines = [
i + 1
for i, line in enumerate(content.split("\n"))
if old_str in line
]
return self.fail_response(
f"Multiple occurrences found in lines {lines}. Please ensure string is unique"
)
# Perform replacement
new_content = content.replace(old_str, new_str)
self.sandbox.fs.upload_file(new_content.encode(), full_path)
# Show snippet around the edit
replacement_line = content.split(old_str)[0].count("\n")
start_line = max(0, replacement_line - self.SNIPPET_LINES)
end_line = replacement_line + self.SNIPPET_LINES + new_str.count("\n")
snippet = "\n".join(new_content.split("\n")[start_line : end_line + 1])
message = f"Replacement successful."
return self.success_response(message)
except Exception as e:
return self.fail_response(f"Error replacing string: {str(e)}")
async def _full_file_rewrite(
self, file_path: str, file_contents: str, permissions: str = "644"
) -> ToolResult:
"""Completely rewrite an existing file with new content"""
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
file_path = self.clean_path(file_path)
full_path = f"{self.workspace_path}/{file_path}"
if not self._file_exists(full_path):
return self.fail_response(
f"File '{file_path}' does not exist. Use create_file to create a new file."
)
self.sandbox.fs.upload_file(file_contents.encode(), full_path)
self.sandbox.fs.set_file_permissions(full_path, permissions)
message = f"File '{file_path}' completely rewritten successfully."
# Check if index.html was rewritten and add 8080 server info (only in root workspace)
if file_path.lower() == "index.html":
try:
website_link = self.sandbox.get_preview_link(8080)
website_url = (
website_link.url
if hasattr(website_link, "url")
else str(website_link).split("url='")[1].split("'")[0]
)
message += f"\n\n[Auto-detected index.html - HTTP server available at: {website_url}]"
message += "\n[Note: Use the provided HTTP server URL above instead of starting a new server]"
except Exception as e:
logger.warning(
f"Failed to get website URL for index.html: {str(e)}"
)
return self.success_response(message)
except Exception as e:
return self.fail_response(f"Error rewriting file: {str(e)}")
async def _delete_file(self, file_path: str) -> ToolResult:
"""Delete a file at the given path"""
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
file_path = self.clean_path(file_path)
full_path = f"{self.workspace_path}/{file_path}"
if not self._file_exists(full_path):
return self.fail_response(f"File '{file_path}' does not exist")
self.sandbox.fs.delete_file(full_path)
return self.success_response(f"File '{file_path}' deleted successfully.")
except Exception as e:
return self.fail_response(f"Error deleting file: {str(e)}")
async def cleanup(self):
"""Clean up sandbox resources."""
@classmethod
def create_with_context(cls, context: Context) -> "SandboxFilesTool[Context]":
"""Factory method to create a SandboxFilesTool with a specific context."""
raise NotImplementedError(
"create_with_context not implemented for SandboxFilesTool"
)
+419
View File
@@ -0,0 +1,419 @@
import asyncio
import time
from typing import Any, Dict, Optional, TypeVar
from uuid import uuid4
from app.daytona.tool_base import Sandbox, SandboxToolsBase
from app.tool.base import ToolResult
from app.utils.logger import logger
Context = TypeVar("Context")
_SHELL_DESCRIPTION = """\
Execute a shell command in the workspace directory.
IMPORTANT: Commands are non-blocking by default and run in a tmux session.
This is ideal for long-running operations like starting servers or build processes.
Uses sessions to maintain state between commands.
This tool is essential for running CLI tools, installing packages, and managing system operations.
"""
class SandboxShellTool(SandboxToolsBase):
"""Tool for executing tasks in a Daytona sandbox with browser-use capabilities.
Uses sessions for maintaining state between commands and provides comprehensive process management.
"""
name: str = "sandbox_shell"
description: str = _SHELL_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"execute_command",
"check_command_output",
"terminate_command",
"list_commands",
],
"description": "The shell action to perform",
},
"command": {
"type": "string",
"description": "The shell command to execute. Use this for running CLI tools, installing packages, "
"or system operations. Commands can be chained using &&, ||, and | operators.",
},
"folder": {
"type": "string",
"description": "Optional relative path to a subdirectory of /workspace where the command should be "
"executed. Example: 'data/pdfs'",
},
"session_name": {
"type": "string",
"description": "Optional name of the tmux session to use. Use named sessions for related commands "
"that need to maintain state. Defaults to a random session name.",
},
"blocking": {
"type": "boolean",
"description": "Whether to wait for the command to complete. Defaults to false for non-blocking "
"execution.",
"default": False,
},
"timeout": {
"type": "integer",
"description": "Optional timeout in seconds for blocking commands. Defaults to 60. Ignored for "
"non-blocking commands.",
"default": 60,
},
"kill_session": {
"type": "boolean",
"description": "Whether to terminate the tmux session after checking. Set to true when you're done "
"with the command.",
"default": False,
},
},
"required": ["action"],
"dependencies": {
"execute_command": ["command"],
"check_command_output": ["session_name"],
"terminate_command": ["session_name"],
"list_commands": [],
},
}
def __init__(
self, sandbox: Optional[Sandbox] = None, thread_id: Optional[str] = None, **data
):
"""Initialize with optional sandbox and thread_id."""
super().__init__(**data)
if sandbox is not None:
self._sandbox = sandbox
async def _ensure_session(self, session_name: str = "default") -> str:
"""Ensure a session exists and return its ID."""
if session_name not in self._sessions:
session_id = str(uuid4())
try:
await self._ensure_sandbox() # Ensure sandbox is initialized
self.sandbox.process.create_session(session_id)
self._sessions[session_name] = session_id
except Exception as e:
raise RuntimeError(f"Failed to create session: {str(e)}")
return self._sessions[session_name]
async def _cleanup_session(self, session_name: str):
"""Clean up a session if it exists."""
if session_name in self._sessions:
try:
await self._ensure_sandbox() # Ensure sandbox is initialized
self.sandbox.process.delete_session(self._sessions[session_name])
del self._sessions[session_name]
except Exception as e:
print(f"Warning: Failed to cleanup session {session_name}: {str(e)}")
async def _execute_raw_command(self, command: str) -> Dict[str, Any]:
"""Execute a raw command directly in the sandbox."""
# Ensure session exists for raw commands
session_id = await self._ensure_session("raw_commands")
# Execute command in session
from app.daytona.sandbox import SessionExecuteRequest
req = SessionExecuteRequest(
command=command, run_async=False, cwd=self.workspace_path
)
response = self.sandbox.process.execute_session_command(
session_id=session_id,
req=req,
timeout=30, # Short timeout for utility commands
)
logs = self.sandbox.process.get_session_command_logs(
session_id=session_id, command_id=response.cmd_id
)
return {"output": logs, "exit_code": response.exit_code}
async def _execute_command(
self,
command: str,
folder: Optional[str] = None,
session_name: Optional[str] = None,
blocking: bool = False,
timeout: int = 60,
) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Set up working directory
cwd = self.workspace_path
if folder:
folder = folder.strip("/")
cwd = f"{self.workspace_path}/{folder}"
# Generate a session name if not provided
if not session_name:
session_name = f"session_{str(uuid4())[:8]}"
# Check if tmux session already exists
check_session = await self._execute_raw_command(
f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'"
)
session_exists = "not_exists" not in check_session.get("output", "")
if not session_exists:
# Create a new tmux session
await self._execute_raw_command(
f"tmux new-session -d -s {session_name}"
)
# Ensure we're in the correct directory and send command to tmux
full_command = f"cd {cwd} && {command}"
wrapped_command = full_command.replace('"', '\\"') # Escape double quotes
# Send command to tmux session
await self._execute_raw_command(
f'tmux send-keys -t {session_name} "{wrapped_command}" Enter'
)
if blocking:
# For blocking execution, wait and capture output
start_time = time.time()
while (time.time() - start_time) < timeout:
# Wait a bit before checking
time.sleep(2)
# Check if session still exists (command might have exited)
check_result = await self._execute_raw_command(
f"tmux has-session -t {session_name} 2>/dev/null || echo 'ended'"
)
if "ended" in check_result.get("output", ""):
break
# Get current output and check for common completion indicators
output_result = await self._execute_raw_command(
f"tmux capture-pane -t {session_name} -p -S - -E -"
)
current_output = output_result.get("output", "")
# Check for prompt indicators that suggest command completion
last_lines = current_output.split("\n")[-3:]
completion_indicators = [
"$",
"#",
">",
"Done",
"Completed",
"Finished",
"",
]
if any(
indicator in line
for indicator in completion_indicators
for line in last_lines
):
break
# Capture final output
output_result = await self._execute_raw_command(
f"tmux capture-pane -t {session_name} -p -S - -E -"
)
final_output = output_result.get("output", "")
# Kill the session after capture
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
return self.success_response(
{
"output": final_output,
"session_name": session_name,
"cwd": cwd,
"completed": True,
}
)
else:
# For non-blocking, just return immediately
return self.success_response(
{
"session_name": session_name,
"cwd": cwd,
"message": f"Command sent to tmux session '{session_name}'. Use check_command_output to view results.",
"completed": False,
}
)
except Exception as e:
# Attempt to clean up session in case of error
if session_name:
try:
await self._execute_raw_command(
f"tmux kill-session -t {session_name}"
)
except:
pass
return self.fail_response(f"Error executing command: {str(e)}")
async def _check_command_output(
self, session_name: str, kill_session: bool = False
) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Check if session exists
check_result = await self._execute_raw_command(
f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'"
)
if "not_exists" in check_result.get("output", ""):
return self.fail_response(
f"Tmux session '{session_name}' does not exist."
)
# Get output from tmux pane
output_result = await self._execute_raw_command(
f"tmux capture-pane -t {session_name} -p -S - -E -"
)
output = output_result.get("output", "")
# Kill session if requested
if kill_session:
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
termination_status = "Session terminated."
else:
termination_status = "Session still running."
return self.success_response(
{
"output": output,
"session_name": session_name,
"status": termination_status,
}
)
except Exception as e:
return self.fail_response(f"Error checking command output: {str(e)}")
async def _terminate_command(self, session_name: str) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Check if session exists
check_result = await self._execute_raw_command(
f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'"
)
if "not_exists" in check_result.get("output", ""):
return self.fail_response(
f"Tmux session '{session_name}' does not exist."
)
# Kill the session
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
return self.success_response(
{"message": f"Tmux session '{session_name}' terminated successfully."}
)
except Exception as e:
return self.fail_response(f"Error terminating command: {str(e)}")
async def _list_commands(self) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# List all tmux sessions
result = await self._execute_raw_command(
"tmux list-sessions 2>/dev/null || echo 'No sessions'"
)
output = result.get("output", "")
if "No sessions" in output or not output.strip():
return self.success_response(
{"message": "No active tmux sessions found.", "sessions": []}
)
# Parse session list
sessions = []
for line in output.split("\n"):
if line.strip():
parts = line.split(":")
if parts:
session_name = parts[0].strip()
sessions.append(session_name)
return self.success_response(
{
"message": f"Found {len(sessions)} active sessions.",
"sessions": sessions,
}
)
except Exception as e:
return self.fail_response(f"Error listing commands: {str(e)}")
async def execute(
self,
action: str,
command: str,
folder: Optional[str] = None,
session_name: Optional[str] = None,
blocking: bool = False,
timeout: int = 60,
kill_session: bool = False,
) -> ToolResult:
"""
Execute a browser action in the sandbox environment.
Args:
timeout:
blocking:
session_name:
folder:
command:
kill_session:
action: The browser action to perform
Returns:
ToolResult with the action's output or error
"""
async with asyncio.Lock():
try:
# Navigation actions
if action == "execute_command":
if not command:
return self.fail_response("command is required for navigation")
return await self._execute_command(
command, folder, session_name, blocking, timeout
)
elif action == "check_command_output":
if session_name is None:
return self.fail_response(
"session_name is required for navigation"
)
return await self._check_command_output(session_name, kill_session)
elif action == "terminate_command":
if session_name is None:
return self.fail_response(
"session_name is required for click_element"
)
return await self._terminate_command(session_name)
elif action == "list_commands":
return await self._list_commands()
else:
return self.fail_response(f"Unknown action: {action}")
except Exception as e:
logger.error(f"Error executing shell action: {e}")
return self.fail_response(f"Error executing shell action: {e}")
async def cleanup(self):
"""Clean up all sessions."""
for session_name in list(self._sessions.keys()):
await self._cleanup_session(session_name)
# Also clean up any tmux sessions
try:
await self._ensure_sandbox()
await self._execute_raw_command("tmux kill-server 2>/dev/null || true")
except Exception as e:
logger.error(f"Error shell box cleanup action: {e}")
+178
View File
@@ -0,0 +1,178 @@
import base64
import mimetypes
import os
from io import BytesIO
from typing import Optional
from PIL import Image
from pydantic import Field
from app.daytona.tool_base import Sandbox, SandboxToolsBase, ThreadMessage
from app.tool.base import ToolResult
# 最大文件大小(原图10MB,压缩后5MB)
MAX_IMAGE_SIZE = 10 * 1024 * 1024
MAX_COMPRESSED_SIZE = 5 * 1024 * 1024
# 压缩设置
DEFAULT_MAX_WIDTH = 1920
DEFAULT_MAX_HEIGHT = 1080
DEFAULT_JPEG_QUALITY = 85
DEFAULT_PNG_COMPRESS_LEVEL = 6
_VISION_DESCRIPTION = """
A sandbox-based vision tool that allows the agent to read image files inside the sandbox using the see_image action.
* Only the see_image action is supported, with the parameter being the relative path of the image under /workspace.
* The image will be compressed and converted to base64 for use in subsequent context.
* Supported formats: JPG, PNG, GIF, WEBP. Maximum size: 10MB.
"""
class SandboxVisionTool(SandboxToolsBase):
name: str = "sandbox_vision"
description: str = _VISION_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["see_image"],
"description": "要执行的视觉动作,目前仅支持 see_image",
},
"file_path": {
"type": "string",
"description": "图片在 /workspace 下的相对路径,如 'screenshots/image.png'",
},
},
"required": ["action", "file_path"],
"dependencies": {"see_image": ["file_path"]},
}
# def __init__(self, project_id: str, thread_id: str, thread_manager: ThreadManager):
# super().__init__(project_id=project_id, thread_manager=thread_manager)
# self.thread_id = thread_id
# self.thread_manager = thread_manager
vision_message: Optional[ThreadMessage] = Field(default=None, exclude=True)
def __init__(
self, sandbox: Optional[Sandbox] = None, thread_id: Optional[str] = None, **data
):
"""Initialize with optional sandbox and thread_id."""
super().__init__(**data)
if sandbox is not None:
self._sandbox = sandbox
def compress_image(self, image_bytes: bytes, mime_type: str, file_path: str):
"""压缩图片,保持合理质量。"""
try:
img = Image.open(BytesIO(image_bytes))
if img.mode in ("RGBA", "LA", "P"):
background = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "P":
img = img.convert("RGBA")
background.paste(
img, mask=img.split()[-1] if img.mode == "RGBA" else None
)
img = background
width, height = img.size
if width > DEFAULT_MAX_WIDTH or height > DEFAULT_MAX_HEIGHT:
ratio = min(DEFAULT_MAX_WIDTH / width, DEFAULT_MAX_HEIGHT / height)
new_width = int(width * ratio)
new_height = int(height * ratio)
img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
output = BytesIO()
if mime_type == "image/gif":
img.save(output, format="GIF", optimize=True)
output_mime = "image/gif"
elif mime_type == "image/png":
img.save(
output,
format="PNG",
optimize=True,
compress_level=DEFAULT_PNG_COMPRESS_LEVEL,
)
output_mime = "image/png"
else:
img.save(
output, format="JPEG", quality=DEFAULT_JPEG_QUALITY, optimize=True
)
output_mime = "image/jpeg"
compressed_bytes = output.getvalue()
return compressed_bytes, output_mime
except Exception:
return image_bytes, mime_type
async def execute(
self, action: str, file_path: Optional[str] = None, **kwargs
) -> ToolResult:
"""
执行视觉动作,目前仅支持 see_image。
参数:
action: 必须为 'see_image'
file_path: 图片相对路径
"""
if action != "see_image":
return self.fail_response(f"未知的视觉动作: {action}")
if not file_path:
return self.fail_response("file_path 参数不能为空")
try:
await self._ensure_sandbox()
cleaned_path = self.clean_path(file_path)
full_path = f"{self.workspace_path}/{cleaned_path}"
try:
file_info = self.sandbox.fs.get_file_info(full_path)
if file_info.is_dir:
return self.fail_response(f"路径 '{cleaned_path}' 是目录,不是图片文件。")
except Exception:
return self.fail_response(f"图片文件未找到: '{cleaned_path}'")
if file_info.size > MAX_IMAGE_SIZE:
return self.fail_response(
f"图片文件 '{cleaned_path}' 过大 ({file_info.size / (1024*1024):.2f}MB),最大允许 {MAX_IMAGE_SIZE / (1024*1024)}MB。"
)
try:
image_bytes = self.sandbox.fs.download_file(full_path)
except Exception:
return self.fail_response(f"无法读取图片文件: {cleaned_path}")
mime_type, _ = mimetypes.guess_type(full_path)
if not mime_type or not mime_type.startswith("image/"):
ext = os.path.splitext(cleaned_path)[1].lower()
if ext == ".jpg" or ext == ".jpeg":
mime_type = "image/jpeg"
elif ext == ".png":
mime_type = "image/png"
elif ext == ".gif":
mime_type = "image/gif"
elif ext == ".webp":
mime_type = "image/webp"
else:
return self.fail_response(
f"不支持或未知的图片格式: '{cleaned_path}'。支持: JPG, PNG, GIF, WEBP。"
)
compressed_bytes, compressed_mime_type = self.compress_image(
image_bytes, mime_type, cleaned_path
)
if len(compressed_bytes) > MAX_COMPRESSED_SIZE:
return self.fail_response(
f"图片文件 '{cleaned_path}' 压缩后仍过大 ({len(compressed_bytes) / (1024*1024):.2f}MB),最大允许 {MAX_COMPRESSED_SIZE / (1024*1024)}MB。"
)
base64_image = base64.b64encode(compressed_bytes).decode("utf-8")
image_context_data = {
"mime_type": compressed_mime_type,
"base64": base64_image,
"file_path": cleaned_path,
"original_size": file_info.size,
"compressed_size": len(compressed_bytes),
}
message = ThreadMessage(
type="image_context", content=image_context_data, is_llm_message=False
)
self.vision_message = message
# return self.success_response(f"成功加载并压缩图片 '{cleaned_path}' (由 {file_info.size / 1024:.1f}KB 压缩到 {len(compressed_bytes) / 1024:.1f}KB)。")
return ToolResult(
output=f"成功加载并压缩图片 '{cleaned_path}'",
base64_image=base64_image,
)
except Exception as e:
return self.fail_response(f"see_image 执行异常: {str(e)}")
+1
View File
@@ -0,0 +1 @@
# Utility functions and constants for agent tools
+87
View File
@@ -0,0 +1,87 @@
import os
# Files to exclude from operations
EXCLUDED_FILES = {
".DS_Store",
".gitignore",
"package-lock.json",
"postcss.config.js",
"postcss.config.mjs",
"jsconfig.json",
"components.json",
"tsconfig.tsbuildinfo",
"tsconfig.json",
}
# Directories to exclude from operations
EXCLUDED_DIRS = {"node_modules", ".next", "dist", "build", ".git"}
# File extensions to exclude from operations
EXCLUDED_EXT = {
".ico",
".svg",
".png",
".jpg",
".jpeg",
".gif",
".bmp",
".tiff",
".webp",
".db",
".sql",
}
def should_exclude_file(rel_path: str) -> bool:
"""Check if a file should be excluded based on path, name, or extension
Args:
rel_path: Relative path of the file to check
Returns:
True if the file should be excluded, False otherwise
"""
# Check filename
filename = os.path.basename(rel_path)
if filename in EXCLUDED_FILES:
return True
# Check directory
dir_path = os.path.dirname(rel_path)
if any(excluded in dir_path for excluded in EXCLUDED_DIRS):
return True
# Check extension
_, ext = os.path.splitext(filename)
if ext.lower() in EXCLUDED_EXT:
return True
return False
def clean_path(path: str, workspace_path: str = "/workspace") -> str:
"""Clean and normalize a path to be relative to the workspace
Args:
path: The path to clean
workspace_path: The base workspace path to remove (default: "/workspace")
Returns:
The cleaned path, relative to the workspace
"""
# Remove any leading slash
path = path.lstrip("/")
# Remove workspace prefix if present
if path.startswith(workspace_path.lstrip("/")):
path = path[len(workspace_path.lstrip("/")) :]
# Remove workspace/ prefix if present
if path.startswith("workspace/"):
path = path[9:]
# Remove any remaining leading slash
path = path.lstrip("/")
return path
+32
View File
@@ -0,0 +1,32 @@
import logging
import os
import structlog
ENV_MODE = os.getenv("ENV_MODE", "LOCAL")
renderer = [structlog.processors.JSONRenderer()]
if ENV_MODE.lower() == "local".lower():
renderer = [structlog.dev.ConsoleRenderer()]
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.dict_tracebacks,
structlog.processors.CallsiteParameterAdder(
{
structlog.processors.CallsiteParameter.FILENAME,
structlog.processors.CallsiteParameter.FUNC_NAME,
structlog.processors.CallsiteParameter.LINENO,
}
),
structlog.processors.TimeStamper(fmt="iso"),
structlog.contextvars.merge_contextvars,
*renderer,
],
cache_logger_on_first_use=True,
)
logger: structlog.stdlib.BoundLogger = structlog.get_logger(level=logging.DEBUG)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 166 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 174 KiB

+114
View File
@@ -0,0 +1,114 @@
# Global LLM configuration
[llm]
model = "claude-3-7-sonnet-20250219" # The LLM model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
# [llm] # Amazon Bedrock
# api_type = "aws" # Required
# model = "us.anthropic.claude-3-7-sonnet-20250219-v1:0" # Bedrock supported modelID
# base_url = "bedrock-runtime.us-west-2.amazonaws.com" # Not used now
# max_tokens = 8192
# temperature = 1.0
# api_key = "bear" # Required but not used for Bedrock
# [llm] #AZURE OPENAI:
# api_type= 'azure'
# model = "YOUR_MODEL_NAME" #"gpt-4o-mini"
# base_url = "{YOUR_AZURE_ENDPOINT.rstrip('/')}/openai/deployments/{AZURE_DEPLOYMENT_ID}"
# api_key = "AZURE API KEY"
# max_tokens = 8096
# temperature = 0.0
# api_version="AZURE API VERSION" #"2024-08-01-preview"
# [llm] #OLLAMA:
# api_type = 'ollama'
# model = "llama3.2"
# base_url = "http://localhost:11434/v1"
# api_key = "ollama"
# max_tokens = 4096
# temperature = 0.0
# Optional configuration for specific LLM models
[llm.vision]
model = "claude-3-7-sonnet-20250219" # The vision model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL for vision model
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
# [llm.vision] #OLLAMA VISION:
# api_type = 'ollama'
# model = "llama3.2-vision"
# base_url = "http://localhost:11434/v1"
# api_key = "ollama"
# max_tokens = 4096
# temperature = 0.0
# Optional configuration for specific browser configuration
# [browser]
# Whether to run browser in headless mode (default: false)
#headless = false
# Disable browser security features (default: true)
#disable_security = true
# Extra arguments to pass to the browser
#extra_chromium_args = []
# Path to a Chrome instance to use to connect to your normal browser
# e.g. '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome'
#chrome_instance_path = ""
# Connect to a browser instance via WebSocket
#wss_url = ""
# Connect to a browser instance via CDP
#cdp_url = ""
# Optional configuration, Proxy settings for the browser
# [browser.proxy]
# server = "http://proxy-server:port"
# username = "proxy-username"
# password = "proxy-password"
# Optional configuration, Search settings.
# [search]
# Search engine for agent to use. Default is "Google", can be set to "Baidu" or "DuckDuckGo" or "Bing".
#engine = "Google"
# Fallback engine order. Default is ["DuckDuckGo", "Baidu", "Bing"] - will try in this order after primary engine fails.
#fallback_engines = ["DuckDuckGo", "Baidu", "Bing"]
# Seconds to wait before retrying all engines again when they all fail due to rate limits. Default is 60.
#retry_delay = 60
# Maximum number of times to retry all engines when all fail. Default is 3.
#max_retries = 3
# Language code for search results. Options: "en" (English), "zh" (Chinese), etc.
#lang = "en"
# Country code for search results. Options: "us" (United States), "cn" (China), etc.
#country = "us"
## Sandbox configuration
#[sandbox]
#use_sandbox = false
#image = "python:3.12-slim"
#work_dir = "/workspace"
#memory_limit = "1g" # 512m
#cpu_limit = 2.0
#timeout = 300
#network_enabled = true
# Daytona configuration
[daytona]
daytona_api_key = ""
#daytona_server_url = "https://app.daytona.io/api"
#daytona_target = "us" #Daytona is currently available in the following regions:United States (us)、Europe (eu)
#sandbox_image_name = "whitezxj/sandbox:0.1.0" #If you don't use this default image,sandbox tools may be useless
#sandbox_entrypoint = "/usr/bin/supervisord -n -c /etc/supervisor/conf.d/supervisord.conf" #If you change this entrypoint,server in sandbox may be useless
#VNC_password = #The password you set to log in sandbox by VNC,it will be 123456 if you don't set
# MCP (Model Context Protocol) configuration
[mcp]
server_reference = "app.mcp.server" # default server module reference
# Optional Runflow configuration
# Your can add additional agents into run-flow workflow to solve different-type tasks.
[runflow]
use_data_analysis_agent = false # The Data Analysi Agent to solve various data analysis tasks
+17
View File
@@ -0,0 +1,17 @@
# Global LLM configuration
[llm] #Jiekou.AI:
api_type = 'jiekou'
model = "claude-sonnet-4-5-20250929" # The LLM model to use
base_url = "https://api.jiekou.ai/openai" # API endpoint URL
api_key = "your Jiekou.AI api key" # Your API key
max_tokens = 64000 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
[llm.vision] #Jiekou.AI VISION:
api_type = 'jiekou'
model = "claude-sonnet-4-5-20250929" # The vision model to use
base_url = "https://api.jiekou.ai/openai" # API endpoint URL for vision model
api_key = "your Jiekou.AI api key" # Your API key for vision model
max_tokens = 64000 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
+8
View File
@@ -31,6 +31,14 @@ temperature = 0.0 # Controls randomness
# max_tokens = 4096
# temperature = 0.0
# [llm] #Jiekou.AI:
# api_type = 'jiekou'
# model = "claude-sonnet-4-5-20250929" # The LLM model to use
# base_url = "https://api.jiekou.ai/openai" # API endpoint URL
# api_key = "your Jiekou.AI api key" # Your API key
# max_tokens = 64000 # Maximum number of tokens in the response
# temperature = 0.0 # Controls randomness
# Optional configuration for specific LLM models
[llm.vision]
model = "claude-3-7-sonnet-20250219" # The vision model to use
-1
View File
@@ -192,4 +192,3 @@ Response:
## Learn More
- [A2A Protocol Documentation](https://google.github.io/A2A/#/documentation)
-1
View File
@@ -192,4 +192,3 @@ Response:
## Learn More
- [A2A Protocol Documentation](https://google.github.io/A2A/#/documentation)
+3 -3
View File
@@ -1,6 +1,7 @@
import httpx
from typing import Any, Dict, AsyncIterable, Literal, List, ClassVar
from typing import Any, AsyncIterable, ClassVar, Dict, List, Literal
from pydantic import BaseModel
from app.agent.manus import Manus
@@ -12,7 +13,6 @@ class ResponseFormat(BaseModel):
class A2AManus(Manus):
async def invoke(self, query, sessionId) -> str:
config = {"configurable": {"thread_id": sessionId}}
response = await self.run(query)
+6 -8
View File
@@ -1,8 +1,8 @@
import logging
from typing import Awaitable, Callable
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import Event, EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.server.events import EventQueue
from a2a.types import (
InvalidParamsError,
Part,
@@ -10,13 +10,11 @@ from a2a.types import (
TextPart,
UnsupportedOperationError,
)
from a2a.utils import (
completed_task,
new_artifact,
)
from .agent import A2AManus
from a2a.utils import completed_task, new_artifact
from a2a.utils.errors import ServerError
from typing import Callable, Awaitable
from .agent import A2AManus
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
+11 -14
View File
@@ -1,25 +1,22 @@
import httpx
import argparse
import asyncio
import logging
from typing import Optional
import httpx
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, InMemoryPushNotifier
from a2a.types import (
AgentCapabilities,
AgentCard,
AgentSkill,
)
from a2a.server.tasks import InMemoryPushNotifier, InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from dotenv import load_dotenv
from .agent_executor import ManusExecutor
from .agent import A2AManus
from app.tool.browser_use_tool import _BROWSER_DESCRIPTION
from app.tool.str_replace_editor import _STR_REPLACE_EDITOR_DESCRIPTION
from app.tool.terminate import _TERMINATE_DESCRIPTION
import logging
from dotenv import load_dotenv
import asyncio
from typing import Optional
from .agent import A2AManus
from .agent_executor import ManusExecutor
load_dotenv()
+1
View File
@@ -36,6 +36,7 @@ boto3~=1.37.18
requests~=2.32.3
beautifulsoup4~=4.13.3
crawl4ai~=0.6.3
huggingface-hub~=0.29.2
setuptools~=75.8.0
+36
View File
@@ -0,0 +1,36 @@
import argparse
import asyncio
from app.agent.sandbox_agent import SandboxManus
from app.logger import logger
async def main():
# Parse command line arguments
parser = argparse.ArgumentParser(description="Run Manus agent with a prompt")
parser.add_argument(
"--prompt", type=str, required=False, help="Input prompt for the agent"
)
args = parser.parse_args()
# Create and initialize Manus agent
agent = await SandboxManus.create()
try:
# Use command line prompt if provided, otherwise ask for input
prompt = args.prompt if args.prompt else input("Enter your prompt: ")
if not prompt.strip():
logger.warning("Empty prompt provided.")
return
logger.warning("Processing your request...")
await agent.run(prompt)
logger.info("Request processing completed.")
except KeyboardInterrupt:
logger.warning("Operation interrupted.")
finally:
# Ensure agent resources are cleaned up before exiting
await agent.cleanup()
if __name__ == "__main__":
asyncio.run(main())