Spaces:
Sleeping
Sleeping
| """Direct REST API endpoints for each tool (non-MCP access).""" | |
| from __future__ import annotations | |
| import logging | |
| from fastapi import APIRouter | |
| from app.environment import configure_environment, get_environment_config | |
| from app.executor import get_executor | |
| from app.file_manager import FileManager | |
| from app.models import ( | |
| AppendToCodeFileRequest, | |
| CheckInstalledPackagesRequest, | |
| ConfigureEnvironmentRequest, | |
| ExecuteCodeRequest, | |
| ExecutionResult, | |
| FilePathRequest, | |
| FileResult, | |
| InitializeCodeFileRequest, | |
| InstallDependenciesRequest, | |
| ) | |
| from app.package_manager import get_package_manager | |
| logger = logging.getLogger(__name__) | |
| router = APIRouter(prefix="/tools", tags=["tools"]) | |
| async def execute_code(request: ExecuteCodeRequest) -> ExecutionResult: | |
| """Execute Python code directly.""" | |
| executor = get_executor() | |
| return await executor.execute_code(request.code, request.filename) | |
| async def install_dependencies(request: InstallDependenciesRequest): | |
| """Install Python packages.""" | |
| pm = get_package_manager() | |
| return await pm.install_packages(request.packages) | |
| async def check_installed_packages(request: CheckInstalledPackagesRequest): | |
| """Check if packages are installed.""" | |
| pm = get_package_manager() | |
| results = await pm.check_packages(request.packages) | |
| return [r.model_dump() for r in results] | |
| async def configure_env(request: ConfigureEnvironmentRequest): | |
| """Reconfigure the execution environment.""" | |
| config = configure_environment( | |
| env_type=request.type, | |
| conda_name=request.conda_name, | |
| venv_path=request.venv_path, | |
| uv_venv_path=request.uv_venv_path, | |
| ) | |
| return config.model_dump() | |
| async def get_env_config(): | |
| """Get current environment configuration.""" | |
| return get_environment_config().model_dump() | |
| async def initialize_code_file(request: InitializeCodeFileRequest) -> FileResult: | |
| """Create a new Python file with initial content.""" | |
| fm = FileManager() | |
| return fm.create_file(request.content, request.filename) | |
| async def append_to_code_file(request: AppendToCodeFileRequest) -> FileResult: | |
| """Append content to an existing code file.""" | |
| fm = FileManager() | |
| return fm.append_to_file(request.file_path, request.content) | |
| async def execute_code_file(request: FilePathRequest) -> ExecutionResult: | |
| """Execute an existing Python file.""" | |
| executor = get_executor() | |
| return await executor.execute_file(request.file_path) | |
| async def read_code_file(request: FilePathRequest) -> FileResult: | |
| """Read content of an existing code file.""" | |
| fm = FileManager() | |
| return fm.read_file(request.file_path) |