plan + fmt

tests/test_demo_workflow.py

@@ -364,14 +364,17 @@ def test_foreach_stress_with_many_documents() -> None:
     assert run.status == RunStatus.COMPLETED
     assert len(run.state["documents"]) == document_count
     assert len(run.state["item_summaries"]) == document_count
-    assert len(
+    assert (
+        len(
             [
                 frame
                 for frame in run.frames.values()
                 if frame.kind == "foreach_iteration"
                 and frame.status == FrameStatus.COMPLETED
             ]
-    ) == document_count
+        )
+        == document_count
+    )
     assert len([entry for entry in run.trace if entry.step_type == "foreach"]) == (
         document_count + 1
     )
@@ -381,9 +384,9 @@ def test_builder_compiles_same_workflow_as_declared_demo() -> None:
     declared = build_demo_workflow()
     built, _registry = build_authoring_demo_workflow()
 
-    assert _strip_schema_titles(built.model_dump(by_alias=True)) == _strip_schema_titles(
-        declared.model_dump(by_alias=True)
-    )
+    assert _strip_schema_titles(
+        built.model_dump(by_alias=True)
+    ) == _strip_schema_titles(declared.model_dump(by_alias=True))
 
 
 def test_builder_compiled_workflow_executes_like_declared_demo() -> None:

tests/test_wf_mcp.py

@@ -65,13 +65,14 @@ def _fixture_server_path() -> str:
 
 
 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]
-    transport = os.environ.get("MCP_EVERYTHING_TRANSPORT", "stdio")
 
         metadata: dict[str, Any] = {
             "transport": transport,
@@ -79,7 +80,7 @@ def _everything_server_connection() -> ConnectionConfig | None:
             "args": args,
         }
 
-    if transport == "streamable_http":
+    elif transport == "streamable_http":
         url = os.environ.get("MCP_EVERYTHING_URL")
         if not url:
             raise AssertionError(
@@ -89,7 +90,8 @@ def _everything_server_connection() -> ConnectionConfig | None:
             "transport": "streamable_http",
             "url": url,
         }
-
+    else:
+        return None
     return ConnectionConfig(
         id="everything.default",
         server="everything",

wf_authoring/mapping.py

@@ -16,15 +16,13 @@ def normalize_path(path: PathArg) -> str:
 
 def bind_fields(**mapping: PathArg) -> dict[str, str]:
     return {
-        normalize_path(source): destination
-        for destination, source in mapping.items()
+        normalize_path(source): destination for destination, source in mapping.items()
     }
 
 
 def bind_state(**mapping: PathArg) -> dict[str, str]:
     return {
-        destination: normalize_path(target)
-        for destination, target in mapping.items()
+        destination: normalize_path(target) for destination, target in mapping.items()
     }
 
 

wf_authoring/spec.py

@@ -3,7 +3,17 @@ from __future__ import annotations
 from inspect import Parameter, iscoroutinefunction, signature
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from typing import Any, Generic, Literal, TypeVar, cast, get_args, get_origin, get_type_hints, overload
+from typing import (
+    Any,
+    Generic,
+    Literal,
+    TypeVar,
+    cast,
+    get_args,
+    get_origin,
+    get_type_hints,
+    overload,
+)
 
 from pydantic import BaseModel
 
@@ -67,9 +77,7 @@ def _infer_models(
     hints = get_type_hints(fn, include_extras=True)
     params = list(signature(fn).parameters.values())
     if len(params) < 2:
-        raise TypeError(
-            "node function must accept at least (payload, ctx) parameters"
-        )
+        raise TypeError("node function must accept at least (payload, ctx) parameters")
 
     payload_param = params[0]
     ctx_param = params[1]
@@ -87,15 +95,11 @@ def _infer_models(
 
     input_model = hints.get(payload_param.name)
     if not _is_basemodel_subclass(input_model):
-        raise TypeError(
-            "node payload annotation must be a pydantic BaseModel subclass"
-        )
+        raise TypeError("node payload annotation must be a pydantic BaseModel subclass")
 
     ctx_type = hints.get(ctx_param.name)
     if ctx_type is not RuntimeContext:
-        raise TypeError(
-            "node context annotation must be wf_core.RuntimeContext"
-        )
+        raise TypeError("node context annotation must be wf_core.RuntimeContext")
 
     return_type = hints.get("return")
     if return_type is None:
@@ -184,12 +188,12 @@ class NodeSpec(Generic[InputT, OutputT]):
 
         return handler
 
+
 @overload
 def node(
     fn: NodeCallable[InputT, OutputT] | AsyncNodeCallable[InputT, OutputT],
     /,
-) -> NodeSpec[InputT, OutputT]:
-    ...
+) -> NodeSpec[InputT, OutputT]: ...
 
 
 @overload
@@ -199,8 +203,7 @@ def node(
 ) -> Callable[
     [NodeCallable[InputT, OutputT] | AsyncNodeCallable[InputT, OutputT]],
     NodeSpec[InputT, OutputT],
-]:
-    ...
+]: ...
 
 
 @overload
@@ -217,12 +220,13 @@ def node(
 ) -> Callable[
     [NodeCallable[InputT, OutputT] | AsyncNodeCallable[InputT, OutputT]],
     NodeSpec[InputT, OutputT],
-]:
-    ...
+]: ...
 
 
 def node(
-    fn: NodeCallable[InputT, OutputT] | AsyncNodeCallable[InputT, OutputT] | None = None,
+    fn: NodeCallable[InputT, OutputT]
+    | AsyncNodeCallable[InputT, OutputT]
+    | None = None,
     *,
     name: str | None = None,
     input_model: type[InputT] | None = None,
@@ -276,8 +280,7 @@ def _build_registry(
     specs: tuple[NodeSpec[Any, Any], ...],
     *,
     export: Literal["sync"],
-) -> dict[str, SyncRegistryHandler]:
-    ...
+) -> dict[str, SyncRegistryHandler]: ...
 
 
 @overload
@@ -285,8 +288,7 @@ def _build_registry(
     specs: tuple[NodeSpec[Any, Any], ...],
     *,
     export: Literal["async"],
-) -> dict[str, AsyncRegistryHandler]:
-    ...
+) -> dict[str, AsyncRegistryHandler]: ...
 
 
 def _build_registry(

wf_core/demo_workflow.py

@@ -192,7 +192,11 @@ def build_demo_workflow() -> Workflow:
             "edges": [
                 {"from": "list_files", "outcome": "ok", "to": "summarize_each"},
                 {"from": "summarize_each", "outcome": "loop", "to": "summarize_one"},
-                {"from": "summarize_each", "outcome": "done", "to": "combine_summaries"},
+                {
+                    "from": "summarize_each",
+                    "outcome": "done",
+                    "to": "combine_summaries",
+                },
                 {"from": "summarize_one", "outcome": "ok", "to": END},
                 {"from": "combine_summaries", "outcome": "ok", "to": "should_email"},
                 {"from": "should_email", "outcome": "true", "to": "approve_email"},

wf_mcp/connections.py

@@ -7,14 +7,10 @@ from .models import ConnectionConfig
 
 def parse_connection_id(connection_id: str) -> tuple[str, str]:
     if "." not in connection_id:
-        raise ValueError(
-            "connection id must look like '<server>.<account>'"
-        )
+        raise ValueError("connection id must look like '<server>.<account>'")
     server, account = connection_id.split(".", 1)
     if not server or not account:
-        raise ValueError(
-            "connection id must look like '<server>.<account>'"
-        )
+        raise ValueError("connection id must look like '<server>.<account>'")
     return server, account
 
 
@@ -37,5 +33,6 @@ class ConnectionRegistry:
         return self.connections[connection_id]
 
     def list_enabled(self) -> list[ConnectionConfig]:
-        return [connection for connection in self.connections.values() if connection.enabled]
-
+        return [
+            connection for connection in self.connections.values() if connection.enabled
+        ]

wf_mcp/models.py

@@ -61,4 +61,3 @@ def dump_catalog_snapshot(snapshot: CatalogSnapshot) -> dict[str, Any]:
         "max_age_seconds": snapshot.max_age_seconds,
         "nodes": [asdict(node) for node in snapshot.nodes],
     }
-

wf_mcp/service.py

@@ -58,7 +58,9 @@ class WfMcpService:
     ) -> None:
         self.connections.get(connection_id)
         qualified_specs = {
-            qualify_node_name(connection_id, spec.name): _qualify_spec(connection_id, spec)
+            qualify_node_name(connection_id, spec.name): _qualify_spec(
+                connection_id, spec
+            )
             for spec in specs
         }
         self.specs_by_connection[connection_id] = qualified_specs

wf_mcp/store.py

@@ -78,4 +78,3 @@ class FileStore(Store):
             max_age_seconds=data["max_age_seconds"],
             nodes=[CatalogNodeEntry(**node) for node in data.get("nodes", [])],
         )
-

wf_mcp_plan.md

@@ -0,0 +1,377 @@
+# `wf_mcp` Plan
+
+This document describes the intended direction of `wf_mcp`.
+
+The important change in scope is this:
+
+- `wf_mcp` is not just a tool wrapper layer
+- `wf_mcp` is a namespaced MCP capability broker plus workflow build/run layer
+
+That means it should be able to face:
+
+- human users
+- client LLMs
+- workflow execution services
+
+without forcing every MCP capability to immediately become a workflow node.
+
+## Goals
+
+`wf_mcp` should:
+
+- manage multiple named MCP backend connections
+- persist auth/session state
+- discover and cache MCP capabilities per connection
+- expose namespaced catalogs to humans and LLMs
+- wrap callable tool capabilities into workflow-executable `NodeSpec`s
+- compile and run workflows against those capabilities
+- preserve traceability between client actions, MCP backend calls, and workflow runs
+
+## Non-goals for the first phase
+
+- full persistence of workflow runs/jobs
+- parallel workflow execution semantics beyond what `wf_core` already supports
+- turning every MCP capability type into a workflow node immediately
+- hiding protocol complexity by inventing vague magic abstractions
+
+## Layering
+
+### `wf_core`
+
+Owns:
+
+- workflow model
+- validation
+- runtime semantics
+- frames
+- trace
+- interrupts
+- foreach
+
+Does not own:
+
+- MCP connections
+- auth/session persistence
+- discovery caching
+- capability brokerage
+
+### `wf_authoring`
+
+Owns:
+
+- `@node`
+- `NodeSpec`
+- `WorkflowBuilder`
+- condition DSL
+- subgraph wrapping
+
+Does not own:
+
+- MCP transport/client logic
+
+### `wf_mcp`
+
+Owns:
+
+- connection registry
+- auth store
+- capability discovery/cache
+- namespaced capability catalog
+- MCP backend adapter layer
+- tool-to-`NodeSpec` wrapping
+- raw plan compilation and workflow execution entrypoints
+- future user-facing service/API surface
+
+## Mental model
+
+Think of `wf_mcp` as having two planes.
+
+### 1. Capability proxy plane
+
+This plane exposes what a backend MCP connection offers.
+
+Capabilities include:
+
+- tools
+- resources
+- prompts
+- notifications/events
+- auth metadata
+- tasks
+- elicitations
+- app/server metadata
+
+This plane is about discovery, namespacing, caching, and proxying.
+
+### 2. Workflow execution plane
+
+This plane decides which capabilities can be used inside workflows and how.
+
+In the first phase:
+
+- tools become executable workflow nodes
+- resources and prompts are exposed in catalog/proxy APIs first
+- tasks, notifications, and elicitations are modeled and surfaced first, then integrated into workflows later
+
+## Connection identity
+
+Connections must be first-class and stable.
+
+Expected shape:
+
+- `<server>.<account>`
+
+Examples:
+
+- `google.personal`
+- `google.work`
+- `github.main`
+- `everything.default`
+
+The same backend/server type may have multiple configured connections with different auth, tool availability, or environment.
+
+## Namespacing
+
+All exposed capabilities should be explicitly namespaced.
+
+Examples:
+
+- `google.personal.tool.list_files`
+- `google.personal.prompt.summarize_folder`
+- `google.personal.resource.drive://folder/abc`
+- `everything.default.tool.echo_tool`
+
+The exact string shape can evolve, but these properties should hold:
+
+- globally unique
+- reversible back to connection id + local capability id
+- understandable by both humans and LLMs
+
+## Current implemented slice
+
+Already present:
+
+- connection config and registry
+- pluggable file-backed auth/catalog store
+- tool catalog snapshots
+- raw workflow plan compilation
+- workflow execution via async runtime
+- lightweight adapter protocol
+- fake adapter tests
+- real MCP SDK adapter for `stdio` and `streamable_http`
+
+This is a good first vertical slice for tool execution.
+
+## Next capability expansion
+
+The next architectural move should be broadening the capability model beyond tools.
+
+### Add models for
+
+- `DiscoveredResource`
+- `DiscoveredPrompt`
+- `DiscoveredNotification`
+- `DiscoveredTaskCapability`
+- `DiscoveredElicitationCapability`
+- maybe `DiscoveredAppMetadata`
+
+These do not all need execution semantics immediately.
+
+They do need:
+
+- namespacing
+- storage/caching
+- catalog exposure
+
+## Catalog direction
+
+We should move from a tool-only catalog to a unified capability catalog.
+
+The catalog should be able to expose, per connection:
+
+- tools
+- resources
+- prompts
+- capability metadata
+- auth requirements or state markers
+- freshness metadata
+
+The client-facing catalog payload should be usable by:
+
+- a human inspector UI
+- an LLM that builds workflows
+- internal service code
+
+## Workflow integration policy
+
+Not every capability becomes a workflow node right away.
+
+### Immediate workflow nodes
+
+- tools
+
+### Catalog/proxy first, workflow later
+
+- resources
+- prompts
+- notifications
+- tasks
+- elicitations
+
+Why:
+
+- tools already match the node call model well
+- the others need more careful semantic mapping
+
+## Elicitations and interrupts
+
+Elicitations are especially interesting because they align with the workflow interrupt model.
+
+Planned stance:
+
+- expose MCP elicitation capability in catalogs first
+- understand the backend semantics first
+- later map appropriate elicitation flows into workflow `InterruptNode` behavior
+
+This should be done deliberately, because workflow interrupt semantics are stronger and more structured than generic protocol-level elicitation.
+
+## Tasks and long-running work
+
+Tasks likely align with future run/job persistence, but they should not force that design immediately.
+
+Near-term stance:
+
+- surface task capability in capability catalogs
+- understand the task API shape
+- leave run persistence mostly out of scope for now
+
+Future direction:
+
+- scheduled jobs
+- persisted runs
+- polling or event-driven task monitoring
+
+## Notifications and traceability
+
+Traceability is a major requirement.
+
+We should eventually distinguish:
+
+- workflow execution trace
+- MCP backend call trace
+- client-facing event/notification stream
+
+These are related but not identical.
+
+The design should make it possible to correlate them through:
+
+- connection id
+- capability id
+- workflow run id
+- frame/node ids where appropriate
+
+## 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`
+
+## 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