wire up stuff to the MCP server

This commit is contained in:
lda
2026-05-11 15:53:30 +07:00 Verified
parent fc7cf887f7
commit 949d461c83
5 changed files with 231 additions and 1 deletions
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
from typing import Any
from mcp.server.fastmcp import FastMCP
from wf_artifacts import (
AvailableCapability,
AvailableSource,
validate_deployment_dependencies,
)
from .service import WfMcpService
def register_artifact_tools(server: FastMCP, service: WfMcpService) -> None:
"""Register stable MCP tools for saved workflow artifact inspection."""
@server.tool()
async def list_workflow_artifacts() -> dict[str, Any]:
if service.artifact_store is None:
return {"nodes": []}
entries = [
service.workflow_artifact_catalog_entry(artifact).model_dump(mode="json")
for artifact in service.artifact_store.list_artifacts()
]
return {"nodes": entries}
@server.tool()
async def inspect_workflow_artifact(
artifact_id: str,
version: int,
) -> dict[str, Any]:
if service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
artifact = service.artifact_store.get_artifact(artifact_id, version)
return artifact.model_dump(mode="json")
@server.tool()
async def validate_workflow_deployment(deployment_id: str) -> dict[str, Any]:
if service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
deployment = service.artifact_store.get_deployment(deployment_id)
artifact = service.artifact_store.get_artifact(
deployment.artifact_id,
deployment.artifact_version,
)
diagnostics = validate_deployment_dependencies(
artifact=artifact,
deployment=deployment,
sources=_available_sources(service),
)
return {
"deployment_id": deployment.id,
"artifact_id": artifact.id,
"artifact_version": artifact.version,
"status": "unrunnable" if diagnostics else "runnable",
"diagnostics": [
diagnostic.model_dump(mode="json") for diagnostic in diagnostics
],
}
def _available_sources(service: WfMcpService) -> list[AvailableSource]:
"""Convert broker capability sources into artifact validation snapshots."""
sources: list[AvailableSource] = []
for source in service.capability_sources.values():
capabilities = {
spec.name.rsplit(".", maxsplit=1)[-1]: AvailableCapability(
name=spec.name.rsplit(".", maxsplit=1)[-1],
kind="node_spec",
input_schema_hash=None,
output_schema_hash=None,
)
for spec in source.capabilities.node_specs.values()
}
sources.append(
AvailableSource(
id=source.id,
enabled=source.enabled,
capabilities=capabilities,
)
)
return sources
+5 -1
View File
@@ -7,6 +7,7 @@ from ..control import BrokerConfigFile
from ..models import BrokerConfig
from ..sdk import McpSdkAdapter
from ..storage import FileStore
from wf_artifacts import FileWorkflowArtifactStore
from .service import WfMcpService
@@ -19,7 +20,10 @@ def load_broker_config(path: str | Path) -> BrokerConfig:
def build_service_from_config(config: BrokerConfig) -> WfMcpService:
"""Create a broker service with SDK adapters for configured connections."""
service = WfMcpService(store=FileStore(config.store_root))
service = WfMcpService(
store=FileStore(config.store_root),
artifact_store=FileWorkflowArtifactStore(config.store_root),
)
for connection in config.connections:
service.register_connection(connection)
if connection.server not in service.adapters:
+2
View File
@@ -6,6 +6,7 @@ from pathlib import Path
from mcp.server.fastmcp import FastMCP
from ..transparent_proxy import create_transparent_proxy_server
from .artifact_tools import register_artifact_tools
from .config import build_service_from_config, load_broker_config
from .prompts import register_broker_prompts
from .resources import register_broker_resources
@@ -25,6 +26,7 @@ def create_broker_server(service: WfMcpService) -> FastMCP:
)
register_broker_tools(server, service)
register_artifact_tools(server, service)
register_broker_resources(server, service)
register_broker_prompts(server, service)
return server
+24
View File
@@ -2,11 +2,19 @@ from __future__ import annotations
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from pydantic import BaseModel
from wf_authoring import NodeReturn, NodeSpec, build_async_registry
from wf_artifacts import (
FileWorkflowArtifactStore,
WorkflowArtifact,
WorkflowArtifactCatalogEntry,
WorkflowArtifactStore,
artifact_catalog_entry,
)
from wf_core import NodeUse, Workflow, execute_workflow_async
from ...connections import ConnectionRegistry, parse_connection_id, qualify_node_name
@@ -40,6 +48,12 @@ from .sources import SpecSource
from .specs import get_qualified_spec, qualify_spec
def _store_root(store: Store) -> Path:
"""Return the file root for stores that expose one, else use local default."""
root = getattr(store, "root", None)
return root if isinstance(root, Path) else Path(".wf_mcp_store")
@dataclass(slots=True)
class WfMcpService:
store: Store
@@ -49,9 +63,12 @@ class WfMcpService:
capability_sources: dict[str, CapabilitySource] = field(default_factory=dict)
events: list[McpEvent] = field(default_factory=list)
include_builtin_specs: bool = True
artifact_store: WorkflowArtifactStore | None = None
def __post_init__(self) -> None:
"""Install broker-local system specs when enabled."""
if self.artifact_store is None:
self.artifact_store = FileWorkflowArtifactStore(_store_root(self.store))
if self.include_builtin_specs:
for source in builtin_sources(self).values():
self.register_spec_source(source)
@@ -230,6 +247,13 @@ class WfMcpService:
"""Return planner-visible node catalog entries from every visible source."""
return self.get_planner_catalog().entries()
def workflow_artifact_catalog_entry(
self,
artifact: WorkflowArtifact,
) -> WorkflowArtifactCatalogEntry:
"""Project a saved workflow artifact as a planner catalog entry."""
return artifact_catalog_entry(artifact)
def get_connection_snapshot(self, connection_id: str) -> CatalogSnapshot | None:
self.connections.get(connection_id)
return self.store.load_catalog(connection_id)