fix(code): harden Agent Plugin runtime values

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Alexander Olsen
2026-08-06 22:03:19 +00:00
co-authored by open-swe[bot]
parent 722c5dca8c
commit 39cc23f042
3 changed files with 65 additions and 9 deletions
@@ -128,6 +128,10 @@ def _load_mcp_server_map(path: Path) -> JsonObject:
return _server_map(raw)
def _agent_plugin_paths(plugin: PluginInstance) -> tuple[Path, Path]:
return plugin.root.resolve(), plugin.data_dir.resolve()
def _plugin_mcp_server_map(plugin: PluginInstance) -> JsonObject:
"""Load a plugin's declared MCP servers without creating runtime state.
@@ -136,13 +140,14 @@ def _plugin_mcp_server_map(plugin: PluginInstance) -> JsonObject:
"""
manifest = plugin.manifest
if manifest is not None and manifest.format == AGENT_PLUGIN_FORMAT:
plugin_root, plugin_data = _agent_plugin_paths(plugin)
servers: JsonObject = {}
for path in plugin.inventory.mcp_files:
servers.update(
load_agent_plugin_mcp(
path,
plugin_root=plugin.root,
plugin_data=plugin.data_dir,
plugin_root=plugin_root,
plugin_data=plugin_data,
)
)
return servers
@@ -191,9 +196,10 @@ def _normalize_server(
if normalized_server.get("type") != "stdio":
return normalized_server
env = normalized_server.get("env")
plugin_root, plugin_data = _agent_plugin_paths(plugin)
plugin_env = plugin_environment(
plugin_root=plugin.root,
plugin_data=plugin.data_dir,
plugin_root=plugin_root,
plugin_data=plugin_data,
project_dir=project_dir,
)
configured_env = env if isinstance(env, dict) else {}
@@ -283,7 +283,7 @@ def _stdio_cwd(value: object, *, plugin_root: Path, plugin_data: Path) -> str:
def _validate_header_value(value: str) -> None:
if any(
(ord(character) < _ASCII_CONTROL_LIMIT and character != "\t")
or ord(character) == _ASCII_DELETE
or ord(character) >= _ASCII_DELETE
for character in value
):
_raise_field("mcpServers.*.headers", "contains an invalid header value")
@@ -1,13 +1,11 @@
from __future__ import annotations
import json
from typing import TYPE_CHECKING, cast
from pathlib import Path
from typing import cast
import pytest
if TYPE_CHECKING:
from pathlib import Path
from deepagents_code.mcp_config import (
MCP_ENV_RESOLUTION_DISABLED,
MCP_REDIRECTS_DISABLED,
@@ -328,6 +326,53 @@ def test_agent_plugin_mcp_adapts_portable_servers(tmp_path: Path) -> None:
}
def test_agent_plugin_mcp_resolves_relative_runtime_paths(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
root = Path("plugin")
data_dir = Path("cache/data")
_write_manifest(root)
_write_json(
root / "mcp.json",
{
"$schema": AGENT_PLUGIN_MCP_SCHEMA,
"mcpServers": {
"local": {
"type": "stdio",
"command": "./bin/server",
"args": ["${PLUGIN_DATA}/state"],
}
},
},
)
manifest, _path, warnings = load_manifest(root)
assert manifest is not None
plugin = PluginInstance(
plugin_id="portable-plugin@tools",
name="portable-plugin",
marketplace="tools",
version=manifest.version,
root=root,
data_dir=data_dir,
manifest=manifest,
inventory=build_inventory(root, manifest, warnings),
)
configs = plugin_mcp_configs((plugin,))
servers = cast("dict[str, JsonObject]", configs[0]["mcpServers"])
local = servers[scoped_mcp_server_name(plugin.plugin_id, "local")]
expected_root = root.resolve()
expected_data = data_dir.resolve()
assert local["command"] == str(expected_root / "bin/server")
assert local["cwd"] == str(expected_root)
assert local["args"] == [str(expected_data / "state")]
env = cast("dict[str, str]", local["env"])
assert env["PLUGIN_ROOT"] == str(expected_root)
assert env["PLUGIN_DATA"] == str(expected_data)
def test_unwritable_plugin_data_skips_only_stdio_servers(tmp_path: Path) -> None:
root = tmp_path / "plugin"
data_dir = tmp_path / "data"
@@ -405,6 +450,11 @@ def test_agent_plugin_mcp_skips_invalid_entries_independently(tmp_path: Path) ->
"url": "https://example.com/sse",
"headers": {"X-Test": "a", "x-test": "b"},
},
"unicode-header": {
"type": "streamable-http",
"url": "https://example.com/mcp",
"headers": {"X-Label": "café"},
},
"unknown-field": {
"type": "stdio",
"command": "python",