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

feat(plugins): map portable Agent Plugins streamable-http entries into the native MCP runtime

Browse Source 
Agent Plugins v1 packages with 'streamable-http' mcp.json entries now load
through Hermes' existing URL-based MCP client instead of being reported and
skipped. The stdio-only limitation was the agreed follow-up slice from
PR #81196.

Boundary rules from the v1 spec (§7.2.1) are enforced:
- URL must be absolute http(s), no user information, no fragment; plain
  HTTP only for localhost/loopback hosts.
- Configured package headers are never forwarded across a cross-origin
  redirect: translation marks entries strict_redirect_headers, and the
  redirect hook in the native runtime strips those headers (plus
  Authorization) whenever a redirect leaves the original origin. On mcp <
  1.24.0, where the client cannot hook redirects, such servers fail closed
  with an actionable upgrade message.
- Legacy 'sse' entries remain reported and skipped.

The redirect hook is extracted into a testable module-level factory
(_make_redirect_header_stripper); default behavior for native config
servers is unchanged (Authorization-only stripping).
This commit is contained in:
Teknium
2026-08-08 21:13:08 -07:00
parent 961f7481a7
commit 471baea520
5 changed files with 363 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/agent_plugins.py
+74 -1
View File
@@ -307,6 +307,74 @@ def _validate_headers(headers: object) -> bool:
return True
def _validate_remote_url(url: object) -> str:
"""Validate a portable remote MCP URL per the v1 spec and return it.
Rules (Agent Plugins v1 §7.2.1): absolute http(s) URL, no user
information, no fragment; non-loopback endpoints must use HTTPS. HTTP is
allowed only when the host is exactly ``localhost`` or an IP literal in a
loopback range. No placeholder or environment expansion is performed.
"""
from urllib.parse import urlsplit
if not isinstance(url, str) or not url:
raise ValueError("url must be a non-empty string")
try:
parsed = urlsplit(url)
except ValueError as exc:
raise ValueError(f"url is not parseable: {exc}") from exc
scheme = parsed.scheme.lower()
if scheme not in {"http", "https"}:
raise ValueError("url scheme must be http or https")
if parsed.username is not None or parsed.password is not None:
raise ValueError("url must not contain user information")
if parsed.fragment:
raise ValueError("url must not contain a fragment")
host = parsed.hostname
if not host:
raise ValueError("url must have a host")
if scheme == "http":
loopback = False
if host == "localhost":
loopback = True
else:
import ipaddress
try:
loopback = ipaddress.ip_address(host).is_loopback
except ValueError:
loopback = False
if not loopback:
raise ValueError("non-loopback url must use https")
return url
def _translate_remote(config: Mapping[str, Any]) -> Dict[str, Any]:
"""Translate a portable ``streamable-http`` entry into native MCP config.
The returned record targets Hermes' existing URL-based MCP runtime.
``strict_redirect_headers`` instructs the runtime to drop the configured
headers on any cross-origin redirect, which the v1 spec requires for
portable packages (configured headers must not be forwarded to a
different origin without explicit user authorization).
"""
if set(config) - _REMOTE_FIELDS:
raise ValueError("unknown remote field")
url = _validate_remote_url(config.get("url"))
if not _validate_headers(config.get("headers")):
raise ValueError("invalid headers")
translated: Dict[str, Any] = {
"url": url,
"strict_redirect_headers": True,
}
headers = config.get("headers")
if headers:
translated["headers"] = dict(headers)
return translated
def _translate_stdio(
config: Mapping[str, Any], plugin_root: Path, data_root: Path
) -> Dict[str, Any]:
@@ -428,7 +496,12 @@ def _discover_mcp(
translated[name] = translated_server
except (OSError, ValueError) as exc:
diagnostics.append(AgentPluginDiagnostic(scope, str(exc)))
elif server_type in {"streamable-http", "sse"}:
elif server_type == "streamable-http":
try:
translated[name] = _translate_remote(server)
except ValueError as exc:
diagnostics.append(AgentPluginDiagnostic(scope, str(exc)))
elif server_type == "sse":
if (
set(server) - _REMOTE_FIELDS
or not isinstance(server.get("url"), str)
tests/hermes_cli/test_agent_plugins.py
+150 -1
View File
@@ -301,6 +301,7 @@ def test_invalid_entries_and_unsupported_remote_preserve_valid_stdio(
"remote": {
"type": "streamable-http",
"url": "https://example.test/mcp",
"headers": {"X-Tenant": "a", "x-tenant": "b"},
},
},
},
@@ -360,7 +361,7 @@ def test_enabled_portable_mcp_probe_does_not_load_plugins(
)
def test_portable_mcp_probe_ignores_unsupported_only_config(
def test_portable_mcp_probe_counts_streamable_http_config(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
@@ -384,11 +385,159 @@ def test_portable_mcp_probe_ignores_unsupported_only_config(
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_BUNDLED_PLUGINS", str(bundled))
assert has_enabled_agent_plugin_mcp(
{"plugins": {"enabled": ["portable.test"]}}
)
def test_portable_mcp_probe_ignores_unsupported_only_config(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
plugin = home / "plugins" / "portable"
plugin.mkdir(parents=True)
_write_json(plugin / "plugin.json", _manifest())
_write_json(
plugin / "mcp.json",
{
"$schema": MCP_SCHEMA_V1,
"mcpServers": {
"remote": {
"type": "sse",
"url": "https://example.test/sse",
}
},
},
)
bundled = tmp_path / "bundled"
bundled.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_BUNDLED_PLUGINS", str(bundled))
assert not has_enabled_agent_plugin_mcp(
{"plugins": {"enabled": ["portable.test"]}}
)
def test_streamable_http_translates_to_native_remote_config(tmp_path: Path) -> None:
_write_json(tmp_path / "plugin.json", _manifest())
_write_json(
tmp_path / "mcp.json",
{
"$schema": MCP_SCHEMA_V1,
"mcpServers": {
"deploy": {
"type": "streamable-http",
"url": "https://deploy.example.test/mcp",
"headers": {"X-Tenant": "public-tenant"},
},
"local": {
"type": "streamable-http",
"url": "http://127.0.0.1:8000/mcp",
},
"bare": {
"type": "streamable-http",
"url": "https://bare.example.test/mcp",
},
},
},
)
package = load_agent_plugin(tmp_path, tmp_path / "data")
assert set(package.mcp_servers) == {"deploy", "local", "bare"}
deploy = package.mcp_servers["deploy"]
assert deploy == {
"url": "https://deploy.example.test/mcp",
"headers": {"X-Tenant": "public-tenant"},
"strict_redirect_headers": True,
}
assert "command" not in deploy
assert package.mcp_servers["bare"] == {
"url": "https://bare.example.test/mcp",
"strict_redirect_headers": True,
}
@pytest.mark.parametrize(
"url",
[
"",
"ftp://example.test/mcp",
"https://user:pass@example.test/mcp",
"https://example.test/mcp#fragment",
"http://example.test/mcp", # non-loopback plain HTTP
"http://10.0.0.5/mcp",
"https:///mcp", # empty host
],
)
def test_streamable_http_rejects_invalid_urls(tmp_path: Path, url: str) -> None:
_write_json(tmp_path / "plugin.json", _manifest())
_write_json(
tmp_path / "mcp.json",
{
"$schema": MCP_SCHEMA_V1,
"mcpServers": {
"remote": {"type": "streamable-http", "url": url},
"valid": {"type": "stdio", "command": "python"},
},
},
)
package = load_agent_plugin(tmp_path, tmp_path / "data")
assert set(package.mcp_servers) == {"valid"}
assert any(d.scope == "mcp:remote" for d in package.diagnostics)
@pytest.mark.parametrize(
"url",
[
"http://localhost:8000/mcp",
"http://127.0.0.1/mcp",
"http://[::1]:9000/mcp",
"https://remote.example.test/mcp",
],
)
def test_streamable_http_accepts_loopback_http_and_https(
tmp_path: Path, url: str
) -> None:
_write_json(tmp_path / "plugin.json", _manifest())
_write_json(
tmp_path / "mcp.json",
{
"$schema": MCP_SCHEMA_V1,
"mcpServers": {"remote": {"type": "streamable-http", "url": url}},
},
)
package = load_agent_plugin(tmp_path, tmp_path / "data")
assert set(package.mcp_servers) == {"remote"}
assert package.mcp_servers["remote"]["url"] == url
def test_sse_remains_unsupported(tmp_path: Path) -> None:
_write_json(tmp_path / "plugin.json", _manifest())
_write_json(
tmp_path / "mcp.json",
{
"$schema": MCP_SCHEMA_V1,
"mcpServers": {
"legacy": {"type": "sse", "url": "https://legacy.example.test/sse"}
},
},
)
package = load_agent_plugin(tmp_path, tmp_path / "data")
assert package.mcp_servers == {}
assert any(
d.scope == "mcp:legacy" and "not supported" in d.message
for d in package.diagnostics
)
def test_portable_mcp_probe_honors_native_precedence(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
tests/tools/test_mcp_tool.py
+70
View File
@@ -2787,3 +2787,73 @@ class TestMCPDiscoveryCrossProcessLock:
os.unlink(lock_path)
except Exception:
pass
class TestRedirectHeaderStripper:
"""Cross-origin redirect header boundary (portable Agent Plugins v1)."""
def _make_response(self, next_headers):
import httpx
next_request = httpx.Request(
"GET", "https://other.example.test/mcp", headers=next_headers
)
response = SimpleNamespace(
is_redirect=True,
next_request=next_request,
)
return response, next_request
def test_default_strips_only_authorization(self):
import httpx
from tools.mcp_tool import _make_redirect_header_stripper
hook = _make_redirect_header_stripper(
httpx.URL("https://origin.example.test/mcp")
)
response, next_request = self._make_response(
{"Authorization": "Bearer x", "X-Tenant": "t"}
)
asyncio.run(hook(response))
assert "authorization" not in next_request.headers
assert next_request.headers["x-tenant"] == "t"
def test_strict_strips_configured_headers_cross_origin(self):
import httpx
from tools.mcp_tool import _make_redirect_header_stripper
hook = _make_redirect_header_stripper(
httpx.URL("https://origin.example.test/mcp"),
strict=True,
configured_header_names={"x-tenant"},
)
response, next_request = self._make_response(
{"Authorization": "Bearer x", "X-Tenant": "t", "Accept": "a"}
)
asyncio.run(hook(response))
assert "authorization" not in next_request.headers
assert "x-tenant" not in next_request.headers
# Client-generated headers unrelated to package config survive.
assert next_request.headers["accept"] == "a"
def test_same_origin_redirect_keeps_headers(self):
import httpx
from tools.mcp_tool import _make_redirect_header_stripper
hook = _make_redirect_header_stripper(
httpx.URL("https://origin.example.test/mcp"),
strict=True,
configured_header_names={"x-tenant"},
)
next_request = httpx.Request(
"GET",
"https://origin.example.test/other",
headers={"Authorization": "Bearer x", "X-Tenant": "t"},
)
response = SimpleNamespace(is_redirect=True, next_request=next_request)
asyncio.run(hook(response))
assert next_request.headers["authorization"] == "Bearer x"
assert next_request.headers["x-tenant"] == "t"
tools/mcp_tool.py
+61 -9
View File
@@ -1303,6 +1303,38 @@ def _apply_identity_header(server_name: str, config: dict, headers: dict) -> dic
return headers
def _make_redirect_header_stripper(
original_url,
*,
strict: bool = False,
configured_header_names: "set[str] | frozenset[str]" = frozenset(),
):
"""Build an httpx response hook that guards cross-origin redirects.
Always strips ``Authorization`` when a redirect leaves the original
origin. When *strict* is true (portable Agent Plugins v1 packages with
``strict_redirect_headers``), every *configured* header (lowercase names
in *configured_header_names*) is stripped as well — the v1 spec forbids
forwarding package-configured headers to a different origin without
explicit user authorization.
"""
async def _strip_on_cross_origin_redirect(response):
if response.is_redirect and response.next_request:
target = response.next_request.url
if (target.scheme, target.host, target.port) != (
original_url.scheme, original_url.host, original_url.port,
):
response.next_request.headers.pop("authorization", None)
response.next_request.headers.pop("Authorization", None)
if strict:
for _name in configured_header_names:
while _name in response.next_request.headers:
del response.next_request.headers[_name]
return _strip_on_cross_origin_redirect
def _format_connect_error(exc: BaseException) -> str:
"""Render nested MCP connection errors into an actionable short message."""
@@ -2890,6 +2922,13 @@ class MCPServerTask:
url = config["url"]
headers = dict(config.get("headers") or {})
# Portable Agent Plugins v1 packages set strict_redirect_headers:
# configured headers are visible package data and MUST NOT be
# forwarded to a different origin through a redirect (spec §7.2.1).
# Capture the configured header names before client-generated
# headers (identity, protocol version) are merged in.
_strict_cfg_headers = bool(config.get("strict_redirect_headers"))
_configured_header_names = {key.lower() for key in headers}
# Optional per-user identity header (config-gated; static or
# profile-derived). Explicit headers of the same name win.
headers = _apply_identity_header(self.name, config, headers)
@@ -2932,6 +2971,14 @@ class MCPServerTask:
# rather than Streamable HTTP). Configure with ``transport: sse`` in the
# mcp_servers entry in config.yaml.
if config.get("transport") == "sse":
if _strict_cfg_headers:
# Portable packages never translate to SSE; if a config
# combines both anyway, fail closed rather than run a
# transport that cannot enforce the redirect boundary.
raise ValueError(
f"MCP server '{self.name}': strict_redirect_headers is "
"not supported on the SSE transport."
)
if sse_client is None:
raise ImportError(
f"MCP server '{self.name}' requires SSE transport but "
@@ -3028,15 +3075,11 @@ class MCPServerTask:
_original_url = httpx.URL(url)
async def _strip_auth_on_cross_origin_redirect(response):
"""Strip Authorization headers when redirected to a different origin."""
if response.is_redirect and response.next_request:
target = response.next_request.url
if (target.scheme, target.host, target.port) != (
_original_url.scheme, _original_url.host, _original_url.port,
):
response.next_request.headers.pop("authorization", None)
response.next_request.headers.pop("Authorization", None)
_strip_auth_on_cross_origin_redirect = _make_redirect_header_stripper(
_original_url,
strict=_strict_cfg_headers,
configured_header_names=_configured_header_names,
)
client_kwargs: dict = {
"follow_redirects": True,
@@ -3085,6 +3128,15 @@ class MCPServerTask:
return reason
else:
# Deprecated API (mcp < 1.24.0): manages httpx client internally.
if _strict_cfg_headers:
# Fail closed: without an owned httpx client we cannot hook
# redirects, so the v1 cross-origin header boundary cannot be
# enforced on this SDK version.
raise ImportError(
f"MCP server '{self.name}' requires mcp >= 1.24.0 to "
"enforce the portable redirect-header boundary "
"(strict_redirect_headers). Upgrade the mcp package."
)
_http_kwargs: dict = {
"headers": headers,
"timeout": float(connect_timeout),
website/docs/developer-guide/plugins/index.md
+8 -4
View File
@@ -85,10 +85,14 @@ profile-scoped writable directory managed by Hermes.
Values declared in portable MCP `env` are visible package data, not a secret
storage mechanism. Do not place credentials in `mcp.json`.
The current portable subset supports stdio MCP only. Portable Streamable HTTP
and legacy SSE entries are reported and skipped because the native remote
client does not yet prove the v1 configured-header redirect boundary end to
end. Agent Plugins v1 does not define trust, permissions, provenance, or a
The current portable subset supports stdio and Streamable HTTP MCP entries.
Portable `streamable-http` entries are routed through Hermes' existing native
remote MCP client (the same runtime that powers URL-based `mcp_servers`
config), with the v1 boundary rules enforced: the URL must be absolute
http(s) with no user information or fragment, plain HTTP is accepted only
for `localhost`/loopback hosts, and configured headers are never forwarded
across a cross-origin redirect. Legacy `sse` entries are reported and
skipped. Agent Plugins v1 does not define trust, permissions, provenance, or a
sandbox. Enabling a package grants its instructions and local executable the
same full-trust posture as other installed Hermes plugins.