More! Events!
This commit is contained in:
@@ -325,10 +325,10 @@ config_reloaded
|
||||
- [ ] Add event kinds for capability changes:
|
||||
- `source_enabled`
|
||||
- `source_disabled`
|
||||
- `catalog_changed`
|
||||
- `tools_changed`
|
||||
- `resources_changed`
|
||||
- `prompts_changed`
|
||||
- [x] `catalog_changed`
|
||||
- [x] `tools_changed`
|
||||
- [x] `resources_changed`
|
||||
- [x] `prompts_changed`
|
||||
- [ ] Do not emit MCP notifications yet unless the server/session API is
|
||||
clearly available. This phase creates the source of truth.
|
||||
|
||||
|
||||
@@ -69,7 +69,9 @@ class BrokerAdminHandlers:
|
||||
params: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
try:
|
||||
return await self.service.invoke_method(connection_id, method, params=params)
|
||||
return await self.service.invoke_method(
|
||||
connection_id, method, params=params
|
||||
)
|
||||
except Exception as exc:
|
||||
return {
|
||||
"connection_id": connection_id,
|
||||
@@ -104,4 +106,4 @@ class BrokerAdminHandlers:
|
||||
}
|
||||
|
||||
def get_broker_events(self) -> list[dict[str, Any]]:
|
||||
return [asdict(event) for event in self.service.list_events()]
|
||||
return [asdict(event) for event in self.service.list_events()]
|
||||
|
||||
@@ -33,4 +33,4 @@ class ConfigManager(Protocol):
|
||||
enabled: bool,
|
||||
) -> dict[str, Any]: ...
|
||||
|
||||
def remove_connection(self, connection_id: str) -> dict[str, Any]: ...
|
||||
def remove_connection(self, connection_id: str) -> dict[str, Any]: ...
|
||||
|
||||
@@ -108,4 +108,4 @@ class TransparentAdminHandlers:
|
||||
)
|
||||
|
||||
def remove_connection(self, connection_id: str) -> dict[str, Any]:
|
||||
return self.runtime.require_manager().remove_connection(connection_id)
|
||||
return self.runtime.require_manager().remove_connection(connection_id)
|
||||
|
||||
@@ -141,6 +141,7 @@ class WfMcpService:
|
||||
connection_id: str,
|
||||
*specs: NodeSpec[Any, Any],
|
||||
max_age_seconds: int | None = None,
|
||||
emit_change_events: bool = True,
|
||||
) -> None:
|
||||
self.connections.get(connection_id)
|
||||
qualified_specs = {
|
||||
@@ -181,6 +182,12 @@ class WfMcpService:
|
||||
payload={"node_count": len(qualified_specs)},
|
||||
)
|
||||
)
|
||||
if emit_change_events:
|
||||
self._record_catalog_change_events(
|
||||
connection_id,
|
||||
snapshot,
|
||||
reason="specs_registered",
|
||||
)
|
||||
|
||||
def get_catalog(self) -> CombinedCatalog:
|
||||
snapshots: dict[str, CatalogSnapshot] = {}
|
||||
@@ -504,6 +511,7 @@ class WfMcpService:
|
||||
connection_id,
|
||||
*specs,
|
||||
max_age_seconds=max_age_seconds,
|
||||
emit_change_events=False,
|
||||
)
|
||||
snapshot = snapshot_from_specs(
|
||||
connection_id,
|
||||
@@ -518,6 +526,11 @@ class WfMcpService:
|
||||
max_age_seconds=max_age_seconds or self.default_catalog_max_age_seconds,
|
||||
)
|
||||
self.store.save_catalog(snapshot)
|
||||
self._record_catalog_change_events(
|
||||
connection_id,
|
||||
snapshot,
|
||||
reason="catalog_refresh",
|
||||
)
|
||||
self._record_event(
|
||||
make_event(
|
||||
"catalog_refresh_completed",
|
||||
@@ -670,3 +683,51 @@ class WfMcpService:
|
||||
|
||||
def _record_event(self, event: McpEvent) -> None:
|
||||
self.event_bus.publish(event)
|
||||
|
||||
def _record_catalog_change_events(
|
||||
self,
|
||||
connection_id: str,
|
||||
snapshot: CatalogSnapshot,
|
||||
*,
|
||||
reason: str,
|
||||
) -> None:
|
||||
"""Emit local change events that future MCP notifications can project."""
|
||||
counts = {
|
||||
"node_count": len(snapshot.nodes),
|
||||
"resource_count": len(snapshot.resources),
|
||||
"prompt_count": len(snapshot.prompts),
|
||||
}
|
||||
if snapshot.nodes:
|
||||
self._record_event(
|
||||
make_event(
|
||||
"tools_changed",
|
||||
connection_id=connection_id,
|
||||
payload={"reason": reason, "node_count": counts["node_count"]},
|
||||
)
|
||||
)
|
||||
if snapshot.resources:
|
||||
self._record_event(
|
||||
make_event(
|
||||
"resources_changed",
|
||||
connection_id=connection_id,
|
||||
payload={
|
||||
"reason": reason,
|
||||
"resource_count": counts["resource_count"],
|
||||
},
|
||||
)
|
||||
)
|
||||
if snapshot.prompts:
|
||||
self._record_event(
|
||||
make_event(
|
||||
"prompts_changed",
|
||||
connection_id=connection_id,
|
||||
payload={"reason": reason, "prompt_count": counts["prompt_count"]},
|
||||
)
|
||||
)
|
||||
self._record_event(
|
||||
make_event(
|
||||
"catalog_changed",
|
||||
connection_id=connection_id,
|
||||
payload={"reason": reason, **counts},
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Sequence
|
||||
|
||||
from wf_mcp.broker import WfMcpService
|
||||
from wf_mcp.events import EventBus, InMemoryEventSink, make_event
|
||||
from wf_mcp.events import EventBus, InMemoryEventSink, McpEvent, make_event
|
||||
from wf_mcp.models import ConnectionConfig
|
||||
from wf_mcp.storage import FileStore
|
||||
|
||||
from .test_support import local_temp_root
|
||||
from .test_support import FakeAdapter, echo_tool, local_temp_root
|
||||
|
||||
|
||||
def test_event_bus_fans_out_to_subscribers() -> None:
|
||||
@@ -35,3 +38,57 @@ def test_service_records_events_through_event_bus() -> None:
|
||||
|
||||
assert service.list_events()[0].kind == "connection_registered"
|
||||
assert sink.list_events()[0] is service.list_events()[0]
|
||||
|
||||
|
||||
def test_register_specs_emits_tool_and_catalog_change_events() -> None:
|
||||
service = WfMcpService(store=FileStore(local_temp_root() / "spec_change_store"))
|
||||
service.register_connection(
|
||||
ConnectionConfig(id="demo.personal", server="demo", account="personal")
|
||||
)
|
||||
|
||||
service.register_specs("demo.personal", echo_tool)
|
||||
|
||||
events = service.list_events()
|
||||
event_kinds = [event.kind for event in events]
|
||||
tools_changed = [
|
||||
event
|
||||
for event in events
|
||||
if event.kind == "tools_changed" and event.connection_id == "demo.personal"
|
||||
]
|
||||
catalog_changed = [
|
||||
event
|
||||
for event in events
|
||||
if event.kind == "catalog_changed" and event.connection_id == "demo.personal"
|
||||
]
|
||||
assert "tools_changed" in event_kinds
|
||||
assert "catalog_changed" in event_kinds
|
||||
assert tools_changed[0].payload["node_count"] == 1
|
||||
assert catalog_changed[0].payload["reason"] == "specs_registered"
|
||||
|
||||
|
||||
def test_refresh_catalog_emits_capability_change_events() -> None:
|
||||
service = WfMcpService(store=FileStore(local_temp_root() / "refresh_change_store"))
|
||||
service.register_connection(
|
||||
ConnectionConfig(id="demo.personal", server="demo", account="personal")
|
||||
)
|
||||
service.register_adapter("demo", FakeAdapter())
|
||||
|
||||
asyncio.run(service.refresh_connection_catalog("demo.personal"))
|
||||
|
||||
events = service.list_events()
|
||||
tools_changed = _first_event(events, "tools_changed")
|
||||
resources_changed = _first_event(events, "resources_changed")
|
||||
prompts_changed = _first_event(events, "prompts_changed")
|
||||
catalog_changed = _first_event(events, "catalog_changed")
|
||||
assert tools_changed.connection_id == "demo.personal"
|
||||
assert tools_changed.payload["node_count"] == 1
|
||||
assert resources_changed.payload["resource_count"] == 1
|
||||
assert prompts_changed.payload["prompt_count"] == 1
|
||||
assert catalog_changed.payload["reason"] == "catalog_refresh"
|
||||
|
||||
|
||||
def _first_event(events: Sequence[McpEvent], kind: str) -> McpEvent:
|
||||
for event in events:
|
||||
if event.kind == kind:
|
||||
return event
|
||||
raise AssertionError(f"expected event {kind!r}")
|
||||
|
||||
Reference in New Issue
Block a user