local source change events to mcp notifs

docs/mcp_protocol_proxy_inventory.md

@@ -220,21 +220,23 @@ WfMcpService
     -> later: McpSessionNotificationSink
 ```
 
-Near-term implementation should add a notification mapping layer with a fake
-sink first. Real MCP emission needs one of these session-aware entry points:
+Implemented local notification pieces:
 
-- a FastMCP tool/resource/prompt handler with injected `Context`
-- middleware that can access the active `Context` or session
-- explicit lower-level session management if FastMCP exposes enough stable API
+- pure internal-event to `mcp.types.ServerNotification` mapping
+- recording sink for protocol projection tests
+- FastMCP `Context` sink for request-scoped notification emission
+- `wf.admin.reload_config` sends tool/resource/prompt list-changed
+  notifications to the current client session
 
-Until we have a session-aware sink, list-changed events should remain internal
-events exposed through `get_broker_events` / `wf-mcp://events`.
+This is still not a global broadcast system. It only emits through an active
+request context, which matches FastMCP's available public API today.
 
 ### Concrete Notification Plan
 
 Implement notifications in this order.
 
-1. Add a pure mapping layer from internal events to MCP notification objects.
+1. Done: add a pure mapping layer from internal events to MCP notification
+   objects.
 
 ```text
 tools_changed     -> ToolListChangedNotification
@@ -245,13 +247,13 @@ prompts_changed   -> PromptListChangedNotification
 This layer should not know about FastMCP sessions. It should be easy to test
 with plain `mcp.types` objects.
 
-2. Add a fake/test notification sink.
+2. Done: add a fake/test notification sink.
 
 The first sink should only record which MCP notification objects would be sent.
 This proves the event-to-notification mapping without depending on Codex,
 Inspector, stdio behavior, or Streamable HTTP behavior.
 
-3. Add a FastMCP `Context` notification sink.
+3. Done: add a FastMCP `Context` notification sink.
 
 This sink can call:
 
@@ -263,7 +265,7 @@ It is only valid while handling a request that has an active FastMCP context.
 This should be treated as a session-scoped projection, not a global broadcast
 system.
 
-4. Wire local admin operations first.
+4. Partly done: wire local admin operations first.
 
 Best first live target:
 
@@ -271,12 +273,17 @@ Best first live target:
 wf.admin.reload_config
 ```
 
-When reload changes mounted capabilities, it should:
+Current behavior:
 
-- perform the reload
-- emit internal catalog/tool/resource/prompt change events
-- send list-changed MCP notifications to the current client session when a
-  FastMCP context is available
+- performs the reload
+- sends tool/resource/prompt list-changed MCP notifications to the current
+  client session when a FastMCP context is available
+
+Remaining cleanup:
+
+- also emit internal catalog/tool/resource/prompt change events from the
+  transparent runtime path, so broker history and protocol notifications share
+  one source of truth
 
 This is intentionally local. It does not require solving upstream notification
 forwarding.

src/wf_mcp/notifications/__init__.py

@@ -0,0 +1,13 @@
+from .mapping import map_event_to_notifications
+from .sink import (
+    FastMcpContextNotificationSink,
+    NotificationSink,
+    RecordingNotificationSink,
+)
+
+__all__ = [
+    "FastMcpContextNotificationSink",
+    "NotificationSink",
+    "RecordingNotificationSink",
+    "map_event_to_notifications",
+]

src/wf_mcp/notifications/mapping.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import mcp.types as mcp_types
+
+from wf_mcp.events import McpEvent
+
+
+def map_event_to_notifications(event: McpEvent) -> list[mcp_types.ServerNotification]:
+    """Project broker-local events into MCP server notifications.
+
+    The event bus is intentionally protocol-neutral. This function is the
+    boundary where local capability changes become official MCP notification
+    payloads that a FastMCP transport can later send to connected clients.
+    """
+    notification = _list_changed_notification(event)
+    if notification is None:
+        return []
+    return [mcp_types.ServerNotification(notification)]
+
+
+def _list_changed_notification(
+    event: McpEvent,
+) -> (
+    mcp_types.ToolListChangedNotification
+    | mcp_types.ResourceListChangedNotification
+    | mcp_types.PromptListChangedNotification
+    | None
+):
+    if event.kind == "tools_changed":
+        return mcp_types.ToolListChangedNotification()
+    if event.kind == "resources_changed":
+        return mcp_types.ResourceListChangedNotification()
+    if event.kind == "prompts_changed":
+        return mcp_types.PromptListChangedNotification()
+    return None

src/wf_mcp/notifications/sink.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Protocol
+
+import mcp.types as mcp_types
+
+from wf_mcp.events import McpEvent
+
+from .mapping import map_event_to_notifications
+
+
+class NotificationSink(Protocol):
+    """Consumes local events and emits or records MCP notifications."""
+
+    def __call__(self, event: McpEvent) -> None: ...
+
+
+class RecordingNotificationSink:
+    """Test sink that records MCP notifications projected from local events."""
+
+    def __init__(self) -> None:
+        self._notifications: list[mcp_types.ServerNotification] = []
+
+    def __call__(self, event: McpEvent) -> None:
+        self._notifications.extend(map_event_to_notifications(event))
+
+    def list_notifications(self) -> list[mcp_types.ServerNotification]:
+        """Return a defensive copy of projected notifications."""
+        return list(self._notifications)
+
+
+class FastMcpNotificationContext(Protocol):
+    """Small protocol for the FastMCP context method we need."""
+
+    async def send_notification(
+        self,
+        notification: mcp_types.ServerNotificationType,
+    ) -> None: ...
+
+
+class FastMcpContextNotificationSink:
+    """Send projected local events through a request-scoped FastMCP context."""
+
+    def __init__(self, context: FastMcpNotificationContext) -> None:
+        self._context = context
+
+    async def send_event(self, event: McpEvent) -> None:
+        for notification in map_event_to_notifications(event):
+            await self._context.send_notification(notification.root)

src/wf_mcp/transparent_proxy/admin.py

@@ -2,9 +2,11 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastmcp import FastMCP
+from fastmcp import Context, FastMCP
 
 from ..admin_surface import ProxyAdminRuntime, TransparentAdminHandlers
+from ..events import make_event
+from ..notifications import FastMcpContextNotificationSink
 
 
 def create_proxy_admin_server(
@@ -44,8 +46,10 @@ def create_proxy_admin_server(
             "Reload the config file and remount enabled upstream MCP connections."
         ),
     )
-    async def reload_config() -> dict[str, Any]:
-        return handlers.reload_config()
+    async def reload_config(ctx: Context) -> dict[str, Any]:
+        result = handlers.reload_config()
+        await _send_reload_notifications(ctx)
+        return result
 
     @admin.tool(
         title="List Proxy Tools",
@@ -134,3 +138,11 @@ def create_proxy_admin_server(
         return handlers.remove_connection(connection_id)
 
     return admin
+
+
+async def _send_reload_notifications(ctx: Context) -> None:
+    """Notify the current client that reload may have changed visible capabilities."""
+    sink = FastMcpContextNotificationSink(ctx)
+    await sink.send_event(make_event("tools_changed"))
+    await sink.send_event(make_event("resources_changed"))
+    await sink.send_event(make_event("prompts_changed"))

tests/wf_mcp/test_notifications.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import asyncio
+
+import mcp.types as mcp_types
+
+from wf_mcp.events import EventBus, make_event
+from wf_mcp.notifications import (
+    FastMcpContextNotificationSink,
+    RecordingNotificationSink,
+    map_event_to_notifications,
+)
+
+
+class FakeFastMcpContext:
+    def __init__(self) -> None:
+        self.sent: list[mcp_types.ServerNotificationType] = []
+
+    async def send_notification(
+        self,
+        notification: mcp_types.ServerNotificationType,
+    ) -> None:
+        self.sent.append(notification)
+
+
+def test_maps_capability_change_events_to_mcp_list_changed_notifications() -> None:
+    tool_event = make_event("tools_changed", connection_id="demo.personal")
+    resource_event = make_event("resources_changed", connection_id="demo.personal")
+    prompt_event = make_event("prompts_changed", connection_id="demo.personal")
+
+    tool_notifications = map_event_to_notifications(tool_event)
+    resource_notifications = map_event_to_notifications(resource_event)
+    prompt_notifications = map_event_to_notifications(prompt_event)
+
+    assert isinstance(tool_notifications[0].root, mcp_types.ToolListChangedNotification)
+    assert tool_notifications[0].root.method == "notifications/tools/list_changed"
+    assert isinstance(
+        resource_notifications[0].root,
+        mcp_types.ResourceListChangedNotification,
+    )
+    assert resource_notifications[0].root.method == "notifications/resources/list_changed"
+    assert isinstance(
+        prompt_notifications[0].root,
+        mcp_types.PromptListChangedNotification,
+    )
+    assert prompt_notifications[0].root.method == "notifications/prompts/list_changed"
+
+
+def test_ignores_events_that_do_not_have_an_mcp_notification_projection() -> None:
+    event = make_event("workflow_artifact_saved", workflow_name="demo")
+
+    assert map_event_to_notifications(event) == []
+
+
+def test_recording_notification_sink_projects_events_from_event_bus() -> None:
+    bus = EventBus()
+    sink = RecordingNotificationSink()
+    bus.subscribe(sink)
+
+    bus.publish(make_event("tools_changed", connection_id="demo.personal"))
+    bus.publish(make_event("workflow_deployment_saved", workflow_name="demo"))
+    bus.publish(make_event("prompts_changed", connection_id="demo.personal"))
+
+    notifications = sink.list_notifications()
+    assert len(notifications) == 2
+    assert notifications[0].root.method == "notifications/tools/list_changed"
+    assert notifications[1].root.method == "notifications/prompts/list_changed"
+
+
+def test_fastmcp_context_notification_sink_sends_projected_notifications() -> None:
+    context = FakeFastMcpContext()
+    sink = FastMcpContextNotificationSink(context)
+
+    async def run() -> None:
+        await sink.send_event(
+            make_event("resources_changed", connection_id="demo.personal")
+        )
+        await sink.send_event(make_event("workflow_artifact_saved", workflow_name="demo"))
+
+    asyncio.run(run())
+
+    assert len(context.sent) == 1
+    assert isinstance(context.sent[0], mcp_types.ResourceListChangedNotification)
+    assert context.sent[0].method == "notifications/resources/list_changed"

tests/wf_mcp/test_transparent_proxy.py

@@ -5,6 +5,7 @@ import json
 import sys
 from typing import Any
 
+import mcp.types as mcp_types
 import pytest
 
 from wf_mcp.models import BrokerConfig, ConnectionConfig
@@ -446,3 +447,37 @@ def test_transparent_proxy_admin_reload_remounts_connections() -> None:
             assert _structured(result) == {"echoed": "reloaded"}
 
     asyncio.run(run_proxy())
+
+
+def test_transparent_proxy_admin_reload_sends_list_changed_notifications() -> None:
+    tmp_path = local_temp_root() / "transparent_proxy_reload_notification_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": [],
+            }
+        ),
+        encoding="utf-8",
+    )
+    config = load_broker_config(config_path)
+    notifications: list[mcp_types.ServerNotification] = []
+
+    async def message_handler(message: object) -> None:
+        if isinstance(message, mcp_types.ServerNotification):
+            notifications.append(message)
+
+    async def run_proxy() -> None:
+        client = create_transparent_proxy_client(config, config_path=config_path)
+        client._session_kwargs["message_handler"] = message_handler
+        async with client:
+            await client.call_tool("wf.admin.reload_config")
+
+    asyncio.run(run_proxy())
+
+    methods = [notification.root.method for notification in notifications]
+    assert "notifications/tools/list_changed" in methods
+    assert "notifications/resources/list_changed" in methods
+    assert "notifications/prompts/list_changed" in methods