Compare commits

..

1 Commits

Author SHA1 Message Date
Soulter
3931c3ca79 feat: implement user agent handling across providers and normalize headers 2026-05-30 21:02:39 +08:00
21 changed files with 213 additions and 577 deletions

1
.gitignore vendored
View File

@@ -9,7 +9,6 @@ uv.lock
# IDE and editors
.vscode
.idea
.zed/
# Logs and temporary files
botpy.log

View File

@@ -10,7 +10,6 @@ from ..utils import (
check_astrbot_root,
get_astrbot_root,
get_git_repo,
install_local_plugin,
manage_plugin,
)
@@ -144,32 +143,12 @@ def list(all: bool) -> None:
@plug.command()
@click.argument("name", required=False)
@click.option(
"--editable",
"-e",
"local_path",
type=click.Path(exists=True, file_okay=False, path_type=Path),
help="Install a plugin from a local directory as a symlink",
)
@click.argument("name")
@click.option("--proxy", help="Proxy server address")
def install(name: str | None, local_path: Path | None, proxy: str | None) -> None:
def install(name: str, proxy: str | None) -> None:
"""Install a plugin"""
base_path = _get_data_path()
plug_path = base_path / "plugins"
if local_path is not None:
install_local_plugin(local_path, plug_path, editable=True)
return
if name is None:
raise click.ClickException("Missing plugin name or local plugin path")
local_name_path = Path(name).expanduser()
if local_name_path.exists() and local_name_path.is_dir():
install_local_plugin(local_name_path, plug_path, editable=False)
return
plugins = build_plug_list(base_path / "plugins")
plugin = next(

View File

@@ -3,13 +3,7 @@ from .basic import (
check_dashboard,
get_astrbot_root,
)
from .plugin import (
PluginStatus,
build_plug_list,
get_git_repo,
install_local_plugin,
manage_plugin,
)
from .plugin import PluginStatus, build_plug_list, get_git_repo, manage_plugin
from .version_comparator import VersionComparator
__all__ = [
@@ -20,6 +14,5 @@ __all__ = [
"check_dashboard",
"get_astrbot_root",
"get_git_repo",
"install_local_plugin",
"manage_plugin",
]

View File

@@ -1,6 +1,5 @@
import shutil
import tempfile
import uuid
from enum import Enum
from io import BytesIO
from pathlib import Path
@@ -20,35 +19,6 @@ class PluginStatus(str, Enum):
NOT_PUBLISHED = "unpublished"
LOCAL_PLUGIN_COPY_IGNORE = shutil.ignore_patterns(
".git",
"__pycache__",
"*.pyc",
".venv",
"venv",
".idea",
".vscode",
".zed",
)
def _validate_plugin_dir_name(plugin_name: str, source_path: Path) -> str:
plugin_name = plugin_name.strip()
plugin_path = Path(plugin_name)
has_separator = "/" in plugin_name or "\\" in plugin_name
if (
not plugin_name
or plugin_name in {".", ".."}
or plugin_path.is_absolute()
or has_separator
or plugin_path.name != plugin_name
):
raise click.ClickException(
f"Local plugin {source_path} metadata.yaml has invalid name: {plugin_name}"
)
return plugin_name
def get_git_repo(url: str, target_path: Path, proxy: str | None = None) -> None:
"""Download code from a Git repository and extract to the specified path"""
temp_dir = Path(tempfile.mkdtemp())
@@ -220,78 +190,6 @@ def build_plug_list(plugins_dir: Path) -> list:
return result
def _cleanup_local_plugin_target(target_path: Path) -> None:
if target_path.is_symlink() or target_path.is_file():
target_path.unlink(missing_ok=True)
elif target_path.exists():
shutil.rmtree(target_path, ignore_errors=True)
def _copy_local_plugin(source_path: Path, plugins_dir: Path, target_path: Path) -> None:
temp_target = plugins_dir / f".{target_path.name}.tmp-{uuid.uuid4().hex}"
try:
shutil.copytree(source_path, temp_target, ignore=LOCAL_PLUGIN_COPY_IGNORE)
temp_target.rename(target_path)
except FileExistsError:
raise click.ClickException(
f"Plugin {target_path.name} already exists"
) from None
except Exception:
raise
finally:
if temp_target.exists() or temp_target.is_symlink():
_cleanup_local_plugin_target(temp_target)
def install_local_plugin(
source_path: Path,
plugins_dir: Path,
editable: bool = False,
) -> None:
"""Install a plugin from a local directory."""
source_path = source_path.expanduser().resolve()
plugins_dir = plugins_dir.resolve()
if not source_path.exists() or not source_path.is_dir():
raise click.ClickException(f"Local plugin path does not exist: {source_path}")
metadata = load_yaml_metadata(source_path)
plugin_name = metadata.get("name")
if not isinstance(plugin_name, str) or not plugin_name.strip():
raise click.ClickException(
f"Local plugin {source_path} must contain metadata.yaml with a valid name"
)
plugin_name = _validate_plugin_dir_name(plugin_name, source_path)
target_path = plugins_dir / plugin_name
if target_path.exists():
raise click.ClickException(f"Plugin {plugin_name} already exists")
try:
plugins_dir.mkdir(parents=True, exist_ok=True)
if editable:
try:
target_path.symlink_to(source_path, target_is_directory=True)
except OSError as e:
raise click.ClickException(
f"Failed to create symlink for editable install: {e}. "
"On Windows, you may need to run as Administrator or enable Developer Mode."
) from e
else:
_copy_local_plugin(source_path, plugins_dir, target_path)
click.echo(f"Plugin {plugin_name} installed successfully from {source_path}")
except FileExistsError:
raise click.ClickException(f"Plugin {plugin_name} already exists") from None
except click.ClickException:
raise
except Exception as e:
if editable and target_path.is_symlink():
_cleanup_local_plugin_target(target_path)
raise click.ClickException(
f"Error installing local plugin {plugin_name}: {e}"
) from e
def manage_plugin(
plugin: dict,
plugins_dir: Path,

View File

@@ -6,6 +6,7 @@ from astrbot.core.computer.booters.cua_defaults import CUA_DEFAULT_CONFIG
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
VERSION = "4.25.2"
ASTRBOT_USER_AGENT = f"astrbot/{VERSION.removeprefix('v')}"
DB_PATH = os.path.join(get_astrbot_data_path(), "data_v4.db")
PERSONAL_WECHAT_CONFIG_METADATA = {
"weixin_oc_base_url": {
@@ -1199,7 +1200,7 @@ CONFIG_METADATA_2 = {
"api_base": "https://api.kimi.com/coding",
"timeout": 120,
"proxy": "",
"custom_headers": {"User-Agent": "claude-code/0.1.0"},
"custom_headers": {"User-Agent": ASTRBOT_USER_AGENT},
"anth_thinking_config": {"type": "", "budget": 0, "effort": ""},
},
"Moonshot": {
@@ -1236,7 +1237,7 @@ CONFIG_METADATA_2 = {
"api_base": "https://api.minimaxi.com/anthropic",
"timeout": 120,
"proxy": "",
"custom_headers": {"User-Agent": "claude-code/0.1.0"},
"custom_headers": {"User-Agent": ASTRBOT_USER_AGENT},
"anth_thinking_config": {"type": "", "budget": 0, "effort": ""},
},
"Xiaomi": {
@@ -1261,7 +1262,7 @@ CONFIG_METADATA_2 = {
"api_base": "https://token-plan-cn.xiaomimimo.com/anthropic",
"timeout": 120,
"proxy": "",
"custom_headers": {"User-Agent": "claude-code/0.1.0"},
"custom_headers": {"User-Agent": ASTRBOT_USER_AGENT},
"anth_thinking_config": {"type": "", "budget": 0, "effort": ""},
},
"xAI": {

View File

@@ -13,9 +13,11 @@ from anthropic.types.usage import Usage
from astrbot import logger
from astrbot.api.provider import Provider
from astrbot.core.agent.message import AudioURLPart, ContentPart, ImageURLPart, TextPart
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
from astrbot.core.provider.entities import LLMResponse, TokenUsage
from astrbot.core.provider.func_tool_manager import ToolSet
from astrbot.core.utils.http_headers import apply_default_headers, normalize_headers
from astrbot.core.utils.io import download_image_by_url
from astrbot.core.utils.network_utils import (
create_proxy_client,
@@ -50,13 +52,12 @@ class ProviderAnthropic(Provider):
@staticmethod
def _normalize_custom_headers(provider_config: dict) -> dict[str, str] | None:
custom_headers = provider_config.get("custom_headers", {})
if not isinstance(custom_headers, dict) or not custom_headers:
normalized_headers = normalize_headers(
provider_config.get("custom_headers", {})
)
if not normalized_headers:
return None
normalized_headers: dict[str, str] = {}
for key, value in custom_headers.items():
normalized_headers[str(key)] = str(value)
return normalized_headers or None
return normalized_headers
@classmethod
def _resolve_custom_headers(
@@ -67,9 +68,7 @@ class ProviderAnthropic(Provider):
) -> dict[str, str] | None:
merged_headers = cls._normalize_custom_headers(provider_config) or {}
if required_headers:
for header_name, header_value in required_headers.items():
if not merged_headers.get(header_name, "").strip():
merged_headers[header_name] = header_value
merged_headers = apply_default_headers(merged_headers, required_headers)
return merged_headers or None
def __init__(
@@ -89,7 +88,10 @@ class ProviderAnthropic(Provider):
if isinstance(self.timeout, str):
self.timeout = int(self.timeout)
self.thinking_config = provider_config.get("anth_thinking_config", {})
self.custom_headers = self._resolve_custom_headers(provider_config)
self.custom_headers = self._resolve_custom_headers(
provider_config,
required_headers={"User-Agent": ASTRBOT_USER_AGENT},
)
if use_api_key:
self._init_api_key(provider_config)

View File

@@ -18,11 +18,13 @@ import astrbot.core.message.components as Comp
from astrbot import logger
from astrbot.api.provider import Provider
from astrbot.core.agent.message import AudioURLPart, ContentPart, ImageURLPart, TextPart
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
from astrbot.core.message.message_event_result import MessageChain
from astrbot.core.provider.entities import LLMResponse, TokenUsage
from astrbot.core.provider.func_tool_manager import ToolSet
from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
from astrbot.core.utils.http_headers import apply_default_headers, normalize_headers
from astrbot.core.utils.io import download_file, download_image_by_url
from astrbot.core.utils.media_utils import ensure_wav
from astrbot.core.utils.network_utils import is_connection_error, log_connection_failure
@@ -76,17 +78,41 @@ class ProviderGoogleGenAI(Provider):
if self.api_base and self.api_base.endswith("/"):
self.api_base = self.api_base[:-1]
self.custom_headers = self._resolve_custom_headers(provider_config)
self._http_client: httpx.AsyncClient | None = None
self._stale_http_clients: list[httpx.AsyncClient] = []
self._init_client()
self.set_model(provider_config.get("model", "unknown"))
self._init_safety_settings()
@staticmethod
def _resolve_custom_headers(provider_config: dict) -> dict[str, str]:
headers = apply_default_headers(
normalize_headers(provider_config.get("custom_headers", {})),
{"user-agent": ASTRBOT_USER_AGENT},
)
return {
"user-agent" if key.lower() == "user-agent" else key: value
for key, value in headers.items()
}
@staticmethod
def _set_gemini_user_agent(client: object, user_agent: str) -> None:
api_client = getattr(client, "_api_client", None)
http_options = getattr(api_client, "_http_options", None)
if http_options is None or http_options.headers is None:
return
for key in list(http_options.headers):
if key.lower() == "user-agent":
http_options.headers.pop(key)
http_options.headers["user-agent"] = user_agent
def _init_client(self) -> None:
"""初始化Gemini客户端"""
proxy = self.provider_config.get("proxy", "")
http_options = types.HttpOptions(
base_url=self.api_base,
headers=dict(self.custom_headers),
timeout=self.timeout * 1000, # 毫秒
)
@@ -94,6 +120,7 @@ class ProviderGoogleGenAI(Provider):
# httpx.AsyncClient 的 timeout 单位为秒(与 HttpOptions 的毫秒不同)
async_client_kwargs: dict = {
"base_url": self.api_base,
"headers": dict(self.custom_headers),
"timeout": self.timeout,
}
if proxy:
@@ -112,10 +139,15 @@ class ProviderGoogleGenAI(Provider):
self._http_client = httpx.AsyncClient(**async_client_kwargs)
http_options.httpx_async_client = self._http_client
self.client = genai.Client(
genai_client = genai.Client(
api_key=self.chosen_api_key,
http_options=http_options,
).aio
)
self._set_gemini_user_agent(
genai_client,
self.custom_headers["user-agent"],
)
self.client = genai_client.aio
def _init_safety_settings(self) -> None:
"""初始化安全设置"""

View File

@@ -1,9 +1,11 @@
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from ..register import register_provider_adapter
from .anthropic_source import ProviderAnthropic
KIMI_CODE_API_BASE = "https://api.kimi.com/coding"
KIMI_CODE_DEFAULT_MODEL = "kimi-for-coding"
KIMI_CODE_USER_AGENT = "claude-code/0.1.0"
KIMI_CODE_USER_AGENT = ASTRBOT_USER_AGENT
@register_provider_adapter(

View File

@@ -34,10 +34,12 @@ from astrbot.core.agent.message import (
TextPart,
)
from astrbot.core.agent.tool import ToolSet
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
from astrbot.core.message.message_event_result import MessageChain
from astrbot.core.provider.entities import LLMResponse, TokenUsage, ToolCallsResult
from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
from astrbot.core.utils.http_headers import apply_default_headers, normalize_headers
from astrbot.core.utils.io import download_file, download_image_by_url
from astrbot.core.utils.media_utils import ensure_wav
from astrbot.core.utils.network_utils import (
@@ -68,6 +70,13 @@ class ProviderOpenAIOfficial(Provider):
"AVIF": "image/avif",
}
@staticmethod
def _resolve_custom_headers(provider_config: dict) -> dict[str, str]:
return apply_default_headers(
normalize_headers(provider_config.get("custom_headers", {})),
{"User-Agent": ASTRBOT_USER_AGENT},
)
@classmethod
def _truncate_error_text_candidate(cls, text: str) -> str:
if len(text) <= cls._ERROR_TEXT_CANDIDATE_MAX_CHARS:
@@ -498,16 +507,10 @@ class ProviderOpenAIOfficial(Provider):
self.api_keys: list = super().get_keys()
self.chosen_api_key = self.api_keys[0] if len(self.api_keys) > 0 else None
self.timeout = provider_config.get("timeout", 120)
self.custom_headers = provider_config.get("custom_headers", {})
self.custom_headers = self._resolve_custom_headers(provider_config)
if isinstance(self.timeout, str):
self.timeout = int(self.timeout)
if not isinstance(self.custom_headers, dict) or not self.custom_headers:
self.custom_headers = None
else:
for key in self.custom_headers:
self.custom_headers[key] = str(self.custom_headers[key])
if "api_version" in provider_config:
# Using Azure OpenAI API
self.client = AsyncAzureOpenAI(

View File

@@ -0,0 +1,31 @@
from collections.abc import Mapping
def normalize_headers(headers: object) -> dict[str, str]:
if not isinstance(headers, dict):
return {}
return {str(key): str(value) for key, value in headers.items()}
def apply_default_headers(
headers: dict[str, str],
default_headers: Mapping[str, str],
) -> dict[str, str]:
merged_headers = dict(headers)
for default_name, default_value in default_headers.items():
existing_name = next(
(
header_name
for header_name in merged_headers
if header_name.lower() == default_name.lower()
),
None,
)
if existing_name is None:
merged_headers[default_name] = default_value
continue
if merged_headers[existing_name].strip():
continue
merged_headers.pop(existing_name)
merged_headers[default_name] = default_value
return merged_headers

View File

@@ -95,7 +95,7 @@ def _validate_template_list(value, meta, path_key, errors, validate_fn) -> None:
validate_fn(
item,
template_meta.get("items", {}),
path=f"{path_key}.templates.{template_key}.",
path=f"{item_path}.",
)

View File

@@ -16,7 +16,6 @@ def get_schema_item(schema: dict | None, key_path: str) -> dict | None:
同时支持:
- 扁平 schema直接 key 命中)
- 嵌套 object schema{type: "object", items: {...}}
- template_list schema<field>.templates.<template>.items
"""
if not isinstance(schema, dict) or not key_path:
@@ -24,31 +23,17 @@ def get_schema_item(schema: dict | None, key_path: str) -> dict | None:
if key_path in schema:
return schema.get(key_path)
parts = key_path.split(".")
current = schema
idx = 0
while idx < len(parts):
part = parts[idx]
parts = key_path.split(".")
for idx, part in enumerate(parts):
if part not in current:
return None
meta = current.get(part)
if idx == len(parts) - 1:
return meta
if not isinstance(meta, dict) or meta.get("type") != "object":
if not isinstance(meta, dict) or meta.get("type") != "template_list":
return None
if idx + 2 >= len(parts) or parts[idx + 1] != "templates":
return None
template_meta = meta.get("templates", {}).get(parts[idx + 2])
if not isinstance(template_meta, dict):
return None
if idx + 2 == len(parts) - 1:
return template_meta
current = template_meta.get("items", {})
idx += 3
continue
return None
current = meta.get("items", {})
idx += 1
return None

View File

@@ -25,25 +25,13 @@ const props = defineProps({
searchKeyword: {
type: String,
default: ''
},
pluginName: {
type: String,
default: ''
},
pluginI18n: {
type: Object,
default: () => ({})
},
pathPrefix: {
type: String,
default: ''
}
})
const { t } = useI18n()
const { getRaw } = useModuleI18n('features/config-metadata')
const { tm: tmConfig } = useModuleI18n('features/config')
const { translateIfKey, resolveConfigText } = useConfigTextResolver(props)
const { translateIfKey } = useConfigTextResolver()
const hintMarkdown = new MarkdownIt({
linkify: true,
@@ -126,18 +114,6 @@ function createSelectorModel(selector) {
})
}
function getItemPath(key) {
return props.pathPrefix ? `${props.pathPrefix}.${key}` : key
}
function getItemDescription(itemKey, itemMeta) {
return resolveConfigText(getItemPath(itemKey), 'description', itemMeta?.description) || itemKey
}
function getItemHint(itemKey, itemMeta) {
return resolveConfigText(getItemPath(itemKey), 'hint', itemMeta?.hint)
}
function openEditorDialog(key, value, theme, language) {
currentEditingKey.value = key
currentEditingLanguage.value = language || 'json'
@@ -167,8 +143,8 @@ function shouldShowItem(itemMeta, itemKey) {
const searchableText = [
itemKey,
getItemDescription(itemKey, itemMeta),
getItemHint(itemKey, itemMeta)
translateIfKey(itemMeta?.description || ''),
translateIfKey(itemMeta?.hint || '')
].join(' ').toLowerCase()
return searchableText.includes(keyword)
@@ -283,13 +259,13 @@ function getSpecialSubtype(value) {
<v-col cols="12" sm="6" class="property-info">
<v-list-item density="compact">
<v-list-item-title class="property-name">
{{ getItemDescription(itemKey, itemMeta) }}
{{ translateIfKey(itemMeta?.description) || itemKey }}
<span class="property-key">({{ itemKey }})</span>
</v-list-item-title>
<v-list-item-subtitle class="property-hint">
<span v-if="itemMeta?.obvious_hint && itemMeta?.hint" class="important-hint"></span>
<span v-html="renderHint(getItemHint(itemKey, itemMeta))"></span>
<span v-html="renderHint(itemMeta?.hint)"></span>
</v-list-item-subtitle>
</v-list-item>
</v-col>
@@ -298,18 +274,12 @@ function getSpecialSubtype(value) {
v-if="itemMeta?.type === 'template_list'"
v-model="createSelectorModel(itemKey).value"
:templates="itemMeta?.templates || {}"
:plugin-name="pluginName"
:plugin-i18n="pluginI18n"
:config-path="getItemPath(itemKey)"
class="config-field"
/>
<ConfigItemRenderer
v-else
v-model="createSelectorModel(itemKey).value"
:item-meta="itemMeta || null"
:plugin-name="pluginName"
:plugin-i18n="pluginI18n"
:config-key="getItemPath(itemKey)"
:show-fullscreen-btn="!!itemMeta?.editor_mode"
@open-fullscreen="openEditorDialog(itemKey, iterable, itemMeta?.editor_theme, itemMeta?.editor_language)"
/>
@@ -369,13 +339,13 @@ function getSpecialSubtype(value) {
<v-col cols="12" sm="6" class="property-info">
<v-list-item density="compact">
<v-list-item-title class="property-name">
{{ getItemDescription(itemKey, itemMeta) }}
{{ translateIfKey(itemMeta?.description) || itemKey }}
<span class="property-key">({{ itemKey }})</span>
</v-list-item-title>
<v-list-item-subtitle class="property-hint">
<span v-if="itemMeta?.obvious_hint && itemMeta?.hint" class="important-hint"></span>
<span v-html="renderHint(getItemHint(itemKey, itemMeta))"></span>
<span v-html="renderHint(itemMeta?.hint)"></span>
</v-list-item-subtitle>
</v-list-item>
</v-col>
@@ -384,18 +354,12 @@ function getSpecialSubtype(value) {
v-if="itemMeta?.type === 'template_list'"
v-model="createSelectorModel(itemKey).value"
:templates="itemMeta?.templates || {}"
:plugin-name="pluginName"
:plugin-i18n="pluginI18n"
:config-path="getItemPath(itemKey)"
class="config-field"
/>
<ConfigItemRenderer
v-else
v-model="createSelectorModel(itemKey).value"
:item-meta="itemMeta || null"
:plugin-name="pluginName"
:plugin-i18n="pluginI18n"
:config-key="getItemPath(itemKey)"
:show-fullscreen-btn="!!itemMeta?.editor_mode"
@open-fullscreen="openEditorDialog(itemKey, iterable, itemMeta?.editor_theme, itemMeta?.editor_language)"
/>

View File

@@ -57,11 +57,8 @@
</v-btn>
<div class="d-flex flex-column">
<v-list-item-title class="property-name">{{ templateLabel(entry.__template_key) }}</v-list-item-title>
<v-list-item-subtitle class="property-hint entry-display-text" v-if="templateDisplayText(entry)">
{{ templateDisplayText(entry) }}
</v-list-item-subtitle>
<v-list-item-subtitle class="property-hint" v-if="templateHintText(entry)">
{{ templateHintText(entry) }}
<v-list-item-subtitle class="property-hint" v-if="getTemplate(entry)?.hint || getTemplate(entry)?.description">
{{ templateText(entry.__template_key, 'hint', getTemplate(entry)?.hint || getTemplate(entry)?.description) }}
</v-list-item-subtitle>
</div>
</div>
@@ -204,7 +201,6 @@ const defaultValueMap = {
string: '',
text: '',
list: [],
file: [],
object: {},
template_list: []
}
@@ -352,49 +348,6 @@ function getTemplate(entry) {
return props.templates?.[key] || null
}
function templateHintText(entry) {
const template = getTemplate(entry)
if (!template || template.hide_hint_in_list) return ''
return templateText(entry.__template_key, 'hint', template.hint || template.description || '')
}
function getItemMetaBySelector(itemsMeta = {}, selector = '') {
const keys = selector.split('.').filter(Boolean)
let currentItems = itemsMeta
let currentMeta = null
for (let i = 0; i < keys.length; i++) {
currentMeta = currentItems?.[keys[i]]
if (!currentMeta) return null
if (i < keys.length - 1) {
if (currentMeta.type !== 'object') return null
currentItems = currentMeta.items || {}
}
}
return currentMeta
}
function templateDisplayText(entry) {
const template = getTemplate(entry)
const displayItem = template?.display_item
if (!template || typeof displayItem !== 'string' || !displayItem) return ''
const displayMeta = getItemMetaBySelector(template.items || {}, displayItem)
if (displayMeta?.type !== 'string') return ''
const value = getValueBySelector(entry, displayItem)
if (typeof value !== 'string' || !value.trim()) return ''
const label = templateItemText(
entry.__template_key,
displayItem,
'description',
displayMeta.description || displayItem,
)
return `${label}: ${value.trim()}`
}
function getValueBySelector(obj, selector) {
const keys = selector.split('.')
let current = obj
@@ -497,11 +450,6 @@ function hasVisibleItemsAfter(entries, currentIndex, entry) {
margin-top: 2px;
}
.entry-display-text {
color: var(--v-theme-primary);
font-weight: 500;
}
.property-key {
font-size: 0.85em;
opacity: 0.7;

View File

@@ -146,14 +146,7 @@ Plugin developers can add a template-style configuration to `_conf_schema` in th
"template_1": {
"name": "Template One",
"hint":"hint",
"display_item": "attr_name",
"hide_hint_in_list": true,
"items": {
"attr_name": {
"description": "Attribute Name",
"type": "string",
"default": ""
},
"attr_a": {
"description": "Attribute A",
"type": "int",
@@ -194,7 +187,6 @@ Saved config example:
"field_id": [
{
"__template_key": "template_1",
"attr_name": "",
"attr_a": 10,
"attr_b": true
},
@@ -206,11 +198,6 @@ Saved config example:
]
```
Templates also support these optional fields:
- `display_item`: Specifies the key of a `string` item inside the template `items`. When set, the WebUI shows that field's current value in the collapsed list of added template entries, for example `Attribute Name: my-adapter`, making it easier to distinguish multiple entries created from the same template. Dot paths are supported for fields inside nested objects, for example `meta.name`.
- `hide_hint_in_list`: When set to `true`, the WebUI hides the template `hint` in the collapsed list of added template entries. The template selection dropdown still shows the `hint`, and hints for fields inside the expanded entry are not affected.
<img width="1000" alt="image" src="https://github.com/user-attachments/assets/74876d30-11a4-491b-a7a0-8ebe8d603782" />

View File

@@ -146,14 +146,7 @@ AstrBot 提供了“强大”的配置解析和可视化功能。能够让用户
"template_1": {
"name": "Template One",
"hint":"hint",
"display_item": "attr_name",
"hide_hint_in_list": true,
"items": {
"attr_name": {
"description": "Attribute Name",
"type": "string",
"default": ""
},
"attr_a": {
"description": "Attribute A",
"type": "int",
@@ -194,7 +187,6 @@ AstrBot 提供了“强大”的配置解析和可视化功能。能够让用户
"field_id": [
{
"__template_key": "template_1",
"attr_name": "",
"attr_a": 10,
"attr_b": true
},
@@ -206,11 +198,6 @@ AstrBot 提供了“强大”的配置解析和可视化功能。能够让用户
]
```
模板本身还支持以下可选字段:
- `display_item`: 指定模板 `items` 中一个 `string` 类型字段的 key。设置后WebUI 会在已添加模板条目的折叠列表中显示该字段当前值,例如 `Attribute Name: my-adapter`,便于添加多个同类型模板时快速区分。支持用点号选择嵌套 object 中的字段,例如 `meta.name`
- `hide_hint_in_list`: 设置为 `true`WebUI 会在已添加模板条目的折叠列表中隐藏该模板的 `hint`。添加模板时的下拉菜单仍会显示 `hint`,展开条目后各配置项自己的 `hint` 也不受影响。
<img width="1000" alt="image" src="https://github.com/user-attachments/assets/74876d30-11a4-491b-a7a0-8ebe8d603782" />
## 在插件中使用配置

View File

@@ -4,6 +4,7 @@ import pytest
import astrbot.core.provider.sources.anthropic_source as anthropic_source
import astrbot.core.provider.sources.kimi_code_source as kimi_code_source
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
from astrbot.core.provider.entities import LLMResponse
@@ -16,6 +17,25 @@ class _FakeAsyncAnthropic:
return None
def test_anthropic_provider_uses_astrbot_default_user_agent(monkeypatch):
monkeypatch.setattr(anthropic_source, "AsyncAnthropic", _FakeAsyncAnthropic)
provider = anthropic_source.ProviderAnthropic(
provider_config={
"id": "anthropic-test",
"type": "anthropic_chat_completion",
"model": "claude-test",
"key": ["test-key"],
},
provider_settings={},
)
assert provider.custom_headers == {"User-Agent": ASTRBOT_USER_AGENT}
assert provider.client.kwargs["default_headers"] == {
"User-Agent": ASTRBOT_USER_AGENT,
}
def test_anthropic_provider_passes_custom_headers_via_default_headers(monkeypatch):
monkeypatch.setattr(anthropic_source, "AsyncAnthropic", _FakeAsyncAnthropic)

View File

@@ -1,190 +0,0 @@
from pathlib import Path
import pytest
from click import ClickException
from click.testing import CliRunner
import astrbot.cli.utils.plugin as plugin_utils
from astrbot.cli.commands.cmd_plug import plug
def _write_plugin(path: Path, name: str = "astrbot_plugin_local_demo") -> None:
path.mkdir(parents=True)
(path / "metadata.yaml").write_text(
"\n".join(
[
f"name: {name}",
"desc: Local plugin",
"version: 1.0.0",
"author: AstrBot",
"repo: https://example.com/local-plugin",
],
),
encoding="utf-8",
)
(path / "main.py").write_text("PLUGIN_LOADED = True\n", encoding="utf-8")
def _write_ignored_plugin_files(path: Path) -> None:
for ignored_dir in [".git", ".venv", "__pycache__", ".idea", ".vscode", ".zed"]:
ignored_path = path / ignored_dir
ignored_path.mkdir()
(ignored_path / "ignored.txt").write_text("ignored\n", encoding="utf-8")
(path / "__pycache__" / "main.pyc").write_bytes(b"ignored")
def _write_astrbot_root(path: Path) -> None:
(path / ".astrbot").touch()
(path / "data" / "plugins").mkdir(parents=True)
def test_plugin_install_editable_symlinks_local_plugin(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
source = tmp_path / "source-plugin"
root.mkdir()
_write_astrbot_root(root)
_write_plugin(source)
monkeypatch.chdir(root)
result = CliRunner().invoke(
plug,
["install", "-e", str(source)],
catch_exceptions=False,
)
target = root / "data" / "plugins" / "astrbot_plugin_local_demo"
target = root / "data" / "plugins" / "astrbot_plugin_local_demo"
assert result.exit_code == 0
assert target.is_symlink()
assert (target / "metadata.yaml").exists()
assert (target / "main.py").read_text(encoding="utf-8") == "PLUGIN_LOADED = True\n"
def test_plugin_install_accepts_local_path_without_editable_flag(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
source = tmp_path / "source-plugin"
root.mkdir()
_write_astrbot_root(root)
_write_plugin(source)
_write_ignored_plugin_files(source)
monkeypatch.chdir(root)
result = CliRunner().invoke(plug, ["install", str(source)])
target = root / "data" / "plugins" / "astrbot_plugin_local_demo"
assert result.exit_code == 0
assert not target.is_symlink()
assert (target / "metadata.yaml").exists()
assert not (target / ".git").exists()
assert not (target / ".venv").exists()
assert not (target / "__pycache__").exists()
assert not (target / ".idea").exists()
assert not (target / ".vscode").exists()
assert not (target / ".zed").exists()
def test_plugin_install_editable_rejects_existing_plugin(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
source = tmp_path / "source-plugin"
root.mkdir()
_write_astrbot_root(root)
_write_plugin(source)
_write_plugin(root / "data" / "plugins" / "astrbot_plugin_local_demo")
monkeypatch.chdir(root)
result = CliRunner().invoke(plug, ["install", "-e", str(source)])
assert result.exit_code != 0
assert "already exists" in result.output
def test_plugin_install_rejects_plugin_name_with_path_separator(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
source = tmp_path / "source-plugin"
root.mkdir()
_write_astrbot_root(root)
_write_plugin(source, name="../bad_plugin")
monkeypatch.chdir(root)
result = CliRunner().invoke(plug, ["install", str(source)])
assert result.exit_code != 0
assert "invalid name" in result.output
assert not (root / "data" / "bad_plugin").exists()
def test_plugin_install_copy_does_not_delete_existing_target_on_race(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
source = tmp_path / "source-plugin"
root.mkdir()
_write_astrbot_root(root)
_write_plugin(source)
monkeypatch.chdir(root)
target = root / "data" / "plugins" / "astrbot_plugin_local_demo"
target.mkdir()
marker = target / "keep.txt"
marker.write_text("keep\n", encoding="utf-8")
result = CliRunner().invoke(plug, ["install", str(source)])
assert result.exit_code != 0
assert "already exists" in result.output
assert marker.read_text(encoding="utf-8") == "keep\n"
def test_plugin_install_copy_does_not_delete_concurrently_created_target(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
source = tmp_path / "source-plugin"
plugins_dir = tmp_path / "plugins"
_write_plugin(source)
target = plugins_dir / "astrbot_plugin_local_demo"
def create_target_then_fail(
_source_path: Path,
_plugins_dir: Path,
_target_path: Path,
) -> None:
target.mkdir(parents=True)
(target / "keep.txt").write_text("keep\n", encoding="utf-8")
raise FileExistsError
monkeypatch.setattr(plugin_utils, "_copy_local_plugin", create_target_then_fail)
with pytest.raises(ClickException, match="already exists"):
plugin_utils.install_local_plugin(source, plugins_dir)
assert (target / "keep.txt").read_text(encoding="utf-8") == "keep\n"
def test_plugin_install_requires_name_or_editable_path(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
root = tmp_path / "root"
root.mkdir()
_write_astrbot_root(root)
monkeypatch.chdir(root)
result = CliRunner().invoke(plug, ["install"])
assert result.exit_code != 0
assert "Missing plugin name or local plugin path" in result.output

View File

@@ -1,10 +1,48 @@
import pytest
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
import astrbot.core.provider.sources.gemini_source as gemini_source
from astrbot.core.provider.entities import LLMResponse
from astrbot.core.provider.sources.gemini_source import ProviderGoogleGenAI
class _FakeGenAIClient:
def __init__(self, **kwargs):
self.kwargs = kwargs
self._api_client = type(
"FakeAPIClient",
(),
{"_http_options": kwargs["http_options"]},
)()
self.aio = type("FakeAioClient", (), {"_api_client": self._api_client})()
@pytest.mark.asyncio
async def test_gemini_provider_uses_astrbot_default_user_agent(monkeypatch):
monkeypatch.setattr(gemini_source.genai, "Client", _FakeGenAIClient)
provider = ProviderGoogleGenAI(
provider_config={
"id": "gemini-test",
"type": "googlegenai_chat_completion",
"model": "gemini-test",
"key": ["test-key"],
"api_base": "https://generativelanguage.googleapis.com/",
},
provider_settings={},
)
try:
assert provider.custom_headers["user-agent"] == ASTRBOT_USER_AGENT
assert provider.client._api_client._http_options.headers["user-agent"] == (
ASTRBOT_USER_AGENT
)
assert provider._http_client.headers["user-agent"] == ASTRBOT_USER_AGENT
finally:
await provider.terminate()
def test_gemini_empty_output_raises_empty_model_output_error():
llm_response = LLMResponse(role="assistant")

View File

@@ -9,6 +9,7 @@ from openai.types.chat.chat_completion_chunk import ChatCompletionChunk
from PIL import Image as PILImage
import astrbot.core.provider.sources.openai_source as openai_source_module
from astrbot.core.config.default import ASTRBOT_USER_AGENT
from astrbot.core.exceptions import EmptyModelOutputError
from astrbot.core.provider.sources.groq_source import ProviderGroq
from astrbot.core.provider.sources.openai_source import ProviderOpenAIOfficial
@@ -26,6 +27,17 @@ class _ErrorWithResponse(Exception):
self.response = SimpleNamespace(text=response_text)
class _FakeChatCompletions:
def create(self):
return None
class _FakeAsyncOpenAI:
def __init__(self, **kwargs):
self.kwargs = kwargs
self.chat = SimpleNamespace(completions=_FakeChatCompletions())
def _make_provider(overrides: dict | None = None) -> ProviderOpenAIOfficial:
provider_config = {
"id": "test-openai",
@@ -56,6 +68,39 @@ def _make_groq_provider(overrides: dict | None = None) -> ProviderGroq:
)
def test_openai_provider_uses_astrbot_default_user_agent(monkeypatch):
monkeypatch.setattr(openai_source_module, "AsyncOpenAI", _FakeAsyncOpenAI)
provider = _make_provider()
assert provider.custom_headers == {"User-Agent": ASTRBOT_USER_AGENT}
assert provider.client.kwargs["default_headers"] == {
"User-Agent": ASTRBOT_USER_AGENT,
}
def test_openai_provider_preserves_custom_user_agent(monkeypatch):
monkeypatch.setattr(openai_source_module, "AsyncOpenAI", _FakeAsyncOpenAI)
provider = _make_provider(
{
"custom_headers": {
"User-Agent": "custom-agent/1.0",
"X-Test-Header": 123,
},
},
)
assert provider.custom_headers == {
"User-Agent": "custom-agent/1.0",
"X-Test-Header": "123",
}
assert provider.client.kwargs["default_headers"] == {
"User-Agent": "custom-agent/1.0",
"X-Test-Header": "123",
}
def test_create_http_client_uses_openai_httpx_module(monkeypatch):
captured: dict[str, object] = {}

View File

@@ -1,88 +0,0 @@
"""Tests for dashboard route utility helpers."""
from astrbot.dashboard.routes.config import validate_config
from astrbot.dashboard.routes.util import get_schema_item
def test_get_schema_item_template_list_file_item():
schema = {
"demo_templates": {
"type": "template_list",
"templates": {
"api_provider": {
"items": {
"tls_certificate_files": {"type": "file"},
},
},
},
},
}
meta = get_schema_item(
schema,
"demo_templates.templates.api_provider.tls_certificate_files",
)
assert meta == {"type": "file"}
def test_get_schema_item_nested_template_list_file_item():
schema = {
"group": {
"type": "object",
"items": {
"demo_templates": {
"type": "template_list",
"templates": {
"nested_profile": {
"items": {
"profile": {
"type": "object",
"items": {
"attachments": {"type": "file"},
},
},
},
},
},
},
},
},
}
meta = get_schema_item(
schema,
"group.demo_templates.templates.nested_profile.profile.attachments",
)
assert meta == {"type": "file"}
def test_validate_config_template_list_file_path_uses_template_schema_path():
schema = {
"demo_templates": {
"type": "template_list",
"templates": {
"api_provider": {
"items": {
"tls_certificate_files": {"type": "file"},
},
},
},
},
}
data = {
"demo_templates": [
{
"__template_key": "api_provider",
"tls_certificate_files": [
"files/demo_templates/templates/api_provider/tls_certificate_files/cert.pem"
],
}
]
}
errors, validated = validate_config(data, schema, is_core=False)
assert errors == []
assert validated == data