add lil peek to in/output

docs/wf_mcp_end_to_end_runbook.md

@@ -162,6 +162,9 @@ arguments:
 }
 ```
 
+The list response is intentionally compact. It includes `source_id`, outcomes,
+and top-level `input_fields` / `output_fields`, but not full JSON schemas.
+
 Then inspect its full contract:
 
 ```yaml

docs/wf_mcp_operator_manual.md

@@ -174,6 +174,11 @@ only when you need its full owned-capability list. The compact response includes
 counts plus small preview lists and `has_more` flags, so clients can usually
 choose the next source to inspect without loading every schema.
 
+Use `wf.workflow.list_capabilities` after that when you need workflow-ready
+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`.
+
 ### 4. Manage Saved Workflows
 
 Use `wf.workflow.*` for artifacts and deployments:

docs/workflow_capabilities.md

@@ -262,6 +262,8 @@ authoring loop:
 - `wf.workflow.list_capabilities`
   - lists compact paged enabled planner-visible workflow-ready node spec
     summaries, with optional query/source filtering
+  - includes the owning `source_id`, outcomes, and top-level input/output field
+    names, but not full schemas
 - `wf.workflow.inspect_capability`
   - returns one full workflow capability contract with schemas and outcomes
 - `wf.workflow.call_capability`
@@ -271,6 +273,15 @@ These are authoring-plane tools. They do not replace the privileged
 `wf.admin.list_sources` source inventory, and older planner-catalog projections
 may remain while callers migrate to the workflow-facing surface.
 
+Recommended discovery order:
+
+1. Use `wf.admin.list_sources` to find capability owners and preview source
+   contents.
+2. Use `wf.workflow.list_capabilities` with `source_id` or `query` to find
+   workflow-ready node specs.
+3. Use `wf.workflow.inspect_capability` only for the selected capability's full
+   schema contract.
+
 ## Relationship To Capability Sources
 
 Sources own capability kinds:

src/wf_mcp/workflow_surface/handlers.py

@@ -59,9 +59,12 @@ class WorkflowSurfaceHandlers:
         capabilities = [
             {
                 "name": detail.name,
+                "source_id": source.id,
                 "description": detail.description,
                 "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),
             }
             for source in sorted(
                 self.service.capability_sources.values(),
@@ -555,6 +558,14 @@ def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
     return observed
 
 
+def _schema_field_names(schema: dict[str, Any]) -> list[str]:
+    """Return top-level JSON object property names for compact discovery rows."""
+    properties = schema.get("properties")
+    if not isinstance(properties, dict):
+        return []
+    return sorted(str(name) for name in properties)
+
+
 def _capability_name(qualified_name: str) -> str | None:
     """Return the local name of one qualified capability ref if it is valid."""
     try:

tests/wf_mcp/test_workflow_surface.py

@@ -79,12 +79,16 @@ def test_workflow_surface_lists_planner_visible_capabilities() -> None:
 
     payload = asyncio.run(handlers.list_capabilities(limit=2))
     names = [capability["name"] for capability in payload["capabilities"]]
+    first = payload["capabilities"][0]
 
     assert len(names) == 2
     assert payload["total"] >= 2
     assert payload["next_cursor"] == "2"
-    assert "description" in payload["capabilities"][0]
-    assert "input_schema" not in payload["capabilities"][0]
+    assert "description" in first
+    assert "source_id" in first
+    assert "input_fields" in first
+    assert "output_fields" in first
+    assert "input_schema" not in first
     assert "wf.admin.list_sources" not in names
 
 
@@ -98,6 +102,7 @@ def test_workflow_surface_filters_capabilities_by_source() -> None:
     assert [capability["name"] for capability in payload["capabilities"]] == [
         "wf.mcp.call_tool"
     ]
+    assert payload["capabilities"][0]["source_id"] == "wf.mcp"
 
 
 def test_workflow_surface_inspects_one_capability() -> None: