From fdce267c55dc365cde89e0c948deb8bf0b77cb82 Mon Sep 17 00:00:00 2001
From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com>
Date: Tue, 15 Sep 2026 22:20:26 +0200
Subject: [PATCH 1/4] core: a protocol can be registered per client, and a
 client owns what it creates
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A communication protocol registered in communication_protocols is one
instance shared by every UtcpClient in the process, and so is any state
it keeps. Right for a credential cache; wrong for connections: a caller
creating a client per tenant, per user or per pooled connection was not
actually isolating them, and no client could tear its own down — the
client had no close() at all.

New registry: communication_protocol_factories, filled through
register_communication_protocol_factory. UtcpClient.create calls a
factory once per client, and the client records the instance as OWNED.
A factory wins over a shared instance of the same type, so a plugin
migrates by moving its registration and callers change nothing. Shared
instances are still looked up live, so late registration keeps working.

Teardown is scoped to what the client owns. UtcpClient.close() (new,
on the interface and the implementation) closes the owned instances and
leaves shared ones to the process — every instance is closed even when
one fails, then the failures are raised together as
UtcpProtocolCloseError. create() adopts the factory instances inside a
cleanup guard: nothing that needs closing is created in the constructor
(it cannot await), a factory that raises part-way leaves the earlier
instances closable, and any initialization failure closes them before
re-raising, a failing cleanup being logged with the original error kept
as the one the caller sees.

CommunicationProtocol.close() is now part of the interface, a no-op by
default, so the client can close any protocol uniformly. All of it
carries REQUIRED docstrings, since the spec is generated from them.

Tests (10, through the public surface): per-client routing, factory over
shared, late shared registration, unknown type names both registries,
close() scoped to own instance, close() waits for every instance,
failed create closes what it created, a raising factory leaves earlier
instances closed, shared survives a failed create, failing cleanup is
reported. Each guard mutation-checked. Core 51/51, http 238, cli 62,
text 11.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 README.md                                     |  23 ++
 core/README.md                                |  23 ++
 core/src/utcp/exceptions/__init__.py          |   4 +-
 .../exceptions/utcp_protocol_close_error.py   |  16 ++
 .../utcp_client_implementation.py             | 108 ++++++--
 .../utcp/interfaces/communication_protocol.py |  37 ++-
 core/src/utcp/plugins/discovery.py            |  26 ++
 core/src/utcp/utcp_client.py                  |  15 ++
 .../client/test_client_protocol_ownership.py  | 242 ++++++++++++++++++
 9 files changed, 472 insertions(+), 22 deletions(-)
 create mode 100644 core/src/utcp/exceptions/utcp_protocol_close_error.py
 create mode 100644 core/tests/client/test_client_protocol_ownership.py

diff --git a/README.md b/README.md
index 400b0da..a9b5196 100644
--- a/README.md
+++ b/README.md
@@ -74,8 +74,31 @@ client = await UtcpClient.create(config={
 
 # Call a tool
 result = await client.call_tool("my_api.get_data", {"id": "123"})
+
+# Release the connections this client opened
+await client.close()
+```
+
+### Protocol Lifetime: Shared vs. Per-Client Instances
+
+A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to.
+
+An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
+
+A protocol that holds **connections** registers a **factory** instead. `UtcpClient.create` calls it once per client, so each client gets its own instance, its own connections, and its own teardown on `close()`:
+
+```python
+from utcp.plugins.discovery import register_communication_protocol_factory
+
+register_communication_protocol_factory("custom_type", CustomCommunicationProtocol)
 ```
 
+This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing.
+
+`utcp-mcp` registers this way — MCP sessions (and, for stdio, child processes) belong to the client that opened them. `utcp-http` stays a shared instance: its OAuth token cache is meant to be reused across clients.
+
+`client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised.
+
 ## Protocol Plugins
 
 UTCP supports multiple communication protocols through dedicated plugins:
diff --git a/core/README.md b/core/README.md
index 400b0da..a9b5196 100644
--- a/core/README.md
+++ b/core/README.md
@@ -74,8 +74,31 @@ client = await UtcpClient.create(config={
 
 # Call a tool
 result = await client.call_tool("my_api.get_data", {"id": "123"})
+
+# Release the connections this client opened
+await client.close()
+```
+
+### Protocol Lifetime: Shared vs. Per-Client Instances
+
+A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to.
+
+An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
+
+A protocol that holds **connections** registers a **factory** instead. `UtcpClient.create` calls it once per client, so each client gets its own instance, its own connections, and its own teardown on `close()`:
+
+```python
+from utcp.plugins.discovery import register_communication_protocol_factory
+
+register_communication_protocol_factory("custom_type", CustomCommunicationProtocol)
 ```
 
+This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing.
+
+`utcp-mcp` registers this way — MCP sessions (and, for stdio, child processes) belong to the client that opened them. `utcp-http` stays a shared instance: its OAuth token cache is meant to be reused across clients.
+
+`client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised.
+
 ## Protocol Plugins
 
 UTCP supports multiple communication protocols through dedicated plugins:
diff --git a/core/src/utcp/exceptions/__init__.py b/core/src/utcp/exceptions/__init__.py
index a33c4f6..d993eb5 100644
--- a/core/src/utcp/exceptions/__init__.py
+++ b/core/src/utcp/exceptions/__init__.py
@@ -1,7 +1,9 @@
 from utcp.exceptions.utcp_variable_not_found_exception import UtcpVariableNotFound
 from utcp.exceptions.utcp_serializer_validation_error import UtcpSerializerValidationError
+from utcp.exceptions.utcp_protocol_close_error import UtcpProtocolCloseError
 
 __all__ = [
     "UtcpVariableNotFound",
-    "UtcpSerializerValidationError"
+    "UtcpSerializerValidationError",
+    "UtcpProtocolCloseError",
 ]
diff --git a/core/src/utcp/exceptions/utcp_protocol_close_error.py b/core/src/utcp/exceptions/utcp_protocol_close_error.py
new file mode 100644
index 0000000..f79304d
--- /dev/null
+++ b/core/src/utcp/exceptions/utcp_protocol_close_error.py
@@ -0,0 +1,16 @@
+from typing import List
+
+
+class UtcpProtocolCloseError(Exception):
+    """REQUIRED
+    Raised when one or more of a client's own protocol instances failed to close.
+
+    Every instance is still asked to close before this is raised, so nothing is
+    left half-torn-down behind it; `failures` carries what each failing close raised.
+    """
+
+    def __init__(self, failures: List[BaseException], total: int):
+        self.failures = failures
+        self.total = total
+        super().__init__(f"{len(failures)} of {total} owned communication protocol(s) failed to close: "
+                         + "; ".join(f"{type(f).__name__}: {f}" for f in failures))
diff --git a/core/src/utcp/implementations/utcp_client_implementation.py b/core/src/utcp/implementations/utcp_client_implementation.py
index b88bead..5f95c4d 100644
--- a/core/src/utcp/implementations/utcp_client_implementation.py
+++ b/core/src/utcp/implementations/utcp_client_implementation.py
@@ -15,7 +15,7 @@
 from utcp.data.utcp_client_config import UtcpClientConfig, UtcpClientConfigSerializer
 from utcp.implementations.default_variable_substitutor import DefaultVariableSubstitutor
 from utcp.implementations.tag_search import TagAndDescriptionWordMatchStrategy
-from utcp.exceptions import UtcpVariableNotFound
+from utcp.exceptions import UtcpVariableNotFound, UtcpProtocolCloseError
 from utcp.data.register_manual_response import RegisterManualResult
 from utcp.interfaces.communication_protocol import CommunicationProtocol
 from utcp.exceptions import UtcpSerializerValidationError
@@ -39,6 +39,62 @@ def __init__(
     ):
         super().__init__(config, root_dir)
         self.variable_substitutor = variable_substitutor
+        # The protocol instances this client CREATED (from the factory registry)
+        # and therefore owns: what its teardown closes. They are adopted by
+        # `create()` after construction, inside its cleanup guard — a
+        # constructor cannot await, so nothing that needs closing on failure is
+        # created here. Shared instances stay in the process registry, looked up
+        # live, and are never closed on this client's behalf.
+        self._owned_comm_protocols: List[CommunicationProtocol] = []
+        self._own_comm_protocols_by_type: Dict[str, CommunicationProtocol] = {}
+
+    def _adopt_factory_protocols(self) -> None:
+        """Instantiate this client's own protocols from the factory registry.
+
+        Each instance is recorded as owned the moment it exists, so a factory
+        that raises part-way leaves the ones before it closable.
+        """
+        for protocol_type, factory in CommunicationProtocol.communication_protocol_factories.items():
+            protocol = factory()
+            self._owned_comm_protocols.append(protocol)
+            self._own_comm_protocols_by_type[protocol_type] = protocol
+
+    def _protocol_for(self, call_template_type: str) -> CommunicationProtocol:
+        """Resolve the protocol this client uses for a call template type.
+
+        This client's own instance wins over a shared one of the same type,
+        which is what lets a plugin migrate from instance to factory without
+        callers changing anything.
+        """
+        protocol = self._own_comm_protocols_by_type.get(call_template_type) or CommunicationProtocol.communication_protocols.get(call_template_type)
+        if protocol is None:
+            available = set(self._own_comm_protocols_by_type) | set(CommunicationProtocol.communication_protocols)
+            raise ValueError(f"No registered communication protocol of type {call_template_type} found, available types: {sorted(available)}")
+        return protocol
+
+    async def _close_owned_protocols(self) -> None:
+        """Close every protocol this client created, waiting for all of them even when one fails."""
+        results = await asyncio.gather(*(protocol.close() for protocol in self._owned_comm_protocols), return_exceptions=True)
+        failures = [result for result in results if isinstance(result, BaseException)]
+        if failures:
+            raise UtcpProtocolCloseError(failures, len(results))
+
+    async def close(self) -> None:
+        """REQUIRED
+        Close the protocol instances this client created and release their resources.
+
+        Only what this client created. A shared instance is in use by every
+        other client in the process — closing it here would clear their state
+        too (a credential cache, a decorator's registry), which is the
+        cross-client damage per-client instances exist to prevent. Shared
+        instances live as long as the process that registered them.
+
+        Raises:
+            UtcpProtocolCloseError: If one or more owned instances failed to close.
+                Every instance is still asked to close first.
+        """
+        await self._close_owned_protocols()
+        logger.info("UTCP client closed, with the protocols it owned.")
 
     @classmethod
     async def create(
@@ -79,16 +135,32 @@ async def create(
         # Create the client
         client = cls(config, DefaultVariableSubstitutor(), root_dir)
 
-        # Substitute variables in the config
-        if client.config.variables:
-            config_without_vars = client_config_serializer.copy(client.config)
-            config_without_vars.variables = None
-            client.config.variables = client.variable_substitutor.substitute(client.config.variables, config_without_vars)
+        # Everything from here on can fail, and the caller never receives a
+        # client it could close — so whatever this client CREATED is closed
+        # here before the failure is re-raised. Only what it created: the shared
+        # instances are in use by every other client and are not this one's.
+        try:
+            client._adopt_factory_protocols()
+
+            # Substitute variables in the config
+            if client.config.variables:
+                config_without_vars = client_config_serializer.copy(client.config)
+                config_without_vars.variables = None
+                client.config.variables = client.variable_substitutor.substitute(client.config.variables, config_without_vars)
+
+            # Load the manuals if any
+            if config.manual_call_templates:
+                await client.register_manuals(config.manual_call_templates)
+        except BaseException:
+            try:
+                await client._close_owned_protocols()
+            except Exception:
+                # The initialization failure stays the error the caller sees;
+                # the cleanup failure is reported rather than swallowed, because
+                # the caller has no client through which to retry it.
+                logger.error("UtcpClient.create failed, and closing the protocols it had created failed too", exc_info=True)
+            raise
 
-        # Load the manuals if any
-        if config.manual_call_templates:
-            await client.register_manuals(config.manual_call_templates)
-        
         return client
 
     async def register_manual(self, manual_call_template: CallTemplate) -> RegisterManualResult:
@@ -121,10 +193,8 @@ async def register_manual(self, manual_call_template: CallTemplate) -> RegisterM
         if await self.config.tool_repository.get_manual(manual_call_template.name) is not None:
             raise ValueError(f"Manual {manual_call_template.name} already registered, please use a different name or deregister the existing manual")
         manual_call_template = self._substitute_call_template_variables(manual_call_template, manual_call_template.name)
-        if manual_call_template.call_template_type not in CommunicationProtocol.communication_protocols:
-            raise ValueError(f"No registered communication protocol of type {manual_call_template.call_template_type} found, available types: {CommunicationProtocol.communication_protocols.keys()}")
-        
-        result = await CommunicationProtocol.communication_protocols[manual_call_template.call_template_type].register_manual(self, manual_call_template)
+
+        result = await self._protocol_for(manual_call_template.call_template_type).register_manual(self, manual_call_template)
 
         if result.success:
             # Determine allowed protocols: use explicit list or default to manual's own protocol
@@ -203,7 +273,7 @@ async def deregister_manual(self, manual_name: str) -> bool:
         manual_call_template = await self.config.tool_repository.get_manual_call_template(manual_name)
         if manual_call_template is None:
             return False
-        await CommunicationProtocol.communication_protocols[manual_call_template.call_template_type].deregister_manual(self, manual_call_template)
+        await self._protocol_for(manual_call_template.call_template_type).deregister_manual(self, manual_call_template)
         return await self.config.tool_repository.remove_manual(manual_name)
 
     async def call_tool(self, tool_name: str, tool_args: Dict[str, Any]) -> Any:
@@ -250,7 +320,7 @@ async def call_tool(self, tool_name: str, tool_args: Dict[str, Any]) -> Any:
                     f"Allowed protocols: {allowed_protocols}"
                 )
         
-        result = await CommunicationProtocol.communication_protocols[tool_call_template.call_template_type].call_tool(self, tool_name, tool_args, tool_call_template)
+        result = await self._protocol_for(tool_call_template.call_template_type).call_tool(self, tool_name, tool_args, tool_call_template)
         
         for post_processor in self.config.post_processing:
             result = post_processor.post_process(self, tool, tool_call_template, result)
@@ -300,7 +370,7 @@ async def call_tool_streaming(self, tool_name: str, tool_args: Dict[str, Any]) -
                     f"Allowed protocols: {allowed_protocols}"
                 )
         
-        async for item in CommunicationProtocol.communication_protocols[tool_call_template.call_template_type].call_tool_streaming(self, tool_name, tool_args, tool_call_template):
+        async for item in self._protocol_for(tool_call_template.call_template_type).call_tool_streaming(self, tool_name, tool_args, tool_call_template):
             for post_processor in self.config.post_processing:
                 item = post_processor.post_process(self, tool, tool_call_template, item)
             yield item
@@ -342,9 +412,7 @@ async def get_required_variables_for_manual_and_tools(self, manual_call_template
             except UtcpVariableNotFound as e:
                 return variables_for_CallTemplate
             return variables_for_CallTemplate
-        if manual_call_template.call_template_type not in CommunicationProtocol.communication_protocols:
-            raise ValueError(f"CallTemplate type not supported: {manual_call_template.call_template_type}")
-        register_manual_result: RegisterManualResult = await CommunicationProtocol.communication_protocols[manual_call_template.call_template_type].register_manual(self, manual_call_template)
+        register_manual_result: RegisterManualResult = await self._protocol_for(manual_call_template.call_template_type).register_manual(self, manual_call_template)
         for tool in register_manual_result.manual.tools:
             variables_for_CallTemplate.extend(self.variable_substitutor.find_required_variables(CallTemplateSerializer().to_dict(tool.tool_call_template), manual_call_template.name))
         return variables_for_CallTemplate
diff --git a/core/src/utcp/interfaces/communication_protocol.py b/core/src/utcp/interfaces/communication_protocol.py
index b12e6ea..b1fc672 100644
--- a/core/src/utcp/interfaces/communication_protocol.py
+++ b/core/src/utcp/interfaces/communication_protocol.py
@@ -6,7 +6,7 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import Dict, Any, AsyncGenerator, TYPE_CHECKING
+from typing import Callable, Dict, Any, AsyncGenerator, TYPE_CHECKING
 from utcp.data.register_manual_response import RegisterManualResult
 from utcp.data.call_template import CallTemplate
 if TYPE_CHECKING:
@@ -24,8 +24,28 @@ class CommunicationProtocol(ABC):
     - Discovering available tools from providers
     - Managing provider lifecycle (registration/deregistration)
     - Executing tool calls through the appropriate protocol
+
+    A protocol is registered in one of two registries, and the choice decides
+    who its state belongs to:
+
+    - `communication_protocols` holds an INSTANCE that is shared by every
+      `UtcpClient` in the process, and so is any state it keeps. That is the
+      right home for state that should be process-wide (a credential cache, a
+      registry a decorator writes into). The instance lives as long as the
+      process that registered it; no client closes it.
+    - `communication_protocol_factories` holds a FACTORY. `UtcpClient.create`
+      calls it once per client, so each client gets its own instance, its own
+      connections, and its own teardown on `close()`. That is what makes a
+      client per tenant, per user, or per pooled connection actually isolate
+      them, rather than every client reaching into one shared instance. A
+      protocol that holds connections or sessions belongs here.
+
+    A type registered as a factory wins over the same type registered as an
+    instance, so a plugin migrates by moving its registration from one
+    registry to the other and callers change nothing.
     """
     communication_protocols: dict[str, 'CommunicationProtocol'] = {}
+    communication_protocol_factories: dict[str, Callable[[], 'CommunicationProtocol']] = {}
 
     @abstractmethod
     async def register_manual(self, caller: 'UtcpClient', manual_call_template: CallTemplate) -> RegisterManualResult:
@@ -118,3 +138,18 @@ async def call_tool_streaming(self, caller: 'UtcpClient', tool_name: str, tool_a
             TimeoutError: If the tool call exceeds the configured timeout.
         """
         pass
+
+    async def close(self) -> None:
+        """REQUIRED
+        Release every connection, session, process or other resource this
+        protocol instance holds.
+
+        `UtcpClient.close()` calls this on each instance the client created
+        from `communication_protocol_factories`, and `UtcpClient.create()`
+        calls it on those instances when initialization fails after they were
+        created. A shared instance from `communication_protocols` is never
+        closed on a client's behalf.
+
+        The default releases nothing, for protocols that hold nothing.
+        """
+        pass
diff --git a/core/src/utcp/plugins/discovery.py b/core/src/utcp/plugins/discovery.py
index 830214e..0c1efe7 100644
--- a/core/src/utcp/plugins/discovery.py
+++ b/core/src/utcp/plugins/discovery.py
@@ -6,6 +6,7 @@
 from utcp.interfaces.tool_post_processor import ToolPostProcessor, ToolPostProcessorConfigSerializer
 from utcp.interfaces.communication_protocol import CommunicationProtocol
 from utcp.data.call_template import CallTemplate, CallTemplateSerializer
+from typing import Callable
 import logging
 
 logger = logging.getLogger(__name__)
@@ -82,6 +83,31 @@ def register_communication_protocol(communication_protocol_type: str, communicat
     logger.info("Registered communication protocol type: " + communication_protocol_type)
     return True
 
+def register_communication_protocol_factory(communication_protocol_type: str, factory: Callable[[], CommunicationProtocol], override: bool = False) -> bool:
+    """REQUIRED
+    Register a communication protocol as a factory, so that every `UtcpClient`
+    gets its own instance of it.
+
+    Use this instead of `register_communication_protocol` for a protocol whose
+    state belongs to one client rather than to the process: connections,
+    sessions, child processes. `UtcpClient.create` calls the factory once per
+    client, and that client's `close()` tears the instance down. A type
+    registered as a factory wins over the same type registered as an instance.
+
+    Args:
+        communication_protocol_type: The communication protocol type identifier.
+        factory: A callable returning a new communication protocol instance.
+        override: Whether to override an existing factory for this type.
+
+    Returns:
+        True if the factory was registered, False otherwise.
+    """
+    if not override and communication_protocol_type in CommunicationProtocol.communication_protocol_factories:
+        return False
+    CommunicationProtocol.communication_protocol_factories[communication_protocol_type] = factory
+    logger.info("Registered communication protocol factory for type: " + communication_protocol_type)
+    return True
+
 def register_tool_repository(tool_repository_type: str, tool_repository: Serializer[ConcurrentToolRepository], override: bool = False) -> bool:
     """REQUIRED
     Register a tool repository implementation.
diff --git a/core/src/utcp/utcp_client.py b/core/src/utcp/utcp_client.py
index 6dd2172..f902d0e 100644
--- a/core/src/utcp/utcp_client.py
+++ b/core/src/utcp/utcp_client.py
@@ -158,3 +158,18 @@ async def get_required_variables_for_registered_tool(self, tool_name: str) -> Li
             A list of required variables for the tool.
         """
         pass
+
+    @abstractmethod
+    async def close(self) -> None:
+        """REQUIRED
+        Close the protocol instances this client created and release their resources.
+
+        Closes every instance the client obtained from
+        `CommunicationProtocol.communication_protocol_factories`. Instances from
+        `CommunicationProtocol.communication_protocols` are shared by every client
+        in the process and are left running.
+
+        Every owned instance is closed even if one of them fails to close; the
+        failures are then raised together as a `UtcpProtocolCloseError`.
+        """
+        pass
diff --git a/core/tests/client/test_client_protocol_ownership.py b/core/tests/client/test_client_protocol_ownership.py
new file mode 100644
index 0000000..8c5a083
--- /dev/null
+++ b/core/tests/client/test_client_protocol_ownership.py
@@ -0,0 +1,242 @@
+"""A client owns the protocol instances it creates from the factory registry.
+
+Everything here goes through the public surface: the two registries on
+``CommunicationProtocol``, ``UtcpClient.create``, ``register_manual``,
+``call_tool`` and ``close``.
+"""
+
+import asyncio
+import logging
+from typing import Any, AsyncGenerator, Dict, List
+
+import pytest
+
+from utcp.data.call_template import CallTemplate
+from utcp.data.register_manual_response import RegisterManualResult
+from utcp.data.tool import JsonSchema, Tool
+from utcp.data.utcp_manual import UtcpManual
+from utcp.exceptions import UtcpProtocolCloseError, UtcpVariableNotFound
+from utcp.interfaces.communication_protocol import CommunicationProtocol
+from utcp.utcp_client import UtcpClient
+from utcp_http.http_call_template import HttpCallTemplate
+
+
+class RecordingProtocol(CommunicationProtocol):
+    """Records which client-facing calls reached this instance, and how often it was closed."""
+
+    def __init__(self):
+        self.registered: List[str] = []
+        self.called: List[str] = []
+        self.closed = 0
+
+    async def register_manual(self, caller: UtcpClient, manual_call_template: CallTemplate) -> RegisterManualResult:
+        self.registered.append(manual_call_template.name)
+        tool = Tool(
+            name="ping",
+            description="answers with the identity of the instance that served it",
+            inputs=JsonSchema(type="object", properties={}),
+            outputs=JsonSchema(type="object", properties={}),
+            tags=[],
+            tool_call_template=manual_call_template,
+        )
+        return RegisterManualResult(
+            manual_call_template=manual_call_template,
+            manual=UtcpManual(manual_version="1.0.0", tools=[tool]),
+            success=True,
+            errors=[],
+        )
+
+    async def deregister_manual(self, caller: UtcpClient, manual_call_template: CallTemplate) -> None:
+        pass
+
+    async def call_tool(self, caller: UtcpClient, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> Any:
+        self.called.append(tool_name)
+        return {"served_by": id(self)}
+
+    async def call_tool_streaming(self, caller: UtcpClient, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> AsyncGenerator[Any, None]:
+        yield await self.call_tool(caller, tool_name, tool_args, tool_call_template)
+
+    async def close(self) -> None:
+        self.closed += 1
+
+
+def http_manual(name: str) -> HttpCallTemplate:
+    return HttpCallTemplate(name=name, url="https://example.test/utcp", http_method="POST", call_template_type="http")
+
+
+@pytest.fixture
+def isolated_registries(monkeypatch):
+    """Both registries start empty and are restored after the test."""
+    monkeypatch.setattr(CommunicationProtocol, "communication_protocols", {})
+    monkeypatch.setattr(CommunicationProtocol, "communication_protocol_factories", {})
+
+
+@pytest.fixture
+def recording_factory(isolated_registries):
+    """Registers a factory for ``http`` and returns the instances it made, in order."""
+    made: List[RecordingProtocol] = []
+
+    def factory() -> RecordingProtocol:
+        protocol = RecordingProtocol()
+        made.append(protocol)
+        return protocol
+
+    CommunicationProtocol.communication_protocol_factories["http"] = factory
+    return made
+
+
+class TestPerClientInstances:
+
+    @pytest.mark.asyncio
+    async def test_each_client_gets_its_own_instance_and_routes_to_it(self, recording_factory):
+        client_a = await UtcpClient.create()
+        client_b = await UtcpClient.create()
+        assert len(recording_factory) == 2
+        own_a, own_b = recording_factory
+
+        await client_a.register_manual(http_manual("a_manual"))
+        await client_b.register_manual(http_manual("b_manual"))
+        result_a = await client_a.call_tool("a_manual.ping", {})
+        result_b = await client_b.call_tool("b_manual.ping", {})
+
+        assert own_a.registered == ["a_manual"] and own_a.called == ["a_manual.ping"]
+        assert own_b.registered == ["b_manual"] and own_b.called == ["b_manual.ping"]
+        assert result_a["served_by"] == id(own_a)
+        assert result_b["served_by"] == id(own_b)
+
+    @pytest.mark.asyncio
+    async def test_a_factory_wins_over_a_shared_instance_of_the_same_type(self, recording_factory):
+        shared = RecordingProtocol()
+        CommunicationProtocol.communication_protocols["http"] = shared
+
+        client = await UtcpClient.create()
+        await client.register_manual(http_manual("m"))
+
+        assert recording_factory[0].registered == ["m"]
+        assert shared.registered == []
+
+    @pytest.mark.asyncio
+    async def test_a_shared_instance_registered_after_the_client_exists_is_still_used(self, isolated_registries):
+        # Late registration in the shared registry keeps working: shared
+        # instances are looked up live, not snapshotted at creation.
+        client = await UtcpClient.create()
+        shared = RecordingProtocol()
+        CommunicationProtocol.communication_protocols["http"] = shared
+
+        await client.register_manual(http_manual("m"))
+
+        assert shared.registered == ["m"]
+
+    @pytest.mark.asyncio
+    async def test_a_type_with_no_protocol_in_either_registry_names_what_is_available(self, isolated_registries):
+        # "http" has a serializer (the plugin is imported) but, with the
+        # registries isolated, no protocol in either of them.
+        CommunicationProtocol.communication_protocol_factories["cli"] = RecordingProtocol
+        CommunicationProtocol.communication_protocols["text"] = RecordingProtocol()
+        client = await UtcpClient.create()
+
+        with pytest.raises(ValueError, match=r"type http found, available types: \['cli', 'text'\]"):
+            await client.register_manual(http_manual("m"))
+
+
+class TestCloseIsScopedToWhatTheClientOwns:
+
+    @pytest.mark.asyncio
+    async def test_close_drains_own_instance_and_leaves_other_clients_and_shared_ones_running(self, recording_factory):
+        shared = RecordingProtocol()
+        CommunicationProtocol.communication_protocols["cli"] = shared
+        client_a = await UtcpClient.create()
+        client_b = await UtcpClient.create()
+        own_a, own_b = recording_factory
+
+        await client_a.close()
+
+        assert own_a.closed == 1
+        assert own_b.closed == 0
+        # Every other client in the process is still using this one.
+        assert shared.closed == 0
+
+        await client_b.close()
+        assert own_b.closed == 1
+        assert shared.closed == 0
+
+    @pytest.mark.asyncio
+    async def test_close_waits_for_every_owned_protocol_even_when_one_fails(self, isolated_registries):
+        # The failing close raises IMMEDIATELY; the healthy one takes a moment.
+        # A first-failure-wins close would return before the healthy one ended.
+        class FailingProtocol(RecordingProtocol):
+            async def close(self) -> None:
+                raise RuntimeError("transport refused to close")
+
+        class SlowProtocol(RecordingProtocol):
+            async def close(self) -> None:
+                await asyncio.sleep(0.02)
+                self.closed += 1
+
+        slow = SlowProtocol()
+        CommunicationProtocol.communication_protocol_factories["http"] = FailingProtocol
+        CommunicationProtocol.communication_protocol_factories["cli"] = lambda: slow
+
+        client = await UtcpClient.create()
+        with pytest.raises(UtcpProtocolCloseError) as raised:
+            await client.close()
+
+        # The failure is still surfaced, with its cause inside...
+        assert len(raised.value.failures) == 1
+        assert isinstance(raised.value.failures[0], RuntimeError)
+        # ...but only after every owned protocol has finished closing.
+        assert slow.closed == 1
+
+
+class TestFailedCreateClosesWhatItCreated:
+
+    @pytest.mark.asyncio
+    async def test_an_instance_made_for_a_create_that_fails_is_closed_not_orphaned(self, recording_factory):
+        # The factory fires, then variable substitution fails on a reference
+        # nothing can resolve — create() raises and the caller never gets a
+        # client to close.
+        with pytest.raises(UtcpVariableNotFound):
+            await UtcpClient.create(config={"variables": {"DERIVED": "${NOWHERE_TO_BE_FOUND}"}})
+
+        assert len(recording_factory) == 1
+        assert recording_factory[0].closed == 1
+
+    @pytest.mark.asyncio
+    async def test_a_factory_that_raises_leaves_the_instances_before_it_closed(self, recording_factory):
+        def broken_factory() -> CommunicationProtocol:
+            raise RuntimeError("no transport available")
+
+        CommunicationProtocol.communication_protocol_factories["cli"] = broken_factory
+
+        with pytest.raises(RuntimeError, match="no transport available"):
+            await UtcpClient.create()
+
+        assert len(recording_factory) == 1
+        assert recording_factory[0].closed == 1
+
+    @pytest.mark.asyncio
+    async def test_a_shared_instance_survives_a_failed_create(self, recording_factory):
+        shared = RecordingProtocol()
+        CommunicationProtocol.communication_protocols["cli"] = shared
+
+        with pytest.raises(UtcpVariableNotFound):
+            await UtcpClient.create(config={"variables": {"DERIVED": "${NOWHERE_TO_BE_FOUND}"}})
+
+        assert recording_factory[0].closed == 1
+        assert shared.closed == 0
+
+    @pytest.mark.asyncio
+    async def test_a_failing_cleanup_is_reported_and_the_original_error_surfaces(self, isolated_registries, caplog):
+        class FailingProtocol(RecordingProtocol):
+            async def close(self) -> None:
+                raise RuntimeError("transport refused to close")
+
+        CommunicationProtocol.communication_protocol_factories["http"] = FailingProtocol
+
+        with caplog.at_level(logging.ERROR, logger="utcp.implementations.utcp_client_implementation"):
+            with pytest.raises(UtcpVariableNotFound):
+                await UtcpClient.create(config={"variables": {"DERIVED": "${NOWHERE_TO_BE_FOUND}"}})
+
+        reported = [record for record in caplog.records if "closing the protocols it had created failed too" in record.getMessage()]
+        assert len(reported) == 1
+        assert "transport refused to close" in reported[0].exc_text

From 40572d863791b830d512415a18e88dff093662cf Mon Sep 17 00:00:00 2001
From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com>
Date: Tue, 15 Sep 2026 22:20:27 +0200
Subject: [PATCH 2/4] mcp: register as a factory, so every client owns its own
 sessions

The MCP protocol holds live sessions and, for stdio, child processes.
Registered as one shared instance, every client in the process dialled
into one session cache and one client's close() drained everyone's.
It now registers through register_communication_protocol_factory, so
each UtcpClient gets its own instance, its own connections and its own
teardown.

The factory registry is a new core feature, so core goes to 1.2.0 and
utcp-mcp to 1.2.0 requiring utcp>=1.2.0. MCP suite 48/48.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 core/pyproject.toml                                   |  2 +-
 plugins/communication_protocols/mcp/pyproject.toml    |  4 ++--
 .../mcp/src/utcp_mcp/__init__.py                      | 11 +++++++++--
 .../mcp/src/utcp_mcp/mcp_communication_protocol.py    | 10 +++++-----
 4 files changed, 17 insertions(+), 10 deletions(-)

diff --git a/core/pyproject.toml b/core/pyproject.toml
index 04c8344..ce07a77 100644
--- a/core/pyproject.toml
+++ b/core/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp"
-version = "1.1.4"
+version = "1.2.0"
 authors = [
   { name = "UTCP Contributors" },
 ]
diff --git a/plugins/communication_protocols/mcp/pyproject.toml b/plugins/communication_protocols/mcp/pyproject.toml
index 9b0641b..8f6a78b 100644
--- a/plugins/communication_protocols/mcp/pyproject.toml
+++ b/plugins/communication_protocols/mcp/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-mcp"
-version = "1.1.4"
+version = "1.2.0"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -14,7 +14,7 @@ requires-python = ">=3.11"
 dependencies = [
     "pydantic>=2.0",
     "mcp>=1.12,<2",
-    "utcp>=1.1.4",
+    "utcp>=1.2.0",
     "mcp-use>=1.3",
     "langchain>=0.3.27,<0.4.0",
 ]
diff --git a/plugins/communication_protocols/mcp/src/utcp_mcp/__init__.py b/plugins/communication_protocols/mcp/src/utcp_mcp/__init__.py
index 85abb78..f81f9f7 100644
--- a/plugins/communication_protocols/mcp/src/utcp_mcp/__init__.py
+++ b/plugins/communication_protocols/mcp/src/utcp_mcp/__init__.py
@@ -1,9 +1,16 @@
 from utcp_mcp.mcp_communication_protocol import McpCommunicationProtocol
 from utcp_mcp.mcp_call_template import McpCallTemplate, McpCallTemplateSerializer
-from utcp.plugins.discovery import register_communication_protocol, register_call_template
+from utcp.plugins.discovery import register_communication_protocol_factory, register_call_template
 
 def register():
-    register_communication_protocol("mcp", McpCommunicationProtocol())
+    # A FACTORY, not an instance: this protocol holds live MCP sessions (and,
+    # for stdio, child processes). Shared, every client in the process would
+    # dial into one session cache — so a caller that creates a client per
+    # tenant, per user, or per pooled connection would not actually be
+    # isolating them, and one client's close() would drain everyone's
+    # sessions. One instance per client gives each its own connections and its
+    # own teardown.
+    register_communication_protocol_factory("mcp", McpCommunicationProtocol)
     register_call_template("mcp", McpCallTemplateSerializer())
 
 __all__ = [
diff --git a/plugins/communication_protocols/mcp/src/utcp_mcp/mcp_communication_protocol.py b/plugins/communication_protocols/mcp/src/utcp_mcp/mcp_communication_protocol.py
index 3406fd6..ec2b032 100644
--- a/plugins/communication_protocols/mcp/src/utcp_mcp/mcp_communication_protocol.py
+++ b/plugins/communication_protocols/mcp/src/utcp_mcp/mcp_communication_protocol.py
@@ -197,10 +197,10 @@ def __init__(self):
         # concurrent first calls for the same server dial once instead of each
         # spawning a session and leaking all but the last.
         self._session_creations: "Dict[Tuple[int, str], asyncio.Task]" = {}
-        # One MCPClient per distinct server configuration. This protocol object is
-        # registered once per process and shared by every manual, so a single
-        # client would make manuals with different configurations evict each
-        # other's sessions, including sessions still in use by a concurrent call.
+        # One MCPClient per distinct server configuration. This protocol object
+        # serves every manual of the UtcpClient that owns it, so a single client
+        # would make manuals with different configurations evict each other's
+        # sessions, including sessions still in use by a concurrent call.
         self._mcp_clients: Dict[str, MCPClient] = {}
         # Which configuration each owner (calling UtcpClient plus manual name)
         # currently uses, so a client nothing references any more can be closed
@@ -844,7 +844,7 @@ async def close(self) -> None:
         """Close all active sessions and clean up resources."""
         self._log_info("Closing MCP communication protocol and cleaning up all sessions")
         await self._cleanup_all_sessions()
-        # Drain the OAuth state so this shared instance holds no credentials past
+        # Drain the OAuth state so this instance holds no credentials past
         # close(). Cancel in-flight fetches (asyncio can) and drop their entries;
         # a fetch that still lands finds it is no longer the current entry and,
         # by the caching rule in _on_oauth_fetch_done, does not repopulate the

From cc76d206c1ee93d483baf7b05e953cae9c8d0f5c Mon Sep 17 00:00:00 2001
From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com>
Date: Wed, 16 Sep 2026 10:58:29 +0200
Subject: [PATCH 3/4] core: a batch registration settles every sibling before
 raising; a late factory is adopted on first use
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two findings from cubic on #108.

register_manuals gathered its registrations first-failure-wins: when one
raised UtcpVariableNotFound the caller got the failure while the sibling
registrations were still running underneath it — and create(), which
closes the client's protocols right after, would close them under a
registration still in flight. Every registration now settles before the
first failure is raised, so a caller holding the error holds a quiet
client.

The resolver consulted the factory registry only at creation, so a
factory registered after a client existed was invisible to it while a
shared instance registered late was not. Both registries are now
consulted live: a late factory is adopted on first use, owned and closed
like the rest. A type resolves the same way whenever it was registered.

Tests: register_manuals raises only once every sibling has finished; a
failed create closes its protocols only after every registration has
finished; a late-registered factory is adopted on first use and owned.
Each mutation-checked. Ownership 13/13, core 54/54, MCP 48/48, http 238.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 README.md                                     |  2 +-
 core/README.md                                |  2 +-
 .../utcp_client_implementation.py             | 52 +++++++++----
 .../utcp/interfaces/communication_protocol.py |  7 +-
 core/src/utcp/plugins/discovery.py            |  7 +-
 .../client/test_client_protocol_ownership.py  | 75 +++++++++++++++++++
 6 files changed, 122 insertions(+), 23 deletions(-)

diff --git a/README.md b/README.md
index a9b5196..9cce106 100644
--- a/README.md
+++ b/README.md
@@ -85,7 +85,7 @@ A protocol plugin registers itself in one of two registries, and the choice deci
 
 An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
 
-A protocol that holds **connections** registers a **factory** instead. `UtcpClient.create` calls it once per client, so each client gets its own instance, its own connections, and its own teardown on `close()`:
+A protocol that holds **connections** registers a **factory** instead. Each `UtcpClient` calls it once — at creation, or on first use if the factory was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
 
 ```python
 from utcp.plugins.discovery import register_communication_protocol_factory
diff --git a/core/README.md b/core/README.md
index a9b5196..9cce106 100644
--- a/core/README.md
+++ b/core/README.md
@@ -85,7 +85,7 @@ A protocol plugin registers itself in one of two registries, and the choice deci
 
 An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
 
-A protocol that holds **connections** registers a **factory** instead. `UtcpClient.create` calls it once per client, so each client gets its own instance, its own connections, and its own teardown on `close()`:
+A protocol that holds **connections** registers a **factory** instead. Each `UtcpClient` calls it once — at creation, or on first use if the factory was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
 
 ```python
 from utcp.plugins.discovery import register_communication_protocol_factory
diff --git a/core/src/utcp/implementations/utcp_client_implementation.py b/core/src/utcp/implementations/utcp_client_implementation.py
index 5f95c4d..c227b63 100644
--- a/core/src/utcp/implementations/utcp_client_implementation.py
+++ b/core/src/utcp/implementations/utcp_client_implementation.py
@@ -4,7 +4,7 @@
 import os
 import json
 import asyncio
-from typing import Dict, Any, List, Union, Optional, AsyncGenerator, TYPE_CHECKING
+from typing import Callable, Dict, Any, List, Union, Optional, AsyncGenerator, TYPE_CHECKING
 
 from utcp.data.call_template import CallTemplate
 from utcp.data.call_template import CallTemplateSerializer
@@ -48,27 +48,42 @@ def __init__(
         self._owned_comm_protocols: List[CommunicationProtocol] = []
         self._own_comm_protocols_by_type: Dict[str, CommunicationProtocol] = {}
 
-    def _adopt_factory_protocols(self) -> None:
-        """Instantiate this client's own protocols from the factory registry.
+    def _adopt(self, protocol_type: str, factory: Callable[[], CommunicationProtocol]) -> CommunicationProtocol:
+        """Create this client's own instance of a protocol and record it as owned.
 
-        Each instance is recorded as owned the moment it exists, so a factory
-        that raises part-way leaves the ones before it closable.
+        Recorded the moment it exists, so a factory that raises part-way
+        through a batch leaves the ones before it closable.
         """
+        protocol = factory()
+        self._owned_comm_protocols.append(protocol)
+        self._own_comm_protocols_by_type[protocol_type] = protocol
+        return protocol
+
+    def _adopt_factory_protocols(self) -> None:
+        """Instantiate this client's own protocols from every factory registered so far."""
         for protocol_type, factory in CommunicationProtocol.communication_protocol_factories.items():
-            protocol = factory()
-            self._owned_comm_protocols.append(protocol)
-            self._own_comm_protocols_by_type[protocol_type] = protocol
+            self._adopt(protocol_type, factory)
 
     def _protocol_for(self, call_template_type: str) -> CommunicationProtocol:
         """Resolve the protocol this client uses for a call template type.
 
-        This client's own instance wins over a shared one of the same type,
-        which is what lets a plugin migrate from instance to factory without
-        callers changing anything.
+        In order: this client's own instance; a factory registered since the
+        client was created (adopted now, so it is owned and closed like the
+        rest); a shared instance. Both registries are consulted live, so a
+        type resolves the same way whenever it was registered — and a factory
+        wins over a shared instance of the same type, which is what lets a
+        plugin migrate from instance to factory without callers changing
+        anything.
         """
-        protocol = self._own_comm_protocols_by_type.get(call_template_type) or CommunicationProtocol.communication_protocols.get(call_template_type)
+        protocol = self._own_comm_protocols_by_type.get(call_template_type)
         if protocol is None:
-            available = set(self._own_comm_protocols_by_type) | set(CommunicationProtocol.communication_protocols)
+            factory = CommunicationProtocol.communication_protocol_factories.get(call_template_type)
+            if factory is not None:
+                protocol = self._adopt(call_template_type, factory)
+        if protocol is None:
+            protocol = CommunicationProtocol.communication_protocols.get(call_template_type)
+        if protocol is None:
+            available = set(self._own_comm_protocols_by_type) | set(CommunicationProtocol.communication_protocol_factories) | set(CommunicationProtocol.communication_protocols)
             raise ValueError(f"No registered communication protocol of type {call_template_type} found, available types: {sorted(available)}")
         return protocol
 
@@ -256,8 +271,15 @@ async def try_register_manual(manual_call_template=manual_call_template):
             
             tasks.append(try_register_manual())
         
-        # Wait for all tasks to complete and collect results
-        results = await asyncio.gather(*tasks)
+        # Wait for EVERY registration to settle, even when one raises. The
+        # caller receives the failure only once no sibling registration is
+        # still running underneath it — create(), for one, closes the
+        # protocols right after, and a registration still in flight would be
+        # using a closed one.
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        failures = [result for result in results if isinstance(result, BaseException)]
+        if failures:
+            raise failures[0]
         return [p for p in results if p is not None]
 
     async def deregister_manual(self, manual_name: str) -> bool:
diff --git a/core/src/utcp/interfaces/communication_protocol.py b/core/src/utcp/interfaces/communication_protocol.py
index b1fc672..7d1f2dd 100644
--- a/core/src/utcp/interfaces/communication_protocol.py
+++ b/core/src/utcp/interfaces/communication_protocol.py
@@ -33,9 +33,10 @@ class CommunicationProtocol(ABC):
       right home for state that should be process-wide (a credential cache, a
       registry a decorator writes into). The instance lives as long as the
       process that registered it; no client closes it.
-    - `communication_protocol_factories` holds a FACTORY. `UtcpClient.create`
-      calls it once per client, so each client gets its own instance, its own
-      connections, and its own teardown on `close()`. That is what makes a
+    - `communication_protocol_factories` holds a FACTORY. Each `UtcpClient`
+      calls it once — at creation, or on first use for a factory registered
+      later — so each client gets its own instance, its own connections, and
+      its own teardown on `close()`. That is what makes a
       client per tenant, per user, or per pooled connection actually isolate
       them, rather than every client reaching into one shared instance. A
       protocol that holds connections or sessions belongs here.
diff --git a/core/src/utcp/plugins/discovery.py b/core/src/utcp/plugins/discovery.py
index 0c1efe7..7647811 100644
--- a/core/src/utcp/plugins/discovery.py
+++ b/core/src/utcp/plugins/discovery.py
@@ -90,9 +90,10 @@ def register_communication_protocol_factory(communication_protocol_type: str, fa
 
     Use this instead of `register_communication_protocol` for a protocol whose
     state belongs to one client rather than to the process: connections,
-    sessions, child processes. `UtcpClient.create` calls the factory once per
-    client, and that client's `close()` tears the instance down. A type
-    registered as a factory wins over the same type registered as an instance.
+    sessions, child processes. Each `UtcpClient` calls the factory once — at
+    creation, or on first use if the factory is registered later — and that
+    client's `close()` tears the instance down. A type registered as a factory
+    wins over the same type registered as an instance.
 
     Args:
         communication_protocol_type: The communication protocol type identifier.
diff --git a/core/tests/client/test_client_protocol_ownership.py b/core/tests/client/test_client_protocol_ownership.py
index 8c5a083..01f00da 100644
--- a/core/tests/client/test_client_protocol_ownership.py
+++ b/core/tests/client/test_client_protocol_ownership.py
@@ -127,6 +127,27 @@ async def test_a_shared_instance_registered_after_the_client_exists_is_still_use
 
         assert shared.registered == ["m"]
 
+    @pytest.mark.asyncio
+    async def test_a_factory_registered_after_the_client_exists_is_adopted_on_first_use_and_owned(self, isolated_registries):
+        client = await UtcpClient.create()
+        made: List[RecordingProtocol] = []
+
+        def factory() -> RecordingProtocol:
+            protocol = RecordingProtocol()
+            made.append(protocol)
+            return protocol
+
+        CommunicationProtocol.communication_protocol_factories["http"] = factory
+
+        await client.register_manual(http_manual("m"))
+        await client.register_manual(http_manual("n"))
+        # One instance, made on first use, serving every later call...
+        assert len(made) == 1
+        assert made[0].registered == ["m", "n"]
+        # ...and owned: the client's close() reaches it.
+        await client.close()
+        assert made[0].closed == 1
+
     @pytest.mark.asyncio
     async def test_a_type_with_no_protocol_in_either_registry_names_what_is_available(self, isolated_registries):
         # "http" has a serializer (the plugin is imported) but, with the
@@ -240,3 +261,57 @@ async def close(self) -> None:
         reported = [record for record in caplog.records if "closing the protocols it had created failed too" in record.getMessage()]
         assert len(reported) == 1
         assert "transport refused to close" in reported[0].exc_text
+
+
+def unresolvable_http_manual(name: str) -> HttpCallTemplate:
+    """A template whose registration raises UtcpVariableNotFound before it reaches any protocol."""
+    return HttpCallTemplate(name=name, url="https://example.test/${NOWHERE_TO_BE_FOUND}", http_method="POST", call_template_type="http")
+
+
+class SlowRegisteringProtocol(RecordingProtocol):
+    """Registration takes a moment and is recorded in a shared event log, as is close()."""
+
+    def __init__(self, events: List[str]):
+        super().__init__()
+        self.events = events
+
+    async def register_manual(self, caller: UtcpClient, manual_call_template: CallTemplate) -> RegisterManualResult:
+        await asyncio.sleep(0.02)
+        self.events.append(f"registered {manual_call_template.name}")
+        return await super().register_manual(caller, manual_call_template)
+
+    async def close(self) -> None:
+        await super().close()
+        self.events.append("closed")
+
+
+class TestABatchRegistrationSettlesEverySiblingBeforeRaising:
+
+    @pytest.mark.asyncio
+    async def test_register_manuals_raises_only_once_every_sibling_registration_has_finished(self, isolated_registries):
+        # The first manual fails immediately (unresolvable variable); the
+        # second is still registering. A first-failure-wins gather would raise
+        # with the second still running underneath the caller.
+        events: List[str] = []
+        CommunicationProtocol.communication_protocols["http"] = SlowRegisteringProtocol(events)
+        client = await UtcpClient.create()
+
+        with pytest.raises(UtcpVariableNotFound):
+            await client.register_manuals([unresolvable_http_manual("bad_manual"), http_manual("slow_manual")])
+
+        assert events == ["registered slow_manual"]
+
+    @pytest.mark.asyncio
+    async def test_a_failed_create_closes_its_protocols_only_after_every_registration_has_finished(self, isolated_registries):
+        # Same batch, through create(): the owned protocol must not be closed
+        # while a sibling registration is still using it.
+        events: List[str] = []
+        CommunicationProtocol.communication_protocol_factories["http"] = lambda: SlowRegisteringProtocol(events)
+
+        with pytest.raises(UtcpVariableNotFound):
+            await UtcpClient.create(config={"manual_call_templates": [
+                unresolvable_http_manual("bad_manual"),
+                http_manual("slow_manual"),
+            ]})
+
+        assert events == ["registered slow_manual", "closed"]

From f4e87bd1efde0f13f363ee91c9d44e32b1ff0d1d Mon Sep 17 00:00:00 2001
From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com>
Date: Wed, 16 Sep 2026 11:09:49 +0200
Subject: [PATCH 4/4] websocket: register as a factory; state the factory
 criterion precisely
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The guidance said a protocol that holds connections registers a
factory, while the WebSocket plugin — one live WebSocket per manual name
and URL — still registered a shared instance: two clients registering
the same manual used, and on deregistration closed, each other's
connection, and one client's close() dropped everyone's. It now
registers through register_communication_protocol_factory, so each
client owns its connections. utcp-websocket goes to 1.2.0 requiring
utcp>=1.2.0. Suite 38/38.

The criterion itself was too blunt. 'Holds connections' would also
sweep in the HTTP plugin's pooled aiohttp session, which is meant to be
shared. The README, the interface docstring and the registry docstring
now say what actually decides it: state that must not be shared between
clients — sessions or connections keyed per manual, child processes,
anything one client's use or close() would take away from another —
goes in a factory; a credential cache or a pooled session stays an
instance. Every shipped plugin is now on the side the criterion puts it:
mcp and websocket are factories; http, sse, streamable_http, gql, cli,
file, text, tcp and udp keep nothing per client and stay shared.

Raised by cubic on #108.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 README.md                                     |  6 ++---
 core/README.md                                |  6 ++---
 .../utcp/interfaces/communication_protocol.py | 22 ++++++++++---------
 core/src/utcp/plugins/discovery.py            |  6 +++--
 .../websocket/pyproject.toml                  |  4 ++--
 .../websocket/src/utcp_websocket/__init__.py  | 10 ++++++---
 6 files changed, 31 insertions(+), 23 deletions(-)

diff --git a/README.md b/README.md
index 9cce106..7929021 100644
--- a/README.md
+++ b/README.md
@@ -83,9 +83,9 @@ await client.close()
 
 A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to.
 
-An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
+An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a pooled HTTP session, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
 
-A protocol that holds **connections** registers a **factory** instead. Each `UtcpClient` calls it once — at creation, or on first use if the factory was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
+A protocol whose state **must not be shared between clients** registers a **factory** instead: live sessions or connections keyed per manual, child processes — anything one client's use or `close()` would take away from another. Each `UtcpClient` calls the factory once — at creation, or on first use if it was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
 
 ```python
 from utcp.plugins.discovery import register_communication_protocol_factory
@@ -95,7 +95,7 @@ register_communication_protocol_factory("custom_type", CustomCommunicationProtoc
 
 This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing.
 
-`utcp-mcp` registers this way — MCP sessions (and, for stdio, child processes) belong to the client that opened them. `utcp-http` stays a shared instance: its OAuth token cache is meant to be reused across clients.
+`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: what they keep is an OAuth token cache and a pooled HTTP session, both meant to be reused across clients.
 
 `client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised.
 
diff --git a/core/README.md b/core/README.md
index 9cce106..7929021 100644
--- a/core/README.md
+++ b/core/README.md
@@ -83,9 +83,9 @@ await client.close()
 
 A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to.
 
-An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that *should* be process-wide — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
+An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a pooled HTTP session, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.
 
-A protocol that holds **connections** registers a **factory** instead. Each `UtcpClient` calls it once — at creation, or on first use if the factory was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
+A protocol whose state **must not be shared between clients** registers a **factory** instead: live sessions or connections keyed per manual, child processes — anything one client's use or `close()` would take away from another. Each `UtcpClient` calls the factory once — at creation, or on first use if it was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`:
 
 ```python
 from utcp.plugins.discovery import register_communication_protocol_factory
@@ -95,7 +95,7 @@ register_communication_protocol_factory("custom_type", CustomCommunicationProtoc
 
 This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing.
 
-`utcp-mcp` registers this way — MCP sessions (and, for stdio, child processes) belong to the client that opened them. `utcp-http` stays a shared instance: its OAuth token cache is meant to be reused across clients.
+`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: what they keep is an OAuth token cache and a pooled HTTP session, both meant to be reused across clients.
 
 `client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised.
 
diff --git a/core/src/utcp/interfaces/communication_protocol.py b/core/src/utcp/interfaces/communication_protocol.py
index 7d1f2dd..56c055b 100644
--- a/core/src/utcp/interfaces/communication_protocol.py
+++ b/core/src/utcp/interfaces/communication_protocol.py
@@ -30,16 +30,18 @@ class CommunicationProtocol(ABC):
 
     - `communication_protocols` holds an INSTANCE that is shared by every
       `UtcpClient` in the process, and so is any state it keeps. That is the
-      right home for state that should be process-wide (a credential cache, a
-      registry a decorator writes into). The instance lives as long as the
-      process that registered it; no client closes it.
-    - `communication_protocol_factories` holds a FACTORY. Each `UtcpClient`
-      calls it once — at creation, or on first use for a factory registered
-      later — so each client gets its own instance, its own connections, and
-      its own teardown on `close()`. That is what makes a
-      client per tenant, per user, or per pooled connection actually isolate
-      them, rather than every client reaching into one shared instance. A
-      protocol that holds connections or sessions belongs here.
+      right home for state that is meant to be shared (a credential cache, a
+      pooled HTTP session, a registry a decorator writes into). The instance
+      lives as long as the process that registered it; no client closes it.
+    - `communication_protocol_factories` holds a FACTORY, for a protocol whose
+      state must not be shared between clients: live sessions or connections
+      keyed per manual, child processes — anything one client's use or
+      `close()` would take away from another. Each `UtcpClient` calls it once
+      — at creation, or on first use for a factory registered later — so each
+      client gets its own instance, its own connections, and its own teardown
+      on `close()`. That is what makes a client per tenant, per user, or per
+      pooled connection actually isolate them, rather than every client
+      reaching into one shared instance.
 
     A type registered as a factory wins over the same type registered as an
     instance, so a plugin migrates by moving its registration from one
diff --git a/core/src/utcp/plugins/discovery.py b/core/src/utcp/plugins/discovery.py
index 7647811..d7aca50 100644
--- a/core/src/utcp/plugins/discovery.py
+++ b/core/src/utcp/plugins/discovery.py
@@ -89,8 +89,10 @@ def register_communication_protocol_factory(communication_protocol_type: str, fa
     gets its own instance of it.
 
     Use this instead of `register_communication_protocol` for a protocol whose
-    state belongs to one client rather than to the process: connections,
-    sessions, child processes. Each `UtcpClient` calls the factory once — at
+    state must not be shared between clients: live sessions or connections
+    keyed per manual, child processes — anything one client's use or `close()`
+    would take away from another. (A credential cache or a pooled HTTP session
+    is meant to be shared and stays an instance.) Each `UtcpClient` calls the factory once — at
     creation, or on first use if the factory is registered later — and that
     client's `close()` tears the instance down. A type registered as a factory
     wins over the same type registered as an instance.
diff --git a/plugins/communication_protocols/websocket/pyproject.toml b/plugins/communication_protocols/websocket/pyproject.toml
index 090df01..f708117 100644
--- a/plugins/communication_protocols/websocket/pyproject.toml
+++ b/plugins/communication_protocols/websocket/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-websocket"
-version = "1.1.5"
+version = "1.2.0"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -14,7 +14,7 @@ requires-python = ">=3.10"
 dependencies = [
     "pydantic>=2.0",
     "aiohttp>=3.8",
-    "utcp>=1.1.4"
+    "utcp>=1.2.0"
 ]
 classifiers = [
     "Development Status :: 4 - Beta",
diff --git a/plugins/communication_protocols/websocket/src/utcp_websocket/__init__.py b/plugins/communication_protocols/websocket/src/utcp_websocket/__init__.py
index 21c5879..33194da 100644
--- a/plugins/communication_protocols/websocket/src/utcp_websocket/__init__.py
+++ b/plugins/communication_protocols/websocket/src/utcp_websocket/__init__.py
@@ -3,14 +3,18 @@
 This plugin provides WebSocket-based real-time bidirectional communication protocol.
 """
 
-from utcp.plugins.discovery import register_communication_protocol, register_call_template
+from utcp.plugins.discovery import register_communication_protocol_factory, register_call_template
 from utcp_websocket.websocket_communication_protocol import WebSocketCommunicationProtocol
 from utcp_websocket.websocket_call_template import WebSocketCallTemplate, WebSocketCallTemplateSerializer
 
 def register():
     """Register the WebSocket communication protocol and call template serializer."""
-    # Register WebSocket communication protocol
-    register_communication_protocol("websocket", WebSocketCommunicationProtocol())
+    # A FACTORY, not an instance: this protocol keeps one live WebSocket per
+    # manual name and URL. Shared, two clients registering the same manual
+    # would use — and on deregistration close — each other's connection, and
+    # one client's close() would drop everyone's. One instance per client
+    # gives each its own connections and its own teardown.
+    register_communication_protocol_factory("websocket", WebSocketCommunicationProtocol)
 
     # Register call template serializer
     register_call_template("websocket", WebSocketCallTemplateSerializer())