mirrors/foundationagents--openmanus
1
0
Fork 0
Code Issues Pull Requests Actions 4 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

remove sb_*.py to tool/sandbox&modify requirements.txt

Browse Source
This commit is contained in:
GhostC
2025-07-30 18:59:33 +08:00
parent 802e212ef2
commit 9cb3bd16c4
15 changed files with 10 additions and 775 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/agent/browser.py
+1 -1
View File
@@ -8,7 +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.sb_browser_tool import SandboxBrowserTool
from app.tool.sandbox.sb_browser_tool import SandboxBrowserTool
# Avoid circular import if BrowserAgent needs BrowserContextHelper
app/agent/sandbox_agent.py
+4 -4
View File
@@ -13,10 +13,10 @@ from app.tool import Terminate, ToolCollection
from app.tool.ask_human import AskHuman
from app.tool.browser_use_tool import BrowserUseTool
from app.tool.mcp import MCPClients, MCPClientTool
from app.tool.sb_browser_tool import SandboxBrowserTool
from app.tool.sb_files_tool import SandboxFilesTool
from app.tool.sb_shell_tool import SandboxShellTool
from app.tool.sb_vision_tool import SandboxVisionTool
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):
app/daytona/api.py
-468
View File
@@ -1,468 +0,0 @@
import os
import urllib.parse
from typing import Optional
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import Response
from pydantic import BaseModel
from sandbox.sandbox import delete_sandbox, get_or_start_sandbox
from services.supabase import DBConnection
from utils.auth_utils import get_optional_user_id
from utils.logger import logger
# Initialize shared resources
router = APIRouter(tags=["sandbox"])
db = None
def initialize(_db: DBConnection):
"""Initialize the sandbox API with resources from the main API."""
global db
db = _db
logger.info("Initialized sandbox API with database connection")
class FileInfo(BaseModel):
"""Model for file information"""
name: str
path: str
is_dir: bool
size: int
mod_time: str
permissions: Optional[str] = None
def normalize_path(path: str) -> str:
"""
Normalize a path to ensure proper UTF-8 encoding and handling.
Args:
path: The file path, potentially containing URL-encoded characters
Returns:
Normalized path with proper UTF-8 encoding
"""
try:
# First, ensure the path is properly URL-decoded
decoded_path = urllib.parse.unquote(path)
# Handle Unicode escape sequences like \u0308
try:
# Replace Python-style Unicode escapes (\u0308) with actual characters
# This handles cases where the Unicode escape sequence is part of the URL
import re
unicode_pattern = re.compile(r"\\u([0-9a-fA-F]{4})")
def replace_unicode(match):
hex_val = match.group(1)
return chr(int(hex_val, 16))
decoded_path = unicode_pattern.sub(replace_unicode, decoded_path)
except Exception as unicode_err:
logger.warning(
f"Error processing Unicode escapes in path '{path}': {str(unicode_err)}"
)
logger.debug(f"Normalized path from '{path}' to '{decoded_path}'")
return decoded_path
except Exception as e:
logger.error(f"Error normalizing path '{path}': {str(e)}")
return path # Return original path if decoding fails
async def verify_sandbox_access(client, sandbox_id: str, user_id: Optional[str] = None):
"""
Verify that a user has access to a specific sandbox based on account membership.
Args:
client: The Supabase client
sandbox_id: The sandbox ID to check access for
user_id: The user ID to check permissions for. Can be None for public resource access.
Returns:
dict: Project data containing sandbox information
Raises:
HTTPException: If the user doesn't have access to the sandbox or sandbox doesn't exist
"""
# Find the project that owns this sandbox
project_result = (
await client.table("projects")
.select("*")
.filter("sandbox->>id", "eq", sandbox_id)
.execute()
)
if not project_result.data or len(project_result.data) == 0:
raise HTTPException(status_code=404, detail="Sandbox not found")
project_data = project_result.data[0]
if project_data.get("is_public"):
return project_data
# For private projects, we must have a user_id
if not user_id:
raise HTTPException(
status_code=401, detail="Authentication required for this resource"
)
account_id = project_data.get("account_id")
# Verify account membership
if account_id:
account_user_result = (
await client.schema("basejump")
.from_("account_user")
.select("account_role")
.eq("user_id", user_id)
.eq("account_id", account_id)
.execute()
)
if account_user_result.data and len(account_user_result.data) > 0:
return project_data
raise HTTPException(status_code=403, detail="Not authorized to access this sandbox")
async def get_sandbox_by_id_safely(client, sandbox_id: str):
"""
Safely retrieve a sandbox object by its ID, using the project that owns it.
Args:
client: The Supabase client
sandbox_id: The sandbox ID to retrieve
Returns:
Sandbox: The sandbox object
Raises:
HTTPException: If the sandbox doesn't exist or can't be retrieved
"""
# Find the project that owns this sandbox
project_result = (
await client.table("projects")
.select("project_id")
.filter("sandbox->>id", "eq", sandbox_id)
.execute()
)
if not project_result.data or len(project_result.data) == 0:
logger.error(f"No project found for sandbox ID: {sandbox_id}")
raise HTTPException(
status_code=404,
detail="Sandbox not found - no project owns this sandbox ID",
)
# project_id = project_result.data[0]['project_id']
# logger.debug(f"Found project {project_id} for sandbox {sandbox_id}")
try:
# Get the sandbox
sandbox = await get_or_start_sandbox(sandbox_id)
# Extract just the sandbox object from the tuple (sandbox, sandbox_id, sandbox_pass)
# sandbox = sandbox_tuple[0]
return sandbox
except Exception as e:
logger.error(f"Error retrieving sandbox {sandbox_id}: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Failed to retrieve sandbox: {str(e)}"
)
@router.post("/sandboxes/{sandbox_id}/files")
async def create_file(
sandbox_id: str,
path: str = Form(...),
file: UploadFile = File(...),
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""Create a file in the sandbox using direct file upload"""
# Normalize the path to handle UTF-8 encoding correctly
path = normalize_path(path)
logger.info(
f"Received file upload request for sandbox {sandbox_id}, path: {path}, user_id: {user_id}"
)
client = await db.client
# Verify the user has access to this sandbox
await verify_sandbox_access(client, sandbox_id, user_id)
try:
# Get sandbox using the safer method
sandbox = await get_sandbox_by_id_safely(client, sandbox_id)
# Read file content directly from the uploaded file
content = await file.read()
# Create file using raw binary content
sandbox.fs.upload_file(content, path)
logger.info(f"File created at {path} in sandbox {sandbox_id}")
return {"status": "success", "created": True, "path": path}
except Exception as e:
logger.error(f"Error creating file in sandbox {sandbox_id}: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/sandboxes/{sandbox_id}/files")
async def list_files(
sandbox_id: str,
path: str,
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""List files and directories at the specified path"""
# Normalize the path to handle UTF-8 encoding correctly
path = normalize_path(path)
logger.info(
f"Received list files request for sandbox {sandbox_id}, path: {path}, user_id: {user_id}"
)
client = await db.client
# Verify the user has access to this sandbox
await verify_sandbox_access(client, sandbox_id, user_id)
try:
# Get sandbox using the safer method
sandbox = await get_sandbox_by_id_safely(client, sandbox_id)
# List files
files = sandbox.fs.list_files(path)
result = []
for file in files:
# Convert file information to our model
# Ensure forward slashes are used for paths, regardless of OS
full_path = (
f"{path.rstrip('/')}/{file.name}" if path != "/" else f"/{file.name}"
)
file_info = FileInfo(
name=file.name,
path=full_path, # Use the constructed path
is_dir=file.is_dir,
size=file.size,
mod_time=str(file.mod_time),
permissions=getattr(file, "permissions", None),
)
result.append(file_info)
logger.info(f"Successfully listed {len(result)} files in sandbox {sandbox_id}")
return {"files": [file.dict() for file in result]}
except Exception as e:
logger.error(f"Error listing files in sandbox {sandbox_id}: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/sandboxes/{sandbox_id}/files/content")
async def read_file(
sandbox_id: str,
path: str,
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""Read a file from the sandbox"""
# Normalize the path to handle UTF-8 encoding correctly
original_path = path
path = normalize_path(path)
logger.info(
f"Received file read request for sandbox {sandbox_id}, path: {path}, user_id: {user_id}"
)
if original_path != path:
logger.info(f"Normalized path from '{original_path}' to '{path}'")
client = await db.client
# Verify the user has access to this sandbox
await verify_sandbox_access(client, sandbox_id, user_id)
try:
# Get sandbox using the safer method
sandbox = await get_sandbox_by_id_safely(client, sandbox_id)
# Read file directly - don't check existence first with a separate call
try:
content = sandbox.fs.download_file(path)
except Exception as download_err:
logger.error(
f"Error downloading file {path} from sandbox {sandbox_id}: {str(download_err)}"
)
raise HTTPException(
status_code=404, detail=f"Failed to download file: {str(download_err)}"
)
# Return a Response object with the content directly
filename = os.path.basename(path)
logger.info(f"Successfully read file {filename} from sandbox {sandbox_id}")
# Ensure proper encoding by explicitly using UTF-8 for the filename in Content-Disposition header
# This applies RFC 5987 encoding for the filename to support non-ASCII characters
encoded_filename = filename.encode("utf-8").decode("latin-1")
content_disposition = f"attachment; filename*=UTF-8''{encoded_filename}"
return Response(
content=content,
media_type="application/octet-stream",
headers={"Content-Disposition": content_disposition},
)
except HTTPException:
# Re-raise HTTP exceptions without wrapping
raise
except Exception as e:
logger.error(f"Error reading file in sandbox {sandbox_id}: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/sandboxes/{sandbox_id}/files")
async def delete_file(
sandbox_id: str,
path: str,
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""Delete a file from the sandbox"""
# Normalize the path to handle UTF-8 encoding correctly
path = normalize_path(path)
logger.info(
f"Received file delete request for sandbox {sandbox_id}, path: {path}, user_id: {user_id}"
)
client = await db.client
# Verify the user has access to this sandbox
await verify_sandbox_access(client, sandbox_id, user_id)
try:
# Get sandbox using the safer method
sandbox = await get_sandbox_by_id_safely(client, sandbox_id)
# Delete file
sandbox.fs.delete_file(path)
logger.info(f"File deleted at {path} in sandbox {sandbox_id}")
return {"status": "success", "deleted": True, "path": path}
except Exception as e:
logger.error(f"Error deleting file in sandbox {sandbox_id}: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/sandboxes/{sandbox_id}")
async def delete_sandbox_route(
sandbox_id: str,
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""Delete an entire sandbox"""
logger.info(
f"Received sandbox delete request for sandbox {sandbox_id}, user_id: {user_id}"
)
client = await db.client
# Verify the user has access to this sandbox
await verify_sandbox_access(client, sandbox_id, user_id)
try:
# Delete the sandbox using the sandbox module function
await delete_sandbox(sandbox_id)
return {"status": "success", "deleted": True, "sandbox_id": sandbox_id}
except Exception as e:
logger.error(f"Error deleting sandbox {sandbox_id}: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
# Should happen on server-side fully
@router.post("/project/{project_id}/sandbox/ensure-active")
async def ensure_project_sandbox_active(
project_id: str,
request: Request = None,
user_id: Optional[str] = Depends(get_optional_user_id),
):
"""
Ensure that a project's sandbox is active and running.
Checks the sandbox status and starts it if it's not running.
"""
logger.info(
f"Received ensure sandbox active request for project {project_id}, user_id: {user_id}"
)
client = await db.client
# Find the project and sandbox information
project_result = (
await client.table("projects")
.select("*")
.eq("project_id", project_id)
.execute()
)
if not project_result.data or len(project_result.data) == 0:
logger.error(f"Project not found: {project_id}")
raise HTTPException(status_code=404, detail="Project not found")
project_data = project_result.data[0]
# For public projects, no authentication is needed
if not project_data.get("is_public"):
# For private projects, we must have a user_id
if not user_id:
logger.error(f"Authentication required for private project {project_id}")
raise HTTPException(
status_code=401, detail="Authentication required for this resource"
)
account_id = project_data.get("account_id")
# Verify account membership
if account_id:
account_user_result = (
await client.schema("basejump")
.from_("account_user")
.select("account_role")
.eq("user_id", user_id)
.eq("account_id", account_id)
.execute()
)
if not (account_user_result.data and len(account_user_result.data) > 0):
logger.error(
f"User {user_id} not authorized to access project {project_id}"
)
raise HTTPException(
status_code=403, detail="Not authorized to access this project"
)
try:
# Get sandbox ID from project data
sandbox_info = project_data.get("sandbox", {})
if not sandbox_info.get("id"):
raise HTTPException(
status_code=404, detail="No sandbox found for this project"
)
sandbox_id = sandbox_info["id"]
# Get or start the sandbox
logger.info(f"Ensuring sandbox is active for project {project_id}")
await get_or_start_sandbox(sandbox_id)
logger.info(
f"Successfully ensured sandbox {sandbox_id} is active for project {project_id}"
)
return {
"status": "success",
"sandbox_id": sandbox_id,
"message": "Sandbox is active",
}
except Exception as e:
logger.error(
f"Error ensuring sandbox is active for project {project_id}: {str(e)}"
)
raise HTTPException(status_code=500, detail=str(e))
app/daytona/sandbox.py
+2 -2
View File
@@ -135,12 +135,12 @@ def create_sandbox(password: str, project_id: str = None):
# Create the sandbox
sandbox = daytona.create(params)
logger.debug(f"Sandbox created with ID: {sandbox.id}")
logger.info(f"Sandbox created with ID: {sandbox.id}")
# Start supervisord in a session for new sandbox
start_supervisord_session(sandbox)
logger.debug(f"Sandbox environment successfully initialized")
logger.info(f"Sandbox environment successfully initialized")
return sandbox
app/tool/sb_browser_tool.py → app/tool/sandbox/sb_browser_tool.py
View File
app/tool/sb_files_tool.py → app/tool/sandbox/sb_files_tool.py
View File
app/tool/sb_shell_tool.py → app/tool/sandbox/sb_shell_tool.py
View File
app/tool/sb_vision_tool.py → app/tool/sandbox/sb_vision_tool.py
View File
requirements.txt
+3
View File
@@ -40,3 +40,6 @@ crawl4ai~=0.6.3
huggingface-hub~=0.29.2
setuptools~=75.8.0
daytona==0.21.8
structlog==25.4.0
tests/daytona/test_computer_use_tool.py
-25
View File
@@ -1,25 +0,0 @@
import asyncio
from app.daytona.sandbox import create_sandbox
from app.tool.computer_use_tool import ComputerUseTool
async def main():
# 创建沙箱和工具
sandbox = create_sandbox(password="123456")
base_url = sandbox.get_preview_link(8000)
print(f"Sandbox base URL: {base_url}")
print(f"Sandbox ID: {sandbox.id}")
computer_tool = ComputerUseTool.create_with_sandbox(sandbox)
# 执行截图操作
result = await computer_tool.execute(action="screenshot")
print(result)
# 清理资源（可选）
await computer_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main()) # 运行异步主函数
tests/daytona/test_daytona.py
-79
View File
@@ -1,79 +0,0 @@
from daytona import CreateSandboxFromImageParams, Daytona, DaytonaConfig, Resources
# Using environment variables (DAYTONA_API_KEY, DAYTONA_API_URL, DAYTONA_TARGET)
# daytona = Daytona()
# Using explicit configuration
config = DaytonaConfig(api_key="", api_url="https://app.daytona.io/api", target="us")
daytona = Daytona(config)
params = CreateSandboxFromImageParams(
image="",
# image=Image.debian_slim("3.12"),
public=True,
labels=None,
env_vars={
"CHROME_PERSISTENT_SESSION": "true",
"RESOLUTION": "1024x768x24",
"RESOLUTION_WIDTH": "1024",
"RESOLUTION_HEIGHT": "768",
"VNC_PASSWORD": "123456",
"ANONYMIZED_TELEMETRY": "false",
"CHROME_PATH": "",
"CHROME_USER_DATA": "",
"CHROME_DEBUGGING_PORT": "9222",
"CHROME_DEBUGGING_HOST": "localhost",
"CHROME_CDP": "",
},
resources=Resources(
cpu=1,
memory=1,
disk=1,
),
auto_stop_interval=15,
auto_archive_interval=24 * 60,
)
# Create the sandbox
sandbox = daytona.create(params)
print(f"Sandbox created with ID: {sandbox.id}")
# from daytona import Daytona
# def main():
# # Initialize the SDK (uses environment variables by default)
# daytona = Daytona(config)
# # Create a new sandbox
# sandbox = daytona.create()
# # Execute a command
# response = sandbox.process.exec("echo 'Hello, World!'")
# print(response.result)
# if __name__ == "__main__":
# main()
# from daytona import Daytona, DaytonaConfig
# Define the configuration
# config = DaytonaConfig(api_key="")
# # Initialize the Daytona client
# daytona = Daytona(config)
# # Create the Sandbox instance
# sandbox = daytona.create()
# # Run the code securely inside the Sandbox
# response = sandbox.process.code_run('print("Hello World from code!")')
# if response.exit_code != 0:
# print(f"Error: {response.exit_code} {response.result}")
# else:
# print(response.result)
# Clean up
# sandbox.delete()
tests/daytona/test_sb_browser_use.py
-82
View File
@@ -1,82 +0,0 @@
import asyncio
from app.daytona.sandbox import create_sandbox
from app.tool.sb_browser_tool import SandboxBrowserTool
async def main():
# 创建沙箱和工具
sandbox = create_sandbox(password="123456")
# config = DaytonaConfig(
# api_key="",
# api_url="https://app.daytona.io/api",
# target="us"
# )
# daytona = Daytona(config)
# sandbox = daytona.find_one("201415a9-28ad-4b6d-8756-13b1e34a70c3")
# if sandbox.state == "archived" or sandbox.state == "stopped":
# logger.info(f"Sandbox is in {sandbox.state} state. Starting...")
# try:
# daytona.start(sandbox)
# start_supervisord_session(sandbox)
# # Wait a moment for the sandbox to initialize
# # sleep(5)
# # Refresh sandbox state after starting
# # sandbox = daytona.get(sandbox.id)
# except Exception as e:
# logger.error(f"Error starting sandbox: {e}")
# raise e
# sandbox.start()
vnc_link = sandbox.get_preview_link(6080)
website_link = sandbox.get_preview_link(8080)
computer_link = sandbox.get_preview_link(8000)
print(f"VNC Link: {vnc_link}")
print(f"Website Link: {website_link}")
print(f"Computer Link: {computer_link}")
# sandbox.start()
# base_url=sandbox.get_preview_link(8000)
# print(f"Sandbox base URL: {base_url}")
# print(f"Sandbox ID: {sandbox.id}")
tool = SandboxBrowserTool(sandbox)
# # 执行截图操作
result, tooresult = await tool.execute(
action="navigate_to", url="https://www.github.com"
)
print(result)
# endpoint = "navigate_to"
# method = "POST"
# params = {
# "url": "https://www.google.com"
# }
# url = f"http://localhost:8003/api/automation/{endpoint}"
# curl_cmd = f"curl -s -X {method} '{url}' -H 'Content-Type: application/json'"
# json_data = json.dumps(params)
# curl_cmd += f" -d '{json_data}'"
# logger.debug(f"Executing curl command: {curl_cmd}")
# response = sandbox.process.exec(curl_cmd, timeout=30)
# print(f"Response: {response}")
# response = sandbox.process.exec("ls -la")
# print(response.result)
# 清理资源（可选）
# await computer_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main()) # 运行异步主函数
# await computer_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main()) # 运行异步主函数
# await computer_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main()) # 运行异步主函数
# await computer_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main()) # 运行异步主函数
tests/daytona/test_sb_files_tool.py
-43
View File
@@ -1,43 +0,0 @@
import asyncio
from app.daytona.sandbox import create_sandbox
from app.tool.sb_files_tool import SandboxFilesTool
async def main():
# 创建沙箱和工具
sandbox = create_sandbox(password="123456")
# base_url=sandbox.get_preview_link(8000)
# print(f"Sandbox base URL: {base_url}")
print(f"Sandbox ID: {sandbox.id}")
vnc_link = sandbox.get_preview_link(6080)
website_link = sandbox.get_preview_link(8080)
print(f"VNC Link: {vnc_link}")
print(f"Website Link: {website_link}")
sb_files_tool = SandboxFilesTool(sandbox)
# 执行创建文件操作
result = await sb_files_tool.execute(
action="create_file", file_path="src/a.txt", file_contents="aaaaa1111"
)
print(result)
# 执行字符串替换操作
# result = await sb_files_tool.execute(action="str_replace", file_path="src/a.txt", old_str="aaaaa", new_str="bbbbb")
# print(result)
# 执行全文件重写操作
result = await sb_files_tool.execute(
action="full_file_rewrite", file_path="src/a.txt", file_contents="1234567"
)
print(result)
# 执行删除文件操作
# result = await sb_files_tool.execute(action="delete_file", file_path="src/a.txt")
# print(result)
# 清理资源（可选）
# await sb_shell_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main())
tests/daytona/test_sb_shell_tools.py
-31
View File
@@ -1,31 +0,0 @@
import asyncio
from app.daytona.sandbox import create_sandbox
from app.tool.sb_shell_tool import SandboxShellTool
async def main():
# 创建沙箱和工具
sandbox = create_sandbox(password="123456")
# base_url=sandbox.get_preview_link(8000)
# print(f"Sandbox base URL: {base_url}")
print(f"Sandbox ID: {sandbox.id}")
vnc_link = sandbox.get_preview_link(6080)
website_link = sandbox.get_preview_link(8080)
print(f"VNC Link: {vnc_link}")
print(f"Website Link: {website_link}")
sb_shell_tool = SandboxShellTool(sandbox)
# 执行截图操作
result = await sb_shell_tool.execute(action="execute_command", command="pwd")
print(result)
# 清理资源（可选）
# await sb_shell_tool.cleanup()
if __name__ == "__main__":
print("123")
asyncio.run(main())
tests/daytona/test_sb_vision_tool.py
-40
View File
@@ -1,40 +0,0 @@
import asyncio
from app.daytona.sandbox import create_sandbox
from app.tool.sb_shell_tool import SandboxShellTool
from app.tool.sb_vision_tool import SandboxVisionTool
async def main():
# 创建沙箱和工具
sandbox = create_sandbox(password="123456")
# base_url=sandbox.get_preview_link(8000)
# print(f"Sandbox base URL: {base_url}")
print(f"Sandbox ID: {sandbox.id}")
vnc_link = sandbox.get_preview_link(6080)
website_link = sandbox.get_preview_link(8080)
print(f"VNC Link: {vnc_link}")
print(f"Website Link: {website_link}")
sb_shell_tool = SandboxShellTool(sandbox)
sb_vision_tool = SandboxVisionTool(sandbox)
# 执行终端命令
result = await sb_shell_tool.execute(
action="execute_command",
command="curl -O http://img.netbian.com/file/2025/0716/091412RIFD9.jpg",
)
print(result)
# 执行see_image操作
result = await sb_vision_tool.execute(
action="see_image", file_path="091412RIFD9.jpg"
)
print(result)
# 清理资源（可选）
# await sb_shell_tool.cleanup()
if __name__ == "__main__":
asyncio.run(main())