proxy runtime to get this thing now

the merge is still merging
This commit is contained in:
lda
2026-05-13 18:44:32 +07:00 Verified
parent 7723d8558e
commit 92f5ca1755
6 changed files with 94 additions and 9 deletions
+18
View File
@@ -59,6 +59,24 @@ provider/proxy unmount lifecycle that we can rely on for safe per-connection
teardown. Until that exists, reload should be treated as best-effort remounting, teardown. Until that exists, reload should be treated as best-effort remounting,
not a fully safe session/subscription lifecycle. not a fully safe session/subscription lifecycle.
Unified mode currently reuses this runtime as its proxy mounting engine. The
`transparent_proxy` package name is therefore partly legacy: the code is still
the place where configured upstream MCP connections become mounted FastMCP
providers.
After a successful reload, the runtime publishes local `tools_changed`,
`resources_changed`, `prompts_changed`, and `catalog_changed` events when an
event bus is supplied. The admin MCP tool projects the same event kinds into
MCP list-changed notifications for the current client session. Config mutation
tools still only stage changes and return `requires_reload`; they do not emit
list-changed notifications until reload remounts the visible capability set.
Do not memoize mounted proxies or clients without an explicit lifecycle design.
The tempting implementation is a dictionary keyed by connection id around
`create_proxy(client, ...)`, but cached clients need clear close/reconnect/error
semantics. Prefer FastMCP's official unmount/provider lifecycle when it becomes
available.
Do not add notification proxying or long-lived subscription handling across Do not add notification proxying or long-lived subscription handling across
reloads without first introducing an explicit mount lifecycle boundary. reloads without first introducing an explicit mount lifecycle boundary.
+2 -1
View File
@@ -24,6 +24,7 @@ def create_unified_proxy_server(
admin_tools: bool = True, admin_tools: bool = True,
) -> FastMCP[Any]: ) -> FastMCP[Any]:
"""Create one MCP server with upstream proxy, admin, and workflow tools.""" """Create one MCP server with upstream proxy, admin, and workflow tools."""
service = build_service_from_config(config)
runtime = TransparentProxyRuntime( runtime = TransparentProxyRuntime(
config, config,
config_path=config_path, config_path=config_path,
@@ -31,8 +32,8 @@ def create_unified_proxy_server(
prompts_as_tools=prompts_as_tools, prompts_as_tools=prompts_as_tools,
search_tools=search_tools, search_tools=search_tools,
admin_tools=admin_tools, admin_tools=admin_tools,
event_bus=service.event_bus,
) )
service = build_service_from_config(config)
_register_workflow_tools(runtime.server, WorkflowSurfaceHandlers(service)) _register_workflow_tools(runtime.server, WorkflowSurfaceHandlers(service))
return runtime.server return runtime.server
+5 -6
View File
@@ -5,8 +5,8 @@ from typing import Any
from fastmcp import Context, FastMCP from fastmcp import Context, FastMCP
from ..admin_surface import ProxyAdminRuntime, TransparentAdminHandlers from ..admin_surface import ProxyAdminRuntime, TransparentAdminHandlers
from ..events import make_event
from ..notifications import FastMcpContextNotificationSink from ..notifications import FastMcpContextNotificationSink
from .reload_events import reload_change_events
def create_proxy_admin_server( def create_proxy_admin_server(
@@ -48,7 +48,7 @@ def create_proxy_admin_server(
) )
async def reload_config(ctx: Context) -> dict[str, Any]: async def reload_config(ctx: Context) -> dict[str, Any]:
result = handlers.reload_config() result = handlers.reload_config()
await _send_reload_notifications(ctx) await _send_reload_notifications(ctx, result)
return result return result
@admin.tool( @admin.tool(
@@ -140,9 +140,8 @@ def create_proxy_admin_server(
return admin return admin
async def _send_reload_notifications(ctx: Context) -> None: async def _send_reload_notifications(ctx: Context, result: dict[str, Any]) -> None:
"""Notify the current client that reload may have changed visible capabilities.""" """Notify the current client that reload may have changed visible capabilities."""
sink = FastMcpContextNotificationSink(ctx) sink = FastMcpContextNotificationSink(ctx)
await sink.send_event(make_event("tools_changed")) for event in reload_change_events(result):
await sink.send_event(make_event("resources_changed")) await sink.send_event(event)
await sink.send_event(make_event("prompts_changed"))
@@ -0,0 +1,24 @@
from __future__ import annotations
from typing import Any
from ..events import McpEvent, make_event
def reload_change_events(result: dict[str, Any]) -> list[McpEvent]:
"""Build local change events for a successful proxy runtime reload."""
event_payload = {
"reason": "transparent_reload",
"mounted_connections": result["mounted_connections"],
"connection_count": result["connection_count"],
"enabled_connection_count": result["enabled_connection_count"],
}
return [
make_event(kind, payload=event_payload)
for kind in (
"tools_changed",
"resources_changed",
"prompts_changed",
"catalog_changed",
)
]
+19 -1
View File
@@ -12,6 +12,7 @@ from fastmcp.server.transforms import Namespace, PromptsAsTools, ResourcesAsTool
from fastmcp.server.transforms.search import BM25SearchTransform from fastmcp.server.transforms.search import BM25SearchTransform
from ..control import BrokerConfigManager, ConfigMutationError from ..control import BrokerConfigManager, ConfigMutationError
from ..events import EventBus
from ..models import BrokerConfig from ..models import BrokerConfig
from ..shared.names import ADMIN_NAMESPACE, LdaNamespace from ..shared.names import ADMIN_NAMESPACE, LdaNamespace
from ..proxy_config import broker_config_to_fastmcp_config from ..proxy_config import broker_config_to_fastmcp_config
@@ -22,6 +23,7 @@ from .tools import (
filter_proxy_tools, filter_proxy_tools,
proxy_tools_page, proxy_tools_page,
) )
from .reload_events import reload_change_events
_ADMIN_TOOL_NAMES = [ _ADMIN_TOOL_NAMES = [
f"{ADMIN_NAMESPACE}.list_connections", f"{ADMIN_NAMESPACE}.list_connections",
@@ -48,6 +50,7 @@ class TransparentProxyRuntime:
prompts_as_tools: bool = False, prompts_as_tools: bool = False,
search_tools: bool = False, search_tools: bool = False,
admin_tools: bool = True, admin_tools: bool = True,
event_bus: EventBus | None = None,
) -> None: ) -> None:
self.config = config self.config = config
self.manager = None if config_path is None else BrokerConfigManager(config_path) self.manager = None if config_path is None else BrokerConfigManager(config_path)
@@ -60,6 +63,7 @@ class TransparentProxyRuntime:
), ),
) )
self.admin_tools = admin_tools self.admin_tools = admin_tools
self.event_bus = event_bus
self.reload() self.reload()
if resources_as_tools: if resources_as_tools:
self.server.add_transform(ResourcesAsTools(self.server)) self.server.add_transform(ResourcesAsTools(self.server))
@@ -107,13 +111,22 @@ class TransparentProxyRuntime:
self.server.mount(proxy) self.server.mount(proxy)
mounted_connections.append(connection.id) mounted_connections.append(connection.id)
return { result = {
"ok": True, "ok": True,
"reloaded": True, "reloaded": True,
"mounted_connections": mounted_connections, "mounted_connections": mounted_connections,
"connection_count": len(config.connections), "connection_count": len(config.connections),
"enabled_connection_count": len(mounted_connections), "enabled_connection_count": len(mounted_connections),
} }
self._publish_reload_events(result)
return result
def _publish_reload_events(self, result: dict[str, Any]) -> None:
"""Publish local change events after a successful best-effort remount."""
if self.event_bus is None:
return
for event in reload_change_events(result):
self.event_bus.publish(event)
async def list_proxy_tools(self) -> list[dict[str, Any]]: async def list_proxy_tools(self) -> list[dict[str, Any]]:
return await self._list_proxy_tools() return await self._list_proxy_tools()
@@ -171,6 +184,7 @@ def create_transparent_proxy_server(
prompts_as_tools: bool = False, prompts_as_tools: bool = False,
search_tools: bool = False, search_tools: bool = False,
admin_tools: bool = True, admin_tools: bool = True,
event_bus: EventBus | None = None,
) -> FastMCP[Any]: ) -> FastMCP[Any]:
validate_transparent_proxy_config( validate_transparent_proxy_config(
config, config,
@@ -184,6 +198,7 @@ def create_transparent_proxy_server(
prompts_as_tools=prompts_as_tools, prompts_as_tools=prompts_as_tools,
search_tools=search_tools, search_tools=search_tools,
admin_tools=admin_tools, admin_tools=admin_tools,
event_bus=event_bus,
).server ).server
@@ -195,6 +210,7 @@ def create_transparent_proxy_client(
prompts_as_tools: bool = False, prompts_as_tools: bool = False,
search_tools: bool = False, search_tools: bool = False,
admin_tools: bool = True, admin_tools: bool = True,
event_bus: EventBus | None = None,
) -> Client[FastMCPTransport]: ) -> Client[FastMCPTransport]:
return Client( return Client(
FastMCPTransport( FastMCPTransport(
@@ -205,6 +221,8 @@ def create_transparent_proxy_client(
prompts_as_tools=prompts_as_tools, prompts_as_tools=prompts_as_tools,
search_tools=search_tools, search_tools=search_tools,
admin_tools=admin_tools, admin_tools=admin_tools,
event_bus=event_bus,
) )
) )
) )
+26 -1
View File
@@ -8,9 +8,10 @@ from typing import Any
import mcp.types as mcp_types import mcp.types as mcp_types
import pytest import pytest
from wf_mcp.events import EventBus, InMemoryEventSink
from wf_mcp.models import BrokerConfig, ConnectionConfig from wf_mcp.models import BrokerConfig, ConnectionConfig
from wf_mcp.proxy_validation import ProxyConfigError, validate_transparent_proxy_config from wf_mcp.proxy_validation import ProxyConfigError, validate_transparent_proxy_config
from wf_mcp.transparent_proxy import create_transparent_proxy_client from wf_mcp.transparent_proxy import TransparentProxyRuntime, create_transparent_proxy_client
from wf_mcp.broker import load_broker_config from wf_mcp.broker import load_broker_config
from .test_support import fixture_server_path, local_temp_root from .test_support import fixture_server_path, local_temp_root
@@ -481,3 +482,27 @@ def test_transparent_proxy_admin_reload_sends_list_changed_notifications() -> No
assert "notifications/tools/list_changed" in methods assert "notifications/tools/list_changed" in methods
assert "notifications/resources/list_changed" in methods assert "notifications/resources/list_changed" in methods
assert "notifications/prompts/list_changed" in methods assert "notifications/prompts/list_changed" in methods
def test_transparent_proxy_runtime_reload_publishes_local_change_events() -> None:
sink = InMemoryEventSink()
event_bus = EventBus(sink)
config = BrokerConfig(
store_root=local_temp_root() / "transparent_proxy_event_store",
connections=[],
)
runtime = TransparentProxyRuntime(config, event_bus=event_bus)
initial_event_count = len(sink.list_events())
result = runtime.reload()
events = sink.list_events()[initial_event_count:]
event_kinds = [event.kind for event in events]
catalog_changed = [
event for event in events if event.kind == "catalog_changed"
]
assert result["reloaded"] is True
assert "tools_changed" in event_kinds
assert "resources_changed" in event_kinds
assert "prompts_changed" in event_kinds
assert catalog_changed[0].payload["reason"] == "transparent_reload"