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

refactor(tools): MCP handlers inline single-use render helpers, drop banners; registration foreign-owner log inlined; body blank squeeze

Browse Source
This commit is contained in:
Teknium
2026-09-03 01:31:18 -07:00
parent 58a993a54d
commit bccfd1de26
7 changed files with 23 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tools/mcp_oauth_manager.py
View File

@@ -111,7 +111,6 @@ class HermesMCPOAuthProvider(HermesProviderMixin, *_SDK_BASES):
except httpx.HTTPError as exc:
logger.debug("MCP OAuth '%s': %s discovery to %s failed: %s", self._hermes_server_name, label, url, exc)
return None
async with httpx.AsyncClient(timeout=10.0) as client:
# PRM discovery to learn the authorization_server URL.
for url in build_protected_resource_metadata_discovery_urls(None, server_url):
@@ -240,7 +239,6 @@ class HermesMCPOAuthProvider(HermesProviderMixin, *_SDK_BASES):
import anyio
with anyio.CancelScope(shield=True):
await self.context.lock.acquire()
if retry_after_concurrent_auth:
yield request
self._persist_oauth_metadata_if_changed()

2
tools/mcp_tool_config.py
View File

@@ -147,7 +147,6 @@ def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
resolved_command = which_hit
elif resolved_command in {"npx", "npm", "node"}:
resolved_command = _node_fallback(resolved_command)
command_dir = os.path.dirname(resolved_command)
if command_dir:
resolved_env = _prepend_path(resolved_env, command_dir)
@@ -204,7 +203,6 @@ def _warn_hidden_whitespace(server_name: str, config: dict) -> List[str]:
elif isinstance(value, list):
for i, v in enumerate(value):
_walk(v, f"{path}[{i}]")
_walk(config, "")
for key_path in flagged:
if (server_name, key_path) not in _whitespace_warned:

5
tools/mcp_tool_errors.py
View File

@@ -90,7 +90,6 @@ def _validate_remote_mcp_url(server_name: str, url: Any) -> str:
stdio servers use ``command`` — or empty host)."""
def _bad(detail: str) -> InvalidMcpUrlError:
return InvalidMcpUrlError(f"Invalid MCP URL for '{server_name}': {detail}")
if not isinstance(url, str):
raise _bad(f"expected a string, got {type(url).__name__}")
stripped = url.strip()
@@ -126,7 +125,6 @@ def _resolve_client_cert(server_name: str, config: dict):
if not os.path.isfile(expanded):
raise FileNotFoundError(f"{prefix}{label} not found at {expanded!r}")
return expanded
if not isinstance(raw_cert, (list, tuple)):
cert_path = _expand(raw_cert, "client_cert")
return (cert_path, _expand(raw_key, "client_key")) if raw_key is not None else cert_path # combined PEM
@@ -153,7 +151,6 @@ def _resolve_identity_header(server_name: str, config: dict):
def _ignore(detail: str, *args):
logger.warning("MCP server '%s': identity_header " + detail + " — ignoring", server_name, *args)
return None
if not isinstance(raw, dict):
return _ignore("must be a mapping with 'name' and 'value'/'value_from' keys (got %s)", type(raw).__name__)
name = raw.get("name")
@@ -202,7 +199,6 @@ def _make_redirect_header_stripper(original_url, *, strict: bool = False,
for _name in configured_header_names if strict else ():
while _name in headers:
del headers[_name]
return _strip_on_cross_origin_redirect
@@ -228,7 +224,6 @@ def _format_connect_error(exc: BaseException) -> str:
text = "" if getattr(current, "exceptions", None) else str(current).strip()
messages = ([text] if text else []) + [m for child in _exc_children(current) for m in _flatten_messages(child)]
return messages or [current.__class__.__name__]
missing = _find_missing(exc)
if not missing:
return _sanitize_error("; ".join(list(dict.fromkeys(_flatten_messages(exc)))[:3]))

49
tools/mcp_tool_handlers.py
View File

@@ -33,8 +33,6 @@ _STDIO_DIED_AGAIN_MSG = (
"cleanly — do NOT retry this tool; ask the user to check the server's command and its stderr log.")
# --------------------------------------------------------------- pre-call gates
def _trust_gate_check(server_name: str, tool_name: str) -> Optional[str]:
"""Approval gate for write-capable tools on ``trust: untrusted`` servers. None to proceed,
else a ``tool_error``. Fail-closed: approval-system errors block."""
@@ -89,8 +87,6 @@ def _acquire_call_server(server_name: str, tool_timeout: float):
return None, not_connected
# ------------------------------------------------------------ breaker bookkeeping
def _result_is_error(result) -> bool:
"""True only for a JSON payload carrying an ``error`` key (non-JSON = success)."""
try:
@@ -138,8 +134,6 @@ def _retry_once(server_name: str, retry_call, op_description: str, what: str):
return result
# --------------------------------------------------------------- recovery ladder
def _handle_auth_error_and_retry(server_name: str, exc: BaseException, retry_call, op_description: str):
"""OAuth recovery + one retry; None when *exc* is not an auth error. ``handle_401`` decides
viability; if viable, signal a reconnect (fresh credentials), wait ready, retry once. Any
@@ -247,8 +241,6 @@ def _dispatch(server_name: str, server: Any, op: str, call, tool_timeout: float,
recoverers, on_final_failure, record_outcome=record_outcome)
# ------------------------------------------------------------- the RPC itself
@asynccontextmanager
async def _track_inflight_rpc(server: Any, server_name: str, op: str):
"""Register the running RPC so teardown can fail it fast. A deliberate teardown
@@ -298,14 +290,6 @@ async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str
await asyncio.gather(rpc_task, watch_task, return_exceptions=True)
# ---------------------------------------------------------- result rendering
def _error_result_text(result) -> str:
"""Concatenated text of an ``isError`` result's blocks (EmbeddedResource payloads: ``.resource.text``)."""
texts = (getattr(b, "text", None) or getattr(getattr(b, "resource", None), "text", None) for b in (result.content or []))
return "".join(str(t) for t in texts if t)
def _render_content_blocks(result, server_name: str) -> str:
"""Text passes through; image/audio blocks are cached (MEDIA: tags); resource blocks are materialized."""
parts: List[str] = []
@@ -325,23 +309,22 @@ def _render_content_blocks(result, server_name: str) -> str:
return _truncate_mcp_text_result("\n".join(parts)) # hard-cap pathological payloads; spillover handles the rest
def _capped_structured_content(result):
"""``structuredContent`` (or None); over the hard cap it degrades to the truncated JSON string (flood guard)."""
def _render_call_tool_result(result, server_name: str) -> str:
"""Pure: ``CallToolResult`` -> handler JSON. ``content`` is primary; ``structuredContent`` supplements it (or
becomes ``result`` without text) and over the hard cap degrades to the truncated JSON string (flood guard);
``_meta`` minus reserved keys. Error text also reads EmbeddedResource payloads (``.resource.text``)."""
if mcp_field(result, "is_error", "isError", False):
texts = (getattr(b, "text", None) or getattr(getattr(b, "resource", None), "text", None) for b in (result.content or []))
error_text = "".join(str(t) for t in texts if t) or "MCP tool returned an error"
return tool_error(_sanitize_error(_truncate_mcp_text_result(error_text)))
text_result = _render_content_blocks(result, server_name)
structured = mcp_field(result, "structured_content", "structuredContent")
try:
as_json = json.dumps(structured, ensure_ascii=False, default=str) if structured is not None else ""
if len(as_json) > _MCP_HARD_RESULT_CAP_CHARS:
structured = _truncate_mcp_text_result(as_json)
except (TypeError, ValueError):
return structured
return _truncate_mcp_text_result(as_json) if len(as_json) > _MCP_HARD_RESULT_CAP_CHARS else structured
def _render_call_tool_result(result, server_name: str) -> str:
"""Pure: ``CallToolResult`` -> handler JSON. ``content`` is primary; ``structuredContent`` supplements it (or
becomes ``result`` without text); ``_meta`` minus reserved keys."""
if mcp_field(result, "is_error", "isError", False):
return tool_error(_sanitize_error(_truncate_mcp_text_result(_error_result_text(result) or "MCP tool returned an error")))
text_result = _render_content_blocks(result, server_name)
structured = _capped_structured_content(result)
pass
meta = _strip_reserved_meta_keys(mcp_field(result, "meta", "meta"))
if structured is None and meta is None:
return json.dumps({"result": text_result}, ensure_ascii=False)
@@ -358,8 +341,6 @@ def _render_call_tool_result(result, server_name: str) -> str:
return json.dumps({"result": text_result}, ensure_ascii=False)
# ------------------------------------------------------------------- handlers
def _make_tool_handler(server_name: str, tool_name: str, tool_timeout: float):
"""Sync registry handler (``handler(args_dict, **kwargs) -> str``) calling an MCP tool via the background loop."""
op = f"tools/call {tool_name}"
@@ -387,12 +368,10 @@ def _make_tool_handler(server_name: str, tool_name: str, tool_timeout: float):
def _on_failure(exc):
_core._bump_server_error(server_name)
logger.error("MCP tool %s/%s call failed: %s", server_name, tool_name, exc)
return _dispatch(
server_name, server, op, _call, tool_timeout,
(_handle_stdio_child_exited_and_retry, _handle_auth_error_and_retry, _handle_session_expired_and_retry),
_on_failure, record_outcome=True)
return _handler
@@ -412,14 +391,11 @@ def _make_utility_handler(op: str, log_label: str, rpc, render, required: Option
async with server._rpc_lock:
result = await rpc(server.session, args, server_name)
return json.dumps(render(result, server_name), ensure_ascii=False)
return _dispatch(
server_name, server, op, _call, tool_timeout,
(_handle_auth_error_and_retry, _handle_session_expired_and_retry),
lambda exc: logger.error("MCP %s/%s failed: %s", server_name, log_label, exc))
return _handler
return _factory
@@ -501,5 +477,4 @@ def _make_check_fn(server_name: str):
server = _core._servers.get(server_name)
return ((server is not None and (server.session is not None or server._is_recycled_stdio()))
or server_name in _core._lazy_server_configs)
return _check

2
tools/mcp_tool_health.py
View File

@@ -66,7 +66,6 @@ class MCPServerHealthMixin:
await self._refresh_tools()
except Exception:
logger.exception("MCP server '%s': dynamic tool refresh failed", self.name)
task = asyncio.create_task(_run())
self._pending_refresh_tasks.add(task)
task.add_done_callback(self._pending_refresh_tasks.discard)
@@ -160,7 +159,6 @@ class MCPServerHealthMixin:
back to ``list_tools`` when the server advertises tools, else the -32601 propagates."""
async def list_tools():
await asyncio.wait_for(self.session.list_tools(), timeout=_KEEPALIVE_RPC_TIMEOUT)
if not self._ping_unsupported:
try:
await asyncio.wait_for(self.session.send_ping(), timeout=_KEEPALIVE_RPC_TIMEOUT)

28
tools/mcp_tool_registration.py
View File

@@ -85,7 +85,6 @@ def _select_utility_schemas(server_name: str, server: "MCPServerTask", config: d
return None
# Legacy gate (no initialize_result): the ClientSession method shares the handler key.
return None if hasattr(server.session, handler_key) else f"session lacks {handler_key}"
selected: List[dict] = []
for entry in _build_utility_schemas(server_name):
reason = _skip_reason(entry["handler_key"])
@@ -205,20 +204,6 @@ def _resolve_name_collisions(name: str, candidates: List[_Candidate]) -> List[_C
return [c for c in unique if c.registry_name not in ambiguous and (c.registry_name, c.origin) not in shadowed]
def _log_foreign_owner(name: str, c: _Candidate, existing_toolset: str, lazy: bool) -> None:
"""Diagnostics for a name already owned by another toolset (skipped to preserve the owner)."""
if lazy:
if not c.is_utility:
logger.warning("MCP server '%s' (lazy): cached tool '%s' collides with toolset '%s' — skipping",
name, c.registry_name, existing_toolset)
elif existing_toolset.startswith("mcp-"):
logger.error("MCP server '%s': %s normalizes to '%s', already owned by MCP toolset '%s' — skipping to "
"preserve the existing owner", name, c.origin, c.registry_name, existing_toolset)
else:
logger.warning("MCP server '%s': %s (→ '%s') collides with built-in tool in toolset '%s' — skipping to "
"preserve built-in", name, c.origin, c.registry_name, existing_toolset)
def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: Callable,
scope: Callable[[], Optional[str]], lazy: bool) -> List[str]:
"""Register candidates under toolset ``mcp-{name}``; returns the names that landed. The
@@ -229,8 +214,17 @@ def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: C
registered: List[str] = []
for c in candidates:
existing_toolset = registry.get_toolset_for_tool(c.registry_name)
if existing_toolset and existing_toolset != toolset_name:
_log_foreign_owner(name, c, existing_toolset, lazy)
if existing_toolset and existing_toolset != toolset_name: # foreign owner: skip, preserve it
if lazy:
if not c.is_utility:
logger.warning("MCP server '%s' (lazy): cached tool '%s' collides with toolset '%s' — skipping",
name, c.registry_name, existing_toolset)
elif existing_toolset.startswith("mcp-"):
logger.error("MCP server '%s': %s normalizes to '%s', already owned by MCP toolset '%s' — skipping to "
"preserve the existing owner", name, c.origin, c.registry_name, existing_toolset)
else:
logger.warning("MCP server '%s': %s (→ '%s') collides with built-in tool in toolset '%s' — skipping to "
"preserve built-in", name, c.origin, c.registry_name, existing_toolset)
continue
registry.register(
name=c.registry_name, toolset=toolset_name, schema=c.schema, handler=c.handler, check_fn=check_fn,

3
tools/mcp_tool_transport.py
View File

@@ -97,7 +97,6 @@ class MCPServerTransportMixin:
raise
logger.info(log_fmt, self.name, exc, *log_extra)
return await call(fallback)
mode = str((self._config or {}).get("protocol", "auto")).lower().strip()
if mode in ("stateless", "modern", "2026-07-28"):
return await attempt("discover", "initialize", lambda exc: True,
@@ -242,7 +241,6 @@ class MCPServerTransportMixin:
# Only judge 2xx (4xx/5xx may be an auth challenge); no content type advertised → trust the SDK.
ct = _content_type_base(resp)
return _is_2xx(resp) and bool(ct) and ct not in self._MCP_CONTENT_TYPES
probe_headers = dict(headers) if headers else {}
try:
async with _httpx.AsyncClient(verify=ssl_verify, follow_redirects=True, timeout=_httpx.Timeout(timeout),
@@ -346,7 +344,6 @@ class MCPServerTransportMixin:
async with httpx.AsyncClient(**client_kwargs) as http_client:
async with _core.streamable_http_client(url, http_client=http_client) as streams:
yield streams
return _owned_client_streams()
async def _run_http(self, config: dict):