config reload bug
This commit is contained in:
@@ -33,6 +33,7 @@ from ...models import (
|
||||
CatalogPromptEntry,
|
||||
CatalogResourceEntry,
|
||||
CatalogSnapshot,
|
||||
BrokerConfig,
|
||||
ConnectionConfig,
|
||||
RawWorkflowPlan,
|
||||
)
|
||||
@@ -90,6 +91,33 @@ class WfMcpService:
|
||||
)
|
||||
)
|
||||
|
||||
def sync_connections_from_config(self, config: BrokerConfig) -> None:
|
||||
"""Reconcile connection sources after the public server reloads config.
|
||||
|
||||
The public server has two cooperating views: proxy mounts read the live
|
||||
file-backed config, while workflow discovery reads this service's source
|
||||
registry. Reload must keep those views aligned, or raw proxy tools can be
|
||||
enabled while planner-visible workflow capabilities remain disabled.
|
||||
"""
|
||||
next_ids = {connection.id for connection in config.connections}
|
||||
previous_ids = set(self.connections.connections)
|
||||
for connection_id in previous_ids - next_ids:
|
||||
del self.connections.connections[connection_id]
|
||||
self.capability_sources.pop(connection_id, None)
|
||||
|
||||
for connection in config.connections:
|
||||
parse_connection_id(connection.id)
|
||||
if connection.id in RESERVED_CONNECTION_IDS:
|
||||
raise ValueError(
|
||||
f"connection id {connection.id!r} is reserved by wf-mcp"
|
||||
)
|
||||
self.connections.register(connection)
|
||||
source = self.capability_sources.get(connection.id)
|
||||
if source is None:
|
||||
self._hydrate_connection_source_from_snapshot(connection)
|
||||
else:
|
||||
source.enabled = connection.enabled
|
||||
|
||||
def register_adapter(self, server: str, adapter: BackendAdapter) -> None:
|
||||
self.adapters[server] = adapter
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ from ..broker.config import build_service_from_config
|
||||
from ..broker.transport import normalize_transport
|
||||
from ..documentation import build_local_documentation_source
|
||||
from ..models import BrokerConfig
|
||||
from ..sdk import McpSdkAdapter
|
||||
from ..transparent_proxy.runtime import ProxyRuntime
|
||||
from ..workflow_surface import register_workflow_tools
|
||||
from .prompts import register_documentation_prompts
|
||||
@@ -29,6 +30,13 @@ def create_server(
|
||||
) -> FastMCP[Any]:
|
||||
"""Create the public MCP server with proxy, admin, and workflow tools."""
|
||||
service = build_service_from_config(config)
|
||||
|
||||
def sync_service(config: BrokerConfig) -> None:
|
||||
service.sync_connections_from_config(config)
|
||||
for connection in config.connections:
|
||||
if connection.server not in service.adapters:
|
||||
service.register_adapter(connection.server, McpSdkAdapter())
|
||||
|
||||
runtime = ProxyRuntime(
|
||||
config,
|
||||
config_path=config_path,
|
||||
@@ -37,6 +45,7 @@ def create_server(
|
||||
search_tools=search_tools,
|
||||
admin_tools=admin_tools,
|
||||
event_bus=service.event_bus,
|
||||
on_reload=sync_service,
|
||||
)
|
||||
if admin_tools:
|
||||
register_service_admin_tools(
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import FastMCP
|
||||
@@ -61,6 +62,7 @@ class ProxyRuntime:
|
||||
search_tools: bool = False,
|
||||
admin_tools: bool = True,
|
||||
event_bus: EventBus | None = None,
|
||||
on_reload: Callable[[BrokerConfig], None] | None = None,
|
||||
) -> None:
|
||||
self.config = config
|
||||
self.manager = None if config_path is None else BrokerConfigManager(config_path)
|
||||
@@ -74,6 +76,7 @@ class ProxyRuntime:
|
||||
)
|
||||
self.admin_tools = admin_tools
|
||||
self.event_bus = event_bus
|
||||
self.on_reload = on_reload
|
||||
self.mounts: ProxyMountRegistry[FastMCP[Any]] = ProxyMountRegistry(
|
||||
create_proxy_mount
|
||||
)
|
||||
@@ -105,6 +108,8 @@ class ProxyRuntime:
|
||||
def reload(self) -> dict[str, Any]:
|
||||
config = self.current_config()
|
||||
validate_transparent_proxy_config(config)
|
||||
if self.on_reload is not None:
|
||||
self.on_reload(config)
|
||||
self.server.providers[:] = [self.server.local_provider]
|
||||
|
||||
mounts = self.mounts.active_mounts_for(config)
|
||||
|
||||
@@ -1,11 +1,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from mcp import types as mcp_types
|
||||
|
||||
from wf_mcp.broker.config import load_broker_config
|
||||
from wf_mcp.models import BrokerConfig, ConnectionConfig
|
||||
from wf_mcp.server import create_server_client
|
||||
|
||||
@@ -225,6 +227,73 @@ def test_server_exposes_platform_documentation_prompts() -> None:
|
||||
asyncio.run(run_proxy())
|
||||
|
||||
|
||||
def test_server_reload_syncs_service_connection_source_enabled_state() -> None:
|
||||
tmp_path = local_temp_root() / "unified_reload_service_source_store"
|
||||
tmp_path.mkdir(parents=True, exist_ok=True)
|
||||
config_path = tmp_path / "wf_mcp.config.json"
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"store_root": ".wf_mcp_store",
|
||||
"connections": [
|
||||
{
|
||||
"id": "fixture.personal",
|
||||
"server": "fixture",
|
||||
"account": "personal",
|
||||
"enabled": False,
|
||||
"metadata": {
|
||||
"transport": "stdio",
|
||||
"command": sys.executable,
|
||||
"args": [fixture_server_path()],
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
config = load_broker_config(config_path)
|
||||
|
||||
async def run_proxy() -> None:
|
||||
client = create_server_client(config, config_path=config_path)
|
||||
async with client:
|
||||
await client.call_tool(
|
||||
"wf.admin.refresh_connection_catalog",
|
||||
{"connection_id": "fixture.personal"},
|
||||
)
|
||||
before = await client.call_tool(
|
||||
"wf.workflow.list_capabilities",
|
||||
{"source_id": "fixture.personal"},
|
||||
)
|
||||
assert _structured(before)["capabilities"] == []
|
||||
|
||||
await client.call_tool(
|
||||
"wf.admin.enable_connection",
|
||||
{"connection_id": "fixture.personal"},
|
||||
)
|
||||
await client.call_tool("wf.admin.reload_config")
|
||||
|
||||
sources = await client.call_tool("wf.admin.list_sources", {"limit": 100})
|
||||
fixture_source = next(
|
||||
source
|
||||
for source in _structured(sources)["sources"]
|
||||
if source["id"] == "fixture.personal"
|
||||
)
|
||||
assert fixture_source["enabled"] is True
|
||||
|
||||
after = await client.call_tool(
|
||||
"wf.workflow.list_capabilities",
|
||||
{"source_id": "fixture.personal"},
|
||||
)
|
||||
names = [
|
||||
capability["name"]
|
||||
for capability in _structured(after)["capabilities"]
|
||||
]
|
||||
assert "fixture.personal.echo_tool" in names
|
||||
|
||||
asyncio.run(run_proxy())
|
||||
|
||||
|
||||
def test_admin_tools_have_human_metadata() -> None:
|
||||
config = BrokerConfig(
|
||||
store_root=local_temp_root() / "unified_admin_metadata_store",
|
||||
|
||||
+3
-2
@@ -19,12 +19,13 @@
|
||||
"id": "playwright.default",
|
||||
"server": "playwright",
|
||||
"account": "default",
|
||||
"enabled": false,
|
||||
"enabled": true,
|
||||
"metadata": {
|
||||
"transport": "stdio",
|
||||
"command": "pnpx",
|
||||
"args": [
|
||||
"@playwright/mcp@latest"
|
||||
"@playwright/mcp@latest",
|
||||
"--isolated"
|
||||
],
|
||||
"env": {}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user