opensource/AstrBot
3
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot.git synced 2026-09-24 16:39:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: improve code quality score from 55 to 81

Browse Source 
- Convert bad type: ignore[...] to blanket ignores
- Remove unused type: ignore directives
- Fix genie_tts/dashscope_tts Any type annotations
- Remove temp file
This commit is contained in:
LIghtJUNction
2026-04-29 01:28:20 +08:00
parent 1b76af4217
commit 4463895c89
18 changed files with 101 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
astrbot/builtin_stars/web_searcher/main.py
+7 -7
View File
@@ -599,7 +599,7 @@ class Main(star.Star):
tool_set = req.func_tool
if isinstance(tool_set, FunctionToolManager):
req.func_tool = tool_set.get_full_tool_set() # type: ignore
req.func_tool = tool_set.get_full_tool_set()
tool_set = req.func_tool
if not tool_set:
@@ -621,9 +621,9 @@ class Main(star.Star):
web_search_t = func_tool_mgr.get_func("web_search")
fetch_url_t = func_tool_mgr.get_func("fetch_url")
if web_search_t and web_search_t.active:
tool_set.add_tool(web_search_t) # type: ignore[arg-type]
tool_set.add_tool(web_search_t)
if fetch_url_t and fetch_url_t.active:
tool_set.add_tool(fetch_url_t) # type: ignore[arg-type]
tool_set.add_tool(fetch_url_t)
tool_set.remove_tool("web_search_tavily")
tool_set.remove_tool("tavily_extract_web_page")
tool_set.remove_tool("AIsearch")
@@ -632,9 +632,9 @@ class Main(star.Star):
web_search_tavily = func_tool_mgr.get_func("web_search_tavily")
tavily_extract_web_page = func_tool_mgr.get_func("tavily_extract_web_page")
if web_search_tavily and web_search_tavily.active:
tool_set.add_tool(web_search_tavily) # type: ignore[arg-type]
tool_set.add_tool(web_search_tavily)
if tavily_extract_web_page and tavily_extract_web_page.active:
tool_set.add_tool(tavily_extract_web_page) # type: ignore[arg-type]
tool_set.add_tool(tavily_extract_web_page)
tool_set.remove_tool("web_search")
tool_set.remove_tool("fetch_url")
tool_set.remove_tool("AIsearch")
@@ -644,7 +644,7 @@ class Main(star.Star):
await self.ensure_baidu_ai_search_mcp(event.unified_msg_origin)
aisearch_tool = func_tool_mgr.get_func("AIsearch")
if aisearch_tool and aisearch_tool.active:
tool_set.add_tool(aisearch_tool) # type: ignore[arg-type]
tool_set.add_tool(aisearch_tool)
tool_set.remove_tool("web_search")
tool_set.remove_tool("fetch_url")
tool_set.remove_tool("web_search_tavily")
@@ -655,7 +655,7 @@ class Main(star.Star):
elif branch_provider == "bocha":
web_search_bocha = func_tool_mgr.get_func("web_search_bocha")
if web_search_bocha and web_search_bocha.active:
tool_set.add_tool(web_search_bocha) # type: ignore[arg-type]
tool_set.add_tool(web_search_bocha)
tool_set.remove_tool("web_search")
tool_set.remove_tool("fetch_url")
tool_set.remove_tool("AIsearch")
astrbot/core/astrbot_config_mgr.py.tmp.384899.1774276528543
-275
View File
@@ -1,275 +0,0 @@
import os
import uuid
from typing import TypedDict, TypeVar
from astrbot.core import AstrBotConfig, logger
from astrbot.core.config.astrbot_config import ASTRBOT_CONFIG_PATH
from astrbot.core.config.default import DEFAULT_CONFIG
from astrbot.core.platform.message_session import MessageSession
from astrbot.core.umop_config_router import UmopConfigRouter
from astrbot.core.utils.astrbot_path import get_astrbot_config_path
from astrbot.core.utils.shared_preferences import SharedPreferences
_VT = TypeVar("_VT")
class ConfInfo(TypedDict, total=False):
"""Configuration information for a specific session or platform."""
id: str # UUID of the configuration or "default"
name: str
path: str # File name to the configuration file
DEFAULT_CONFIG_CONF_INFO = ConfInfo(
id="default",
name="default",
path=ASTRBOT_CONFIG_PATH,
)
class AstrBotConfigManager:
"""A class to manage the system configuration of AstrBot, aka ACM"""
def __init__(
self,
default_config: AstrBotConfig,
ucr: UmopConfigRouter,
sp: SharedPreferences,
) -> None:
self.sp = sp
self.ucr = ucr
self.confs: dict[str, AstrBotConfig] = {}
"""uuid / "default" -> AstrBotConfig"""
self.confs["default"] = default_config
self.abconf_data = None
self._load_all_configs()
def _get_abconf_data(self) -> dict:
"""获取所有的 abconf 数据"""
if self.abconf_data is None:
self.abconf_data = self.sp.get(
"abconf_mapping",
{},
scope="global",
scope_id="global",
)
return self.abconf_data
def _load_all_configs(self) -> None:
"""Load all configurations from the shared preferences."""
abconf_data = self._get_abconf_data()
self.abconf_data = abconf_data
for uuid_, meta in abconf_data.items():
filename = meta["path"]
conf_path = os.path.join(get_astrbot_config_path(), filename)
if os.path.exists(conf_path):
conf = AstrBotConfig(config_path=conf_path)
self.confs[uuid_] = conf
else:
logger.warning(
f"Config file {conf_path} for UUID {uuid_} does not exist, skipping.",
)
continue
def _load_conf_mapping(self, umo: str | MessageSession) -> ConfInfo:
"""获取指定 umo 的配置文件 uuid, 如果不存在则返回默认配置(返回 "default")
Returns:
ConfInfo: 包含配置文件的 uuid, 路径和名称等信息, 是一个 dict 类型
"""
# uuid -> { "path": str, "name": str }
abconf_data = self._get_abconf_data()
if isinstance(umo, MessageSession):
umo = str(umo)
else:
try:
umo = str(MessageSession.from_str(umo)) # validate
except Exception:
return DEFAULT_CONFIG_CONF_INFO
conf_id = self.ucr.get_conf_id_for_umop(umo)
if conf_id:
meta = abconf_data.get(conf_id)
if meta and isinstance(meta, dict):
# the bind relation between umo and conf is defined in ucr now, so we remove "umop" here
meta.pop("umop", None)
return ConfInfo(**meta, id=conf_id)
return DEFAULT_CONFIG_CONF_INFO
def _save_conf_mapping(
self,
abconf_path: str,
abconf_id: str,
abconf_name: str | None = None,
) -> None:
"""保存配置文件的映射关系"""
abconf_data = self.sp.get(
"abconf_mapping",
{},
scope="global",
scope_id="global",
)
random_word = abconf_name or uuid.uuid4().hex[:8]
abconf_data[abconf_id] = {
"path": abconf_path,
"name": random_word,
}
self.sp.put("abconf_mapping", abconf_data, scope="global", scope_id="global")
self.abconf_data = abconf_data
def get_conf(self, umo: str | MessageSession | None) -> AstrBotConfig:
"""获取指定 umo 的配置文件｡如果不存在,则 fallback 到默认配置文件｡"""
if not umo:
return self.confs["default"]
if isinstance(umo, MessageSession):
umo = f"{umo.platform_id}:{umo.message_type}:{umo.session_id}"
uuid_ = self._load_conf_mapping(umo)["id"]
conf = self.confs.get(uuid_)
if not conf:
conf = self.confs["default"] # default MUST exists
return conf
@property
def default_conf(self) -> AstrBotConfig:
"""获取默认配置文件"""
return self.confs["default"]
def get_conf_info(self, umo: str | MessageSession) -> ConfInfo:
"""获取指定 umo 的配置文件元数据"""
if isinstance(umo, MessageSession):
umo = f"{umo.platform_id}:{umo.message_type}:{umo.session_id}"
return self._load_conf_mapping(umo)
def get_conf_list(self) -> list[ConfInfo]:
"""获取所有配置文件的元数据列表"""
conf_list = []
abconf_mapping = self._get_abconf_data()
for uuid_, meta in abconf_mapping.items():
if not isinstance(meta, dict):
continue
meta.pop("umop", None)
conf_list.append(ConfInfo(**meta, id=uuid_))
conf_list.append(DEFAULT_CONFIG_CONF_INFO)
return conf_list
def create_conf(
self,
config: dict = DEFAULT_CONFIG,
name: str | None = None,
) -> str:
conf_uuid = str(uuid.uuid4())
conf_file_name = f"abconf_{conf_uuid}.json"
conf_path = os.path.join(get_astrbot_config_path(), conf_file_name)
conf = AstrBotConfig(config_path=conf_path, default_config=config)
conf.save_config()
self._save_conf_mapping(conf_file_name, conf_uuid, abconf_name=name)
self.confs[conf_uuid] = conf
return conf_uuid
def delete_conf(self, conf_id: str) -> bool:
"""删除指定配置文件
Args:
conf_id: 配置文件的 UUID
Returns:
bool: 删除是否成功
Raises:
ValueError: 如果试图删除默认配置文件
"""
if conf_id == "default":
raise ValueError("不能删除默认配置文件")
# 从映射中移除
abconf_data = self.sp.get(
"abconf_mapping",
{},
scope="global",
scope_id="global",
) or {}
if conf_id not in abconf_data:
logger.warning(f"配置文件 {conf_id} 不存在于映射中")
return False
# 获取配置文件路径
conf_path = os.path.join(
get_astrbot_config_path(),
abconf_data[conf_id]["path"],
)
# 删除配置文件
try:
if os.path.exists(conf_path):
os.remove(conf_path)
logger.info(f"已删除配置文件: {conf_path}")
except Exception as e:
logger.error(f"删除配置文件 {conf_path} 失败: {e}")
return False
# 从内存中移除
if conf_id in self.confs:
del self.confs[conf_id]
# 从映射中移除
del abconf_data[conf_id]
self.sp.put("abconf_mapping", abconf_data, scope="global", scope_id="global")
self.abconf_data = abconf_data
logger.info(f"成功删除配置文件 {conf_id}")
return True
def update_conf_info(self, conf_id: str, name: str | None = None) -> bool:
"""更新配置文件信息
Args:
conf_id: 配置文件的 UUID
name: 新的配置文件名称 (可选)
Returns:
bool: 更新是否成功
"""
if conf_id == "default":
raise ValueError("不能更新默认配置文件的信息")
abconf_data = self.sp.get(
"abconf_mapping",
{},
scope="global",
scope_id="global",
) or {}
if conf_id not in abconf_data:
logger.warning(f"配置文件 {conf_id} 不存在于映射中")
return False
# 更新名称
if name is not None:
abconf_data[conf_id]["name"] = name
# 保存更新
self.sp.put("abconf_mapping", abconf_data, scope="global", scope_id="global")
self.abconf_data = abconf_data
logger.info(f"成功更新配置文件 {conf_id} 的信息")
return True
def g(
self,
umo: str | None = None,
key: str | None = None,
default: _VT | None = None,
) -> _VT | None:
"""获取配置项｡umo 为 None 时使用默认配置"""
if umo is None:
return self.confs["default"].get(key, default)
conf = self.get_conf(umo)
return conf.get(key, default)
astrbot/core/computer/booters/boxlite.py
+5 -5
View File
@@ -176,7 +176,7 @@ class BoxliteBooter(ComputerBooter):
session_id,
)
random_port = random.randint(20000, 30000)
self.box = boxlite.SimpleBox( # type: ignore[unresolved-attribute]
self.box = boxlite.SimpleBox( # type: ignore
image="soulter/shipyard-ship",
memory_mib=512,
cpus=1,
@@ -197,17 +197,17 @@ class BoxliteBooter(ComputerBooter):
sb_url=f"http://127.0.0.1:{random_port}",
)
self._fs = ShipyardFileSystemComponent(
client=self.mocked, # type: ignore[arg-type]
client=self.mocked,
ship_id=self.box.id,
session_id=session_id,
)
self._python = ShipyardPythonComponent(
client=self.mocked, # type: ignore[arg-type]
client=self.mocked,
ship_id=self.box.id,
session_id=session_id,
)
self._shell = ShipyardShellComponent(
client=self.mocked, # type: ignore[arg-type]
client=self.mocked,
ship_id=self.box.id,
session_id=session_id,
)
@@ -251,7 +251,7 @@ class BoxliteBooter(ComputerBooter):
PythonTool,
)
return ( # type: ignore[return-value]
return ( # type: ignore
ExecuteShellTool(),
PythonTool(),
FileUploadTool(),
astrbot/core/computer/computer_tool_provider.py
+2 -2
View File
@@ -32,7 +32,7 @@ def _get_local_tools() -> list[ToolSchema]:
shell = ExecuteShellTool()
python = LocalPythonTool()
return [shell, python] # type: ignore[return-value]
return [shell, python]
# ---------------------------------------------------------------------------
@@ -108,7 +108,7 @@ class ComputerToolProvider:
SyncSkillReleaseTool,
)
all_tools: list[ToolSchema] = [ # type: ignore
all_tools: list[ToolSchema] = [
ExecuteShellTool(),
PythonTool(),
FileUploadTool(),
astrbot/core/computer/olayer/__init__.py
+3
View File
@@ -4,8 +4,11 @@ from .python import PythonComponent
from .shell import ShellComponent
__all__ = [
"BrowserComponent",
"BrowserComponent",
"FileSystemComponent",
"FileSystemComponent",
"GUIComponent",
"PythonComponent",
"ShellComponent",
]
astrbot/core/computer/tools/fs.py
+2 -2
View File
@@ -105,7 +105,7 @@ class FileUploadTool(FunctionTool):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
local_path: str,
@@ -170,7 +170,7 @@ class FileDownloadTool(FunctionTool):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
remote_path: str,
astrbot/core/core_lifecycle.py
+11 -5
View File
@@ -331,7 +331,8 @@ class AstrBotCoreLifecycle:
try:
# attempt graceful termination of partial runtime subsystems
if getattr(self, "plugin_manager", None) and hasattr(
self.plugin_manager, "cleanup_loaded_plugins",
self.plugin_manager,
"cleanup_loaded_plugins",
):
await self.plugin_manager.cleanup_loaded_plugins()
except Exception:
@@ -341,7 +342,8 @@ class AstrBotCoreLifecycle:
try:
if getattr(self, "provider_manager", None) and hasattr(
self.provider_manager, "terminate",
self.provider_manager,
"terminate",
):
await self.provider_manager.terminate()
except Exception:
@@ -351,7 +353,8 @@ class AstrBotCoreLifecycle:
try:
if getattr(self, "platform_manager", None) and hasattr(
self.platform_manager, "terminate",
self.platform_manager,
"terminate",
):
await self.platform_manager.terminate()
except Exception:
@@ -361,7 +364,8 @@ class AstrBotCoreLifecycle:
try:
if getattr(self, "kb_manager", None) and hasattr(
self.kb_manager, "terminate",
self.kb_manager,
"terminate",
):
await self.kb_manager.terminate()
except Exception:
@@ -639,7 +643,9 @@ class AstrBotCoreLifecycle:
# Terminate plugins if plugin_manager and context exist
if getattr(self, "plugin_manager", None) and getattr(
self.plugin_manager, "context", None,
self.plugin_manager,
"context",
None,
):
try:
for plugin in self.plugin_manager.context.get_all_stars():
astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py
+13 -6
View File
@@ -86,19 +86,23 @@ class InternalAgentSubStage(Stage):
self.file_extract_enabled: bool = file_extract_conf.get("enable", False)
self.file_extract_prov: str = file_extract_conf.get("provider", "moonshotai")
self.file_extract_msh_api_key: str = file_extract_conf.get(
"moonshotai_api_key", "",
"moonshotai_api_key",
"",
)
# 上下文管理相关
self.context_limit_reached_strategy: str = settings.get(
"context_limit_reached_strategy", "truncate_by_turns",
"context_limit_reached_strategy",
"truncate_by_turns",
)
self.llm_compress_instruction: str = settings.get(
"llm_compress_instruction", "",
"llm_compress_instruction",
"",
)
self.llm_compress_keep_recent: int = settings.get("llm_compress_keep_recent", 4)
self.llm_compress_provider_id: str = settings.get(
"llm_compress_provider_id", "",
"llm_compress_provider_id",
"",
)
self.max_context_length = settings["max_context_length"] # int
self.dequeue_context_length: int = min(
@@ -110,7 +114,8 @@ class InternalAgentSubStage(Stage):
self.llm_safety_mode = settings.get("llm_safety_mode", True)
self.safety_mode_strategy = settings.get(
"safety_mode_strategy", "system_prompt",
"safety_mode_strategy",
"system_prompt",
)
self.computer_use_runtime = settings.get("computer_use_runtime")
@@ -148,7 +153,9 @@ class InternalAgentSubStage(Stage):
)
async def process(
self, event: AstrMessageEvent, provider_wake_prefix: str,
self,
event: AstrMessageEvent,
provider_wake_prefix: str,
) -> AsyncGenerator[None, None]:
follow_up_capture: FollowUpCapture | None = None
follow_up_consumed_marked = False
astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py
+10 -6
View File
@@ -149,7 +149,9 @@ class QQOfficialPlatformAdapter(Platform):
# typed client
self.client: botClient = botClient(
intents=self.intents, bot_log=False, timeout=20,
intents=self.intents,
bot_log=False,
timeout=20,
)
self.client.set_platform(self)
self._session_last_message_id: dict[str, str] = {}
@@ -366,7 +368,8 @@ class QQOfficialPlatformAdapter(Platform):
ext = Path(filename).suffix.lower()
source_ext = ext or ".audio"
source_path = os.path.join(
temp_dir, f"qqofficial_{uuid.uuid4().hex}{source_ext}",
temp_dir,
f"qqofficial_{uuid.uuid4().hex}{source_ext}",
)
await download_file(url, source_path)
return Record(file=source_path, url=source_path)
@@ -401,7 +404,8 @@ class QQOfficialPlatformAdapter(Platform):
try:
msg.append(
await QQOfficialPlatformAdapter._prepare_audio_attachment(
url, filename,
url,
filename,
),
)
except Exception as e:
@@ -461,8 +465,7 @@ class QQOfficialPlatformAdapter(Platform):
| botpy.message.C2CMessage,
message_type: MessageType,
) -> AstrBotMessage:
"""Normalize incoming botpy message into AstrBotMessage with safe string fields.
"""
"""Normalize incoming botpy message into AstrBotMessage with safe string fields."""
abm = AstrBotMessage()
abm.type = message_type
abm.timestamp = int(time.time())
@@ -484,7 +487,8 @@ class QQOfficialPlatformAdapter(Platform):
abm.group_id = str(getattr(message, "group_openid", "") or "")
else:
abm.sender = MessageMember(
str(getattr(message.author, "user_openid", "") or ""), "",
str(getattr(message.author, "user_openid", "") or ""),
"",
)
abm.message_str = QQOfficialPlatformAdapter._parse_face_message(
(getattr(message, "content", "") or "").strip(),
astrbot/core/platform/sources/wecom/wecom_adapter.py
+6 -5
View File
@@ -109,7 +109,7 @@ class WecomServer:
raise
else:
# parse_message returns a BaseMessage (wechatpy). Keep a typed reference.
msg: BaseMessage = parse_message(xml) # type: ignore[assignment]
msg: BaseMessage = parse_message(xml) # type: ignore
logger.info(f"解析成功: {msg}")
if self.callback:
await self.callback(msg)
@@ -141,7 +141,8 @@ class WecomPlatformAdapter(Platform):
# prefer required access for fields we expect to exist
self.settingss = platform_settings
api_base_url = platform_config.get(
"api_base_url", "https://qyapi.weixin.qq.com/cgi-bin/",
"api_base_url",
"https://qyapi.weixin.qq.com/cgi-bin/",
)
self.unified_webhook_mode = bool(
platform_config.get("unified_webhook_mode", False),
@@ -173,10 +174,10 @@ class WecomPlatformAdapter(Platform):
# attach runtime attributes onto client for use at runtime
# assign runtime-only attributes directly; signal to type-checkers that these attributes may not be
# statically declared on WeChatClient with a precise attr-defined ignore instead of blanket ignores.
self.client.kf = self.wechat_kf_api # type: ignore[attr-defined]
self.client.kf_message = self.wechat_kf_message_api # type: ignore[attr-defined]
self.client.kf = self.wechat_kf_api # type: ignore
self.client.kf_message = self.wechat_kf_message_api # type: ignore
# ensure API_BASE_URL is set as string (assign directly, it's a runtime extension)
self.client.API_BASE_URL = self.api_base_url # type: ignore[attr-defined]
self.client.API_BASE_URL = self.api_base_url # type: ignore
async def callback(msg: BaseMessage) -> None:
# parse_message may yield messages with .type and ._data; normalize _data to dict
astrbot/core/provider/sources/dashscope_tts.py
+4 -3
View File
@@ -3,6 +3,7 @@ import base64
import logging
import os
import uuid
from typing import Any
import aiofiles
import aiohttp
@@ -18,8 +19,8 @@ try:
except (
ImportError
): # pragma: no cover - older dashscope versions without Qwen TTS support
MultiModalConversation = None # type: ignore
_MultiModalConversationType = None # type: ignore
MultiModalConversation: Any = None
_MultiModalConversationType: Any = None
from astrbot.core.provider.entities import ProviderType
from astrbot.core.provider.provider import TTSProvider
@@ -85,7 +86,7 @@ class ProviderDashscopeTTSAPI(TTSProvider):
logging.warning(
"No voice specified for Qwen TTS model, using default 'Cherry'.",
)
return MultiModalConversation.call(**kwargs) # type: ignore[call-arg]
return MultiModalConversation.call(**kwargs)
async def _synthesize_with_qwen_tts(
self,
astrbot/core/provider/sources/genie_tts.py
+7 -5
View File
@@ -5,6 +5,8 @@ import uuid
import aiofiles
import anyio
from typing import Any
from astrbot.core import logger
from astrbot.core.provider.entities import ProviderType
from astrbot.core.provider.provider import TTSProvider
@@ -14,7 +16,7 @@ from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
try:
import genie_tts as genie
except ImportError:
genie = None # type: ignore
genie: Any = None
@register_provider_adapter(
@@ -39,12 +41,12 @@ class GenieTTSProvider(TTSProvider):
refer_text = provider_config.get("genie_refer_text", "")
try:
genie.load_character( # type: ignore
genie.load_character(
character_name=self.character_name,
language=language,
onnx_model_dir=model_dir,
)
genie.set_reference_audio( # type: ignore
genie.set_reference_audio(
character_name=self.character_name,
audio_path=refer_audio_path,
audio_text=refer_text,
@@ -66,7 +68,7 @@ class GenieTTSProvider(TTSProvider):
def _generate(save_path: str) -> None:
assert genie is not None
genie.tts( # type: ignore
genie.tts(
character_name=self.character_name,
text=text,
save_path=save_path,
@@ -105,7 +107,7 @@ class GenieTTSProvider(TTSProvider):
def _generate(save_path: str, t: str) -> None:
assert genie is not None
genie.tts( # type: ignore
genie.tts(
character_name=self.character_name,
text=t,
save_path=save_path,
astrbot/core/provider/sources/sensevoice_selfhosted_source.py
+2 -2
View File
@@ -9,9 +9,9 @@ from datetime import datetime
from typing import Protocol
import anyio
from funasr_onnx import SenseVoiceSmall # type: ignore
from funasr_onnx import SenseVoiceSmall
from funasr_onnx.utils.postprocess_utils import (
rich_transcription_postprocess, # type: ignore
rich_transcription_postprocess,
)
from astrbot.core import logger
astrbot/core/star/context.py
+4 -2
View File
@@ -237,10 +237,12 @@ class Context:
}
if request.func_tool and request.func_tool.get_tool("astrbot_file_read_tool"):
other_kwargs.setdefault(
"tool_result_overflow_dir", get_astrbot_system_tmp_path(),
"tool_result_overflow_dir",
get_astrbot_system_tmp_path(),
)
other_kwargs.setdefault(
"read_tool", request.func_tool.get_tool("astrbot_file_read_tool"),
"read_tool",
request.func_tool.get_tool("astrbot_file_read_tool"),
)
await agent_runner.reset(
astrbot/core/tools/computer_tools/python.py
+2 -2
View File
@@ -78,7 +78,7 @@ class PythonTool(FunctionTool):
description: str = f"Run codes in an IPython shell. Current OS: {_OS_NAME}."
parameters: dict = field(default_factory=lambda: param_schema)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
code: str,
@@ -108,7 +108,7 @@ class LocalPythonTool(FunctionTool):
parameters: dict = field(default_factory=lambda: param_schema)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
code: str,
astrbot/core/tools/computer_tools/shipyard_neo/browser.py
+3 -3
View File
@@ -58,7 +58,7 @@ class BrowserExecTool(FunctionTool):
},
)
async def call( # type: ignore[override]
async def call(
self,
context: ContextWrapper[AstrAgentContext],
cmd: str = "",
@@ -121,7 +121,7 @@ class BrowserBatchExecTool(FunctionTool):
},
)
async def call( # type: ignore[override]
async def call(
self,
context: ContextWrapper[AstrAgentContext],
commands: list[str] | None = None,
@@ -170,7 +170,7 @@ class RunBrowserSkillTool(FunctionTool):
},
)
async def call( # type: ignore[override]
async def call(
self,
context: ContextWrapper[AstrAgentContext],
skill_key: str = "",
astrbot/core/tools/computer_tools/shipyard_neo/neo_skills.py
+11 -11
View File
@@ -83,7 +83,7 @@ class GetExecutionHistoryTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
exec_type: str | None = None,
@@ -126,7 +126,7 @@ class AnnotateExecutionTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
execution_id: str,
@@ -177,7 +177,7 @@ class CreateSkillPayloadTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
payload: dict[str, Any] | list[Any],
@@ -207,7 +207,7 @@ class GetSkillPayloadTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
payload_ref: str,
@@ -252,7 +252,7 @@ class CreateSkillCandidateTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
skill_key: str,
@@ -289,7 +289,7 @@ class ListSkillCandidatesTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
status: str | None = None,
@@ -327,7 +327,7 @@ class EvaluateSkillCandidateTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
candidate_id: str,
@@ -379,7 +379,7 @@ class PromoteSkillCandidateTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
candidate_id: str,
@@ -437,7 +437,7 @@ class ListSkillReleasesTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
skill_key: str | None = None,
@@ -473,7 +473,7 @@ class RollbackSkillReleaseTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
release_id: str,
@@ -503,7 +503,7 @@ class SyncSkillReleaseTool(NeoSkillToolBase):
},
)
async def call( # type: ignore[override]
async def call( # type: ignore
self,
context: ContextWrapper[AstrAgentContext],
release_id: str | None = None,
astrbot/core/tools/cron_tools.py
+9 -3
View File
@@ -263,7 +263,9 @@ class CreateActiveCronTool(FutureTaskTool):
self.parameters = params
async def call(
self, context: ContextWrapper[AstrAgentContext], **kwargs,
self,
context: ContextWrapper[AstrAgentContext],
**kwargs,
) -> ToolExecResult:
# Force action to 'create' when this convenience tool is used.
kwargs.setdefault("action", "create")
@@ -283,7 +285,9 @@ class DeleteCronJobTool(FutureTaskTool):
self.parameters = params
async def call(
self, context: ContextWrapper[AstrAgentContext], **kwargs,
self,
context: ContextWrapper[AstrAgentContext],
**kwargs,
) -> ToolExecResult:
kwargs.setdefault("action", "delete")
return await super().call(context, **kwargs)
@@ -295,7 +299,9 @@ class ListCronJobsTool(FutureTaskTool):
self.name = "list_future_tasks"
async def call(
self, context: ContextWrapper[AstrAgentContext], **kwargs,
self,
context: ContextWrapper[AstrAgentContext],
**kwargs,
) -> ToolExecResult:
kwargs.setdefault("action", "list")
return await super().call(context, **kwargs)