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

discovery

Browse Source
This commit is contained in:
lda
2026-04-29 22:05:41 +07:00 Verified
parent c2681a6976
commit 0461459106
10 changed files with 605 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2 -2
View File
@@ -14,5 +14,5 @@ dev = [
"pytest>=8",
]
# [tool.pytest.ini_options]
# addopts = "-p no:cacheprovider"
[tool.pytest.ini_options]
addopts = "-p no:cacheprovider"
tests/test_wf_mcp.py
-419
View File
@@ -1,419 +0,0 @@
from __future__ import annotations
import asyncio
import os
import sys
from pathlib import Path
from typing import Any
import pytest
from pydantic import BaseModel
from wf_authoring import NodeReturn, node
from wf_core import END, RuntimeContext, RunStatus
from wf_mcp import (
AuthRecord,
ConnectionConfig,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
FileStore,
McpSdkAdapter,
RawWorkflowPlan,
ToolCallResult,
WfMcpService,
)
class EchoInput(BaseModel):
text: str
class EchoOutput(BaseModel):
echoed: str
class FinalizeInput(BaseModel):
echoed: str
class FinalizeOutput(BaseModel):
result: str
@node()
async def echo_tool(payload: EchoInput, ctx: RuntimeContext) -> EchoOutput:
return EchoOutput(echoed=payload.text)
@node(outcomes=("done",))
def finalize_tool(
payload: FinalizeInput, ctx: RuntimeContext
) -> NodeReturn[FinalizeOutput]:
return NodeReturn(
outcome="done",
output=FinalizeOutput(result=f"final:{payload.echoed}"),
)
def _local_temp_root() -> Path:
root = Path("test-artifacts") / "wf_mcp_store"
root.mkdir(parents=True, exist_ok=True)
return root
def _fixture_server_path() -> str:
return str((Path(__file__).parent / "fixtures" / "mcp_echo_server.py").resolve())
def _everything_server_connection() -> ConnectionConfig | None:
transport = os.environ.get("MCP_EVERYTHING_TRANSPORT", "stdio")
if transport == "stdio":
command = os.environ.get("MCP_EVERYTHING_COMMAND")
if not command:
return None
raw_args = os.environ.get("MCP_EVERYTHING_ARGS", "")
args = [arg for arg in raw_args.split(" ") if arg]
metadata: dict[str, Any] = {
"transport": transport,
"command": command,
"args": args,
}
elif transport == "streamable_http":
url = os.environ.get("MCP_EVERYTHING_URL")
if not url:
raise AssertionError(
"MCP_EVERYTHING_URL must be set when MCP_EVERYTHING_TRANSPORT=streamable_http"
)
metadata = {
"transport": "streamable_http",
"url": url,
}
else:
return None
return ConnectionConfig(
id="everything.default",
server="everything",
account="default",
metadata=metadata,
)
class FakeAdapter:
async def list_tools(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredTool]:
return [
DiscoveredTool(
name="echo_tool",
description="Echo text back",
input_schema={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
output_schema={
"type": "object",
"properties": {"echoed": {"type": "string"}},
"required": ["echoed"],
},
)
]
async def list_resources(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredResource]:
return [
DiscoveredResource(
uri="demo://docs/welcome",
name="resource.welcome",
description="Welcome resource",
mime_type="text/plain",
metadata={"kind": "static"},
)
]
async def list_prompts(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]:
return [
DiscoveredPrompt(
name="prompt.summarize",
description="Summarize text",
arguments=[
{
"name": "text",
"required": True,
"description": "Text to summarize",
}
],
metadata={"kind": "template"},
)
]
async def get_connection_metadata(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> dict[str, Any]:
return {
"server": connection.server,
"account": connection.account,
"auth_scheme": auth.scheme if auth is not None else None,
}
async def call_tool(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
tool_name: str,
payload: dict[str, Any],
) -> ToolCallResult:
if tool_name != "echo_tool":
raise KeyError(tool_name)
return ToolCallResult(
outcome="ok",
output={"echoed": str(payload["text"])},
)
def test_file_store_round_trips_auth() -> None:
store = FileStore(_local_temp_root() / "auth_store")
record = AuthRecord(
connection_id="demo.personal",
scheme="oauth",
payload={"token": "secret"},
)
store.save_auth(record)
loaded = store.load_auth("demo.personal")
assert loaded == record
def test_service_builds_namespaced_catalog() -> None:
service = WfMcpService(store=FileStore(_local_temp_root() / "catalog_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool, finalize_tool)
payload = service.get_catalog().as_payload()
names = [node["qualified_name"] for node in payload["nodes"]]
assert names == [
"demo.personal.echo_tool",
"demo.personal.finalize_tool",
]
def test_service_compiles_and_runs_raw_plan() -> None:
service = WfMcpService(store=FileStore(_local_temp_root() / "run_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool, finalize_tool)
plan = RawWorkflowPlan(
name="demo_plan",
input_schema={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
state_schema={
"fields": {
"echoed": {"type": "string"},
"result": {"type": "string"},
}
},
output_schema={
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
start="echo",
nodes=[
{
"id": "echo",
"type": "node",
"node": "demo.personal.echo_tool",
"in_map": {"input.text": "text"},
"out_map": {"echoed": "state.echoed"},
},
{
"id": "finalize",
"type": "node",
"node": "demo.personal.finalize_tool",
"in_map": {"state.echoed": "echoed"},
"out_map": {"result": "state.result"},
},
],
edges=[
{"from": "echo", "outcome": "ok", "to": "finalize"},
{"from": "finalize", "outcome": "done", "to": END},
],
)
run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
assert run.status == RunStatus.COMPLETED
assert run.output == {"result": "final:hello"}
def test_service_refreshes_catalog_from_adapter() -> None:
service = WfMcpService(store=FileStore(_local_temp_root() / "adapter_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.save_auth(
AuthRecord(
connection_id="demo.personal",
scheme="token",
payload={"token": "abc"},
)
)
service.register_adapter("demo", FakeAdapter())
asyncio.run(service.refresh_connection_catalog("demo.personal"))
payload = service.get_catalog().as_payload()
assert payload["nodes"] == [
{
"qualified_name": "demo.personal.echo_tool",
"connection_id": "demo.personal",
"local_name": "echo_tool",
"description": "Echo text back",
"outcomes": ["ok"],
"input_schema": {
"additionalProperties": True,
"properties": {"text": {"title": "Text"}},
"required": ["text"],
"title": "demo.personal_echo_tool_Input",
"type": "object",
},
"output_schema": {
"additionalProperties": True,
"properties": {"echoed": {"title": "Echoed"}},
"required": ["echoed"],
"title": "demo.personal_echo_tool_Output",
"type": "object",
},
}
]
assert payload["resources"] == [
{
"qualified_name": "demo.personal.resource.welcome",
"connection_id": "demo.personal",
"local_name": "resource.welcome",
"uri": "demo://docs/welcome",
"description": "Welcome resource",
"mime_type": "text/plain",
"metadata": {"kind": "static"},
}
]
assert payload["prompts"] == [
{
"qualified_name": "demo.personal.prompt.summarize",
"connection_id": "demo.personal",
"local_name": "prompt.summarize",
"description": "Summarize text",
"arguments": [
{
"name": "text",
"required": True,
"description": "Text to summarize",
}
],
"metadata": {"kind": "template"},
}
]
assert payload["connections"] == [
{
"connection_id": "demo.personal",
"fetched_at_epoch_ms": payload["connections"][0]["fetched_at_epoch_ms"],
"max_age_seconds": 300,
"metadata": {
"server": "demo",
"account": "personal",
"auth_scheme": "token",
},
}
]
def test_mcp_sdk_adapter_lists_and_calls_stdio_tool() -> None:
service = WfMcpService(store=FileStore(_local_temp_root() / "sdk_adapter_store"))
service.register_connection(
ConnectionConfig(
id="fixture.personal",
server="fixture",
account="personal",
metadata={
"transport": "stdio",
"command": sys.executable,
"args": [_fixture_server_path()],
},
)
)
service.register_adapter("fixture", McpSdkAdapter())
try:
asyncio.run(service.refresh_connection_catalog("fixture.personal"))
except PermissionError as exc:
pytest.skip(f"stdio MCP transport is not permitted in this environment: {exc}")
payload = service.get_catalog().as_payload()
assert payload["nodes"][0]["qualified_name"] == "fixture.personal.echo_tool"
adapter = McpSdkAdapter()
try:
result = asyncio.run(
adapter.call_tool(
connection=service.connections.get("fixture.personal"),
auth=None,
tool_name="echo_tool",
payload={"text": "hello"},
)
)
except PermissionError as exc:
pytest.skip(f"stdio MCP transport is not permitted in this environment: {exc}")
assert result.outcome == "ok"
assert result.output == {"echoed": "hello"}
def test_mcp_sdk_adapter_can_probe_everything_server() -> None:
connection = _everything_server_connection()
if connection is None:
pytest.skip(
"set MCP_EVERYTHING_COMMAND to enable the live everything-server integration test"
)
service = WfMcpService(
store=FileStore(_local_temp_root() / "everything_server_store")
)
service.register_connection(connection)
service.register_adapter("everything", McpSdkAdapter())
try:
asyncio.run(service.refresh_connection_catalog("everything.default"))
except PermissionError as exc:
pytest.skip(f"live MCP transport is not permitted in this environment: {exc}")
payload = service.get_catalog().as_payload()
assert payload["nodes"], "everything-server should expose at least one tool"
assert all(
node["qualified_name"].startswith("everything.default.")
for node in payload["nodes"]
)
assert "resources" in payload
assert "prompts" in payload
tests/test_wf_mcp_sdk_adapter.py
+82
View File
@@ -0,0 +1,82 @@
from __future__ import annotations
import asyncio
import pytest
from wf_mcp import ConnectionConfig, FileStore, McpSdkAdapter, WfMcpService
from test_wf_mcp_support import (
everything_server_connection,
fixture_server_path,
local_temp_root,
sys,
)
def test_mcp_sdk_adapter_lists_and_calls_stdio_tool() -> None:
service = WfMcpService(store=FileStore(local_temp_root() / "sdk_adapter_store"))
service.register_connection(
ConnectionConfig(
id="fixture.personal",
server="fixture",
account="personal",
metadata={
"transport": "stdio",
"command": sys.executable,
"args": [fixture_server_path()],
},
)
)
service.register_adapter("fixture", McpSdkAdapter())
try:
asyncio.run(service.refresh_connection_catalog("fixture.personal"))
except PermissionError as exc:
pytest.skip(f"stdio MCP transport is not permitted in this environment: {exc}")
payload = service.get_catalog().as_payload()
assert payload["nodes"][0]["qualified_name"] == "fixture.personal.echo_tool"
adapter = McpSdkAdapter()
try:
result = asyncio.run(
adapter.call_tool(
connection=service.connections.get("fixture.personal"),
auth=None,
tool_name="echo_tool",
payload={"text": "hello"},
)
)
except PermissionError as exc:
pytest.skip(f"stdio MCP transport is not permitted in this environment: {exc}")
assert result.outcome == "ok"
assert result.output == {"echoed": "hello"}
def test_mcp_sdk_adapter_can_probe_everything_server() -> None:
connection = everything_server_connection()
if connection is None:
pytest.skip(
"set MCP_EVERYTHING_COMMAND to enable the live everything-server integration test"
)
service = WfMcpService(
store=FileStore(local_temp_root() / "everything_server_store")
)
service.register_connection(connection)
service.register_adapter("everything", McpSdkAdapter())
try:
asyncio.run(service.refresh_connection_catalog("everything.default"))
except PermissionError as exc:
pytest.skip(f"live MCP transport is not permitted in this environment: {exc}")
payload = service.get_catalog().as_payload()
assert payload["nodes"], "everything-server should expose at least one tool"
assert all(
node["qualified_name"].startswith("everything.default.")
for node in payload["nodes"]
)
assert "resources" in payload
assert "prompts" in payload
tests/test_wf_mcp_service.py
+165
View File
@@ -0,0 +1,165 @@
from __future__ import annotations
import asyncio
from wf_core import END, RunStatus
from wf_mcp import (
AuthRecord,
ConnectionConfig,
FileStore,
RawWorkflowPlan,
WfMcpService,
)
from test_wf_mcp_support import FakeAdapter, echo_tool, finalize_tool, local_temp_root
def test_service_builds_namespaced_catalog() -> None:
service = WfMcpService(store=FileStore(local_temp_root() / "catalog_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool, finalize_tool)
payload = service.get_catalog().as_payload()
names = [node["qualified_name"] for node in payload["nodes"]]
assert names == [
"demo.personal.echo_tool",
"demo.personal.finalize_tool",
]
def test_service_compiles_and_runs_raw_plan() -> None:
service = WfMcpService(store=FileStore(local_temp_root() / "run_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool, finalize_tool)
plan = RawWorkflowPlan(
name="demo_plan",
input_schema={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
state_schema={
"fields": {
"echoed": {"type": "string"},
"result": {"type": "string"},
}
},
output_schema={
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
start="echo",
nodes=[
{
"id": "echo",
"type": "node",
"node": "demo.personal.echo_tool",
"in_map": {"input.text": "text"},
"out_map": {"echoed": "state.echoed"},
},
{
"id": "finalize",
"type": "node",
"node": "demo.personal.finalize_tool",
"in_map": {"state.echoed": "echoed"},
"out_map": {"result": "state.result"},
},
],
edges=[
{"from": "echo", "outcome": "ok", "to": "finalize"},
{"from": "finalize", "outcome": "done", "to": END},
],
)
run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
assert run.status == RunStatus.COMPLETED
assert run.output == {"result": "final:hello"}
def test_service_refreshes_catalog_from_adapter() -> None:
service = WfMcpService(store=FileStore(local_temp_root() / "adapter_store"))
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.save_auth(
AuthRecord(
connection_id="demo.personal",
scheme="token",
payload={"token": "abc"},
)
)
service.register_adapter("demo", FakeAdapter())
asyncio.run(service.refresh_connection_catalog("demo.personal"))
payload = service.get_catalog().as_payload()
assert payload["nodes"] == [
{
"qualified_name": "demo.personal.echo_tool",
"connection_id": "demo.personal",
"local_name": "echo_tool",
"description": "Echo text back",
"outcomes": ["ok"],
"input_schema": {
"additionalProperties": True,
"properties": {"text": {"title": "Text"}},
"required": ["text"],
"title": "demo.personal_echo_tool_Input",
"type": "object",
},
"output_schema": {
"additionalProperties": True,
"properties": {"echoed": {"title": "Echoed"}},
"required": ["echoed"],
"title": "demo.personal_echo_tool_Output",
"type": "object",
},
}
]
assert payload["resources"] == [
{
"qualified_name": "demo.personal.resource.welcome",
"connection_id": "demo.personal",
"local_name": "resource.welcome",
"uri": "demo://docs/welcome",
"description": "Welcome resource",
"mime_type": "text/plain",
"metadata": {"kind": "static"},
}
]
assert payload["prompts"] == [
{
"qualified_name": "demo.personal.prompt.summarize",
"connection_id": "demo.personal",
"local_name": "prompt.summarize",
"description": "Summarize text",
"arguments": [
{
"name": "text",
"required": True,
"description": "Text to summarize",
}
],
"metadata": {"kind": "template"},
}
]
assert payload["connections"] == [
{
"connection_id": "demo.personal",
"fetched_at_epoch_ms": payload["connections"][0]["fetched_at_epoch_ms"],
"max_age_seconds": 300,
"metadata": {
"server": "demo",
"account": "personal",
"auth_scheme": "token",
},
}
]
tests/test_wf_mcp_store.py
+19
View File
@@ -0,0 +1,19 @@
from __future__ import annotations
from wf_mcp import AuthRecord, FileStore
from test_wf_mcp_support import local_temp_root
def test_file_store_round_trips_auth() -> None:
store = FileStore(local_temp_root() / "auth_store")
record = AuthRecord(
connection_id="demo.personal",
scheme="oauth",
payload={"token": "secret"},
)
store.save_auth(record)
loaded = store.load_auth("demo.personal")
assert loaded == record
tests/test_wf_mcp_support.py
+191
View File
@@ -0,0 +1,191 @@
from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import Any
from pydantic import BaseModel
from wf_authoring import NodeReturn, node
from wf_core import RuntimeContext
from wf_mcp import (
AuthRecord,
ConnectionConfig,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
ToolCallResult,
)
class EchoInput(BaseModel):
text: str
class EchoOutput(BaseModel):
echoed: str
class FinalizeInput(BaseModel):
echoed: str
class FinalizeOutput(BaseModel):
result: str
@node()
async def echo_tool(payload: EchoInput, ctx: RuntimeContext) -> EchoOutput:
return EchoOutput(echoed=payload.text)
@node(outcomes=("done",))
def finalize_tool(
payload: FinalizeInput, ctx: RuntimeContext
) -> NodeReturn[FinalizeOutput]:
return NodeReturn(
outcome="done",
output=FinalizeOutput(result=f"final:{payload.echoed}"),
)
def local_temp_root() -> Path:
root = Path("test-artifacts") / "wf_mcp_store"
root.mkdir(parents=True, exist_ok=True)
return root
def fixture_server_path() -> str:
return str((Path(__file__).resolve().parent / "fixtures" / "mcp_echo_server.py"))
def everything_server_connection() -> ConnectionConfig | None:
transport = os.environ.get("MCP_EVERYTHING_TRANSPORT", "stdio")
if transport == "stdio":
command = os.environ.get("MCP_EVERYTHING_COMMAND")
if not command:
return None
raw_args = os.environ.get("MCP_EVERYTHING_ARGS", "")
args = [arg for arg in raw_args.split(" ") if arg]
metadata: dict[str, Any] = {
"transport": transport,
"command": command,
"args": args,
}
elif transport == "streamable_http":
url = os.environ.get("MCP_EVERYTHING_URL")
if not url:
raise AssertionError(
"MCP_EVERYTHING_URL must be set when MCP_EVERYTHING_TRANSPORT=streamable_http"
)
metadata = {
"transport": "streamable_http",
"url": url,
}
else:
return None
return ConnectionConfig(
id="everything.default",
server="everything",
account="default",
metadata=metadata,
)
class FakeAdapter:
async def list_tools(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredTool]:
return [
DiscoveredTool(
name="echo_tool",
description="Echo text back",
input_schema={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
output_schema={
"type": "object",
"properties": {"echoed": {"type": "string"}},
"required": ["echoed"],
},
)
]
async def list_resources(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredResource]:
return [
DiscoveredResource(
uri="demo://docs/welcome",
name="resource.welcome",
description="Welcome resource",
mime_type="text/plain",
metadata={"kind": "static"},
)
]
async def list_prompts(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]:
return [
DiscoveredPrompt(
name="prompt.summarize",
description="Summarize text",
arguments=[
{
"name": "text",
"required": True,
"description": "Text to summarize",
}
],
metadata={"kind": "template"},
)
]
async def get_connection_metadata(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> dict[str, Any]:
return {
"server": connection.server,
"account": connection.account,
"auth_scheme": auth.scheme if auth is not None else None,
}
async def call_tool(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
tool_name: str,
payload: dict[str, Any],
) -> ToolCallResult:
if tool_name != "echo_tool":
raise KeyError(tool_name)
return ToolCallResult(
outcome="ok",
output={"echoed": str(payload["text"])},
)
__all__ = [
"FakeAdapter",
"echo_tool",
"everything_server_connection",
"finalize_tool",
"fixture_server_path",
"local_temp_root",
"sys",
]
wf_mcp/__init__.py
+8
View File
@@ -7,6 +7,11 @@ from .adapters import (
)
from .catalog import CombinedCatalog
from .connections import ConnectionRegistry, parse_connection_id, qualify_node_name
from .discovery import (
DiscoveredConnectionCapabilities,
discover_connection_capabilities,
specs_from_discovered_tools,
)
from .models import (
AuthRecord,
CatalogNodeEntry,
@@ -31,6 +36,7 @@ __all__ = [
"CombinedCatalog",
"ConnectionConfig",
"ConnectionRegistry",
"DiscoveredConnectionCapabilities",
"DiscoveredPrompt",
"DiscoveredResource",
"DiscoveredTool",
@@ -40,7 +46,9 @@ __all__ = [
"Store",
"ToolCallResult",
"WfMcpService",
"discover_connection_capabilities",
"parse_connection_id",
"qualify_node_name",
"specs_from_discovered_tools",
"wrap_discovered_tool",
]
wf_mcp/discovery.py
+59
View File
@@ -0,0 +1,59 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from wf_authoring import NodeSpec
from .adapters import (
BackendAdapter,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
)
from .models import AuthRecord, ConnectionConfig
from .wrappers import wrap_discovered_tool
@dataclass(slots=True)
class DiscoveredConnectionCapabilities:
tools: list[DiscoveredTool] = field(default_factory=list)
resources: list[DiscoveredResource] = field(default_factory=list)
prompts: list[DiscoveredPrompt] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
async def discover_connection_capabilities(
*,
connection: ConnectionConfig,
auth: AuthRecord | None,
adapter: BackendAdapter,
) -> DiscoveredConnectionCapabilities:
tools = await adapter.list_tools(connection, auth)
resources = await adapter.list_resources(connection, auth)
prompts = await adapter.list_prompts(connection, auth)
metadata = await adapter.get_connection_metadata(connection, auth)
return DiscoveredConnectionCapabilities(
tools=tools,
resources=resources,
prompts=prompts,
metadata=metadata,
)
def specs_from_discovered_tools(
*,
connection: ConnectionConfig,
auth: AuthRecord | None,
adapter: BackendAdapter,
tools: list[DiscoveredTool],
) -> list[NodeSpec[Any, Any]]:
return [
wrap_discovered_tool(
connection=connection,
auth=auth,
adapter=adapter,
tool=tool,
)
for tool in tools
]
wf_mcp/service.py
+15 -17
View File
@@ -10,9 +10,9 @@ from wf_core import NodeUse, Workflow, execute_workflow_async
from .adapters import BackendAdapter
from .catalog import CombinedCatalog, snapshot_from_specs
from .connections import ConnectionRegistry, parse_connection_id, qualify_node_name
from .discovery import discover_connection_capabilities, specs_from_discovered_tools
from .models import AuthRecord, CatalogSnapshot, ConnectionConfig, RawWorkflowPlan
from .store import Store
from .wrappers import wrap_discovered_tool
def _qualify_spec(connection_id: str, spec: NodeSpec[Any, Any]) -> NodeSpec[Any, Any]:
@@ -92,19 +92,17 @@ class WfMcpService:
raise KeyError(f"no adapter registered for server {connection.server!r}")
auth = self.load_auth(connection_id)
tools = await adapter.list_tools(connection, auth)
resources = await adapter.list_resources(connection, auth)
prompts = await adapter.list_prompts(connection, auth)
metadata = await adapter.get_connection_metadata(connection, auth)
specs = [
wrap_discovered_tool(
connection=connection,
auth=auth,
adapter=adapter,
tool=tool,
)
for tool in tools
]
capabilities = await discover_connection_capabilities(
connection=connection,
auth=auth,
adapter=adapter,
)
specs = specs_from_discovered_tools(
connection=connection,
auth=auth,
adapter=adapter,
tools=capabilities.tools,
)
self.register_specs(
connection_id,
*specs,
@@ -113,9 +111,9 @@ class WfMcpService:
snapshot = snapshot_from_specs(
connection_id,
specs=self.specs_by_connection.get(connection_id, {}),
resources=resources,
prompts=prompts,
metadata=metadata,
resources=capabilities.resources,
prompts=capabilities.prompts,
metadata=capabilities.metadata,
fetched_at_epoch_ms=int(time.time() * 1000),
max_age_seconds=max_age_seconds or self.default_catalog_max_age_seconds,
)
wf_mcp_plan.md
+64
View File
@@ -380,6 +380,70 @@ If we follow the plan above, the next likely modules are:
Whether these stay as separate files or fold back into `models.py` / `service.py` should be driven by clarity, not purity.
## Auth and storage
Auth should remain behind a pluggable store interface.
First implementation:
- file-backed store
Expected future replacements:
- database-backed store
- encrypted local store
- secret-manager-backed store
The store should persist:
- auth records
- cached capability snapshots
It may later persist:
- saved plans/workflows
- job specs
- run metadata
## Execution model
Near-term execution stance:
- async-first at the MCP layer
- use existing async workflow runtime
- keep workflow runs mostly in memory
- leave room for future scheduled/offline execution
The important offline use case is:
- build workflow once
- execute it later without the LLM in the loop
This is closer to scheduled automation than to interactive planning.
## Public API direction
`wf_mcp` should expose two explicit entrypoints.
### 1. Convenient/build-style API
For human or higher-level service use.
Examples:
- build workflow from selected catalog items
- helper methods around namespaced tools/resources/prompts
### 2. Raw plan API
For client LLM use.
This should accept plans that:
- reference namespaced capabilities directly
- avoid raw `NodeDef` authoring
- still compile down to `wf_core.Workflow`
## Test organization direction
The current test suite is still small enough to live in two top-level files, but it will get noisy if `wf_mcp` grows beyond tools.