fix: address rpc and mcp review followups

docs/current_roadmap.md

@@ -100,6 +100,12 @@ implementation state.
       public `RpcWorkflowApiClient` still satisfies `WorkflowApiSurface`, while
       client methods and server JSON-RPC registrations live in focused
       capability, draft, artifact, deployment, and run modules.
+   - Completed: JSON-RPC and CLI now expose `call_capability` through
+      `workflow.capabilities.call` and `wf cap call`, so local and remote users
+      can smoke-test a capability before creating draft workspaces.
+   - Completed: RPC client domain mixins now share a single typed
+      `RpcCaller._call(method, params)` transport primitive instead of
+      repeating `_call` stubs in every mixin.
    - Completed: read-only source inventory now has a protocol-neutral
       `WorkflowSourceAdminApi` / `WorkflowSourceAdminSurface`; MCP admin source
       tools delegate through it while connection/raw MCP operations remain

docs/wf_api_architecture.md

@@ -16,7 +16,9 @@ frontends can share.
 | `wf_artifacts` | Saved definitions and persistence contracts: workflow artifacts, deployments, draft workspaces, run records, checkpoints, artifact/deployment validation models. |
 | `wf_platform` | Shared capability/source concepts and platform-facing contracts such as capability refs, source inventory models, documentation source models, and JSON schema helpers. |
 | `wf_api` | Application workflow contract and process-local implementation over core/artifacts/platform: capability discovery, wrapper hints, draft editing, artifact/deployment operations, run/resume operations, next actions, and progressive response shaping. |
-| `wf_mcp` | MCP-specific transport, tool schemas, upstream MCP adapters, broker services, proxy/admin tools, config reload, and `WorkflowApi` context construction for MCP. |
+| `wf_server` | Durable server composition boundary that hosts a `WorkflowApi` plus optional admin/source-registry surfaces. |
+| `wf_sources_mcp` | MCP-as-upstream-source implementation: source ids, source registry DTOs, auth/catalog stores, discovery, SDK client/facade, persistent runtime pool, and tool-wrapper helpers. |
+| `wf_mcp` | MCP frontend/compatibility package: old `wf-mcp` server entry points, broker glue around MCP-hosted services, proxy/admin tools, and compatibility shims while callers migrate. |
 | `wf_transport_rpc_http` | JSON-RPC-over-HTTP transport adapter and remote client over `WorkflowApiSurface`, not a reimplementation of workflow business logic. |
 | future `wf_http` / WebSocket / MCP server transports | Additional transports over `WorkflowApiSurface`, not new workflow application APIs. |
 | `wf_cli` | CLI frontend over `WorkflowApiSurface`; it may run locally against process-local stores or target a remote JSON-RPC backend. |
@@ -117,6 +119,21 @@ can satisfy it by method shape:
 - Future auth, cache, recording, WebSocket, or MCP-server adapters should also
   target the same surface instead of importing concrete implementation classes.
 
+Current remote CLI flow:
+
+```text
+wf_cli
+  -> wf_transport_rpc_http.RpcWorkflowApiClient
+  -> wf_server.WorkflowServer
+  -> wf_api.WorkflowApi / admin surfaces
+  -> wf_sources_mcp or other source implementations
+```
+
+`RpcWorkflowApiClient` is intentionally composed from domain mixins over one
+transport primitive, `RpcCaller._call(method, params)`. Domain mixins must not
+own HTTP state or duplicate `_call` stubs; they are method bundles over the
+shared JSON-RPC request primitive.
+
 The surface is split into domain protocols:
 
 ```text

docs/wf_cli.md

@@ -153,6 +153,18 @@ wf cap inspect wf.std.concat
 `inspect` returns the full contract, including `wrapper_hints` when available.
 Hints are scaffolding, not semantic guarantees.
 
+Call one capability once before creating a draft:
+
+```bash
+wf cap call wf.std.constant --input '{"value": "hello"}'
+wf --url http://127.0.0.1:8765/rpc cap call everything.default.echo --input '{"message": "hello"}'
+```
+
+`cap call` is an authoring/runtime smoke test. It uses the same local or remote
+target selection as the rest of the CLI and returns a normalized outcome,
+output, source id, and diagnostics. Use it to confirm payload shape and upstream
+source reachability before spending time on a draft workspace.
+
 ## Draft Workspaces
 
 Create a draft from a capability:

skills/wf-cli/SKILL.md

@@ -28,6 +28,7 @@ Canonical docs:
 ```bash
 wf cap list --format ids
 wf cap inspect <capability>
+wf cap call <capability> --input '{"field":"value"}'
 
 wf draft create-from-capability <workspace_id> <capability>
 wf draft inspect <workspace_id> --include-draft
@@ -45,6 +46,7 @@ wf run trace <run_id> --from 0 --limit 25
 
 - Prefer `--input-file` for large JSON.
 - Prefer `--format ids` or `--format compact` for discovery.
+- Use `wf cap call` as a cheap smoke test before creating a draft.
 - Do not request unbounded traces.
 - Do not treat wrapper hints as semantic guarantees.
 - If validation fails, run `wf explain <code>` or `wf explain --input-file <validation-output.json>`.

skills/wf-workflow/SKILL.md

@@ -20,21 +20,25 @@ low-level escape hatch.
 1. Discover sources with `wf.admin.list_sources` or `wf cap list`.
 2. Discover workflow-ready capabilities with `wf.workflow.list_capabilities`.
 3. Inspect one candidate with `wf.workflow.inspect_capability`.
-4. Create a patchable draft workspace with
+4. Call one candidate with `wf.workflow.call_capability` or `wf cap call` when
+   payload shape or upstream source reachability is uncertain.
+5. Create a patchable draft workspace with
    `wf.workflow.create_draft_workspace_from_capability`.
-5. Patch targeted fields with focused helpers or JSON Patch.
+6. Patch targeted fields with focused helpers or JSON Patch.
-6. Validate with `wf.workflow.validate_draft_workspace`.
+7. Validate with `wf.workflow.validate_draft_workspace`.
-7. Save with `wf.workflow.create_artifact_from_workspace` or
+8. Save with `wf.workflow.create_artifact_from_workspace` or
    `wf.workflow.create_wrapper_from_workspace`.
-8. Save a deployment with `wf.workflow.save_deployment`.
+9. Save a deployment with `wf.workflow.save_deployment`.
-9. Validate with `wf.workflow.validate_deployment`.
+10. Validate with `wf.workflow.validate_deployment`.
-10. Run with `wf.workflow.run_deployment`.
+11. Run with `wf.workflow.run_deployment`.
-11. Inspect stopped runs with `wf.workflow.inspect_run`; read bounded trace
+12. Inspect stopped runs with `wf.workflow.inspect_run`; read bounded trace
     slices only when debugging.
 
 ## Rules
 
 - Use workflow capabilities, not raw MCP tools, when building graphs.
+- Use `call_capability` for single-call probes; use deployments/runs for durable
+  lifecycle behavior.
 - Treat wrapper hints as scaffolding, not semantic truth.
 - Use draft workspaces for iterative authoring; avoid rewriting full drafts.
 - Use explicit source bindings at deployment time.

src/wf_cli/context.py

@@ -282,7 +282,7 @@ def _rpc_timeout_from_optional_config(
         return override
     try:
         config = load_workflow_config(path)
-    except FileNotFoundError, json.JSONDecodeError, ValidationError:
+    except (FileNotFoundError, json.JSONDecodeError, ValidationError):
         return 30.0
     target = config.client.target
     if isinstance(target, RpcHttpTargetConfig):

src/wf_mcp/broker/service/connection_service.py

@@ -83,7 +83,7 @@ class ConnectionService:
                     continue
                 if connection.id in registry_entries:
                     continue
-                seeded = connection_config_to_registry_entry(connection)  # type: ignore[arg-type]
+                seeded = connection_config_to_registry_entry(connection)
                 registry_entries[seeded.id] = seeded
                 registry_changed = True
                 self.events.record_kind(

src/wf_mcp/workflow_surface/refs.py

@@ -4,6 +4,8 @@ New code should import from `wf_api.refs`. This module stays so older MCP
 workflow-surface imports keep working until callers migrate.
 """
 
+from __future__ import annotations
+
 from wf_api.refs import (
     WorkflowSurfaceCapabilityId,
     parse_workflow_surface_capability_id,

src/wf_sources_mcp/auth.py

@@ -61,7 +61,8 @@ def mcp_auth_headers(auth: AuthRecord | None) -> dict[str, str]:
         else {}
     )
     token = auth.payload.get("token")
-    if isinstance(token, str) and "Authorization" not in headers:
+    has_authorization = any(key.lower() == "authorization" for key in headers)
+    if isinstance(token, str) and not has_authorization:
         headers["Authorization"] = f"Bearer {token}"
     return headers
 

src/wf_sources_mcp/discovery.py

@@ -69,11 +69,7 @@ def _root_exception(exc: BaseException) -> BaseException:
     """Unwrap the first nested exception from MCP task-group ExceptionGroups."""
     current: BaseException = exc
     while isinstance(current, ExceptionGroup) and current.exceptions:
-        nested = current.exceptions[0]
+        current = current.exceptions[0]
-        if isinstance(nested, BaseException):
-            current = nested
-            continue
-        break
     return current
 
 

src/wf_sources_mcp/runtime/pool.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 from collections.abc import Awaitable, Callable
 from dataclasses import asdict, dataclass, field
@@ -30,6 +31,8 @@ def connection_runtime_fingerprint(
     command, URL, account, or auth payload must create a fresh session.
     """
 
+    # Expected payloads are dataclasses/primitive mappings today. `default=str`
+    # is only a fallback for SDK URL/path-like leaf values in auth/transport data.
     return json.dumps(
         {
             "connection": asdict(connection),
@@ -53,6 +56,7 @@ class McpRuntimePool:
 
     session_factory: SessionFactory
     _sessions: dict[str, tuple[str, PersistentMcpSession]] = field(default_factory=dict)
+    _session_locks: dict[str, asyncio.Lock] = field(default_factory=dict)
 
     async def get_session(
         self,
@@ -63,16 +67,22 @@ class McpRuntimePool:
         current = self._sessions.get(connection.id)
         if current is not None and current[0] == fingerprint:
             return current[1]
-        if current is not None:
-            await current[1].close()
 
-        created = self.session_factory(connection, auth)
+        lock = self._session_locks.setdefault(connection.id, asyncio.Lock())
-        if isawaitable(created):
+        async with lock:
-            session = await created
+            current = self._sessions.get(connection.id)
-        else:
+            if current is not None and current[0] == fingerprint:
-            session = cast(PersistentMcpSession, created)
+                return current[1]
-        self._sessions[connection.id] = (fingerprint, session)
+            if current is not None:
-        return session
+                await current[1].close()
+
+            created = self.session_factory(connection, auth)
+            if isawaitable(created):
+                session = await created
+            else:
+                session = cast(PersistentMcpSession, created)
+            self._sessions[connection.id] = (fingerprint, session)
+            return session
 
     async def call_tool(
         self,
@@ -157,6 +167,7 @@ class McpRuntimePool:
 
     async def close_connection(self, connection_id: str) -> None:
         current = self._sessions.pop(connection_id, None)
+        self._session_locks.pop(connection_id, None)
         if current is not None:
             await current[1].close()
 
@@ -164,5 +175,6 @@ class McpRuntimePool:
         """Close all live runtimes; useful for server shutdown and tests."""
         sessions = list(self._sessions.values())
         self._sessions.clear()
+        self._session_locks.clear()
         for _fingerprint, session in sessions:
             await session.close()

src/wf_sources_mcp/schema_models.py

@@ -38,7 +38,7 @@ def _python_type_from_schema(schema: object) -> object:
 
     if schema_type == "array":
         item_type = _python_type_from_schema(schema.get("items", {}))
-        return list[item_type] if isinstance(item_type, type) else list[Any]
+        return list[item_type]
 
     if not isinstance(schema_type, str):
         return Any
@@ -52,7 +52,7 @@ def _optional_type(annotation: object) -> object:
     origin = get_origin(annotation)
     if origin in {Union, UnionType} and NoneType in get_args(annotation):
         return annotation
-    return annotation | None if isinstance(annotation, type) else Any
+    return cast(Any, annotation) | None
 
 
 def _field_default(

src/wf_sources_mcp/source_registry.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Literal, Protocol, cast
+from typing import Any, Literal, Protocol, cast
 
 from pydantic import Field, field_validator, model_validator
 
@@ -60,7 +60,7 @@ class LegacyConnectionConfigLike(Protocol):
     def enabled(self) -> bool: ...
 
     @property
-    def metadata(self) -> Mapping[str, object]: ...
+    def metadata(self) -> Mapping[str, Any]: ...
 
 
 class McpSourceRegistryEntry(SourceRegistryBaseModel):

src/wf_transport_rpc_http/client/admin.py

@@ -3,32 +3,32 @@ from __future__ import annotations
 from collections.abc import Mapping
 from typing import Any
 
+from .base import RpcCaller
+
 
 class RpcAdminClientMixin:
     """JSON-RPC implementation of read-only admin/config surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
+    async def list_connections(self: RpcCaller) -> dict[str, Any]:
-
-    async def list_connections(self) -> dict[str, Any]:
         return await self._call("workflow.admin.connections.list", {})
 
-    async def get_connection_statuses(self) -> dict[str, Any]:
+    async def get_connection_statuses(self: RpcCaller) -> dict[str, Any]:
         return await self._call("workflow.admin.connection_statuses.list", {})
 
-    async def list_events(self) -> dict[str, Any]:
+    async def list_events(self: RpcCaller) -> dict[str, Any]:
         return await self._call("workflow.admin.events.list", {})
 
-    async def list_auth_records(self) -> dict[str, Any]:
+    async def list_auth_records(self: RpcCaller) -> dict[str, Any]:
         return await self._call("workflow.admin.auth.list", {})
 
-    async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
+    async def inspect_auth_record(self: RpcCaller, auth_ref: str) -> dict[str, Any]:
         return await self._call(
             "workflow.admin.auth.inspect",
             {"auth_ref": auth_ref},
         )
 
     async def save_auth_record(
-        self,
+        self: RpcCaller,
         *,
         auth_ref: str,
         scheme: str,
@@ -45,7 +45,7 @@ class RpcAdminClientMixin:
             },
         )
 
-    async def delete_auth_record(self, auth_ref: str) -> dict[str, Any]:
+    async def delete_auth_record(self: RpcCaller, auth_ref: str) -> dict[str, Any]:
         return await self._call(
             "workflow.admin.auth.delete",
             {"auth_ref": auth_ref},

src/wf_transport_rpc_http/client/artifacts.py

@@ -2,14 +2,14 @@ from __future__ import annotations
 
 from typing import Any, Literal
 
+from .base import RpcCaller
+
 
 class RpcArtifactClientMixin:
     """JSON-RPC implementation of workflow artifact surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
-
     async def list_artifacts(
-        self,
+        self: RpcCaller,
         *,
         query: str | None = None,
         kind: Literal["workflow", "wrapper"] | None = None,
@@ -27,12 +27,14 @@ class RpcArtifactClientMixin:
         )
 
     async def inspect_artifact(
-        self, *, artifact_id: str, version: int
+        self: RpcCaller, *, artifact_id: str, version: int
     ) -> dict[str, Any]:
         return await self._call(
             "workflow.artifacts.inspect",
             {"artifact_id": artifact_id, "version": version},
         )
 
-    async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
+    async def save_artifact(
+        self: RpcCaller, artifact: dict[str, Any]
+    ) -> dict[str, Any]:
         return await self._call("workflow.artifacts.save", {"artifact": artifact})

src/wf_transport_rpc_http/client/base.py

@@ -1,12 +1,18 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Protocol
 from uuid import uuid4
 
 import httpx
 
 
+class RpcCaller(Protocol):
+    """Transport primitive required by domain RPC client mixins."""
+
+    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
+
+
 @dataclass(slots=True)
 class RpcClientTransport:
     """Shared JSON-RPC request plumbing for workflow RPC client mixins.

src/wf_transport_rpc_http/client/capabilities.py

@@ -2,14 +2,14 @@ from __future__ import annotations
 
 from typing import Any
 
+from .base import RpcCaller
+
 
 class RpcCapabilityClientMixin:
     """JSON-RPC implementation of workflow capability surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
-
     async def list_capabilities(
-        self,
+        self: RpcCaller,
         *,
         query: str | None = None,
         source_id: str | None = None,
@@ -26,14 +26,16 @@ class RpcCapabilityClientMixin:
             },
         )
 
-    async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
+    async def inspect_capability(
+        self: RpcCaller, *, qualified_name: str
+    ) -> dict[str, Any]:
         return await self._call(
             "workflow.capabilities.inspect",
             {"qualified_name": qualified_name},
         )
 
     async def call_capability(
-        self,
+        self: RpcCaller,
         *,
         qualified_name: str,
         payload: dict[str, Any],

src/wf_transport_rpc_http/client/deployments.py

@@ -2,33 +2,39 @@ from __future__ import annotations
 
 from typing import Any
 
+from .base import RpcCaller
+
 
 class RpcDeploymentClientMixin:
     """JSON-RPC implementation of workflow deployment surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
+    async def list_deployments(self: RpcCaller) -> dict[str, Any]:
-
-    async def list_deployments(self) -> dict[str, Any]:
         return await self._call("workflow.deployments.list", {})
 
-    async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+    async def inspect_deployment(
+        self: RpcCaller, *, deployment_id: str
+    ) -> dict[str, Any]:
         return await self._call(
             "workflow.deployments.inspect",
             {"deployment_id": deployment_id},
         )
 
     async def validate_deployment(
-        self, *, deployment_id: str, live_check: bool = False
+        self: RpcCaller, *, deployment_id: str, live_check: bool = False
     ) -> dict[str, Any]:
         return await self._call(
             "workflow.deployments.validate",
             {"deployment_id": deployment_id, "live_check": live_check},
         )
 
-    async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
+    async def save_deployment(
+        self: RpcCaller, deployment: dict[str, Any]
+    ) -> dict[str, Any]:
         return await self._call("workflow.deployments.save", {"deployment": deployment})
 
-    async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+    async def delete_deployment(
+        self: RpcCaller, *, deployment_id: str
+    ) -> dict[str, Any]:
         return await self._call(
             "workflow.deployments.delete",
             {"deployment_id": deployment_id},

src/wf_transport_rpc_http/client/drafts.py

@@ -3,17 +3,17 @@ from __future__ import annotations
 from collections.abc import Sequence
 from typing import Any, Literal
 
+from .base import RpcCaller
+
 
 class RpcDraftClientMixin:
     """JSON-RPC implementation of workflow draft workspace surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
+    async def list_draft_workspaces(self: RpcCaller) -> dict[str, Any]:
-
-    async def list_draft_workspaces(self) -> dict[str, Any]:
         return await self._call("workflow.draft_workspaces.list", {})
 
     async def get_draft_workspace(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
         include_draft: bool = False,
@@ -24,7 +24,7 @@ class RpcDraftClientMixin:
         )
 
     async def create_draft_workspace_from_capability(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
         capability_name: str,
@@ -58,7 +58,7 @@ class RpcDraftClientMixin:
         )
 
     async def patch_draft_workspace(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
         revision: int,
@@ -70,7 +70,7 @@ class RpcDraftClientMixin:
         )
 
     async def validate_draft_workspace(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
     ) -> dict[str, Any]:
@@ -80,7 +80,7 @@ class RpcDraftClientMixin:
         )
 
     async def create_artifact_from_workspace(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
         artifact_id: str,
@@ -110,7 +110,7 @@ class RpcDraftClientMixin:
         )
 
     async def create_wrapper_from_workspace(
-        self,
+        self: RpcCaller,
         *,
         workspace_id: str,
         artifact_id: str,

src/wf_transport_rpc_http/client/runs.py

@@ -4,14 +4,14 @@ from typing import Any
 
 from wf_api.runs import TraceRangeLike
 
+from .base import RpcCaller
+
 
 class RpcRunClientMixin:
     """JSON-RPC implementation of workflow run lifecycle surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
-
     async def run_deployment(
-        self,
+        self: RpcCaller,
         *,
         deployment_id: str,
         workflow_input: dict[str, Any],
@@ -27,7 +27,7 @@ class RpcRunClientMixin:
         )
 
     async def resume_run(
-        self,
+        self: RpcCaller,
         *,
         run_id: str,
         resume_payload: dict[str, Any],
@@ -44,11 +44,11 @@ class RpcRunClientMixin:
             },
         )
 
-    async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
+    async def inspect_run(self: RpcCaller, *, run_id: str) -> dict[str, Any]:
         return await self._call("workflow.runs.inspect", {"run_id": run_id})
 
     async def read_run_trace(
-        self,
+        self: RpcCaller,
         *,
         run_id: str,
         trace_range: TraceRangeLike,

src/wf_transport_rpc_http/client/source_registry.py

@@ -2,14 +2,14 @@ from __future__ import annotations
 
 from typing import Any
 
+from .base import RpcCaller
+
 
 class RpcSourceRegistryClientMixin:
     """JSON-RPC implementation of source registry surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
-
     async def list_registry_entries(
-        self,
+        self: RpcCaller,
         *,
         cursor: str | None = None,
         limit: int = 50,
@@ -20,7 +20,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def inspect_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         source_id: str,
     ) -> dict[str, Any]:
@@ -30,7 +30,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def add_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         entry: dict[str, Any],
     ) -> dict[str, Any]:
@@ -40,7 +40,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def update_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         source_id: str,
         patch: dict[str, Any],
@@ -51,7 +51,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def enable_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         source_id: str,
     ) -> dict[str, Any]:
@@ -61,7 +61,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def disable_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         source_id: str,
     ) -> dict[str, Any]:
@@ -71,7 +71,7 @@ class RpcSourceRegistryClientMixin:
         )
 
     async def remove_registry_entry(
-        self,
+        self: RpcCaller,
         *,
         source_id: str,
     ) -> dict[str, Any]:
@@ -80,7 +80,7 @@ class RpcSourceRegistryClientMixin:
             {"source_id": source_id},
         )
 
-    async def apply_registry_changes(self) -> dict[str, Any]:
+    async def apply_registry_changes(self: RpcCaller) -> dict[str, Any]:
         return await self._call(
             "workflow.admin.source_registry.apply",
             {},

src/wf_transport_rpc_http/client/sources.py

@@ -2,14 +2,14 @@ from __future__ import annotations
 
 from typing import Any
 
+from .base import RpcCaller
+
 
 class RpcSourceAdminClientMixin:
     """JSON-RPC implementation of read-only source admin surface methods."""
 
-    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
-
     async def list_sources(
-        self,
+        self: RpcCaller,
         *,
         cursor: str | None = None,
         limit: int = 50,
@@ -22,7 +22,7 @@ class RpcSourceAdminClientMixin:
             },
         )
 
-    async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
+    async def inspect_source(self: RpcCaller, *, source_id: str) -> dict[str, Any]:
         return await self._call(
             "workflow.sources.inspect",
             {"source_id": source_id},

src/wf_transport_rpc_http/methods/source_registry.py

@@ -26,11 +26,12 @@ def _require_source_registry_admin(
 ) -> WorkflowSourceRegistrySurface:
     admin = server.source_registry_admin
     if admin is None:
+        verb = "are" if operation in {"reads", "mutations"} else "is"
         raise WorkflowRpcError(
             data={
                 "code": "source_registry_unavailable",
                 "message": (
-                    f"source registry admin {operation} are not available "
+                    f"source registry admin {operation} {verb} not available "
                     "for this server"
                 ),
             }

tests/wf_mcp/service/conftest.py

@@ -28,8 +28,8 @@ class ContentOnlyOutputAdapter(FakeAdapter):
 
     async def list_tools(
         self,
-        connection,
+        connection: ConnectionConfig,
-        auth,
+        auth: AuthRecord | None,
     ) -> list[DiscoveredTool]:
         return [
             DiscoveredTool(
@@ -51,8 +51,8 @@ class ContentOnlyOutputAdapter(FakeAdapter):
 
     async def call_tool(
         self,
-        connection,
+        connection: ConnectionConfig,
-        auth,
+        auth: AuthRecord | None,
         tool_name: str,
         payload: dict[str, Any],
     ) -> ToolCallResult:

tests/wf_mcp/test_source_registry.py

@@ -142,7 +142,7 @@ def test_connection_config_to_registry_entry_preserves_transport_metadata() -> N
         },
     )
 
-    entry = connection_config_to_registry_entry(connection)  # type: ignore[arg-type]
+    entry = connection_config_to_registry_entry(connection)
 
     assert entry.id == "github.work"
     assert entry.provider == "github"
@@ -168,7 +168,7 @@ def test_connection_config_to_registry_entry_accepts_flat_stdio_metadata() -> No
         },
     )
 
-    entry = connection_config_to_registry_entry(connection)  # type: ignore[arg-type]
+    entry = connection_config_to_registry_entry(connection)
 
     assert entry.transport.kind == "stdio"
     assert isinstance(entry.transport, StdioSourceTransport)
@@ -191,7 +191,7 @@ def test_connection_config_to_registry_entry_accepts_flat_http_metadata() -> Non
         },
     )
 
-    entry = connection_config_to_registry_entry(connection)  # type: ignore[arg-type]
+    entry = connection_config_to_registry_entry(connection)
 
     assert entry.transport.kind == "http"
     assert isinstance(entry.transport, HttpSourceTransport)
@@ -204,7 +204,7 @@ def test_connection_config_to_registry_entry_requires_transport_metadata() -> No
     connection = ConnectionConfig(id="github.work", server="github", account="work")
 
     with pytest.raises(ValueError, match="requires metadata.transport"):
-        connection_config_to_registry_entry(connection)  # type: ignore[arg-type]
+        connection_config_to_registry_entry(connection)
 
 
 class _McpSource:

tests/wf_sources_mcp/test_auth_storage_exports.py

@@ -47,6 +47,19 @@ def test_wf_sources_mcp_auth_adapters_interpret_mcp_payload() -> None:
     assert mcp_auth_env(auth) == {"GITHUB_TOKEN": "secret"}
 
 
+def test_wf_sources_mcp_auth_headers_preserve_existing_authorization_case() -> None:
+    auth = AuthRecord(
+        connection_id="github.work",
+        scheme="bearer",
+        payload={
+            "token": "secret",
+            "headers": {"authorization": "Bearer custom"},
+        },
+    )
+
+    assert mcp_auth_headers(auth) == {"authorization": "Bearer custom"}
+
+
 def test_wf_sources_mcp_file_stores_keep_existing_disk_shape(tmp_path) -> None:
     auth_store = FileAuthStore(tmp_path / "auth-root")
     catalog_store = FileCatalogStore(tmp_path / "catalog-root")

tests/wf_sources_mcp/test_runtime.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from contextlib import AsyncExitStack
 from typing import Any
 
@@ -259,6 +260,41 @@ async def test_runtime_pool_reuses_unchanged_connection() -> None:
     assert created == [connection]
 
 
+@pytest.mark.asyncio
+async def test_runtime_pool_serializes_concurrent_session_creation() -> None:
+    created: list[McpSourceConnection] = []
+    release = asyncio.Event()
+
+    async def create_session(
+        connection: McpSourceConnection, auth: AuthRecord | None
+    ) -> PersistentMcpSession:
+        created.append(connection)
+        await release.wait()
+
+        async def _call(tool_name: str, payload: dict[str, Any]) -> ToolCallResult:
+            return ToolCallResult(outcome="ok", output={"echoed": payload["text"]})
+
+        return PersistentMcpSession(
+            connection=connection,
+            auth=auth,
+            call_callback=_call,
+        )
+
+    pool = McpRuntimePool(session_factory=create_session)
+    connection = _connection()
+
+    first = asyncio.create_task(pool.get_session(connection, None))
+    second = asyncio.create_task(pool.get_session(connection, None))
+    await asyncio.sleep(0)
+    release.set()
+
+    first_session, second_session = await asyncio.gather(first, second)
+    await pool.close_all()
+
+    assert first_session is second_session
+    assert created == [connection]
+
+
 def test_runtime_fingerprint_changes_when_transport_changes() -> None:
     original = _connection()
     changed = McpSourceConnection(

tests/wf_sources_mcp/test_schema_models.py

@@ -111,6 +111,45 @@ def test_model_from_schema_allows_extra_fields_and_tolerates_unknown_shapes() ->
     assert dumped["extra"] == "kept"
 
 
+def test_model_from_schema_preserves_complex_array_item_annotations() -> None:
+    model = model_from_schema(
+        "NestedArrayInput",
+        {
+            "type": "object",
+            "properties": {
+                "items": {
+                    "type": "array",
+                    "items": {
+                        "type": ["object", "null"],
+                    },
+                },
+            },
+        },
+    )
+
+    annotation = model.model_fields["items"].annotation
+
+    assert str(annotation) == "list[dict[str, typing.Any] | None]"
+
+
+def test_model_from_schema_preserves_optional_complex_annotations() -> None:
+    model = model_from_schema(
+        "OptionalObjectInput",
+        {
+            "type": "object",
+            "properties": {
+                "metadata": {
+                    "type": ["object", "null"],
+                },
+            },
+        },
+    )
+
+    annotation = model.model_fields["metadata"].annotation
+
+    assert str(annotation) == "dict[str, typing.Any] | None"
+
+
 def test_model_from_schema_exports_from_package_root() -> None:
     from wf_sources_mcp import model_from_schema as root_model_from_schema
     from wf_sources_mcp.schema_models import model_from_schema

tests/wf_transport_rpc_http/test_structure.py

@@ -2,8 +2,11 @@ from __future__ import annotations
 
 import importlib
 import inspect
+import pkgutil
 
+import wf_transport_rpc_http.client as rpc_client_package
 from wf_transport_rpc_http.client import RpcWorkflowApiClient
+from wf_transport_rpc_http.client.base import RpcCaller, RpcClientTransport
 
 
 def test_rpc_transport_has_domain_method_modules() -> None:
@@ -26,3 +29,20 @@ def test_rpc_transport_client_stays_thin() -> None:
     line_count = len(inspect.getsource(RpcWorkflowApiClient).splitlines())
 
     assert line_count < 40
+
+
+def test_rpc_client_mixins_share_one_call_contract() -> None:
+    call_owners = {
+        RpcClientTransport._call,
+        RpcCaller._call,
+    }
+
+    for module_info in pkgutil.iter_modules(rpc_client_package.__path__):
+        if module_info.name in {"__init__", "base"}:
+            continue
+        module = importlib.import_module(f"wf_transport_rpc_http.client.{module_info.name}")
+        for _name, value in inspect.getmembers(module, inspect.isclass):
+            if value.__module__ == module.__name__:
+                assert "_call" not in value.__dict__
+
+    assert len(call_owners) == 2