example use, live MCP

told chat i dont want to use the fake ahh server we have (FakeAdapter). switch over, boom, errors.

examples/mcp_tool_workflow.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+import sys
+
+from wf_core import END
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig, RawWorkflowPlan
+from wf_mcp.sdk import McpSdkAdapter
+from wf_mcp.storage import FileStore
+
+FIXTURE_SERVER = (
+    Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "mcp_echo_server.py"
+)
+
+
+async def run_example() -> dict[str, object]:
+    """Discover an MCP tool, compile it into a workflow, and execute it."""
+    service = WfMcpService(store=FileStore(Path(".wf_mcp_store") / "example"))
+    service.register_connection(
+        ConnectionConfig(
+            id="fixture.personal",
+            server="fixture",
+            account="personal",
+            metadata={
+                "transport": "stdio",
+                "command": sys.executable,
+                "args": [str(FIXTURE_SERVER)],
+            },
+        )
+    )
+    service.register_adapter("fixture", McpSdkAdapter())
+
+    await service.refresh_connection_catalog("fixture.personal")
+
+    plan = RawWorkflowPlan(
+        name="mcp_echo_workflow",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"fields": {"echoed": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "fixture.personal.echo_tool",
+                "in_map": {"input.text": "text"},
+                "out_map": {"echoed": "state.echoed"},
+            }
+        ],
+        edges=[
+            {"from": "echo", "outcome": "ok", "to": END},
+            {"from": "echo", "outcome": "error", "to": END}, # workaround for now; i need wf_authoring.ops::runtime_error
+        ],
+    )
+
+    run = await service.run_workflow_from_plan(plan, {"text": "hello from MCP"})
+    return {
+        "status": run.status.value,
+        "output": run.output,
+        "catalog": service.get_catalog().as_payload(),
+    }
+
+
+def main() -> None:
+    """Run the MCP tool workflow example and print the result."""
+    import json
+
+    print(json.dumps(asyncio.run(run_example()), indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/authoring/test_nodes.py

@@ -177,11 +177,9 @@ def test_pydantic_field_descriptions_survive_node_catalog_schema() -> None:
     assert entry.description == "Echoes a documented message."
     assert entry.input_schema["type"] == "object"
     assert (
-        entry.input_schema["properties"]["message"]["description"]
-        == "Message to echo"
+        entry.input_schema["properties"]["message"]["description"] == "Message to echo"
     )
     assert entry.output_schema["type"] == "object"
     assert (
-        entry.output_schema["properties"]["echoed"]["description"]
-        == "Echoed message"
+        entry.output_schema["properties"]["echoed"]["description"] == "Echoed message"
     )

tests/fixtures/mcp_echo_server.py

@@ -1,13 +1,20 @@
 from __future__ import annotations
 
+from typing import Annotated, TypedDict
+
 from mcp.server.fastmcp import FastMCP
+from pydantic import Field
 
 
 server = FastMCP("echo-fixture")
 
+class EchoToolResult(TypedDict):
+    echoed: str
 
 @server.tool(title="Echo tool")
-async def echo_tool(text: str) -> dict[str, str]:
+async def echo_tool(
+    text: Annotated[str, Field(description="Text to echo")],
+) -> EchoToolResult:
     return {"echoed": text}
 
 

tests/rewrite/models.py

@@ -47,6 +47,29 @@ class Context(TypedDict):  # dataclass support? no. i mean langgraph doesnt.
 # class Input(TypedDict, total=False):
 
 
+## context is a hard thing
+# According to https://docs.langchain.com/oss/python/concepts/context, there are three types:
+#
+# | type | mut | lifetime |
+# | --- | --- | --- |
+# |static runtime (context) | static | single run |
+# |dynamic runtime (state) | mut | single run |
+# |dynamic cross-convo (store) | mut | cross-conversation |
+#
+# now what the hell is store
+### store
+#
+# store is used in langgraph-demo for debugging. but it can be used for more things.
+# it saves every turn. every graph nodes. I use InMemoryStore, you can use psql store!
+# This allows for picking the work up again after a while for example.
+# a lot more versatility there.
+#
+### what about us? how should we handle context?
+#
+# in the future if wed like, we could handle context. This could be useful for lda.chat!
+# i want lda.chat to be/have a meta-agent. So i could spin up ideas! a 
+
+
 class ContextInput(BaseModel):
     context: Context  # final!
 

tests/wf_mcp/test_mcp_tool_workflow_example.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, cast
+
+import pytest
+
+from examples.mcp_tool_workflow import run_example
+
+
+def test_mcp_tool_workflow_example_runs_discovered_tool() -> None:
+    try:
+        result = asyncio.run(run_example())
+    except PermissionError as exc:
+        pytest.skip(f"stdio MCP transport is not permitted in this environment: {exc}")
+
+    assert result["status"] == "completed"
+    assert result["output"] == {"echoed": "hello from MCP"}
+    catalog = cast(dict[str, Any], result["catalog"])
+    node = catalog["nodes"][0]
+    assert node["qualified_name"] == "fixture.personal.echo_tool"
+    assert node["input_schema"]["properties"]["text"]["description"] == "Text to echo"

tests/wf_mcp/test_service.py

@@ -196,10 +196,7 @@ def test_service_catalog_preserves_json_schema_description_metadata() -> None:
     node = service.get_catalog().as_payload()["nodes"][0]
     assert node["input_schema"]["type"] == "object"
     assert isinstance(node["input_schema"]["properties"], dict)
-    assert (
-        node["input_schema"]["properties"]["text"]["description"]
-        == "Text to echo"
-    )
+    assert node["input_schema"]["properties"]["text"]["description"] == "Text to echo"
     assert node["output_schema"]["type"] == "object"
     assert isinstance(node["output_schema"]["properties"], dict)
 
@@ -226,7 +223,9 @@ def test_service_wrapped_tool_adapter_model_validates_simple_schema_types() -> N
     except ValueError as exc:
         assert "text" in str(exc)
     else:
-        raise AssertionError("expected generated adapter model to reject non-string text")
+        raise AssertionError(
+            "expected generated adapter model to reject non-string text"
+        )
 
 
 def test_service_records_tool_call_events() -> None: