Resource Backend Development
About 703 wordsAbout 2 min
2025-02-04
This article introduces how to develop heavyweight Backend services.
Features
- ✅ Inherits
Backendbase class - ✅ Uses
@tooldecorator to mark tool methods - ✅ Optionally implements lifecycle methods
- ✅ Supports Session management
Backend Types
1. Shared Resource Backend (e.g., RAG)
Only implements warmup() and shutdown(), no Session needed:
from sandbox.server.backends.base import Backend
from sandbox.server.core.decorators import tool
class RAGBackend(Backend):
name = "rag"
description = "RAG Backend"
version = "1.0.0"
def __init__(self, config=None):
super().__init__(config)
self._model = None
async def warmup(self) -> None:
"""Warmup: Load Embedding model"""
self._model = load_model()
async def shutdown(self) -> None:
"""Shutdown: Release model"""
self._model = None
@tool("rag:search")
async def search(self, query: str, top_k: int = 10) -> Dict:
results = self._model.search(query, top_k)
return {"code": 0, "data": {"results": results}}2. Session Resource Backend (e.g., VM)
Only implements initialize() and cleanup(), requires Session:
class VMBackend(Backend):
name = "vm"
description = "VM Backend"
version = "1.0.0"
async def initialize(self, worker_id: str, config: Dict) -> Dict:
"""Create Session - Allocate VM instance"""
screen_size = config.get("screen_size", [1920, 1080])
vm = create_vm(screen_size)
return {"vm": vm, "screen_size": screen_size}
async def cleanup(self, worker_id: str, session_info: Dict) -> None:
"""Destroy Session - Release VM"""
vm = session_info.get("data", {}).get("vm")
if vm:
vm.close()
@tool("vm:screenshot")
async def screenshot(self, session_info: Dict) -> Dict:
vm = session_info["data"]["vm"]
image = vm.screenshot()
return {"code": 0, "data": {"image": image}}
@tool("vm:click")
async def click(self, x: int, y: int, session_info: Dict) -> Dict:
vm = session_info["data"]["vm"]
vm.click(x, y)
return {"code": 0, "data": {"clicked": [x, y]}}3. Hybrid Backend (e.g., Browser)
Implements all lifecycle methods:
class BrowserBackend(Backend):
name = "browser"
description = "Browser Backend"
version = "1.0.0"
async def warmup(self) -> None:
"""Warmup: Launch browser process"""
self._browser = await launch_browser()
async def shutdown(self) -> None:
"""Shutdown: Stop browser process"""
await self._browser.close()
async def initialize(self, worker_id: str, config: Dict) -> Dict:
"""Create Session - Open new page"""
page = await self._browser.new_page()
return {"page": page}
async def cleanup(self, worker_id: str, session_info: Dict) -> None:
"""Destroy Session - Close page"""
page = session_info.get("data", {}).get("page")
if page:
await page.close()
@tool("browser:goto")
async def goto(self, url: str, session_info: Dict) -> Dict:
page = session_info["data"]["page"]
await page.goto(url)
return {"code": 0, "data": {"url": url}}Development Steps
1. Create Backend File
# sandbox/server/backends/resources/my_backend.py
from typing import Dict, Any, Optional
from ..base import Backend, BackendConfig
from ...core.decorators import tool
class MyBackend(Backend):
name = "my"
description = "My Custom Backend"
version = "1.0.0"
def __init__(self, config: Optional[BackendConfig] = None):
super().__init__(config)
# Initialization
# Implement the lifecycle methods you need...
@tool("my:action")
async def action(self, param: str, session_info: Dict = None) -> Dict:
return {"code": 0, "data": {"result": "..."}}2. Add Configuration
{
"resources": {
"my": {
"enabled": true,
"backend_class": "sandbox.server.backends.resources.my_backend.MyBackend",
"config": {
"option1": "value1"
}
}
}
}3. Use the Backend
async with Sandbox() as sandbox:
# If Session is needed
await sandbox.create_session("my", {"option": "value"})
# Execute tool
result = await sandbox.execute("my:action", {"param": "value"})
# Destroy Session
await sandbox.destroy_session("my")@tool Decorator
@tool(name: str, resource_type: str = None)| Parameter | Description |
|---|---|
name | Tool name (e.g., "vm:screenshot") |
resource_type | Resource type (optional, automatically parsed from name) |
session_info Parameter
If the Backend implements initialize(), tool functions can receive the session_info parameter:
@tool("my:action")
async def action(self, param: str, session_info: Dict) -> Dict:
# session_info structure
# {
# "session_id": "xxx",
# "worker_id": "sandbox_xxx",
# "resource_type": "my",
# "data": {...} # Data returned by initialize()
# }
data = session_info["data"]
resource = data["resource"]
# Use the resource...Lifecycle Sequence
sandbox.start()
└── backend.warmup() # If implemented
sandbox.create_session("my", config)
└── backend.initialize(worker_id, config)
sandbox.execute("my:action", params)
└── backend.action(**params, session_info=...)
sandbox.destroy_session("my")
└── backend.cleanup(worker_id, session_info)
sandbox.shutdown_server()
└── backend.shutdown() # If implementedBest Practices
1. Error Handling
@tool("my:action")
async def action(self, param: str, session_info: Dict) -> Dict:
try:
result = do_something(param)
return {"code": 0, "data": result}
except ValueError as e:
return {"code": 1002, "message": str(e)}
except TimeoutError:
return {"code": 2002, "message": "Timeout"}
except Exception as e:
return {"code": 5000, "message": str(e)}2. Resource Cleanup
async def cleanup(self, worker_id: str, session_info: Dict) -> None:
try:
resource = session_info.get("data", {}).get("resource")
if resource:
resource.close()
except Exception as e:
logger.warning(f"Cleanup error: {e}")3. Logging
import logging
logger = logging.getLogger("MyBackend")
class MyBackend(Backend):
async def initialize(self, worker_id: str, config: Dict) -> Dict:
logger.info(f"📦 [{worker_id}] Creating session...")
# ...
logger.info(f"✅ [{worker_id}] Session created")