fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: code formatting and type improvements

Browse Source 
- Format bwrap.py with ruff, clean up imports
- Remove unused cast import in tool.py
- Add getattr fallbacks in context.py for handler name resolution
- Fix param_type annotation to allow Any in command.py
This commit is contained in:
LIghtJUNction
2026-03-23 02:29:33 +08:00
parent 371ff24de1
commit 04e9bf8ca8
5 changed files with 94 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
astrbot/cli/utils/dashboard.py
View File

@@ -1,5 +1,5 @@
from importlib import resources
import os
from importlib import resources
from pathlib import Path
import click
@@ -26,7 +26,9 @@ class DashboardManager:
click.echo("Dashboard is not installed")
# Skip interactive prompt when running under systemd
if os.environ.get("ASTRBOT_SYSTEMD") == "1":
click.echo("Skipping interactive dashboard installation in systemd mode.")
click.echo(
"Skipping interactive dashboard installation in systemd mode."
)
return
if click.confirm(
"Install dashboard?",

2
astrbot/core/agent/tool.py
View File

@@ -1,6 +1,6 @@
import copy
from collections.abc import AsyncGenerator, Awaitable, Callable
from typing import Any, Generic, cast
from typing import Any, Generic
import jsonschema
import mcp

126
astrbot/core/computer/booters/bwrap.py
View File

@@ -3,15 +3,13 @@ from __future__ import annotations
import asyncio
import locale
import os
import shlex
import shutil
import subprocess
import shlex
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from astrbot.api import logger
from astrbot.core.utils.astrbot_path import (
get_astrbot_temp_path,
)
@@ -19,6 +17,7 @@ from astrbot.core.utils.astrbot_path import (
from ..olayer import FileSystemComponent, PythonComponent, ShellComponent
from .base import ComputerBooter
def _decode_shell_output(output: bytes | None) -> str:
if output is None:
return ""
@@ -36,6 +35,7 @@ def _decode_shell_output(output: bytes | None) -> str:
return output.decode("utf-8", errors="replace")
@dataclass
class BwrapConfig:
workspace_dir: str
@@ -50,10 +50,11 @@ class BwrapConfig:
if p not in self.ro_binds:
self.ro_binds.append(p)
def build_bwrap_cmd(config: BwrapConfig, script_cmd: list[str]) -> list[str]:
"""Helper to build a bubblewrap command."""
cmd = ["bwrap"]
if not config.share_net:
cmd.append("--unshare-net")
@@ -61,7 +62,7 @@ def build_bwrap_cmd(config: BwrapConfig, script_cmd: list[str]) -> list[str]:
for path in config.ro_binds:
if os.path.exists(path):
cmd.extend(["--ro-bind", path, path])
for path in config.rw_binds:
# Avoid bind mounting dangerous host paths
if path == "/" or path.startswith("/root"):
@@ -70,26 +71,35 @@ def build_bwrap_cmd(config: BwrapConfig, script_cmd: list[str]) -> list[str]:
cmd.extend(["--bind", path, path])
# Make system binds the last to avoid issues about ro `/`
cmd.extend([
"--unshare-pid",
"--unshare-ipc",
"--unshare-uts",
"--die-with-parent",
"--dir", "/tmp",
"--dir", "/var/tmp",
"--proc", "/proc",
"--dev", "/dev",
"--bind", config.workspace_dir, config.workspace_dir,
])
cmd.extend(
[
"--unshare-pid",
"--unshare-ipc",
"--unshare-uts",
"--die-with-parent",
"--dir",
"/tmp",
"--dir",
"/var/tmp",
"--proc",
"/proc",
"--dev",
"/dev",
"--bind",
config.workspace_dir,
config.workspace_dir,
]
)
cmd.extend(["--"])
cmd.extend(script_cmd)
return cmd
@dataclass
class BwrapShellComponent(ShellComponent):
config: BwrapConfig
async def exec(
self,
command: str,
@@ -99,19 +109,18 @@ class BwrapShellComponent(ShellComponent):
shell: bool = True,
background: bool = False,
) -> dict[str, Any]:
def _run() -> dict[str, Any]:
run_env = os.environ.copy()
if env:
run_env.update({str(k): str(v) for k, v in env.items()})
working_dir = cwd if cwd else self.config.workspace_dir
# Use /bin/sh -c to run the evaluated command
# The command must be run inside bwrap
script_cmd = ["/bin/sh", "-c", command] if shell else shlex.split(command)
bwrap_cmd = build_bwrap_cmd(self.config, script_cmd)
if background:
proc = subprocess.Popen(
bwrap_cmd,
@@ -121,7 +130,7 @@ class BwrapShellComponent(ShellComponent):
stderr=subprocess.DEVNULL,
)
return {"pid": proc.pid, "stdout": "", "stderr": "", "exit_code": None}
result = subprocess.run(
bwrap_cmd,
cwd=working_dir,
@@ -137,10 +146,11 @@ class BwrapShellComponent(ShellComponent):
return await asyncio.to_thread(_run)
@dataclass
class BwrapPythonComponent(PythonComponent):
config: BwrapConfig
async def exec(
self,
code: str,
@@ -149,7 +159,9 @@ class BwrapPythonComponent(PythonComponent):
silent: bool = False,
) -> dict[str, Any]:
def _run() -> dict[str, Any]:
bwrap_cmd = build_bwrap_cmd(self.config, [os.environ.get("PYTHON", "python3"), "-c", code])
bwrap_cmd = build_bwrap_cmd(
self.config, [os.environ.get("PYTHON", "python3"), "-c", code]
)
try:
result = subprocess.run(
bwrap_cmd,
@@ -165,7 +177,9 @@ class BwrapPythonComponent(PythonComponent):
}
except subprocess.TimeoutExpired as e:
return {
"stdout": e.stdout.decode() if isinstance(e.stdout, bytes) else str(e.stdout or ""),
"stdout": e.stdout.decode()
if isinstance(e.stdout, bytes)
else str(e.stdout or ""),
"stderr": f"Execution timed out after {timeout} seconds.",
"exit_code": 1,
}
@@ -175,14 +189,16 @@ class BwrapPythonComponent(PythonComponent):
"stderr": str(e),
"exit_code": 1,
}
return await asyncio.to_thread(_run)
@dataclass
class HostBackedFileSystemComponent(FileSystemComponent):
"""File operations happen safely on host mapping to workspace, making I/O extremely fast."""
workspace_dir: str
def _safe_path(self, path: str) -> str:
# Simply maps it. In a stricter implementation, we could verify it's inside workspace_dir.
# But for this implementation, we trust the agent or restrict to workspace_dir.
@@ -190,7 +206,9 @@ class HostBackedFileSystemComponent(FileSystemComponent):
path = os.path.join(self.workspace_dir, path)
return path
async def create_file(self, path: str, content: str = "", mode: int = 0o644) -> dict[str, Any]:
async def create_file(
self, path: str, content: str = "", mode: int = 0o644
) -> dict[str, Any]:
p = self._safe_path(path)
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "w", encoding="utf-8") as f:
@@ -201,13 +219,15 @@ class HostBackedFileSystemComponent(FileSystemComponent):
async def read_file(self, path: str, encoding: str = "utf-8") -> dict[str, Any]:
p = self._safe_path(path)
try:
with open(p, "r", encoding=encoding) as f:
with open(p, encoding=encoding) as f:
content = f.read()
return {"success": True, "content": content}
except Exception as e:
return {"success": False, "error": str(e)}
async def write_file(self, path: str, content: str, mode: str = "w", encoding: str = "utf-8") -> dict[str, Any]:
async def write_file(
self, path: str, content: str, mode: str = "w", encoding: str = "utf-8"
) -> dict[str, Any]:
p = self._safe_path(path)
os.makedirs(os.path.dirname(p), exist_ok=True)
try:
@@ -228,7 +248,9 @@ class HostBackedFileSystemComponent(FileSystemComponent):
except Exception as e:
return {"success": False, "error": str(e)}
async def list_dir(self, path: str = ".", show_hidden: bool = False) -> dict[str, Any]:
async def list_dir(
self, path: str = ".", show_hidden: bool = False
) -> dict[str, Any]:
p = self._safe_path(path)
try:
items = os.listdir(p)
@@ -238,8 +260,9 @@ class HostBackedFileSystemComponent(FileSystemComponent):
except Exception as e:
return {"success": False, "error": str(e), "items": []}
class BwrapBooter(ComputerBooter):
def __init__(self, rw_binds: list[str] = None, ro_binds: list[str] = None):
def __init__(self, rw_binds: list[str] | None = None, ro_binds: list[str] | None = None):
self._rw_binds = rw_binds or []
self._ro_binds = ro_binds or []
self._fs: HostBackedFileSystemComponent | None = None
@@ -248,15 +271,15 @@ class BwrapBooter(ComputerBooter):
self.config: BwrapConfig | None = None
@property
def fs(self) -> FileSystemComponent:
def fs(self) -> FileSystemComponent | None:
return self._fs
@property
def python(self) -> PythonComponent:
def python(self) -> PythonComponent | None:
return self._python
@property
def shell(self) -> ShellComponent:
def shell(self) -> ShellComponent | None:
return self._shell
@property
@@ -264,37 +287,42 @@ class BwrapBooter(ComputerBooter):
return ("python", "shell", "filesystem")
async def boot(self, session_id: str) -> None:
workspace_dir = os.path.join(get_astrbot_temp_path(), f"sandbox_workspace_{session_id}")
workspace_dir = os.path.join(
get_astrbot_temp_path(), f"sandbox_workspace_{session_id}"
)
os.makedirs(workspace_dir, exist_ok=True)
self.config = BwrapConfig(
workspace_dir=os.path.abspath(workspace_dir),
rw_binds=self._rw_binds,
ro_binds=self._ro_binds
ro_binds=self._ro_binds,
)
self._fs = HostBackedFileSystemComponent(self.config.workspace_dir)
self._python = BwrapPythonComponent(self.config)
self._shell = BwrapShellComponent(self.config)
if not await self.available():
if not await self.available():
raise RuntimeError(
"BubbleWrap sandbox unavailable on current machine for no bwrap executable.")
test_shl = await self._shell.exec(command = "ls > /dev/null")
"BubbleWrap sandbox unavailable on current machine for no bwrap executable."
)
test_shl = await self._shell.exec(command="ls > /dev/null")
if test_shl["exit_code"] != 0:
raise RuntimeError(
'''BubbleWrap sandbox fails to exec test shell command "ls > /dev/null" with stderr:
{}'''.format(test_shl["stderr"]))
test_py = await self._python.exec(code = "print('Yes')")
"""BubbleWrap sandbox fails to exec test shell command "ls > /dev/null" with stderr:
{}""".format(test_shl["stderr"])
)
test_py = await self._python.exec(code="print('Yes')")
if test_py["exit_code"] != 0:
raise RuntimeError(
'''BubbleWrap sandbox fails to exec test python code "print('Yes')" with stderr:
{}'''.format(test_py["stderr"]))
"""BubbleWrap sandbox fails to exec test python code "print('Yes')" with stderr:
{}""".format(test_py["stderr"])
)
async def shutdown(self) -> None:
if self.config and os.path.exists(self.config.workspace_dir):
shutil.rmtree(self.config.workspace_dir, ignore_errors=True)
async def upload_file(self, path: str, file_name: str) -> dict:
if not self._fs:
if not self._fs or not self.config:
return {"success": False, "error": "Not booted"}
target = os.path.join(self.config.workspace_dir, file_name)
try:
@@ -304,7 +332,7 @@ class BwrapBooter(ComputerBooter):
return {"success": False, "error": str(e)}
async def download_file(self, remote_path: str, local_path: str) -> None:
if not self._fs:
if not self._fs or not self.config:
return
if not remote_path.startswith("/"):
remote_path = os.path.join(self.config.workspace_dir, remote_path)

17
astrbot/core/star/context.py
View File

@@ -46,6 +46,7 @@ from .star_handler import EventType, StarHandlerMetadata, star_handlers_registry
logger = logging.getLogger("astrbot")
if TYPE_CHECKING:
from astrbot.core.astr_agent_context import AstrAgentContext
from astrbot.core.cron.manager import CronJobManager
@@ -608,9 +609,11 @@ class Context:
"""
md = StarHandlerMetadata(
event_type=EventType.OnLLMRequestEvent,
handler_full_name=func_obj.__module__ + "_" + func_obj.__name__,
handler_name=func_obj.__name__,
handler_module_path=func_obj.__module__,
handler_full_name=getattr(func_obj, "__module__", "")
+ "_"
+ getattr(func_obj, "__name__", ""),
handler_name=getattr(func_obj, "__name__", ""),
handler_module_path=getattr(func_obj, "__module__", ""),
handler=func_obj,
event_filters=[],
desc=desc,
@@ -656,9 +659,11 @@ class Context:
"""
md = StarHandlerMetadata(
event_type=EventType.AdapterMessageEvent,
handler_full_name=awaitable.__module__ + "_" + awaitable.__name__,
handler_name=awaitable.__name__,
handler_module_path=awaitable.__module__,
handler_full_name=getattr(awaitable, "__module__", "")
+ "_"
+ getattr(awaitable, "__name__", ""),
handler_name=getattr(awaitable, "__name__", ""),
handler_module_path=getattr(awaitable, "__module__", ""),
handler=awaitable,
event_filters=[],
desc=desc,

2
astrbot/core/star/filter/command.py
View File

@@ -93,7 +93,7 @@ class CommandFilter(HandlerFilter):
def validate_and_convert_params(
self,
params: list[Any],
param_type: dict[str, type],
param_type: dict[str, type | Any],
) -> dict[str, Any]:
"""将参数列表 params 根据 param_type 转换为参数字典｡"""
result = {}