feat(sdk): project-level .mcp.json discovery and merge#2772
feat(sdk): project-level .mcp.json discovery and merge#2772nehaaprasad wants to merge 3 commits intoOpenHands:mainfrom
Conversation
nehaaprasad
changed the title
VascoSch92
left a comment
VascoSch92
left a comment
There was a problem hiding this comment.
Hey @nehaaprasad
thanks for your contribution.
I left some comment.
Moreover, Env variable expansion is missing. The issue explicitly requires ${VAR} and ${VAR:-default} syntax support, but try_load_project_mcp_config doesn't expand env variables in the loaded config.
The good news is this already exists in the codebase: load_mcp_config() in skills/utils.py calls expand_mcp_variables() (line 144), which handles both ${VAR} (checks provided variables, then os.environ) and ${VAR:-default} fallbacks.
Therefore, you can reuse load_mcp_config() inside try_load_project_mcp_config instead of loading the JSON manually, this should give you env var expansion, validation via MCPConfig, and consistent error handling for free.
| settings. UIs should set this only after the user approves | ||
| project-scoped servers. |
There was a problem hiding this comment.
| settings. UIs should set this only after the user approves | |
| project-scoped servers. | |
| settings. |
| all_plugin_hooks: list[HookConfig] = [] | ||
| all_plugin_agents: list[AgentDefinition] = [] | ||
|
|
||
| project_dir = Path(self.workspace.working_dir) |
There was a problem hiding this comment.
_ensure_plugins_loaded is about loading plugins, discovering and merging project-level MCP config is a separate concern that happens to run at the same time.
Can you move that into another method ?
Something like
def _load_project_mcp_config(self) -> dict[str, Any] | None:
"""Load project .mcp.json if trusted, return merged config or None."""
if not self._trust_project_mcp:
return None
project_dir = Path(self.workspace.working_dir)
project_mcp = try_load_project_mcp_config(project_dir)
if project_mcp is None:
return None
merged = merge_mcp_configs(project_mcp, self.agent.mcp_config)
if merged != self.agent.mcp_config:
self.agent = self.agent.model_copy(update={"mcp_config": merged})
with self._state:
self._state.agent = self.agent
return mergedand then you can call it before _ensure_plugins_loaded() in the _ensure_agent_ready method
| @@ -0,0 +1,38 @@ | |||
| """Project-level .mcp.json discovery and loading.""" | |||
|
|
|||
| from __future__ import annotations | |||
There was a problem hiding this comment.
you don't need this import.
| project_dir / ".openhands" / ".mcp.json", | ||
| project_dir / ".mcp.json", |
There was a problem hiding this comment.
This should be constant at the top of the file.
_PROJECT_MCP_CANDIDATES: Final[tuple[str]] = (".openhands/.mcp.json", ".mcp.json")| logger = get_logger(__name__) | ||
|
|
||
|
|
||
| def find_project_mcp_json(project_dir: Path) -> Path | None: |
| return None | ||
|
|
||
|
|
||
| def try_load_project_mcp_config(project_dir: Path) -> dict[str, Any] | None: |
There was a problem hiding this comment.
| def try_load_project_mcp_config(project_dir: Path) -> dict[str, Any] | None: | |
| def load_project_mcp_config(project_dir: Path) -> dict[str, Any] | None: |
just make clear in the docstring that it returns None if there is a problem
| logger = get_logger(__name__) | ||
|
|
||
|
|
||
| def merge_mcp_configs( |
There was a problem hiding this comment.
Move this method in another place. It is a general MCP utility. Perhaps something like openhands/sdk/mcp/merge.py
| if base is None and overlay is None: | ||
| return {} | ||
| if base is None: | ||
| return dict(overlay) if overlay is not None else {} | ||
| if overlay is None: | ||
| return dict(base) | ||
|
|
There was a problem hiding this comment.
You can write that with a match :-)
match (base, overlay):
case (None, None):
return {}
case (None, _):
return dict(overlay)
case (_, None):
return dict(base)| if overlay is None: | ||
| return dict(base) | ||
|
|
||
| result = dict(base) |
There was a problem hiding this comment.
Two things:
- The loop at the end skips "mcpServers" but overwrites every other key — that's just result.update() with one exclusion. You can simplify by merging everything first, then deep-merging mcpServers separately.
- With the match rewrite:
def merge_mcp_configs(
base: dict[str, Any] | None,
overlay: dict[str, Any] | None,
) -> dict[str, Any]:
"""Merge two MCP config dicts; overlay wins on key conflicts."""
match (base, overlay):
case (None, None):
return {}
case (None, _):
return dict(overlay)
case (_, None):
return dict(base)
result = {**base, **overlay}
if "mcpServers" in base and "mcpServers" in overlay:
result["mcpServers"] = {**base["mcpServers"], **overlay["mcpServers"]}
return resultThe {**base, **overlay} handles all top-level keys (overlay wins). Then the mcpServers fix-up deep-merges server entries instead of letting overlay completely replace the dict. If only one side has mcpServers, the spread already did the right thing.
There was a problem hiding this comment.
Use TestLLM instead of LLM with SecretStr.
The tests create LLM(model="test/model", api_key=SecretStr("test-key")) which is fragile — it may try to resolve model config or validate against real providers. TestLLM exists exactly for this and is already used in similar LocalConversation integration tests (test_switch_model.py, test_agent_status_transition.py).
2 participants
why
Summary
How to Test
Type
[x] Feature
Notes
fix : #2754