start handling the rest of MCP

its a BIG THING god damn.
This commit is contained in:
lda
2026-04-29 21:39:27 +07:00 Verified
parent 6bff79c304
commit c2681a6976
12 changed files with 492 additions and 227 deletions
+2 -2
View File
@@ -14,5 +14,5 @@ dev = [
"pytest>=8", "pytest>=8",
] ]
[tool.pytest.ini_options] # [tool.pytest.ini_options]
addopts = "-p no:cacheprovider" # addopts = "-p no:cacheprovider"
+6
View File
@@ -1,5 +1,11 @@
# lda.chat - Running Design Notes # lda.chat - Running Design Notes
## Related docs
- [authoring_sketch.md](authoring_sketch.md): `wf_authoring` direction, including `@node`, `NodeSpec`, builder ergonomics, async registry seams, and subgraph-as-node.
- [wf_mcp_plan.md](wf_mcp_plan.md): `wf_mcp` direction as a namespaced MCP capability broker plus workflow build/run layer.
- [scratchpad.md](scratchpad.md): rougher design history and intermediate spec notes that fed the current model.
## What is this ## What is this
`lda.chat` is an AI agent that turns natural language requests into executable `lda.chat` is an AI agent that turns natural language requests into executable
+89
View File
@@ -14,6 +14,8 @@ from wf_core import END, RuntimeContext, RunStatus
from wf_mcp import ( from wf_mcp import (
AuthRecord, AuthRecord,
ConnectionConfig, ConnectionConfig,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool, DiscoveredTool,
FileStore, FileStore,
McpSdkAdapter, McpSdkAdapter,
@@ -123,6 +125,52 @@ class FakeAdapter:
) )
] ]
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( async def call_tool(
self, self,
connection: ConnectionConfig, connection: ConnectionConfig,
@@ -262,6 +310,45 @@ def test_service_refreshes_catalog_from_adapter() -> None:
}, },
} }
] ]
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: def test_mcp_sdk_adapter_lists_and_calls_stdio_tool() -> None:
@@ -328,3 +415,5 @@ def test_mcp_sdk_adapter_can_probe_everything_server() -> None:
node["qualified_name"].startswith("everything.default.") node["qualified_name"].startswith("everything.default.")
for node in payload["nodes"] for node in payload["nodes"]
) )
assert "resources" in payload
assert "prompts" in payload
+13 -1
View File
@@ -1,9 +1,17 @@
from .adapters import BackendAdapter, DiscoveredTool, ToolCallResult from .adapters import (
BackendAdapter,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
ToolCallResult,
)
from .catalog import CombinedCatalog from .catalog import CombinedCatalog
from .connections import ConnectionRegistry, parse_connection_id, qualify_node_name from .connections import ConnectionRegistry, parse_connection_id, qualify_node_name
from .models import ( from .models import (
AuthRecord, AuthRecord,
CatalogNodeEntry, CatalogNodeEntry,
CatalogPromptEntry,
CatalogResourceEntry,
CatalogSnapshot, CatalogSnapshot,
ConnectionConfig, ConnectionConfig,
RawWorkflowPlan, RawWorkflowPlan,
@@ -17,10 +25,14 @@ __all__ = [
"AuthRecord", "AuthRecord",
"BackendAdapter", "BackendAdapter",
"CatalogNodeEntry", "CatalogNodeEntry",
"CatalogPromptEntry",
"CatalogResourceEntry",
"CatalogSnapshot", "CatalogSnapshot",
"CombinedCatalog", "CombinedCatalog",
"ConnectionConfig", "ConnectionConfig",
"ConnectionRegistry", "ConnectionRegistry",
"DiscoveredPrompt",
"DiscoveredResource",
"DiscoveredTool", "DiscoveredTool",
"FileStore", "FileStore",
"McpSdkAdapter", "McpSdkAdapter",
+35
View File
@@ -16,6 +16,23 @@ class DiscoveredTool:
metadata: dict[str, Any] = field(default_factory=dict) metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True)
class DiscoveredResource:
uri: str
name: str
description: str | None
mime_type: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True)
class DiscoveredPrompt:
name: str
description: str | None
arguments: list[dict[str, Any]] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True) @dataclass(slots=True)
class ToolCallResult: class ToolCallResult:
outcome: str outcome: str
@@ -30,6 +47,24 @@ class BackendAdapter(Protocol):
auth: AuthRecord | None, auth: AuthRecord | None,
) -> list[DiscoveredTool]: ... ) -> list[DiscoveredTool]: ...
async def list_resources(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredResource]: ...
async def list_prompts(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]: ...
async def get_connection_metadata(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> dict[str, Any]: ...
async def call_tool( async def call_tool(
self, self,
connection: ConnectionConfig, connection: ConnectionConfig,
+84 -2
View File
@@ -5,14 +5,23 @@ from typing import Any
from wf_authoring import NodeCatalog, NodeSpec from wf_authoring import NodeCatalog, NodeSpec
from .adapters import DiscoveredPrompt, DiscoveredResource
from .connections import qualify_node_name from .connections import qualify_node_name
from .models import CatalogNodeEntry, CatalogSnapshot from .models import (
CatalogNodeEntry,
CatalogPromptEntry,
CatalogResourceEntry,
CatalogSnapshot,
)
def snapshot_from_specs( def snapshot_from_specs(
connection_id: str, connection_id: str,
*, *,
specs: dict[str, NodeSpec[Any, Any]], specs: dict[str, NodeSpec[Any, Any]],
resources: list[DiscoveredResource] | None = None,
prompts: list[DiscoveredPrompt] | None = None,
metadata: dict[str, Any] | None = None,
fetched_at_epoch_ms: int, fetched_at_epoch_ms: int,
max_age_seconds: int, max_age_seconds: int,
) -> CatalogSnapshot: ) -> CatalogSnapshot:
@@ -31,11 +40,37 @@ def snapshot_from_specs(
) )
for entry in catalog.entries() for entry in catalog.entries()
] ]
resource_entries = [
CatalogResourceEntry(
qualified_name=qualify_node_name(connection_id, resource.name),
connection_id=connection_id,
local_name=resource.name,
uri=resource.uri,
description=resource.description,
mime_type=resource.mime_type,
metadata=resource.metadata,
)
for resource in resources or []
]
prompt_entries = [
CatalogPromptEntry(
qualified_name=qualify_node_name(connection_id, prompt.name),
connection_id=connection_id,
local_name=prompt.name,
description=prompt.description,
arguments=prompt.arguments,
metadata=prompt.metadata,
)
for prompt in prompts or []
]
return CatalogSnapshot( return CatalogSnapshot(
connection_id=connection_id, connection_id=connection_id,
fetched_at_epoch_ms=fetched_at_epoch_ms, fetched_at_epoch_ms=fetched_at_epoch_ms,
max_age_seconds=max_age_seconds, max_age_seconds=max_age_seconds,
nodes=nodes, nodes=nodes,
resources=resource_entries,
prompts=prompt_entries,
metadata=metadata or {},
) )
@@ -49,6 +84,18 @@ class CombinedCatalog:
result.extend(snapshot.nodes) result.extend(snapshot.nodes)
return sorted(result, key=lambda entry: entry.qualified_name) return sorted(result, key=lambda entry: entry.qualified_name)
def resource_entries(self) -> list[CatalogResourceEntry]:
result: list[CatalogResourceEntry] = []
for snapshot in self.snapshots.values():
result.extend(snapshot.resources)
return sorted(result, key=lambda entry: entry.qualified_name)
def prompt_entries(self) -> list[CatalogPromptEntry]:
result: list[CatalogPromptEntry] = []
for snapshot in self.snapshots.values():
result.extend(snapshot.prompts)
return sorted(result, key=lambda entry: entry.qualified_name)
def as_payload(self) -> dict[str, Any]: def as_payload(self) -> dict[str, Any]:
return { return {
"nodes": [ "nodes": [
@@ -62,5 +109,40 @@ class CombinedCatalog:
"output_schema": entry.output_schema, "output_schema": entry.output_schema,
} }
for entry in self.entries() for entry in self.entries()
] ],
"resources": [
{
"qualified_name": entry.qualified_name,
"connection_id": entry.connection_id,
"local_name": entry.local_name,
"uri": entry.uri,
"description": entry.description,
"mime_type": entry.mime_type,
"metadata": entry.metadata,
}
for entry in self.resource_entries()
],
"prompts": [
{
"qualified_name": entry.qualified_name,
"connection_id": entry.connection_id,
"local_name": entry.local_name,
"description": entry.description,
"arguments": entry.arguments,
"metadata": entry.metadata,
}
for entry in self.prompt_entries()
],
"connections": [
{
"connection_id": snapshot.connection_id,
"fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms,
"max_age_seconds": snapshot.max_age_seconds,
"metadata": snapshot.metadata,
}
for snapshot in sorted(
self.snapshots.values(),
key=lambda snapshot: snapshot.connection_id,
)
],
} }
+60 -1
View File
@@ -8,9 +8,18 @@ from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.client.streamable_http import streamable_http_client from mcp.client.streamable_http import streamable_http_client
from mcp.types import CallToolResult as McpCallToolResult from mcp.types import CallToolResult as McpCallToolResult
from mcp.types import ListPromptsResult, ListResourcesResult
from mcp.types import ListToolsResult, Tool as McpTool from mcp.types import ListToolsResult, Tool as McpTool
from mcp.types import Prompt as McpPrompt
from mcp.types import Resource as McpResource
from .adapters import BackendAdapter, DiscoveredTool, ToolCallResult from .adapters import (
BackendAdapter,
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
ToolCallResult,
)
from .models import AuthRecord, ConnectionConfig from .models import AuthRecord, ConnectionConfig
@@ -39,6 +48,28 @@ def _tool_to_discovered(tool: McpTool) -> DiscoveredTool:
) )
def _resource_to_discovered(resource: McpResource) -> DiscoveredResource:
return DiscoveredResource(
uri=str(resource.uri),
name=str(resource.uri),
description=resource.description,
mime_type=resource.mimeType,
metadata=resource.model_dump(by_alias=True),
)
def _prompt_to_discovered(prompt: McpPrompt) -> DiscoveredPrompt:
arguments = [
argument.model_dump(by_alias=True) for argument in prompt.arguments or []
]
return DiscoveredPrompt(
name=prompt.name,
description=prompt.description,
arguments=arguments,
metadata=prompt.model_dump(by_alias=True),
)
def _tool_result_to_call_result(result: McpCallToolResult) -> ToolCallResult: def _tool_result_to_call_result(result: McpCallToolResult) -> ToolCallResult:
if result.structuredContent is not None: if result.structuredContent is not None:
output = result.structuredContent output = result.structuredContent
@@ -107,6 +138,34 @@ class McpSdkAdapter(BackendAdapter):
result: ListToolsResult = await session.list_tools() result: ListToolsResult = await session.list_tools()
return [_tool_to_discovered(tool) for tool in result.tools] return [_tool_to_discovered(tool) for tool in result.tools]
async def list_resources(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredResource]:
async with self._session(connection, auth) as session:
result: ListResourcesResult = await session.list_resources()
return [_resource_to_discovered(resource) for resource in result.resources]
async def list_prompts(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]:
async with self._session(connection, auth) as session:
result: ListPromptsResult = await session.list_prompts()
return [_prompt_to_discovered(prompt) for prompt in result.prompts]
async def get_connection_metadata(
self,
connection: ConnectionConfig,
auth: AuthRecord | None,
) -> dict[str, Any]:
return {
"server": connection.server,
"transport": connection.metadata.get("transport", "stdio"),
}
async def call_tool( async def call_tool(
self, self,
connection: ConnectionConfig, connection: ConnectionConfig,
+27
View File
@@ -31,12 +31,36 @@ class CatalogNodeEntry:
output_schema: dict[str, Any] output_schema: dict[str, Any]
@dataclass(slots=True)
class CatalogResourceEntry:
qualified_name: str
connection_id: str
local_name: str
uri: str
description: str | None
mime_type: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True)
class CatalogPromptEntry:
qualified_name: str
connection_id: str
local_name: str
description: str | None
arguments: list[dict[str, Any]] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True) @dataclass(slots=True)
class CatalogSnapshot: class CatalogSnapshot:
connection_id: str connection_id: str
fetched_at_epoch_ms: int fetched_at_epoch_ms: int
max_age_seconds: int max_age_seconds: int
nodes: list[CatalogNodeEntry] = field(default_factory=list) nodes: list[CatalogNodeEntry] = field(default_factory=list)
resources: list[CatalogResourceEntry] = field(default_factory=list)
prompts: list[CatalogPromptEntry] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
def is_stale(self, now_epoch_ms: int) -> bool: def is_stale(self, now_epoch_ms: int) -> bool:
age_ms = now_epoch_ms - self.fetched_at_epoch_ms age_ms = now_epoch_ms - self.fetched_at_epoch_ms
@@ -60,4 +84,7 @@ def dump_catalog_snapshot(snapshot: CatalogSnapshot) -> dict[str, Any]:
"fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms, "fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms,
"max_age_seconds": snapshot.max_age_seconds, "max_age_seconds": snapshot.max_age_seconds,
"nodes": [asdict(node) for node in snapshot.nodes], "nodes": [asdict(node) for node in snapshot.nodes],
"resources": [asdict(resource) for resource in snapshot.resources],
"prompts": [asdict(prompt) for prompt in snapshot.prompts],
"metadata": snapshot.metadata,
} }
+13
View File
@@ -93,6 +93,9 @@ class WfMcpService:
auth = self.load_auth(connection_id) auth = self.load_auth(connection_id)
tools = await adapter.list_tools(connection, auth) 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 = [ specs = [
wrap_discovered_tool( wrap_discovered_tool(
connection=connection, connection=connection,
@@ -107,6 +110,16 @@ class WfMcpService:
*specs, *specs,
max_age_seconds=max_age_seconds, max_age_seconds=max_age_seconds,
) )
snapshot = snapshot_from_specs(
connection_id,
specs=self.specs_by_connection.get(connection_id, {}),
resources=resources,
prompts=prompts,
metadata=metadata,
fetched_at_epoch_ms=int(time.time() * 1000),
max_age_seconds=max_age_seconds or self.default_catalog_max_age_seconds,
)
self.store.save_catalog(snapshot)
def compile_plan(self, plan: RawWorkflowPlan) -> Workflow: def compile_plan(self, plan: RawWorkflowPlan) -> Workflow:
node_defs: dict[str, Any] = {} node_defs: dict[str, Any] = {}
+25 -10
View File
@@ -1,10 +1,16 @@
from __future__ import annotations from __future__ import annotations
import json import json
from dataclasses import asdict
from pathlib import Path from pathlib import Path
from .models import AuthRecord, CatalogNodeEntry, CatalogSnapshot from .models import (
AuthRecord,
CatalogNodeEntry,
CatalogPromptEntry,
CatalogResourceEntry,
CatalogSnapshot,
dump_catalog_snapshot,
)
class Store: class Store:
@@ -44,7 +50,14 @@ class FileStore(Store):
def save_auth(self, record: AuthRecord) -> None: def save_auth(self, record: AuthRecord) -> None:
self._auth_path(record.connection_id).write_text( self._auth_path(record.connection_id).write_text(
json.dumps(asdict(record), indent=2), json.dumps(
{
"connection_id": record.connection_id,
"scheme": record.scheme,
"payload": record.payload,
},
indent=2,
),
encoding="utf-8", encoding="utf-8",
) )
@@ -56,14 +69,8 @@ class FileStore(Store):
return AuthRecord(**data) return AuthRecord(**data)
def save_catalog(self, snapshot: CatalogSnapshot) -> None: def save_catalog(self, snapshot: CatalogSnapshot) -> None:
payload = {
"connection_id": snapshot.connection_id,
"fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms,
"max_age_seconds": snapshot.max_age_seconds,
"nodes": [asdict(node) for node in snapshot.nodes],
}
self._catalog_path(snapshot.connection_id).write_text( self._catalog_path(snapshot.connection_id).write_text(
json.dumps(payload, indent=2), json.dumps(dump_catalog_snapshot(snapshot), indent=2),
encoding="utf-8", encoding="utf-8",
) )
@@ -77,4 +84,12 @@ class FileStore(Store):
fetched_at_epoch_ms=data["fetched_at_epoch_ms"], fetched_at_epoch_ms=data["fetched_at_epoch_ms"],
max_age_seconds=data["max_age_seconds"], max_age_seconds=data["max_age_seconds"],
nodes=[CatalogNodeEntry(**node) for node in data.get("nodes", [])], nodes=[CatalogNodeEntry(**node) for node in data.get("nodes", [])],
resources=[
CatalogResourceEntry(**resource)
for resource in data.get("resources", [])
],
prompts=[
CatalogPromptEntry(**prompt) for prompt in data.get("prompts", [])
],
metadata=data.get("metadata", {}),
) )
+138 -101
View File
@@ -162,6 +162,31 @@ Already present:
This is a good first vertical slice for tool execution. This is a good first vertical slice for tool execution.
## What "good proxy" means
For `wf_mcp`, "proxy" should not only mean "call tools on behalf of a client."
It should mean:
- discover backend MCP capabilities and expose them with stable namespacing
- preserve enough metadata that a client can understand what exists and how to use it
- forward or mirror backend-facing events and responses in a traceable way
- keep auth/session boundaries explicit per connection
- separate "what the backend offers" from "what becomes a workflow node"
So the proxy layer should eventually broker:
- tools
- resources
- prompts
- notifications/events
- auth-related state markers
- tasks / long-running operations
- elicitations / human-input capabilities
- app/server metadata
Not all of these need workflow semantics immediately, but they should still have a place in the package model.
## Next capability expansion ## Next capability expansion
The next architectural move should be broadening the capability model beyond tools. The next architectural move should be broadening the capability model beyond tools.
@@ -202,6 +227,81 @@ The client-facing catalog payload should be usable by:
- an LLM that builds workflows - an LLM that builds workflows
- internal service code - internal service code
## Near-term implementation order
This is the intended order of work from here.
### Milestone 1: Broaden discovery models
Add normalized discovery models for:
- resources
- prompts
- notifications
- tasks
- elicitations
- app/server metadata
Outcome:
- `wf_mcp` can describe more of an MCP backend than just tools
- namespacing rules apply uniformly across capability types
### Milestone 2: Unified capability snapshots
Move from tool-only snapshots to connection-scoped capability snapshots that can store:
- tools
- resources
- prompts
- capability metadata
- freshness timestamps
- auth-state hints
Outcome:
- the catalog becomes useful to a real client or inspector
- capability caching is no longer tool-specific
### Milestone 3: Service APIs for non-tool capabilities
Expose service methods that let a client:
- inspect resources
- inspect prompts
- read capability metadata
- refresh one connection or all connections
Outcome:
- `wf_mcp` becomes a usable broker, not only a workflow launcher
### Milestone 4: Trace/event correlation
Add a light event model that can connect:
- client action
- backend MCP request/response
- workflow run/step trace
Outcome:
- better observability
- cleaner debugging
- future notification/task support has a home
### Milestone 5: Selective workflow mapping
Only after the above is stable:
- keep tools as workflow nodes
- evaluate whether prompts/resources should become helper nodes or stay catalog-only
- map elicitation to workflow interrupts carefully
Outcome:
- workflow integration grows from understood protocol behavior, not guesses
## Workflow integration policy ## Workflow integration policy
Not every capability becomes a workflow node right away. Not every capability becomes a workflow node right away.
@@ -258,120 +358,57 @@ Traceability is a major requirement.
We should eventually distinguish: We should eventually distinguish:
- workflow execution trace - workflow execution trace
- MCP backend call trace - backend MCP call trace
- client-facing event/notification stream - client-facing event stream
These are related but not identical. The important design rule is that these should be related, but not collapsed into one giant anonymous log.
The design should make it possible to correlate them through: ## Concrete next files
- connection id If we follow the plan above, the next likely modules are:
- capability id
- workflow run id
- frame/node ids where appropriate
## Auth and storage - `wf_mcp/discovery.py`
- orchestrate refresh and cache policy for all capability types
- `wf_mcp/capabilities.py`
- broader normalized capability models if `models.py` starts getting crowded
- `wf_mcp/events.py`
- event / trace correlation shapes
- `wf_mcp/resources.py`
- resource-facing service helpers
- `wf_mcp/prompts.py`
- prompt-facing service helpers
Auth should remain behind a pluggable store interface. Whether these stay as separate files or fold back into `models.py` / `service.py` should be driven by clarity, not purity.
First implementation: ## Test organization direction
- file-backed store 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.
Expected future replacements: Recommended direction once the next capability work starts:
- database-backed store - `tests/wf_core/`
- encrypted local store - `tests/wf_authoring/`
- secret-manager-backed store - `tests/wf_mcp/`
- `tests/fixtures/`
The store should persist: For `wf_mcp`, likely split by concern:
- auth records - `tests/wf_mcp/test_store.py`
- cached capability snapshots - `tests/wf_mcp/test_catalog.py`
- `tests/wf_mcp/test_service.py`
- `tests/wf_mcp/test_adapters.py`
- `tests/wf_mcp/test_sdk_adapter.py`
It may later persist: This does not need to happen immediately, but it should happen before one `test_wf_mcp.py` turns into a junk drawer.
- saved plans/workflows ## Working rule for the next phase
- job specs
- run metadata
## Execution model Before making a new MCP capability executable inside workflows, first answer:
Near-term execution stance: - is this primarily catalog/proxy surface, or workflow surface?
- what is the stable namespaced identifier?
- what is the trace/event story?
- does it need auth/session context?
- does it map cleanly to `wf_core`, or does it stay above it?
- async-first at the MCP layer That rule should keep us from forcing everything through the node model too early.
- 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`
## Proposed modules
Existing:
- `models.py`
- `connections.py`
- `store.py`
- `catalog.py`
- `service.py`
- `adapters.py`
- `wrappers.py`
- `mcp_sdk_adapter.py`
Likely next:
- `discovery.py`
- `capabilities.py`
- `events.py`
- `auth.py`
- `plans.py`
- `jobs.py`
## Recommended next implementation order
1. Broaden capability models beyond tools
2. Introduce unified capability snapshots/catalog payloads
3. Add discovery/cache orchestration policy
4. Expose prompt/resource inspection through service APIs
5. Add trace/event correlation hooks
6. Revisit workflow integration for elicitation/tasks
## Guiding rule
`wf_mcp` should not collapse protocol richness into fake simplicity too early.
It should:
- preserve namespacing
- preserve capability boundaries
- preserve traceability
- only turn protocol features into workflow features when the semantic mapping is clear
File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 33 KiB