misc: use structured path for errors

docs/durable_run_operations.md

@@ -36,6 +36,10 @@ wf.workflow.run_deployment
 projected as newly-created MCP tools in the current session; many clients do
 not rebuild callable schemas after `tools/list` changes.
 
+These run tools are MCP control tools, not graph-usable workflow capabilities.
+They are discovered through MCP `tools/list` or harness search-tools, not through
+`wf.workflow.list_capabilities`.
+
 ## `run_deployment`
 
 Starts one deployment execution:

docs/wf_mcp_end_to_end_runbook.md

@@ -447,6 +447,11 @@ wf.workflow.validate_deployment
 wf.workflow.run_deployment
 ```
 
+`wf.workflow.list_capabilities` only lists graph-usable workflow capabilities.
+Control tools such as `wf.workflow.inspect_run`, `wf.workflow.read_run_trace`,
+draft workspace mutation helpers, and admin operations are discovered through
+MCP `tools/list` or the client's search-tools surface.
+
 ## Common Failure Points
 
 ### The connection exists but nothing is discoverable

docs/wf_mcp_operator_manual.md

@@ -102,6 +102,15 @@ Typical tools:
 The workflow surface is intentionally split by job. Use the primary path first;
 the advanced tools exist for debugging, compatibility, or focused repair.
 
+There are two discovery paths:
+
+- `wf.workflow.list_capabilities` / `inspect_capability` discover
+  workflow-facing node specs and saved wrappers that can be placed in graphs.
+- MCP `tools/list` or harness search-tools discover control tools such as
+  `wf.workflow.inspect_run`, `wf.workflow.read_run_trace`, draft workspace
+  mutators, and admin operations. These are not workflow capabilities and will
+  not appear in `list_capabilities`.
+
 ### Discovery
 
 Primary:
@@ -279,6 +288,11 @@ nodes rather than source ownership. Its rows include `source_id`, outcomes, and
 top-level input/output field names, while full JSON schemas stay behind
 `wf.workflow.inspect_capability`.
 
+Do not use `list_capabilities` to find MCP control tools. Run/debug helpers such
+as `wf.workflow.inspect_run` and `wf.workflow.read_run_trace` are ordinary MCP
+tools, not graph nodes. Discover them through MCP `tools/list`, search-tools, or
+the tool map in this manual.
+
 `wf.workflow.call_capability` is the REPL-style test step. Its result is
 self-describing: `kind` is either `node_spec` or `wrapper_artifact`,
 `source_id` identifies the owner when applicable, and `diagnostics` is empty for
@@ -317,6 +331,10 @@ wf.workflow.inspect_capability
 
 Use the compact list first, then inspect only the likely candidates.
 
+This list intentionally excludes control-plane MCP tools. If the client needs
+to inspect a run, patch a workspace, or call an admin operation, use MCP
+tool/search discovery instead of workflow capability discovery.
+
 ### 3. Test One Capability Directly
 
 ```text
@@ -446,6 +464,12 @@ A raw tool can be directly callable and still be an awkward workflow node if it
 uses provider-specific result envelopes, status strings, or transport-level
 errors where a graph wants explicit outcomes.
 
+For the common MCP shape `content: [{type: "text", text: "..."}]`, generated
+workflow capabilities keep `output.content` raw. `content` may contain text,
+images, resources, or mixed blocks, so wrapper hints do not invent a top-level
+`output.text`. Add an explicit wrapper/extraction node when a workflow wants a
+specific content-block field.
+
 ### Artifact Versus Deployment
 
 An artifact is the immutable saved workflow definition.

docs/workflow_capabilities.md

@@ -199,6 +199,13 @@ while the graph wants:
 }
 ```
 
+MCP text content blocks are a common provider-envelope case. When an MCP tool
+returns exactly one text content block, the SDK keeps the raw `content` list and
+also exposes a convenience `text` field. Wrapper hints prefer `text` so a string
+state field can map the actual text without writing the raw content-block list.
+Multiple content blocks or non-text content still require explicit wrapper
+decisions.
+
 ### Human-Friendly Versus Graph-Friendly Inputs
 
 A raw tool may be good for interactive human use but awkward for stateful graph

docs/workflow_drafts.md

@@ -607,6 +607,11 @@ It does not guess that a normal output state path is also an error message.
 Provider-specific error envelopes still belong in saved wrapper artifacts or
 follow-up patches.
 
+`error_message_source` accepts the same structural graph path shape used by
+other mapping fields, for example
+`{"root": "state", "parts": ["error_message"]}`. Legacy strings such as
+`state.error_message` remain accepted for compatibility.
+
 In MCP Inspector, workspace mutation tools accept a single `request` object.
 This is deliberate: the request object carries descriptions and validation for
 the authoring envelope while raw JSON Schema fields remain plain JSON objects.

src/wf_mcp/broker/catalog.py

@@ -14,6 +14,7 @@ from ..capabilities import (
 )
 from ..connections import qualify_node_name
 from ..models import CatalogSnapshot
+from ..sdk.converters import workflow_output_schema_from_mcp_tool_schema
 
 
 def snapshot_from_specs(
@@ -42,7 +43,9 @@ def snapshot_from_specs(
             description=entry.description,
             outcomes=entry.outcomes,
             input_schema=entry.input_schema,
-            output_schema=entry.output_schema,
+            output_schema=workflow_output_schema_from_mcp_tool_schema(
+                entry.output_schema
+            ),
         )
         for entry in catalog.entries()
     ]
@@ -127,7 +130,9 @@ class CombinedCatalog:
                     "description": entry.description,
                     "outcomes": list(entry.outcomes),
                     "input_schema": entry.input_schema,
-                    "output_schema": entry.output_schema,
+                    "output_schema": workflow_output_schema_from_mcp_tool_schema(
+                        entry.output_schema
+                    ),
                 }
                 for entry in self.entries()
             ],

src/wf_mcp/broker/service/core.py

@@ -50,6 +50,7 @@ from ...models import (
     RawWorkflowPlan,
 )
 from ...runtime import ToolExecutor
+from ...sdk.converters import workflow_output_schema_from_mcp_tool_schema
 from ...sdk import BackendAdapter
 from ...shared.errors import error_payload
 from ...shared.names import RESERVED_CONNECTION_IDS
@@ -875,7 +876,8 @@ class WfMcpService:
         """
         model_prefix = entry.qualified_name.replace(".", "_").replace("-", "_")
         input_model = _model_from_schema(f"{model_prefix}_Input", entry.input_schema)
-        output_model = _model_from_schema(f"{model_prefix}_Output", entry.output_schema)
+        output_schema = workflow_output_schema_from_mcp_tool_schema(entry.output_schema)
+        output_model = _model_from_schema(f"{model_prefix}_Output", output_schema)
 
         async def invoke_tool(payload: BaseModel) -> NodeReturn[BaseModel]:
             connection = self.connections.get(entry.connection_id)
@@ -901,7 +903,7 @@ class WfMcpService:
             is_async=True,
             accepts_context=False,
             input_schema_contract=entry.input_schema,
-            output_schema_contract=entry.output_schema,
+            output_schema_contract=output_schema,
         )
 
     def _get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:

src/wf_mcp/sdk/converters.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 from mcp.types import CallToolResult as McpCallToolResult
 from mcp.types import Prompt as McpPrompt
 from mcp.types import Resource as McpResource
@@ -11,10 +13,7 @@ from .base import ToolCallResult
 
 def tool_to_discovered(tool: McpTool) -> DiscoveredTool:
     """Convert an MCP SDK tool into the broker discovery model."""
-    output_schema = tool.outputSchema or {
+    output_schema = workflow_output_schema_from_mcp_tool_schema(tool.outputSchema)
-        "type": "object",
-        "properties": {"content": {"type": "array"}},
-    }
     display_name = (
         tool.annotations.title
         if tool.annotations is not None and tool.annotations.title
@@ -31,6 +30,22 @@ def tool_to_discovered(tool: McpTool) -> DiscoveredTool:
     )
 
 
+def workflow_output_schema_from_mcp_tool_schema(
+    schema: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Return the MCP tool output schema without inventing workflow fields.
+
+    MCP tools without structured output expose raw content blocks. Those blocks
+    can be text, images, resource links, or mixed results, so wf_mcp must not
+    pretend there is a stable top-level ``text`` field. Workflow authors should
+    add an explicit wrapper/extraction node for the block shape they expect.
+    """
+    return schema or {
+        "type": "object",
+        "properties": {"content": {"type": "array"}},
+    }
+
+
 def resource_to_discovered(resource: McpResource) -> DiscoveredResource:
     """Convert an MCP SDK resource into the broker discovery model."""
     local_name = resource.name or str(resource.uri)
@@ -64,7 +79,7 @@ def tool_result_to_call_result(result: McpCallToolResult) -> ToolCallResult:
     if result.structuredContent is not None:
         output = result.structuredContent
     else:
-        output = {
+        output: dict[str, Any] = {
             "content": [item.model_dump(by_alias=True) for item in result.content]
         }
     return ToolCallResult(

src/wf_mcp/workflow/wrappers.py

@@ -13,6 +13,7 @@ from wf_mcp.broker.events import McpEvent, make_event
 from ..capabilities import DiscoveredTool
 from ..models import AuthRecord, ConnectionConfig
 from ..runtime import ToolExecutor
+from ..sdk.converters import workflow_output_schema_from_mcp_tool_schema
 
 
 _JSON_TYPE_MAP: dict[str, object] = {
@@ -128,9 +129,10 @@ def wrap_discovered_tool(
         f"{connection.id}_{tool.name}_Input",
         tool.input_schema,
     )
+    output_schema = workflow_output_schema_from_mcp_tool_schema(tool.output_schema)
     output_model = _model_from_schema(
         f"{connection.id}_{tool.name}_Output",
-        tool.output_schema,
+        output_schema,
     )
 
     async def invoke_tool(
@@ -180,5 +182,5 @@ def wrap_discovered_tool(
         description=tool.description,
         is_async=True,
         input_schema_contract=tool.input_schema,
-        output_schema_contract=tool.output_schema,
+        output_schema_contract=output_schema,
     )

src/wf_mcp/workflow_surface/handlers.py

@@ -69,7 +69,10 @@ from .run_lifecycle import (
     restore_interrupted_run,
     validate_pinned_resume_environment,
 )
-from .wrapper_hints import wrapper_hints_for_capability
+from .wrapper_hints import (
+    workflow_output_schema_for_authoring,
+    wrapper_hints_for_capability,
+)
 
 if TYPE_CHECKING:
     from wf_core import RunState
@@ -138,7 +141,9 @@ class WorkflowSurfaceHandlers:
                 "outcomes": list(detail.outcomes),
                 "is_async": detail.is_async,
                 "input_fields": _schema_field_names(detail.input_schema),
-                "output_fields": _schema_field_names(detail.output_schema),
+                "output_fields": _schema_field_names(
+                    workflow_output_schema_for_authoring(detail.output_schema)
+                ),
             }
             for source in sorted(
                 self.service.capability_sources.values(),
@@ -716,7 +721,7 @@ class WorkflowSurfaceHandlers:
         output: Sequence[OutputBinding] | None = None,
         input_map: dict[str, str] | None = None,
         output_map: dict[str, str] | None = None,
-        error_message_source: str | None = None,
+        error_message_source: str | GraphSourcePath | None = None,
         title: str | None = None,
     ) -> dict[str, Any]:
         """Bootstrap the smallest patchable draft around one workflow capability."""
@@ -787,7 +792,7 @@ class WorkflowSurfaceHandlers:
         output: Sequence[OutputBinding] | None = None,
         input_map: dict[str, str] | None = None,
         output_map: dict[str, str] | None = None,
-        error_message_source: str | None = None,
+        error_message_source: str | GraphSourcePath | None = None,
     ) -> dict[str, Any]:
         """Create a patchable draft workspace from inspect_capability hints."""
         capability = await self.inspect_capability(qualified_name=capability_name)
@@ -1339,8 +1344,9 @@ def _draft_output_bindings_payload(output_map: dict[str, str]) -> list[dict[str,
     ]
 
 
-def _graph_path_payload(value: str) -> dict[str, str | list[str]]:
+def _graph_path_payload(value: str | GraphSourcePath) -> dict[str, str | list[str]]:
-    return GraphSourcePath._serialize(GraphSourcePath.parse(value))
+    path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
+    return GraphSourcePath._serialize(path)
 
 
 def _local_path_payload(value: str) -> dict[str, str | list[str]]:

src/wf_mcp/workflow_surface/models.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field
 from wf_artifacts import ArtifactKind
 from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
 from wf_core.models.steps import InputBinding, OutputBinding
+from wf_core.paths import GraphSourcePath
 
 WorkspaceId = Annotated[
     str,
@@ -67,6 +68,16 @@ SourceBindings = Annotated[
         )
     ),
 ]
+ErrorMessageSource = Annotated[
+    GraphSourcePath | str,
+    Field(
+        description=(
+            "State path for runtime_error.message. Prefer structural paths such "
+            "as {'root': 'state', 'parts': ['error_message']}; strings like "
+            "state.error_message remain compatibility input."
+        )
+    ),
+]
 
 
 class CallCapabilityResult(BaseModel):
@@ -276,7 +287,7 @@ class CreateMinimalDraftWorkspaceRequest(BaseModel):
             "workflow state paths, for example {'echoed': 'state.echoed'}."
         ),
     )
-    error_message_source: str | None = Field(
+    error_message_source: ErrorMessageSource | None = Field(
         default=None,
         description=(
             "Optional state path used as runtime_error.message when the capability "
@@ -332,7 +343,7 @@ class CreateDraftWorkspaceFromCapabilityRequest(BaseModel):
             "Deprecated compatibility override for the hinted capability output map."
         ),
     )
-    error_message_source: str | None = Field(
+    error_message_source: ErrorMessageSource | None = Field(
         default=None,
         description=(
             "Optional state path used as runtime_error.message when the capability "

src/wf_mcp/workflow_surface/wrapper_hints.py

@@ -103,22 +103,26 @@ def wrapper_hints_for_capability(
     create routes by itself.
     """
     input_properties = _object_properties(input_schema)
-    output_properties = _object_properties(output_schema)
+    hint_output_schema = workflow_output_schema_for_authoring(output_schema)
+    output_properties = _object_properties(hint_output_schema)
     input_map = {f"input.{name}": name for name in sorted(input_properties)}
-    output_map = {name: f"state.{name}" for name in sorted(output_properties)}
+    output_map_properties = _default_output_map_properties(
+        output_schema, output_properties
+    )
+    output_map = {name: f"state.{name}" for name in sorted(output_map_properties)}
     state_schema = {
         "type": "object",
         "properties": {
-            name: schema for name, schema in sorted(output_properties.items())
+            name: schema for name, schema in sorted(output_map_properties.items())
         },
     }
     wrapper_output_schema = {
         "type": "object",
         "properties": {
-            name: schema for name, schema in sorted(output_properties.items())
+            name: schema for name, schema in sorted(output_map_properties.items())
         },
     }
-    missing_decisions = _missing_decisions_for_output(output_schema)
+    missing_decisions = _missing_decisions_for_output(hint_output_schema)
     outcome_candidates = _boolean_outcome_candidates(output_properties)
     if outcome_candidates:
         missing_decisions.append(
@@ -132,10 +136,23 @@ def wrapper_hints_for_capability(
         )
     confidence = _confidence_for_hint(
         input_schema=input_schema,
-        output_schema=output_schema,
+        output_schema=hint_output_schema,
         missing_decisions=missing_decisions,
         outcome_candidates=outcome_candidates,
     )
+    notes = [
+        "Hints are scaffolding, not semantic guarantees.",
+        (
+            "Declared outcomes are preserved; output-field outcome "
+            "inference is not automatic."
+        ),
+    ]
+    if _has_raw_mcp_content(output_schema):
+        notes.append(
+            "Raw MCP content blocks are not workflow-shaped. Use an explicit "
+            "wrapper or extraction node to handle TextContent, ResourceLink, "
+            "images, or mixed content before writing to typed state."
+        )
     return WrapperAuthoringHints(
         capability_name=capability_name,
         confidence=confidence,
@@ -149,16 +166,20 @@ def wrapper_hints_for_capability(
         output_map=output_map,
         outcome_candidates=outcome_candidates,
         missing_decisions=missing_decisions,
-        notes=[
+        notes=notes,
-            "Hints are scaffolding, not semantic guarantees.",
-            (
-                "Declared outcomes are preserved; output-field outcome "
-                "inference is not automatic."
-            ),
-        ],
     )
 
 
+def workflow_output_schema_for_authoring(output_schema: JsonObject) -> JsonObject:
+    """Return the workflow-author-facing output schema for one capability.
+
+    Raw MCP ``content`` blocks stay raw. A wrapper may choose to extract
+    ``content[0].text`` or handle resources/images, but the authoring surface
+    must not invent that decision as a top-level schema field.
+    """
+    return {"type": "object", "properties": _object_properties(output_schema)}
+
+
 def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
     """Return object properties that are themselves JSON Schema objects."""
     properties = schema.get("properties")
@@ -171,6 +192,32 @@ def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
     }
 
 
+def _has_raw_mcp_content(schema: JsonObject) -> bool:
+    """Return true when schema exposes MCP's raw content-block envelope."""
+    properties = _object_properties(schema)
+    content_schema = properties.get("content")
+    return isinstance(content_schema, dict) and content_schema.get("type") == "array"
+
+
+def _default_output_map_properties(
+    raw_output_schema: JsonObject,
+    output_properties: dict[str, JsonObject],
+) -> dict[str, JsonObject]:
+    """Return fields safe enough to wire by default in generated hints.
+
+    Raw MCP ``content`` is a protocol envelope, not a workflow value. Keeping it
+    out of the default map prevents the scaffold from writing text/image/resource
+    blocks into typed state without an explicit extraction wrapper.
+    """
+    if _has_raw_mcp_content(raw_output_schema):
+        return {
+            name: schema
+            for name, schema in output_properties.items()
+            if name != "content"
+        }
+    return output_properties
+
+
 def _missing_decisions_for_output(output_schema: JsonObject) -> list[MissingDecision]:
     """Return explicit decisions required by output schema shape."""
     properties = _object_properties(output_schema)

tests/openapi/test_executor.py

@@ -11,7 +11,7 @@ from wf_openapi.executor import OpenApiOperationOutput
 from wf_openapi.spec import load_openapi_operations
 from wf_openapi.validation import load_openapi_app
 
-FIXTURE = Path("tests/openapi/fixtures/petstore_minimal.openapi.json")
+FIXTURE = Path(__file__).parent / "fixtures" / "petstore_minimal.openapi.json"
 
 
 def test_call_openapi_operation_maps_success() -> None:

tests/wf_mcp/test_sdk_converters.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from mcp.types import CallToolResult, TextContent, Tool
+
+from wf_mcp.sdk.converters import tool_result_to_call_result, tool_to_discovered
+
+
+def test_tool_without_output_schema_exposes_raw_content_schema() -> None:
+    tool = Tool(
+        name="echo",
+        inputSchema={"type": "object", "properties": {}},
+    )
+
+    discovered = tool_to_discovered(tool)
+
+    properties = discovered.output_schema["properties"]
+    assert properties["content"]["type"] == "array"
+    assert "text" not in properties
+
+
+def test_tool_with_content_only_output_schema_stays_raw() -> None:
+    tool = Tool(
+        name="echo",
+        inputSchema={"type": "object", "properties": {}},
+        outputSchema={
+            "type": "object",
+            "properties": {
+                "content": {
+                    "type": "array",
+                    "description": "Raw MCP content blocks.",
+                }
+            },
+            "required": ["content"],
+        },
+    )
+
+    discovered = tool_to_discovered(tool)
+
+    properties = discovered.output_schema["properties"]
+    assert properties["content"]["type"] == "array"
+    assert "text" not in properties
+    assert discovered.output_schema["required"] == ["content"]
+
+
+def test_tool_result_single_text_content_block_stays_in_content() -> None:
+    result = CallToolResult(
+        content=[TextContent(type="text", text="Echo: hello")],
+    )
+
+    converted = tool_result_to_call_result(result)
+
+    assert converted.outcome == "ok"
+    assert "text" not in converted.output
+    assert converted.output["content"][0]["type"] == "text"
+    assert converted.output["content"][0]["text"] == "Echo: hello"
+
+
+def test_tool_result_structured_content_is_not_rewritten() -> None:
+    result = CallToolResult(
+        content=[TextContent(type="text", text="ignored")],
+        structuredContent={"value": "structured"},
+    )
+
+    converted = tool_result_to_call_result(result)
+
+    assert converted.output["value"] == "structured"
+    assert "text" not in converted.output

tests/wf_mcp/test_service.py

@@ -8,6 +8,7 @@ from wf_artifacts import FileDraftWorkspaceStore, WorkflowDeployment
 from wf_authoring import NodeSpec, build_async_registry, node
 from wf_core import END, NodeUse, RunStatus, RuntimeContext
 from wf_mcp.broker import WfMcpService
+from wf_mcp.capabilities import DiscoveredTool
 from wf_mcp.models import AuthRecord, ConnectionConfig, RawWorkflowPlan
 from wf_mcp.runtime import ToolExecutor
 from wf_mcp.sdk import ToolCallResult
@@ -37,6 +38,46 @@ def pro_dotted_echo_tool(payload: EchoInput) -> EchoOutput:
     return EchoOutput(echoed=f"pro:{payload.text}")
 
 
+class ContentOnlyOutputAdapter(FakeAdapter):
+    """Adapter fixture for MCP tools that expose the raw content envelope."""
+
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        return [
+            DiscoveredTool(
+                name="echo_tool",
+                title="Echo Tool",
+                description="Echo text back",
+                input_schema={
+                    "type": "object",
+                    "properties": {"message": {"type": "string"}},
+                    "required": ["message"],
+                },
+                output_schema={
+                    "type": "object",
+                    "properties": {"content": {"type": "array"}},
+                    "required": ["content"],
+                },
+            )
+        ]
+
+    async def call_tool(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        tool_name: str,
+        payload: dict[str, Any],
+    ) -> ToolCallResult:
+        message = payload.get("message", "")
+        return ToolCallResult(
+            outcome="ok",
+            output={"content": [{"type": "text", "text": f"Echo: {message}"}]},
+        )
+
+
 def _single_echo_plan(plan_name: str, node_name: str) -> RawWorkflowPlan:
     return _raw_plan(
         name=plan_name,
@@ -855,6 +896,28 @@ def test_service_catalog_preserves_json_schema_description_metadata() -> None:
     assert isinstance(node["output_schema"]["properties"], dict)
 
 
+def test_service_preserves_content_only_tool_output_schema_for_workflows() -> None:
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "content_only_output_schema_store")
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", ContentOnlyOutputAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    catalog_node = service.get_catalog().as_payload()["nodes"][0]
+    source = service.capability_sources["demo.personal"]
+    spec = source.capabilities.node_specs["demo.personal.echo_tool"]
+
+    assert catalog_node["output_schema"]["properties"]["content"]["type"] == "array"
+    assert "text" not in catalog_node["output_schema"]["properties"]
+    assert spec.output_schema_contract is not None
+    assert spec.output_schema_contract["properties"]["content"]["type"] == "array"
+    assert "text" not in spec.output_schema_contract["properties"]
+
+
 def test_service_wrapped_tool_adapter_model_validates_simple_schema_types() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "schema_model_store"))
     service.register_connection(
@@ -939,6 +1002,55 @@ def test_service_records_tool_call_events() -> None:
     assert tool_events[1].payload["outcome"] == "ok"
 
 
+def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "raw_content_contract"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", ContentOnlyOutputAdapter())
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    plan = _raw_plan(
+        name="raw_content_contract",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"properties": {"outline": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"outline": {"type": "string"}},
+            "required": ["outline"],
+        },
+        output=[
+            {
+                "target": {"root": "local", "parts": ["outline"]},
+                "path": {"root": "state", "parts": ["outline"]},
+            }
+        ],
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [input_binding("input.text", "message")],
+                "output": [output_binding("text", "state.outline")],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": END}],
+    )
+
+    workflow = service.compile_plan(plan)
+    report = workflow.validate_structure()
+
+    assert not report.ok
+    assert any(
+        "source field 'text' is not declared in node output schema" in issue.message
+        for issue in report.errors
+    )
+
+
 def test_service_can_inspect_resources_and_prompts() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "inspect_store"))
     service.register_connection(

tests/wf_mcp/test_workflow_surface.py

@@ -13,11 +13,14 @@ from wf_artifacts import (
 )
 from wf_authoring import node, reducer
 from wf_mcp.broker import WfMcpService
-from wf_mcp.models import ConnectionConfig, RawWorkflowPlan
+from wf_mcp.models import AuthRecord, ConnectionConfig, RawWorkflowPlan
+from wf_mcp.sdk import ToolCallResult
 from wf_mcp.storage import FileStore
 from wf_mcp.workflow_surface import TraceRange, WorkflowSurfaceHandlers
+from wf_mcp.workflow_surface.models import CreateMinimalDraftWorkspaceRequest
 from wf_core.models.steps import InputPathBinding, OutputBinding
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+from wf_mcp.capabilities import DiscoveredTool
 from wf_platform import (
     CapabilityBuckets,
     CapabilitySource,
@@ -65,6 +68,102 @@ def failing_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
     raise RuntimeError("upstream exploded")
 
 
+class ContentOnlyOutputAdapter:
+    """MCP-like adapter whose tool exposes raw content blocks as output schema."""
+
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        return [
+            DiscoveredTool(
+                name="echo",
+                title="Echo",
+                description="Echo a message as an MCP text content block.",
+                input_schema={
+                    "type": "object",
+                    "properties": {"message": {"type": "string"}},
+                    "required": ["message"],
+                },
+                output_schema={
+                    "type": "object",
+                    "properties": {"content": {"type": "array"}},
+                    "required": ["content"],
+                },
+            )
+        ]
+
+    async def list_resources(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[Any]:
+        return []
+
+    async def list_prompts(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[Any]:
+        return []
+
+    async def get_connection_metadata(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> dict[str, Any]:
+        return {"server": connection.server}
+
+    async def call_tool(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        tool_name: str,
+        payload: dict[str, Any],
+    ) -> ToolCallResult:
+        message = payload.get("message", "")
+        return ToolCallResult(
+            outcome="ok",
+            output={"content": [{"type": "text", "text": f"Echo: {message}"}]},
+        )
+
+    async def read_resource(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        uri: str,
+    ) -> dict[str, Any]:
+        raise KeyError(uri)
+
+    async def get_prompt(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        prompt_name: str,
+        arguments: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        raise KeyError(prompt_name)
+
+    async def invoke_method(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        method: str,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        raise KeyError(method)
+
+    async def send_notification(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        method: str,
+        params: dict[str, Any] | None = None,
+    ) -> None:
+        raise KeyError(method)
+
+
 @reducer(name="custom.default.multiply")
 def multiply(current: int | None, incoming: int) -> int:
     return (current or 1) * incoming
@@ -830,7 +929,7 @@ def test_workflow_surface_minimal_draft_honors_explicit_error_message_source() -
             output_schema={"type": "object"},
             input_map={"input.text": "text"},
             output_map={"echoed": "state.echoed"},
-            error_message_source="state.error_message",
+            error_message_source=GraphSourcePath.state("error_message"),
         )
     )
     assert service.draft_workspace_store is not None
@@ -844,6 +943,27 @@ def test_workflow_surface_minimal_draft_honors_explicit_error_message_source() -
     ]
 
 
+def test_minimal_draft_request_accepts_structural_error_message_source() -> None:
+    request = CreateMinimalDraftWorkspaceRequest.model_validate(
+        {
+            "workspace_id": "echo_draft_structural_error",
+            "name": "echo",
+            "capability_name": "demo.personal.mcp_echo_tool",
+            "input_schema": {"type": "object"},
+            "state_schema": {"type": "object"},
+            "output_schema": {"type": "object"},
+            "error_message_source": {
+                "root": "state",
+                "parts": ["error_message"],
+            },
+        }
+    )
+
+    assert isinstance(request.error_message_source, GraphSourcePath)
+    assert request.error_message_source.root == "state"
+    assert request.error_message_source.parts == ("error_message",)
+
+
 def test_workflow_surface_accepts_canonical_bindings_for_minimal_workspace() -> None:
     service = WfMcpService(
         store=FileStore(local_temp_root() / "surface_minimal_canonical_mcp"),
@@ -937,6 +1057,42 @@ def test_workflow_surface_creates_draft_workspace_from_capability_hints() -> Non
     ]
 
 
+def test_workflow_surface_does_not_auto_map_raw_mcp_content_blocks() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_content_only_content_hint"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_content_only_content_hint_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(
+            id="everything.default", server="everything", account="default"
+        )
+    )
+    service.register_adapter("everything", ContentOnlyOutputAdapter())
+    asyncio.run(service.refresh_connection_catalog("everything.default"))
+    handlers = WorkflowSurfaceHandlers(service)
+
+    inspected = asyncio.run(
+        handlers.inspect_capability(qualified_name="everything.default.echo")
+    )
+    created = asyncio.run(
+        handlers.create_draft_workspace_from_capability(
+            workspace_id="content_blocks",
+            capability_name="everything.default.echo",
+            name="content_blocks",
+        )
+    )
+
+    assert inspected["wrapper_hints"]["output_map"] == {}
+    assert created["wrapper_hints"]["output_map"] == {}
+    assert inspected["wrapper_hints"]["missing_decisions"][0]["kind"] == (
+        "review_nested_output"
+    )
+    assert "Raw MCP content blocks" in inspected["wrapper_hints"]["notes"][2]
+
+
 def test_workflow_surface_creates_artifact_from_workspace() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_workspace_artifact"

tests/wf_mcp/test_workflow_wrapper_hints.py

@@ -154,6 +154,58 @@ def test_wrapper_hints_mark_nested_outputs_as_low_confidence() -> None:
     assert dumped["output_map"] == {"results": "state.results"}
 
 
+def test_wrapper_hints_do_not_auto_map_raw_mcp_content_blocks() -> None:
+    hints = wrapper_hints_for_capability(
+        capability_name="everything.default.echo",
+        input_schema={
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {
+                "content": {"type": "array"},
+            },
+        },
+        outcomes=["ok", "error"],
+    )
+
+    dumped = hints.model_dump(mode="json")
+
+    assert dumped["confidence"] == "low"
+    assert dumped["output_map"] == {}
+    assert "content" not in dumped["state_schema"]["properties"]
+    assert dumped["missing_decisions"][0]["kind"] == "review_nested_output"
+    assert "Raw MCP content blocks" in dumped["notes"][2]
+
+
+def test_wrapper_hints_keep_content_only_mcp_output_explicit() -> None:
+    hints = wrapper_hints_for_capability(
+        capability_name="everything.default.echo",
+        input_schema={
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {"content": {"type": "array"}},
+            "required": ["content"],
+        },
+        outcomes=["ok", "error"],
+    )
+
+    dumped = hints.model_dump(mode="json")
+
+    assert dumped["confidence"] == "low"
+    assert dumped["output_map"] == {}
+    assert "content" not in dumped["state_schema"]["properties"]
+    assert dumped["output_schema"]["properties"] == {}
+    assert dumped["missing_decisions"][0]["kind"] == "review_nested_output"
+    assert "TextContent" in dumped["notes"][2]
+
+
 def test_wrapper_hints_mark_empty_output_schema_as_low_confidence() -> None:
     hints = wrapper_hints_for_capability(
         capability_name="demo.personal.no_output",

tests/wf_mcp/test_workflow_wrappers.py

@@ -29,6 +29,28 @@ class RecordingAdapter:
         return ToolCallResult(outcome="ok", output={})
 
 
+class TextContentAdapter:
+    async def call_tool(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        tool_name: str,
+        payload: dict[str, Any],
+    ) -> ToolCallResult:
+        message = payload.get("message", "")
+        return ToolCallResult(
+            outcome="ok",
+            output={
+                "content": [
+                    {
+                        "type": "text",
+                        "text": f"Echo: {message}",
+                    }
+                ],
+            },
+        )
+
+
 def test_discovered_tool_wrapper_omits_unset_optional_arguments() -> None:
     adapter = RecordingAdapter()
     spec = wrap_discovered_tool(
@@ -66,3 +88,43 @@ def test_discovered_tool_wrapper_omits_unset_optional_arguments() -> None:
 
     assert adapter.payloads[0] == {}
     assert adapter.payloads[1] == {"target": "main"}
+
+
+def test_discovered_tool_wrapper_preserves_raw_mcp_content_output() -> None:
+    spec = wrap_discovered_tool(
+        connection=ConnectionConfig(
+            id="everything.default",
+            server="everything",
+            account="default",
+        ),
+        auth=None,
+        executor=cast(ToolExecutor, TextContentAdapter()),
+        tool=DiscoveredTool(
+            name="echo",
+            title="Echo",
+            description=None,
+            input_schema={
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+                "required": ["message"],
+            },
+            output_schema={
+                "type": "object",
+                "properties": {"content": {"type": "array"}},
+            },
+        ),
+    )
+    handler = build_async_registry(spec)[spec.name]
+
+    async def run_call() -> dict[str, Any]:
+        return await handler(
+            {"message": "hello"},
+            RuntimeContext(current_node_id="echo"),
+        )
+
+    result = asyncio.run(run_call())
+
+    assert result["outcome"] == "ok"
+    assert "text" not in result["output"]
+    assert result["output"]["content"][0]["type"] == "text"
+    assert result["output"]["content"][0]["text"] == "Echo: hello"