lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: build mcp server from workflow config

Browse Source
This commit is contained in:
lda
2026-06-05 01:29:10 +07:00 Verified
parent 0b9ff5fcd1
commit f0346c6735
11 changed files with 285 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+4
View File
@@ -181,6 +181,10 @@ implementation state.
`kind: "mcp"` entries with stdio/http transport, auth reference, metadata,
enabled flag, and `locked` / `seed` ownership policy. Runtime composition
from these entries is the next slice.
Runtime bridge complete: neutral `kind: "mcp"` source entries can now build
the MCP-backed `WorkflowServer`. `--mcp-config` remains supported as a
legacy compatibility path while new configs should prefer
`server.sources[]`.
- Transport package boundary cleanup follows the config migration. The current
`wf-rpc-server --mcp-config` hookup proves the product path but makes
`wf_transport_rpc_http.cli` import `wf_mcp.broker`, tripping the existing
docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md
+4
View File
@@ -118,6 +118,10 @@ Model slice complete when `wf_config.server.sources[]` accepts `kind: "mcp"`
entries. The next slice converts those neutral source entries into MCP
broker runtime connections and server composition.
Runtime bridge complete when `wf-rpc-server --config <path>` can compose an
MCP-backed server from neutral `server.sources[]` entries. `--mcp-config`
remains a compatibility alias until existing users migrate.
First slice should not include:
- live upstream MCP source management
docs/wf_cli.md
+23
View File
@@ -21,6 +21,29 @@ Start a local/static JSON-RPC workflow server:
wf-rpc-server --store-root .wf_store --host 127.0.0.1 --port 8765
```
Prefer neutral workflow config for new MCP-backed servers:
```json
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": ".wf_store"},
"transports": [{"kind": "rpc_http", "host": "127.0.0.1", "port": 8765}],
"sources": [
{
"kind": "mcp",
"id": "everything.default",
"provider": "everything",
"account": "default",
"transport": {"kind": "stdio", "command": "uvx", "args": ["mcp-server-everything"]}
}
]
}
}
```
`--mcp-config` is still accepted for legacy broker config files.
Start a JSON-RPC server backed by MCP broker config and MCP-capable sources:
```bash
src/wf_mcp/broker/__init__.py
+2
View File
@@ -7,6 +7,7 @@ from .discovery import (
from .events import McpEvent, make_event
from .server import (
build_workflow_server_from_config,
build_workflow_server_from_workflow_config,
create_broker_server,
workflow_server_from_service,
)
@@ -21,6 +22,7 @@ __all__ = [
"WfMcpService",
"build_service_from_config",
"build_workflow_server_from_config",
"build_workflow_server_from_workflow_config",
"create_broker_server",
"discover_connection_capabilities",
"load_broker_config",
src/wf_mcp/broker/config.py
+21 -1
View File
@@ -4,12 +4,13 @@ import json
from pathlib import Path
from wf_api import file_workflow_stores
from wf_config import WorkflowConfigFile
from ..control import BrokerConfigFile
from ..models import BrokerConfig
from ..runtime import McpRuntimePool, PersistentSessionFactory
from ..sdk import McpSdkAdapter
from ..source_registry import FileSourceRegistryStore
from ..source_registry import FileSourceRegistryStore, workflow_mcp_source_to_connection_config
from ..storage import FileStore
from .service import WfMcpService
@@ -21,6 +22,18 @@ def load_broker_config(path: str | Path) -> BrokerConfig:
return BrokerConfigFile.model_validate(data).to_runtime(config_path=config_path)
def broker_config_from_workflow_config(config: WorkflowConfigFile) -> BrokerConfig:
"""Create MCP broker runtime config from neutral workflow server config."""
return BrokerConfig(
store_root=config.server.store.root,
connections=[
workflow_mcp_source_to_connection_config(source)
for source in config.server.sources
if getattr(source, "kind", None) == "mcp"
],
)
def build_service_from_config(config: BrokerConfig) -> WfMcpService:
"""Create a broker service with SDK adapters for configured connections."""
runtime_factory = PersistentSessionFactory()
@@ -44,3 +57,10 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
if connection.server not in service.adapters:
service.register_adapter(connection.server, McpSdkAdapter())
return service
__all__ = [
"broker_config_from_workflow_config",
"build_service_from_config",
"load_broker_config",
]
src/wf_mcp/broker/server.py
+12 -1
View File
@@ -12,8 +12,9 @@ from wf_api import (
from wf_api.stores import WorkflowStores
from wf_server import WorkflowServer, WorkflowServerConfig
from wf_config import WorkflowConfigFile
from .artifact_tools import register_artifact_tools
from .config import build_service_from_config
from .config import broker_config_from_workflow_config, build_service_from_config
from .prompts import register_broker_prompts
from .resources import register_broker_resources
from .service import WfMcpService
@@ -101,8 +102,18 @@ def build_workflow_server_from_config(config: BrokerConfig) -> WorkflowServer:
)
def build_workflow_server_from_workflow_config(
config: WorkflowConfigFile,
) -> WorkflowServer:
"""Build an MCP-backed WorkflowServer from neutral workflow config sources."""
return build_workflow_server_from_config(
broker_config_from_workflow_config(config)
)
__all__ = [
"build_workflow_server_from_config",
"build_workflow_server_from_workflow_config",
"create_broker_server",
"workflow_server_from_service",
]
src/wf_mcp/source_registry.py
+39
View File
@@ -160,6 +160,44 @@ def connection_config_to_registry_entry(
)
def workflow_mcp_source_to_connection_config(source: object) -> ConnectionConfig:
"""Convert neutral wf_config MCP source config into a broker connection.
Keep this adapter in wf_mcp because the output is MCP broker runtime state.
The input is intentionally typed as object to avoid making wf_mcp's public
registry module part of wf_config's import graph.
"""
from .models import ConnectionConfig
if getattr(source, "kind", None) != "mcp":
raise ValueError("expected wf_config MCP source")
for field in ("id", "provider", "account", "enabled", "ownership"):
if getattr(source, field, None) is None:
raise ValueError(f"wf_config MCP source missing required field: {field}")
transport = getattr(source, "transport")
metadata = dict(getattr(source, "metadata", {}))
metadata.update(
{
"transport": transport.model_dump(mode="json"),
"source_registry": False,
}
)
profile = getattr(source, "profile", None)
if profile is not None:
metadata["profile"] = profile
auth_ref = getattr(source, "auth_ref", None)
if auth_ref is not None:
metadata["auth_ref"] = auth_ref
return ConnectionConfig(
id=getattr(source, "id"),
server=getattr(source, "provider"),
account=getattr(source, "account"),
enabled=getattr(source, "enabled"),
metadata=metadata,
source_config_ownership=getattr(source, "ownership"),
)
__all__ = [
"FileSourceRegistryStore",
"HttpSourceTransport",
@@ -170,4 +208,5 @@ __all__ = [
"StdioSourceTransport",
"connection_config_to_registry_entry",
"registry_entry_to_connection_config",
"workflow_mcp_source_to_connection_config",
]
src/wf_transport_rpc_http/cli.py
+11 -1
View File
@@ -11,7 +11,11 @@ from wf_config import (
load_workflow_config,
)
from wf_mcp.broker import build_workflow_server_from_config, load_broker_config
from wf_mcp.broker import (
build_workflow_server_from_config,
build_workflow_server_from_workflow_config,
load_broker_config,
)
from wf_server import build_local_static_workflow_server
@@ -66,6 +70,12 @@ def serve(
if config is not None:
workflow_config = load_workflow_config(config)
has_mcp_sources = any(
getattr(source, "kind", None) == "mcp"
for source in workflow_config.server.sources
)
if server is None and has_mcp_sources:
server = build_workflow_server_from_workflow_config(workflow_config)
store = workflow_config.server.store
if server is None and not isinstance(store, FilesystemStoreConfig):
raise typer.BadParameter(
tests/wf_mcp/test_workflow_config_bridge.py
+71
View File
@@ -0,0 +1,71 @@
from __future__ import annotations
from wf_config import WorkflowConfigFile
from wf_mcp.broker.config import broker_config_from_workflow_config
def test_broker_config_from_workflow_config_converts_mcp_sources(tmp_path) -> None:
workflow_config = WorkflowConfigFile.model_validate(
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": str(tmp_path / "store")},
"sources": [
{
"kind": "mcp",
"id": "everything.default",
"enabled": True,
"provider": "everything",
"account": "default",
"profile": "dev",
"ownership": "seed",
"transport": {
"kind": "stdio",
"command": "uvx",
"args": ["mcp-server-everything"],
"env": {"DEBUG": "1"},
},
"auth_ref": "auth.everything.default",
"metadata": {"description": "Everything test server"},
}
],
},
}
)
broker_config = broker_config_from_workflow_config(workflow_config)
assert broker_config.store_root == tmp_path / "store"
assert len(broker_config.connections) == 1
connection = broker_config.connections[0]
assert connection.id == "everything.default"
assert connection.server == "everything"
assert connection.account == "default"
assert connection.enabled is True
assert connection.source_config_ownership == "seed"
assert connection.metadata["profile"] == "dev"
assert connection.metadata["auth_ref"] == "auth.everything.default"
assert connection.metadata["transport"] == {
"kind": "stdio",
"command": "uvx",
"args": ["mcp-server-everything"],
"env": {"DEBUG": "1"},
}
assert connection.metadata["description"] == "Everything test server"
def test_broker_config_from_workflow_config_ignores_non_mcp_sources(tmp_path) -> None:
workflow_config = WorkflowConfigFile.model_validate(
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": str(tmp_path / "store")},
"sources": [{"kind": "stdlib", "id": "wf.std"}],
},
}
)
broker_config = broker_config_from_workflow_config(workflow_config)
assert broker_config.store_root == tmp_path / "store"
assert broker_config.connections == []
tests/wf_transport_rpc_http/test_cli.py
+62
View File
@@ -281,3 +281,65 @@ def test_rpc_server_cli_mcp_config_with_config_uses_transport_settings(monkeypat
assert captured["host"] == "127.0.0.3"
assert captured["port"] == 7777
assert captured["access_log"] is False
def test_rpc_server_cli_config_with_mcp_source_uses_mcp_builder(
monkeypatch, tmp_path
) -> None:
captured = {}
def fake_build_from_workflow_config(config):
captured["source_kinds"] = [source.kind for source in config.server.sources]
return object()
def fake_create_rpc_app(server, *, rpc_path="/rpc"):
captured["server"] = server
captured["rpc_path"] = rpc_path
return "app"
def fake_run(app, *, host, port, access_log):
captured["run"] = {
"app": app,
"host": host,
"port": port,
"access_log": access_log,
}
monkeypatch.setattr(
"wf_transport_rpc_http.cli.build_workflow_server_from_workflow_config",
fake_build_from_workflow_config,
)
monkeypatch.setattr("wf_transport_rpc_http.cli.create_rpc_app", fake_create_rpc_app)
monkeypatch.setattr("wf_transport_rpc_http.cli.uvicorn.run", fake_run)
config_path = tmp_path / "wf.json"
config_path.write_text(
"""
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": ".wf_store"},
"transports": [{"kind": "rpc_http", "host": "127.0.0.1", "port": 8765}],
"sources": [
{
"kind": "mcp",
"id": "everything.default",
"provider": "everything",
"account": "default",
"transport": {"kind": "stdio", "command": "uvx"}
}
]
}
}
""",
encoding="utf-8",
)
from wf_transport_rpc_http.cli import app
from typer.testing import CliRunner
result = CliRunner().invoke(app, ["--config", str(config_path)])
assert result.exit_code == 0, result.output
assert captured["source_kinds"] == ["mcp"]
assert captured["run"]["app"] == "app"
tests/wf_transport_rpc_http/test_mcp_backed_server_rpc.py
+36 -1
View File
@@ -2,7 +2,8 @@ from __future__ import annotations
import httpx
from wf_mcp.broker.server import build_workflow_server_from_config
from wf_config import WorkflowConfigFile
from wf_mcp.broker.server import build_workflow_server_from_config, build_workflow_server_from_workflow_config
from wf_mcp.models import BrokerConfig, ConnectionConfig
from wf_mcp.source_registry import (
FileSourceRegistryStore,
@@ -109,3 +110,37 @@ async def test_mcp_backed_rpc_reports_connections_and_events(tmp_path) -> None:
event["kind"] == "connection_registered"
for event in events["result"]["events"]
)
async def test_mcp_backed_rpc_can_be_built_from_neutral_workflow_config(
tmp_path,
) -> None:
workflow_config = WorkflowConfigFile.model_validate(
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": str(tmp_path / "store")},
"sources": [
{
"kind": "mcp",
"id": "demo.default",
"provider": "demo",
"account": "default",
"transport": {"kind": "stdio", "command": "demo-server"},
}
],
},
}
)
server = build_workflow_server_from_workflow_config(workflow_config)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
connections = await _rpc(
http_client, "workflow.admin.connections.list", {}
)
assert connections["result"]["connections"][0]["id"] == "demo.default"