ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-35-C' into simp/r3-35

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:44:13 -07:00
parent 20e4603bf7 54df4e53d4
commit 09fa4f5d26
4 changed files with 331 additions and 552 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/x_search_tool.py
+93 -178
View File
@@ -1,19 +1,12 @@
#!/usr/bin/env python3
"""X Search tool backed by xAI's built-in ``x_search`` Responses API tool.
Registers when either xAI credential path is available (``XAI_API_KEY`` or
``hermes auth add xai-oauth``). At call time an explicit ``XAI_API_KEY`` wins
(``prefer_api_key=True``): x_search is API-metered and the subscription OAuth
bearer answers ``/v1/responses`` in a degraded no-citation mode (#88040).
Defensive output: ``from_date``/``to_date`` are validated client-side (strict
``YYYY-MM-DD``, ``from <= to``, ``from`` not in the future) so malformed windows
fail fast instead of burning a billable call. Successful responses carry
``degraded``/``degraded_reason``: True when a narrowing filter was active AND
xAI returned no citations in either channel, meaning the answer came from the
model's own knowledge rather than the X index.
Salvaged from PR #10786 (originally by @Jaaneek).
Registers when either xAI credential path is available (``XAI_API_KEY`` or ``hermes auth add
xai-oauth``). At call time an explicit ``XAI_API_KEY`` wins (``prefer_api_key=True``): x_search
is API-metered and the subscription OAuth bearer answers ``/v1/responses`` without citations.
Date filters are validated client-side so malformed windows fail fast instead of burning a
billable call. Results carry ``degraded``: True when a narrowing filter was active AND xAI
returned no citations in either channel (answer came from model knowledge, not the X index).
"""
from __future__ import annotations
@@ -38,10 +31,6 @@ X_SEARCH_REASONING_EFFORTS = ("low", "medium", "high", "xhigh")
MAX_HANDLES = 10
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
def _load_x_search_config() -> Dict[str, Any]:
try:
from hermes_cli.config import load_config
@@ -51,51 +40,29 @@ def _load_x_search_config() -> Dict[str, Any]:
return {}
def _get_x_search_model() -> str:
return str(_load_x_search_config().get("model") or "").strip() or DEFAULT_X_SEARCH_MODEL
def _get_x_search_reasoning_effort() -> Optional[str]:
raw_value = _load_x_search_config().get("reasoning_effort")
if raw_value is None or not str(raw_value).strip():
return None
effort = str(raw_value).strip().lower()
if effort not in X_SEARCH_REASONING_EFFORTS:
allowed = ", ".join(X_SEARCH_REASONING_EFFORTS)
raise ValueError(
f"x_search.reasoning_effort must be one of: {allowed} "
f"(got {raw_value!r})"
)
raise ValueError(f"x_search.reasoning_effort must be one of: {allowed} (got {raw_value!r})")
return effort
def _get_x_search_int(key: str, default: int, floor: int) -> int:
raw_value = _load_x_search_config().get(key, default)
try:
return max(floor, int(raw_value))
return max(floor, int(_load_x_search_config().get(key, default)))
except Exception:
return default
def _get_x_search_timeout_seconds() -> int:
return _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
def _get_x_search_retries() -> int:
return _get_x_search_int("retries", DEFAULT_X_SEARCH_RETRIES, 0)
# ---------------------------------------------------------------------------
# Credential resolution
# ---------------------------------------------------------------------------
def _resolve_xai_bearer() -> Tuple[str, str, str]:
"""Return ``(api_key, base_url, source)``; ``source`` is ``"xai-oauth"`` or ``"xai"``.
Raises ``RuntimeError`` when no credential is usable so a credential that
expires between registration and invocation yields a clean tool error, not
a 401. ``prefer_api_key=True``: see module docstring (#88040).
Raises ``RuntimeError`` when no credential is usable so a credential that expires between
registration and invocation yields a clean tool error, not a 401.
"""
creds = resolve_xai_http_credentials(prefer_api_key=True)
api_key = str(creds.get("api_key") or "").strip()
@@ -105,23 +72,17 @@ def _resolve_xai_bearer() -> Tuple[str, str, str]:
"to sign in with your SuperGrok subscription, or set XAI_API_KEY."
)
base_url = str(creds.get("base_url") or DEFAULT_XAI_BASE_URL).strip().rstrip("/")
source = str(creds.get("provider") or "xai")
return api_key, base_url, source
return api_key, base_url, str(creds.get("provider") or "xai")
def check_x_search_requirements() -> bool:
"""True when xAI credentials resolve to a non-empty bearer (OAuth auto-refreshed)."""
try:
creds = resolve_xai_http_credentials()
return bool(str(creds.get("api_key") or "").strip())
return bool(str(resolve_xai_http_credentials().get("api_key") or "").strip())
except Exception:
return False
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _normalize_handles(handles: Optional[List[str]], field_name: str) -> List[str]:
cleaned = [h for h in (str(handle or "").strip().lstrip("@") for handle in handles or []) if h]
if len(cleaned) > MAX_HANDLES:
@@ -129,34 +90,30 @@ def _normalize_handles(handles: Optional[List[str]], field_name: str) -> List[st
return cleaned
def _parse_iso_date(value: str, field_name: str) -> date:
"""Parse a strict YYYY-MM-DD string (xAI silently accepts malformed dates and returns no citations)."""
def _parse_iso_date(value: str, field_name: str) -> Optional[date]:
"""Strict YYYY-MM-DD or None for blank (xAI silently accepts malformed dates and returns no citations)."""
raw = value.strip()
if not raw:
return None
try:
return datetime.strptime(raw, "%Y-%m-%d").date()
except ValueError as exc:
raise ValueError(
f"{field_name} must be YYYY-MM-DD (got {raw!r})"
) from exc
raise ValueError(f"{field_name} must be YYYY-MM-DD (got {raw!r})") from exc
def _validate_date_range(from_date: str, to_date: str) -> None:
"""Both parse as YYYY-MM-DD; from <= to; from not after today UTC (to may be in the future)."""
parsed_from = _parse_iso_date(from_date, "from_date") if from_date.strip() else None
parsed_to = _parse_iso_date(to_date, "to_date") if to_date.strip() else None
parsed_from, parsed_to = _parse_iso_date(from_date, "from_date"), _parse_iso_date(to_date, "to_date")
if parsed_from and parsed_to and parsed_from > parsed_to:
raise ValueError(
f"from_date ({parsed_from.isoformat()}) must be on or before "
f"to_date ({parsed_to.isoformat()})"
f"from_date ({parsed_from.isoformat()}) must be on or before to_date ({parsed_to.isoformat()})"
)
today_utc = datetime.now(timezone.utc).date()
if parsed_from is not None and parsed_from > today_utc:
raise ValueError(
f"from_date ({parsed_from.isoformat()}) is in the future; "
f"X Search only indexes past posts (today UTC is {today_utc.isoformat()})"
)
if parsed_from is not None:
today_utc = datetime.now(timezone.utc).date()
if parsed_from > today_utc:
raise ValueError(
f"from_date ({parsed_from.isoformat()}) is in the future; "
f"X Search only indexes past posts (today UTC is "
f"{today_utc.isoformat()})"
)
def _message_contents(payload: Dict[str, Any]):
@@ -169,14 +126,12 @@ def _extract_response_text(payload: Dict[str, Any]) -> str:
output_text = str(payload.get("output_text") or "").strip()
if output_text:
return output_text
parts: List[str] = []
for content in _message_contents(payload):
if content.get("type") in {"output_text", "text"}:
text = str(content.get("text") or "").strip()
if text:
parts.append(text)
return "\n\n".join(parts).strip()
parts = [
str(content.get("text") or "").strip()
for content in _message_contents(payload)
if content.get("type") in {"output_text", "text"}
]
return "\n\n".join(p for p in parts if p).strip()
def _extract_inline_citations(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
@@ -197,79 +152,76 @@ def _http_error_message(exc: requests.HTTPError) -> str:
response = getattr(exc, "response", None)
if response is None:
return str(exc)
try:
payload = response.json()
except Exception:
payload = None
if isinstance(payload, dict):
code = str(payload.get("code") or "").strip()
error = str(payload.get("error") or "").strip()
message = error or str(payload)
message = str(payload.get("error") or "").strip() or str(payload)
if code and code not in message:
message = f"{code}: {message}"
return message or str(exc)
text = str(getattr(response, "text", "") or "").strip()
if text:
return text[:500]
return str(exc)
return text[:500] if text else str(exc)
def _error_json(error: str, exc: BaseException) -> str:
return json.dumps(
{
"success": False,
"provider": "xai",
"tool": "x_search",
"error": error,
"error_type": type(exc).__name__,
},
ensure_ascii=False,
)
body = {"success": False, "provider": "xai", "tool": "x_search", "error": error}
return json.dumps({**body, "error_type": type(exc).__name__}, ensure_ascii=False)
def _post_with_retries(url: str, headers: Dict[str, str], payload: Dict[str, Any]) -> requests.Response:
"""POST with retries on 5xx / timeout / connection errors; re-raises the last failure."""
timeout_seconds = _get_x_search_timeout_seconds()
max_retries = _get_x_search_retries()
response: Optional[requests.Response] = None
timeout_seconds = _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
max_retries = _get_x_search_int("retries", DEFAULT_X_SEARCH_RETRIES, 0)
for attempt in range(max_retries + 1):
try:
response = requests.post(url, headers=headers, json=payload, timeout=timeout_seconds)
response.raise_for_status()
break
return response
except requests.HTTPError as e:
status_code = getattr(getattr(e, "response", None), "status_code", None)
if status_code is None or status_code < 500 or attempt >= max_retries:
raise
logger.warning(
"x_search upstream failure on attempt %s/%s: %s",
attempt + 1,
max_retries + 1,
_http_error_message(e),
)
time.sleep(min(5.0, 1.5 * (attempt + 1)))
kind, detail = "upstream", _http_error_message(e)
except (requests.ReadTimeout, requests.ConnectionError) as e:
if attempt >= max_retries:
raise
logger.warning(
"x_search transient failure on attempt %s/%s: %s",
attempt + 1,
max_retries + 1,
e,
)
time.sleep(min(5.0, 1.5 * (attempt + 1)))
if response is None:
raise RuntimeError("x_search request did not return a response")
return response
kind, detail = "transient", e
logger.warning("x_search %s failure on attempt %s/%s: %s", kind, attempt + 1, max_retries + 1, detail)
time.sleep(min(5.0, 1.5 * (attempt + 1)))
raise RuntimeError("x_search request did not return a response")
# ---------------------------------------------------------------------------
# Tool implementation
# ---------------------------------------------------------------------------
def _build_x_search_tool_def(
allowed_x_handles, excluded_x_handles, from_date: str, to_date: str,
enable_image_understanding: bool, enable_video_understanding: bool,
) -> Tuple[Dict[str, Any], List[str]]:
"""Return ``(tool_def, active_filters)``; raises ValueError on invalid filters."""
allowed = _normalize_handles(allowed_x_handles, "allowed_x_handles")
excluded = _normalize_handles(excluded_x_handles, "excluded_x_handles")
if allowed and excluded:
raise ValueError("allowed_x_handles and excluded_x_handles cannot be used together")
_validate_date_range(from_date, to_date)
tool_def: Dict[str, Any] = {"type": "x_search"}
active_filters: List[str] = []
for key, value in (
("allowed_x_handles", allowed),
("excluded_x_handles", excluded),
("from_date", from_date.strip()),
("to_date", to_date.strip()),
):
if value:
tool_def[key] = value
active_filters.append(key)
if enable_image_understanding:
tool_def["enable_image_understanding"] = True
if enable_video_understanding:
tool_def["enable_video_understanding"] = True
return tool_def, active_filters
def x_search_tool(
query: str,
@@ -282,66 +234,38 @@ def x_search_tool(
) -> str:
if not query or not query.strip():
return tool_error("query is required for x_search")
try:
api_key, base_url, source = _resolve_xai_bearer()
except RuntimeError as exc:
return tool_error(str(exc))
try:
allowed = _normalize_handles(allowed_x_handles, "allowed_x_handles")
excluded = _normalize_handles(excluded_x_handles, "excluded_x_handles")
if allowed and excluded:
return tool_error("allowed_x_handles and excluded_x_handles cannot be used together")
try:
_validate_date_range(from_date, to_date)
tool_def, active_filters = _build_x_search_tool_def(
allowed_x_handles, excluded_x_handles, from_date, to_date,
enable_image_understanding, enable_video_understanding,
)
reasoning_effort = _get_x_search_reasoning_effort()
except ValueError as exc:
return tool_error(str(exc))
from_date, to_date = from_date.strip(), to_date.strip()
tool_def: Dict[str, Any] = {"type": "x_search"}
active_filters: List[str] = []
for key, value in (
("allowed_x_handles", allowed),
("excluded_x_handles", excluded),
("from_date", from_date),
("to_date", to_date),
):
if value:
tool_def[key] = value
active_filters.append(key)
if enable_image_understanding:
tool_def["enable_image_understanding"] = True
if enable_video_understanding:
tool_def["enable_video_understanding"] = True
payload = {
"model": _get_x_search_model(),
"model": str(_load_x_search_config().get("model") or "").strip() or DEFAULT_X_SEARCH_MODEL,
"input": [{"role": "user", "content": query.strip()}],
"tools": [tool_def],
"store": False,
}
if reasoning_effort:
payload["reasoning"] = {"effort": reasoning_effort}
response = _post_with_retries(
f"{base_url}/responses",
{
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"User-Agent": hermes_xai_user_agent(),
},
payload,
)
data = response.json()
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"User-Agent": hermes_xai_user_agent(),
}
data = _post_with_retries(f"{base_url}/responses", headers, payload).json()
citations = list(data.get("citations") or [])
inline_citations = _extract_inline_citations(data)
# xAI returns 200 with a synthesized answer even when no posts match the
# narrowing filters; with both citation channels empty the answer came
# from training data, so flag it as degraded.
# xAI returns 200 with a synthesized answer even when no posts match the narrowing
# filters; with both citation channels empty the answer came from training data.
degraded = bool(active_filters) and not citations and not inline_citations
return json.dumps(
{
@@ -356,9 +280,7 @@ def x_search_tool(
"inline_citations": inline_citations,
"degraded": degraded,
"degraded_reason": (
f"no citations returned despite filters: {', '.join(active_filters)}"
if degraded
else None
f"no citations returned despite filters: {', '.join(active_filters)}" if degraded else None
),
},
ensure_ascii=False,
@@ -368,7 +290,8 @@ def x_search_tool(
return _error_json(_http_error_message(e), e)
except requests.ReadTimeout as e:
logger.error("x_search timed out: %s", e, exc_info=True)
return _error_json(f"xAI x_search timed out after {_get_x_search_timeout_seconds()} seconds", e)
timeout = _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
return _error_json(f"xAI x_search timed out after {timeout} seconds", e)
except Exception as e:
logger.error("x_search failed: %s", e, exc_info=True)
return _error_json(str(e), e)
@@ -428,23 +351,15 @@ X_SEARCH_SCHEMA = {
def _handle_x_search(args, **kw):
return x_search_tool(
query=args.get("query", ""),
allowed_x_handles=args.get("allowed_x_handles"),
excluded_x_handles=args.get("excluded_x_handles"),
from_date=args.get("from_date", ""),
to_date=args.get("to_date", ""),
enable_image_understanding=bool(args.get("enable_image_understanding", False)),
enable_video_understanding=bool(args.get("enable_video_understanding", False)),
args.get("query", ""), args.get("allowed_x_handles"), args.get("excluded_x_handles"),
args.get("from_date", ""), args.get("to_date", ""),
bool(args.get("enable_image_understanding", False)),
bool(args.get("enable_video_understanding", False)),
)
registry.register(
name="x_search",
toolset="x_search",
schema=X_SEARCH_SCHEMA,
handler=_handle_x_search,
check_fn=check_x_search_requirements,
requires_env=["XAI_API_KEY"],
emoji="🐦",
name="x_search", toolset="x_search", schema=X_SEARCH_SCHEMA, handler=_handle_x_search,
check_fn=check_x_search_requirements, requires_env=["XAI_API_KEY"], emoji="🐦",
max_result_size_chars=100_000,
)
tools/xai_http.py
+39 -64
View File
@@ -21,14 +21,12 @@ def _dict_get(obj: Any, key: str) -> Any:
def has_xai_credentials() -> bool:
"""Cheap probe: True when xAI credentials are *likely* usable.
Deliberately avoids :func:`resolve_xai_http_credentials` (disk locks, OAuth
network refresh) — hot-paint callers and ``WebSearchProvider.is_available()``
must not do network I/O. Checks, fast-to-slow: ``XAI_API_KEY``; a non-empty
``providers.xai-oauth.tokens.access_token`` in ``auth.json``; any
``credential_pool.xai-oauth`` entry with an ``access_token`` (pool-only
multi-account grants never write the providers singleton). Returns False on
any exception so a corrupted auth store can't block other availability
scans; truthful refresh/expiry handling happens in the caller's request.
Deliberately avoids :func:`resolve_xai_http_credentials` (disk locks, OAuth network
refresh) — hot-paint callers and ``WebSearchProvider.is_available()`` must not do network
I/O. Checks, fast-to-slow: ``XAI_API_KEY``; ``providers.xai-oauth.tokens.access_token`` in
``auth.json``; any ``credential_pool.xai-oauth`` entry with an ``access_token`` (pool-only
multi-account grants never write the providers singleton). Returns False on any exception
so a corrupted auth store can't block other availability scans.
"""
try:
from agent.secret_scope import get_secret
@@ -49,11 +47,9 @@ def has_xai_credentials() -> bool:
if str(_dict_get(tokens, "access_token") or "").strip():
return True
entries = _dict_get(_dict_get(store, "credential_pool"), "xai-oauth")
if isinstance(entries, list):
for entry in entries:
if isinstance(entry, dict) and str(entry.get("access_token", "") or "").strip():
return True
return False
return isinstance(entries, list) and any(
isinstance(e, dict) and str(e.get("access_token", "") or "").strip() for e in entries
)
except Exception:
return False
@@ -61,14 +57,12 @@ def has_xai_credentials() -> bool:
def get_env_value(name: str, default=None):
"""Read ``name`` from ``~/.hermes/.env`` first, then ``os.environ``.
Wraps :func:`hermes_cli.config.get_env_value` so tests can patch
``tools.xai_http.get_env_value`` to inject dotenv-only secrets.
Wraps :func:`hermes_cli.config.get_env_value` so tests can patch ``tools.xai_http.get_env_value``.
"""
try:
from hermes_cli.config import get_env_value as _hermes_get_env_value
except ImportError:
return os.environ.get(name, default)
value = _hermes_get_env_value(name)
return value if value is not None else default
@@ -98,15 +92,17 @@ def _load_config_section(section_name: str) -> Dict[str, Any]:
return {}
_TRUE_WORDS = {"1", "true", "yes", "on", "enabled"}
_FALSE_WORDS = {"0", "false", "no", "off", "disabled"}
def _coerce_bool(value: Any, default: bool) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on", "enabled"}:
return True
if normalized in {"0", "false", "no", "off", "disabled"}:
return False
if normalized in _TRUE_WORDS | _FALSE_WORDS:
return normalized in _TRUE_WORDS
return default
@@ -124,18 +120,15 @@ def _coerce_expires_after(value: Any) -> Optional[int]:
return SAFE_XAI_STORAGE_EXPIRES_AFTER_SECONDS
if isinstance(value, (int, float)):
seconds = int(value)
if seconds <= 0:
return None
return min(seconds, MAX_XAI_STORAGE_EXPIRES_AFTER_SECONDS)
return None if seconds <= 0 else min(seconds, MAX_XAI_STORAGE_EXPIRES_AFTER_SECONDS)
return SAFE_XAI_STORAGE_EXPIRES_AFTER_SECONDS
def read_xai_imagine_storage_config(section_name: str) -> Dict[str, Any]:
"""Read ``<section_name>.xai.storage`` (``image_gen`` or ``video_gen``).
"""Read ``<section_name>.xai.storage`` (``image_gen``/``video_gen``) -> {enabled, public_url, expires_after}.
Shape: ``storage: {enabled: true, public_url: true, expires_after: null}``.
Storage is on by default so xAI returns permanent public URLs instead of
short-lived CDN URLs; ``expires_after: null`` means permanent.
Storage is on by default so xAI returns permanent public URLs instead of short-lived CDN
URLs; ``expires_after: null`` means permanent.
"""
storage = _dict_get(_dict_get(_load_config_section(section_name), "xai"), "storage")
storage = storage if isinstance(storage, dict) else {}
@@ -147,22 +140,17 @@ def read_xai_imagine_storage_config(section_name: str) -> Dict[str, Any]:
def build_xai_storage_options(
section_name: str,
*,
filename_prefix: str,
extension: str,
section_name: str, *, filename_prefix: str, extension: str,
) -> Optional[Dict[str, Any]]:
"""Return an xAI ``storage_options`` payload, or None when disabled."""
cfg = read_xai_imagine_storage_config(section_name)
if not cfg["enabled"]:
return None
ts = datetime.datetime.now(datetime.UTC).strftime("%Y%m%d-%H%M%S")
short = uuid.uuid4().hex[:8]
ext = extension.lstrip(".") or "bin"
payload: Dict[str, Any] = {
"filename": f"{filename_prefix}-{ts}-{short}.{ext}",
"public_url": bool(cfg["public_url"]),
"filename": f"{filename_prefix}-{ts}-{short}.{ext}", "public_url": bool(cfg["public_url"]),
}
if cfg["expires_after"] is not None:
payload["expires_after"] = cfg["expires_after"]
@@ -174,9 +162,8 @@ def xai_storage_notice_text(section_name: str) -> str:
cfg = read_xai_imagine_storage_config(section_name)
if not cfg["enabled"]:
return ""
if cfg["expires_after"] is None:
retention = "without an automatic expiry"
else:
retention = "without an automatic expiry"
if cfg["expires_after"] is not None:
days = cfg["expires_after"] / (24 * 60 * 60)
retention = f"for about {days:g} day{'s' if days != 1 else ''}"
return (
@@ -209,9 +196,8 @@ def maybe_mark_xai_storage_notice_seen(section_name: str) -> Optional[str]:
def _resolve_explicit_xai_api_key() -> str:
"""Read ``XAI_API_KEY`` via ``resolve_provider_secret`` (config → profile scope → env/.env → pool).
Both the preferred-key and the no-OAuth fallback paths go through here so
scope policy (incl. failing closed in a multiplexed gateway turn) is never
re-implemented per caller.
Both the preferred-key and the no-OAuth fallback paths go through here so scope policy
(incl. failing closed in a multiplexed gateway turn) is never re-implemented per caller.
"""
try:
from tools.tool_backend_helpers import resolve_provider_secret
@@ -227,12 +213,8 @@ def _xai_base_url_override() -> str:
def _resolve_explicit_xai_base_url(default: str = DEFAULT_XAI_BASE_URL) -> str:
"""Base URL for the explicit-API-key path.
Origin-pinned via ``_xai_validate_inference_base_url`` so a tampered env
override can't exfiltrate the bearer; on rejection falls back to ``default``
rather than raising.
"""
"""Base URL for the explicit-API-key path, origin-pinned via ``_xai_validate_inference_base_url``
so a tampered env override can't exfiltrate the bearer; rejection falls back to ``default``."""
override = _xai_base_url_override()
try:
import hermes_cli.auth as auth_mod
@@ -243,21 +225,17 @@ def _resolve_explicit_xai_base_url(default: str = DEFAULT_XAI_BASE_URL) -> str:
def resolve_xai_http_credentials(
*,
force_refresh: bool = False,
api_key_hint: Optional[str] = None,
prefer_api_key: bool = False,
*, force_refresh: bool = False, api_key_hint: Optional[str] = None, prefer_api_key: bool = False,
) -> Dict[str, str]:
"""Resolve bearer credentials for direct xAI HTTP endpoints.
Default order: Hermes-managed xAI OAuth, then ``XAI_API_KEY`` (read via
``get_env_value`` so ``~/.hermes/.env`` keys count). ``prefer_api_key=True``
inverts that for API-metered endpoints where the subscription OAuth bearer
authorizes but misbehaves (x_search answers without citations, TTS 403s).
Both branches honor ``HERMES_XAI_BASE_URL``/``XAI_BASE_URL`` behind the same
origin-pinning validation. ``force_refresh=True`` forces an OAuth refresh;
pass the rejected bearer as ``api_key_hint`` so a multi-account pool
refreshes the issuing entry, not whichever its strategy selects first.
Default order: Hermes-managed xAI OAuth, then ``XAI_API_KEY`` (via ``get_env_value`` so
``~/.hermes/.env`` keys count). ``prefer_api_key=True`` inverts that for API-metered
endpoints where the subscription OAuth bearer authorizes but misbehaves (x_search answers
without citations, TTS 403s). Both branches honor ``HERMES_XAI_BASE_URL``/``XAI_BASE_URL``
behind the same origin-pinning validation. ``force_refresh=True`` forces an OAuth refresh;
pass the rejected bearer as ``api_key_hint`` so a multi-account pool refreshes the issuing
entry, not whichever its strategy selects first.
"""
if prefer_api_key:
explicit_key = str(_resolve_explicit_xai_api_key() or "").strip()
@@ -275,18 +253,15 @@ def resolve_xai_http_credentials(
# the next healthy account rather than resurrecting the stale row.
entry = pool.select()
access_token = str(
getattr(entry, "runtime_api_key", None)
or getattr(entry, "access_token", "")
getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
).strip()
fallback_base_url = str(
getattr(entry, "runtime_base_url", None)
or getattr(entry, "base_url", "")
or auth_mod.DEFAULT_XAI_OAUTH_BASE_URL
).strip().rstrip("/")
base_url = auth_mod._xai_validate_inference_base_url(
_xai_base_url_override(),
fallback=fallback_base_url,
)
override = _xai_base_url_override()
base_url = auth_mod._xai_validate_inference_base_url(override, fallback=fallback_base_url)
if access_token:
return {"provider": "xai-oauth", "api_key": access_token, "base_url": base_url}
except Exception:
tools/xai_video_tools.py
+10 -30
View File
@@ -7,11 +7,7 @@ import json
from typing import Any, Dict, Optional
from hermes_cli.config import load_config
from plugins.video_gen.xai import (
has_xai_video_credentials,
run_xai_video_edit,
run_xai_video_extend,
)
from plugins.video_gen.xai import has_xai_video_credentials, run_xai_video_edit, run_xai_video_extend
from tools.registry import registry, tool_error
@@ -29,14 +25,11 @@ def _check_xai_video_requirements() -> bool:
def _clean_string(value: Any) -> Optional[str]:
if isinstance(value, str) and value.strip():
return value.strip()
return None
return value.strip() if isinstance(value, str) and value.strip() else None
def _coerce_int(value: Any) -> Optional[int]:
# bool is rejected (unlike video_generation_tool._coerce_int) so
# duration=true never becomes 1.
# bool is rejected (unlike video_generation_tool._coerce_int) so duration=true never becomes 1.
if value is None or isinstance(value, bool):
return None
try:
@@ -48,9 +41,7 @@ def _coerce_int(value: Any) -> Optional[int]:
def _normalize_public_video_url(video_url: Any) -> Optional[str]:
"""Require a public HTTPS MP4 URL (``http``/``https`` only)."""
cleaned = _clean_string(video_url)
if cleaned and cleaned.lower().startswith(("http://", "https://")):
return cleaned
return None
return cleaned if cleaned and cleaned.lower().startswith(("http://", "https://")) else None
_VIDEO_URL_PARAM = {
@@ -60,10 +51,7 @@ _VIDEO_URL_PARAM = {
"`public_url` from a prior xAI Imagine result."
),
}
_MODEL_PARAM = {
"type": "string",
"description": "Optional xAI Imagine model override.",
}
_MODEL_PARAM = {"type": "string", "description": "Optional xAI Imagine model override."}
def _xai_video_schema(name: str, verb: str, noun: str, prompt_verb: str, extra: Dict[str, Any]) -> Dict[str, Any]:
@@ -111,8 +99,6 @@ XAI_VIDEO_EXTEND_SCHEMA: Dict[str, Any] = _xai_video_schema(
def _run_xai_video_tool(args: Dict[str, Any], op: str, run, **extra: Any) -> str:
prompt = _clean_string(args.get("prompt"))
video_url = _normalize_public_video_url(args.get("video_url"))
model = _clean_string(args.get("model"))
if not prompt:
return tool_error(f"prompt is required for xAI video {op}")
if not video_url:
@@ -131,6 +117,7 @@ def _run_xai_video_tool(args: Dict[str, Any], op: str, run, **extra: Any) -> str
"provider": "xai",
})
model = _clean_string(args.get("model"))
return json.dumps(run(prompt=prompt, video_url=video_url, model=model, **extra))
@@ -139,9 +126,8 @@ def _handle_xai_video_edit(args: Dict[str, Any], **_kw: Any) -> str:
def _handle_xai_video_extend(args: Dict[str, Any], **_kw: Any) -> str:
return _run_xai_video_tool(
args, "extend", run_xai_video_extend, duration=_coerce_int(args.get("duration")),
)
duration = _coerce_int(args.get("duration"))
return _run_xai_video_tool(args, "extend", run_xai_video_extend, duration=duration)
for _name, _schema, _handler in (
@@ -149,12 +135,6 @@ for _name, _schema, _handler in (
("xai_video_extend", XAI_VIDEO_EXTEND_SCHEMA, _handle_xai_video_extend),
):
registry.register(
name=_name,
toolset="video_gen",
schema=_schema,
handler=_handler,
check_fn=_check_xai_video_requirements,
requires_env=[],
is_async=False,
emoji="video",
name=_name, toolset="video_gen", schema=_schema, handler=_handler,
check_fn=_check_xai_video_requirements, requires_env=[], is_async=False, emoji="video",
)
tools/yuanbao_tools.py
+189 -280
View File
@@ -8,9 +8,10 @@ The active adapter singleton lives in ``gateway.platforms.yuanbao.get_active_ada
from __future__ import annotations
import functools
import logging
from pathlib import Path
from typing import List, Optional, Tuple
from typing import Tuple
from tools.registry import registry, tool_result
@@ -26,6 +27,35 @@ MENTION_HINT = (
# Image extensions for media dispatch (mirrors MessageSender.IMAGE_EXTS)
_IMAGE_EXTS = frozenset({".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"})
class _YbError(Exception):
"""Tool-level failure; ``payload`` is the error envelope returned to the model."""
def __init__(self, msg: str, **extra):
super().__init__(msg)
self.payload = {"success": False, "error": msg, **extra}
def _err(msg: str) -> dict:
return {"success": False, "error": msg}
def _yb_tool(label: str):
"""Handler decorator: ``fn(args)`` → tool_result; ``_YbError`` → its envelope; else logged ``_err``."""
def deco(fn):
@functools.wraps(fn)
async def handler(args, **kw):
try:
return tool_result(await fn(args))
except _YbError as exc:
return tool_result(exc.payload)
except Exception as exc:
logger.exception("[yuanbao_tools] %s error", label)
return tool_result(_err(str(exc)))
return handler
return deco
def _get_active_adapter():
"""Lazy import to avoid ImportError when gateway.platforms.yuanbao is unavailable."""
try:
@@ -35,333 +65,193 @@ def _get_active_adapter():
return None
def _err(msg: str) -> dict:
return {"success": False, "error": msg}
def _adapter():
adapter = _get_active_adapter()
if adapter is None:
raise _YbError("Yuanbao adapter is not connected")
return adapter
def _session_env(name: str) -> str:
try:
from gateway.session_context import get_session_env
return get_session_env(name, "")
except Exception:
return ""
def _nick(m: dict) -> str:
return m.get("nickname", m.get("nick_name", ""))
async def _resolve_dm_recipient(adapter, group_code: str, name: str) -> Tuple[str, str, Optional[dict]]:
async def _members(adapter, group_code: str) -> list:
raw = await adapter.get_group_member_list(group_code)
if raw is None:
raise _YbError("get_group_member_list returned None")
return raw.get("members", [])
async def _resolve_dm_recipient(adapter, group_code: str, name: str) -> Tuple[str, str]:
"""Resolve ``name`` to (user_id, nickname) via the group member list.
Returns ``("", "", error_dict)`` on failure; >1 partial match yields candidates
for disambiguation instead of guessing.
>1 partial match raises with ``candidates`` for disambiguation instead of guessing.
"""
if not group_code:
return "", "", _err("group_code is required when user_id is not provided")
raise _YbError("group_code is required when user_id is not provided")
if not name:
return "", "", _err("name is required when user_id is not provided")
try:
raw = await adapter.get_group_member_list(group_code)
if raw is None:
return "", "", _err("get_group_member_list returned None")
filt = name.strip().lower()
matched = [
m for m in raw.get("members", [])
if filt in (m.get("nickname") or m.get("nick_name") or "").lower()
]
if not matched:
return "", "", _err(f'No member matching "{name}" found in group {group_code}.')
if len(matched) > 1:
return "", "", {
"success": False,
"error": f'Multiple members match "{name}". Please specify which one.',
"candidates": [
{"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched
],
}
m = matched[0]
return m.get("user_id", ""), m.get("nickname", m.get("nick_name", name)), None
except Exception as exc:
logger.exception("[yuanbao_tools] send_dm member lookup error")
return "", "", _err(str(exc))
raise _YbError("name is required when user_id is not provided")
filt = name.strip().lower()
matched = [
m for m in await _members(adapter, group_code)
if filt in (m.get("nickname") or m.get("nick_name") or "").lower()
]
if not matched:
raise _YbError(f'No member matching "{name}" found in group {group_code}.')
if len(matched) > 1:
raise _YbError(
f'Multiple members match "{name}". Please specify which one.',
candidates=[{"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched],
)
m = matched[0]
return m.get("user_id", ""), m.get("nickname", m.get("nick_name", name))
async def get_group_info(group_code: str) -> dict:
@_yb_tool("get_group_info")
async def get_group_info(args) -> dict:
"""查询群基本信息（群名、群主、成员数）。"""
group_code = args.get("group_code", "")
if not group_code:
return _err("group_code is required")
adapter = _get_active_adapter()
if adapter is None:
return _err("Yuanbao adapter is not connected")
try:
gi = await adapter.query_group_info(group_code)
if gi is None:
return _err("query_group_info returned None")
return {
"success": True,
"group_code": group_code,
"group_name": gi.get("group_name", ""),
"member_count": gi.get("member_count", 0),
"owner": {
"user_id": gi.get("owner_id", ""),
"nickname": gi.get("owner_nickname", ""),
},
"note": 'The group is called "派 (Pai)" in the app.',
}
except Exception as exc:
logger.exception("[yuanbao_tools] get_group_info error")
return _err(str(exc))
gi = await _adapter().query_group_info(group_code)
if gi is None:
return _err("query_group_info returned None")
return {
"success": True,
"group_code": group_code,
"group_name": gi.get("group_name", ""),
"member_count": gi.get("member_count", 0),
"owner": {"user_id": gi.get("owner_id", ""), "nickname": gi.get("owner_nickname", "")},
"note": 'The group is called "派 (Pai)" in the app.',
}
async def query_group_members(
group_code: str,
action: str = "list_all",
name: str = "",
mention: bool = False,
) -> dict:
@_yb_tool("query_group_members")
async def query_group_members(args) -> dict:
"""统一的群成员查询（对齐 TS query_session_members）。
action: find (按昵称模糊搜索; 无 name 时等同 list_all) / list_bots / list_all (默认).
"""
group_code, name = args.get("group_code", ""), args.get("name", "")
action = args.get("action", "list_all")
if not group_code:
return _err("group_code is required")
adapter = _get_active_adapter()
if adapter is None:
return _err("Yuanbao adapter is not connected")
try:
raw = await adapter.get_group_member_list(group_code)
if raw is None:
return _err("get_group_member_list returned None")
all_members = [
{
"user_id": m.get("user_id", ""),
"nickname": _nick(m),
"role": _USER_TYPE_LABEL.get(m.get("user_type", m.get("role", 0)), "unknown"),
}
for m in await _members(_adapter(), group_code)
]
if not all_members:
return _err("No members found in this group.")
all_members = [
{
"user_id": m.get("user_id", ""),
"nickname": _nick(m),
"role": _USER_TYPE_LABEL.get(m.get("user_type", m.get("role", 0)), "unknown"),
}
for m in raw.get("members", [])
]
if not all_members:
return _err("No members found in this group.")
hint = {"mention_hint": MENTION_HINT} if args.get("mention", False) else {}
hint = {"mention_hint": MENTION_HINT} if mention else {}
def _listing(ok: bool, msg: str, members: list) -> dict:
return {"success": ok, "msg": msg, "members": members, **hint}
def _listing(ok: bool, msg: str, members: list) -> dict:
return {"success": ok, "msg": msg, "members": members, **hint}
if action == "list_bots":
bots = [m for m in all_members if m["role"] in {"yuanbao_ai", "bot"}]
if not bots:
return _err("No bots found in this group.")
return _listing(True, f"Found {len(bots)} bot(s).", bots)
if action == "list_bots":
bots = [m for m in all_members if m["role"] in {"yuanbao_ai", "bot"}]
if not bots:
return _err("No bots found in this group.")
return _listing(True, f"Found {len(bots)} bot(s).", bots)
if action == "find" and name:
filt = name.strip().lower()
matched = [m for m in all_members if filt in m["nickname"].lower()]
if matched:
return _listing(True, f'Found {len(matched)} member(s) matching "{name}".', matched)
return _listing(False, f'No match for "{name}". All members listed below.', all_members)
if action == "find" and name:
filt = name.strip().lower()
matched = [m for m in all_members if filt in m["nickname"].lower()]
if matched:
return _listing(True, f'Found {len(matched)} member(s) matching "{name}".', matched)
return _listing(False, f'No match for "{name}". All members listed below.', all_members)
return _listing(True, f"Found {len(all_members)} member(s).", all_members)
except Exception as exc:
logger.exception("[yuanbao_tools] query_group_members error")
return _err(str(exc))
return _listing(True, f"Found {len(all_members)} member(s).", all_members)
async def search_sticker(query: str = "", limit: int = 10) -> dict:
@_yb_tool("search_sticker")
async def search_sticker(args) -> dict:
"""在内置贴纸表中按关键词模糊搜索，返回 Top-N 候选（空 query 返回前 N 条）。"""
from gateway.platforms.yuanbao_sticker import search_stickers
query, limit = args.get("query", ""), args.get("limit", 10)
try:
safe_limit = max(1, min(50, int(limit) if limit else 10))
except (TypeError, ValueError):
safe_limit = 10
try:
matches = search_stickers(query or "", limit=safe_limit)
except Exception as exc:
logger.exception("[yuanbao_tools] search_sticker error")
return _err(str(exc))
matches = search_stickers(query or "", limit=safe_limit)
return {
"success": True,
"query": query or "",
"count": len(matches),
"results": [
{
"sticker_id": s.get("sticker_id", ""),
"name": s.get("name", ""),
"description": s.get("description", ""),
"package_id": s.get("package_id", ""),
}
{k: s.get(k, "") for k in ("sticker_id", "name", "description", "package_id")}
for s in matches
],
}
async def send_sticker(
sticker: str = "",
chat_id: str = "",
reply_to: str = "",
) -> dict:
@_yb_tool("send_sticker")
async def send_sticker(args) -> dict:
"""向 chat_id（缺省取当前会话 HERMES_SESSION_CHAT_ID）发送一张内置贴纸（TIMFaceElem）。
``sticker``: 名称（如 "六六六"）或 sticker_id（如 "278"）；为空时随机发送。
``chat_id``: ``direct:{account_id}`` / ``group:{group_code}`` / 裸 account_id。
"""
from gateway.session_context import get_session_env
from gateway.platforms.yuanbao_sticker import (
get_sticker_by_id,
get_sticker_by_name,
get_random_sticker,
)
from gateway.platforms.yuanbao_sticker import get_sticker_by_id, get_sticker_by_name, get_random_sticker
target = (chat_id or "").strip() or get_session_env("HERMES_SESSION_CHAT_ID", "")
target = (args.get("chat_id", "") or "").strip() or _session_env("HERMES_SESSION_CHAT_ID")
if not target:
return _err("chat_id is required (no active yuanbao session detected)")
adapter = _get_active_adapter()
if adapter is None:
return _err("Yuanbao adapter is not connected")
adapter = _adapter()
raw = (sticker or "").strip()
sticker_obj: Optional[dict] = None
raw = (args.get("sticker", "") or "").strip()
if not raw:
sticker_obj = get_random_sticker()
else:
if raw.isdigit():
sticker_obj = get_sticker_by_id(raw)
sticker_obj = get_sticker_by_id(raw) if raw.isdigit() else None
if sticker_obj is None:
sticker_obj = get_sticker_by_name(raw)
if sticker_obj is None:
return _err(
f"Sticker not found: {raw!r}. "
f"Use search_sticker first to discover available stickers."
)
try:
result = await adapter.send_sticker(
chat_id=target,
sticker_name=sticker_obj.get("name", ""),
reply_to=reply_to or None,
)
except Exception as exc:
logger.exception("[yuanbao_tools] send_sticker error")
return _err(str(exc))
return _err(f"Sticker not found: {raw!r}. Use search_sticker first to discover available stickers.")
result = await adapter.send_sticker(
chat_id=target, sticker_name=sticker_obj.get("name", ""), reply_to=args.get("reply_to", "") or None,
)
if getattr(result, "success", False):
return {
"success": True,
"chat_id": target,
"sticker": {
"sticker_id": sticker_obj.get("sticker_id", ""),
"name": sticker_obj.get("name", ""),
},
"sticker": {"sticker_id": sticker_obj.get("sticker_id", ""), "name": sticker_obj.get("name", "")},
"message_id": getattr(result, "message_id", None),
"note": "Sticker delivered to the chat. If you have additional text to say, reply now; otherwise end your turn without generating text.",
}
return _err(getattr(result, "error", "send_sticker failed"))
async def send_dm(
group_code: str,
name: str,
message: str,
user_id: str = "",
media_files: Optional[List[Tuple[str, bool]]] = None,
) -> dict:
@_yb_tool("send_dm")
async def send_dm(args) -> dict:
"""Send a DM to a group member, with optional media.
Without ``user_id`` the member list of ``group_code`` is searched by ``name``
(partial, case-insensitive; >1 match returns candidates). Text goes via
adapter.send_dm; media_files (path, is_voice) go via send_image_file (images) or
send_document. Partial media failures are reported in ``note``, not as failure.
group_code defaults to the session's "group:<code>" chat_id. Without ``user_id`` the member
list is searched by ``name`` (partial, case-insensitive; >1 match returns candidates).
media_files items are {"path", "is_voice"} dicts or (path, is_voice) pairs; ``MEDIA:<path>``
tags in the text count too. Partial media failures are reported in ``note``, not as failure.
"""
if not message and not media_files:
return _err("message or media_files is required")
adapter = _get_active_adapter()
if adapter is None:
return _err("Yuanbao adapter is not connected")
resolved_user_id = user_id.strip() if user_id else ""
resolved_nickname = name.strip()
if not resolved_user_id:
resolved_user_id, resolved_nickname, err = await _resolve_dm_recipient(adapter, group_code, name)
if err is not None:
return err
if not resolved_user_id:
return _err("Could not resolve user_id")
chat_id = f"direct:{resolved_user_id}"
last_result = None
errors: list[str] = []
try:
if message and message.strip():
last_result = await adapter.send_dm(resolved_user_id, message, group_code=group_code)
if not last_result.success:
errors.append(last_result.error or "text send failed")
for media_path, _is_voice in media_files or []:
send = adapter.send_image_file if Path(media_path).suffix.lower() in _IMAGE_EXTS else adapter.send_document
last_result = await send(chat_id, media_path, group_code=group_code)
if not last_result.success:
errors.append(last_result.error or "media send failed")
if last_result is None:
return _err("No deliverable text or media remained")
if errors and not last_result.success:
return _err("; ".join(errors))
result = {
"success": True,
"user_id": resolved_user_id,
"nickname": resolved_nickname,
"message_id": last_result.message_id,
"note": f'DM sent to "{resolved_nickname}" successfully.',
}
if errors:
result["note"] += f" (partial failure: {'; '.join(errors)})"
return result
except Exception as exc:
logger.exception("[yuanbao_tools] send_dm error")
return _err(str(exc))
# ---------------------------------------------------------------------------
# Registry registration
# ---------------------------------------------------------------------------
def _check_yuanbao():
"""Toolset availability check — True when running in a yuanbao gateway session."""
try:
from gateway.session_context import get_session_env
if get_session_env("HERMES_SESSION_PLATFORM", "") == "yuanbao":
return True
except Exception:
pass
return _get_active_adapter() is not None
async def _handle_yb_query_group_info(args, **kw):
return tool_result(await get_group_info(group_code=args.get("group_code", "")))
async def _handle_yb_query_group_members(args, **kw):
return tool_result(await query_group_members(
group_code=args.get("group_code", ""),
action=args.get("action", "list_all"),
name=args.get("name", ""),
mention=bool(args.get("mention", False)),
))
async def _handle_yb_send_dm(args, **kw):
# group_code: explicit arg, else the session's "group:<code>" chat_id.
group_code = args.get("group_code", "")
if not group_code:
try:
from gateway.session_context import get_session_env
chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
if chat_id.startswith("group:"):
group_code = chat_id.split(":", 1)[1]
except Exception:
pass
chat_id = _session_env("HERMES_SESSION_CHAT_ID")
if chat_id.startswith("group:"):
group_code = chat_id.split(":", 1)[1]
# media_files items: {"path", "is_voice"} dicts or (path, is_voice) pairs.
media_files = []
for item in args.get("media_files") or []:
if isinstance(item, dict):
@@ -369,32 +259,56 @@ async def _handle_yb_send_dm(args, **kw):
elif isinstance(item, (list, tuple)) and len(item) >= 2:
media_files.append((str(item[0]), bool(item[1])))
# LLMs often embed MEDIA:<path> tags in the text instead of using media_files.
from gateway.platforms.base import BasePlatformAdapter
embedded_media, message = BasePlatformAdapter.extract_media(args.get("message", ""))
if embedded_media:
media_files.extend(embedded_media)
media_files.extend(embedded_media or [])
media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
return tool_result(await send_dm(
group_code=group_code,
name=args.get("name", ""),
message=message,
user_id=args.get("user_id", ""),
media_files=media_files or None,
))
if not message and not media_files:
return _err("message or media_files is required")
adapter = _adapter()
name, user_id = args.get("name", ""), args.get("user_id", "")
resolved_user_id, resolved_nickname = user_id.strip() if user_id else "", name.strip()
if not resolved_user_id:
resolved_user_id, resolved_nickname = await _resolve_dm_recipient(adapter, group_code, name)
if not resolved_user_id:
return _err("Could not resolve user_id")
chat_id = f"direct:{resolved_user_id}"
last_result = None
errors: list[str] = []
if message and message.strip():
last_result = await adapter.send_dm(resolved_user_id, message, group_code=group_code)
if not last_result.success:
errors.append(last_result.error or "text send failed")
for media_path, _is_voice in media_files:
send = adapter.send_image_file if Path(media_path).suffix.lower() in _IMAGE_EXTS else adapter.send_document
last_result = await send(chat_id, media_path, group_code=group_code)
if not last_result.success:
errors.append(last_result.error or "media send failed")
if last_result is None:
return _err("No deliverable text or media remained")
if errors and not last_result.success:
return _err("; ".join(errors))
result = {
"success": True,
"user_id": resolved_user_id,
"nickname": resolved_nickname,
"message_id": last_result.message_id,
"note": f'DM sent to "{resolved_nickname}" successfully.',
}
if errors:
result["note"] += f" (partial failure: {'; '.join(errors)})"
return result
async def _handle_yb_search_sticker(args, **kw):
return tool_result(await search_sticker(query=args.get("query", ""), limit=args.get("limit", 10)))
async def _handle_yb_send_sticker(args, **kw):
return tool_result(await send_sticker(
sticker=args.get("sticker", ""),
chat_id=args.get("chat_id", ""),
reply_to=args.get("reply_to", ""),
))
def _check_yuanbao():
"""Toolset availability check — True when running in a yuanbao gateway session."""
return _session_env("HERMES_SESSION_PLATFORM") == "yuanbao" or _get_active_adapter() is not None
# (schema, handler, emoji); the tool name is schema["name"].
@@ -417,7 +331,7 @@ _TOOLS = (
"required": ["group_code"],
},
},
_handle_yb_query_group_info,
get_group_info,
"👥",
),
(
@@ -464,7 +378,7 @@ _TOOLS = (
"required": ["group_code", "action"],
},
},
_handle_yb_query_group_members,
query_group_members,
"📋",
),
(
@@ -532,7 +446,7 @@ _TOOLS = (
"required": [],
},
},
_handle_yb_send_dm,
send_dm,
"✉️",
),
(
@@ -563,7 +477,7 @@ _TOOLS = (
"required": [],
},
},
_handle_yb_search_sticker,
search_sticker,
"🔍",
),
(
@@ -606,18 +520,13 @@ _TOOLS = (
"required": [],
},
},
_handle_yb_send_sticker,
send_sticker,
"🎨",
),
)
for _schema, _handler, _emoji in _TOOLS:
registry.register(
name=_schema["name"],
toolset="hermes-yuanbao",
schema=_schema,
handler=_handler,
check_fn=_check_yuanbao,
is_async=True,
emoji=_emoji,
name=_schema["name"], toolset="hermes-yuanbao", schema=_schema, handler=_handler,
check_fn=_check_yuanbao, is_async=True, emoji=_emoji,
)