From c151ebab7e193dfd14a6a1a7c8c4e49e9369d95f Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:00:38 +0200 Subject: [PATCH 1/9] Pre-release fixes: reuse the MCP client across calls; error bodies on streaming calls Pre-release review of dev: - mcp: _ensure_mcp_client compared the client's whole config dict with its mcpServers entry, which was always unequal, so every tool call built a new MCPClient and spawned a fresh server process that was never closed. Compare the mcpServers entry, and close the previous client's sessions when the configuration really changes. Test asserts one client and one session across two calls. - sse / streamable_http: the streaming call paths now raise with the server's error body like discovery and the TypeScript SDK already do. - _errors: decode the bounded body once. Co-Authored-By: Claude Fable 5.1 --- .../http/src/utcp_http/_errors.py | 6 ++---- .../src/utcp_http/sse_communication_protocol.py | 2 +- .../streamable_http_communication_protocol.py | 2 +- .../src/utcp_mcp/mcp_communication_protocol.py | 15 ++++++++++++++- .../mcp/tests/test_mcp_transport.py | 11 +++++++++++ 5 files changed, 29 insertions(+), 7 deletions(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/_errors.py b/plugins/communication_protocols/http/src/utcp_http/_errors.py index b3a0e90..71ddb7e 100644 --- a/plugins/communication_protocols/http/src/utcp_http/_errors.py +++ b/plugins/communication_protocols/http/src/utcp_http/_errors.py @@ -69,11 +69,9 @@ async def _read_body_bounded(response: aiohttp.ClientResponse, limit: int) -> st # body was read directly, so aiohttp's own buffered-body machinery must not # be relied on, and an unknown charset name must not lose the detail. try: - encoding = response.charset or "utf-8" - raw.decode(encoding, errors="replace") + return raw.decode(response.charset or "utf-8", errors="replace") except (LookupError, RuntimeError, ValueError): - encoding = "utf-8" - return raw.decode(encoding, errors="replace") + return raw.decode("utf-8", errors="replace") async def raise_for_status_with_body(response: aiohttp.ClientResponse) -> None: diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index fdb2ff5..f10c936 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -292,7 +292,7 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, f"handshakes; update the call template to point at " f"the final URL directly." ) - response.raise_for_status() + await raise_for_status_with_body(response) except Exception as e: if reconnect_attempts == 0: # The initial handshake failing (refused, timed out, non-2xx) is a diff --git a/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py index df4b94f..0b93639 100644 --- a/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py @@ -293,7 +293,7 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, f"followed during streaming handshakes; update the " f"call template to point at the final URL directly." ) - response.raise_for_status() + await raise_for_status_with_body(response) async for chunk in self._process_http_stream(response, tool_call_template.chunk_size, tool_call_template.name): yield chunk 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 f86f4ab..b488b2a 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 @@ -100,7 +100,20 @@ def _log_error(self, message: str): async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate'): """Ensure MCPClient is initialized with the current configuration.""" - if self._mcp_client is None or self._mcp_client.config != manual_call_template.config.mcpServers: + # ``MCPClient.config`` is the whole ``{"mcpServers": ...}`` dict, so it must + # be compared against its ``mcpServers`` entry. Comparing the whole dict to + # the servers mapping was always unequal, which rebuilt the client and + # spawned a fresh server process on every call without ever closing the + # previous ones. + current_servers = self._mcp_client.config.get("mcpServers") if self._mcp_client is not None else None + if self._mcp_client is None or current_servers != manual_call_template.config.mcpServers: + if self._mcp_client is not None: + # The configuration changed: release the previous client's sessions + # so their child processes do not outlive it. + try: + await self._mcp_client.close_all_sessions() + except Exception as e: + self._log_warning(f"Failed to close sessions of the previous MCP client: {e}") # Create a new MCPClient with the server configuration config = {"mcpServers": manual_call_template.config.mcpServers} self._mcp_client = _QuietStdioMCPClient.from_dict(config) diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index d2e2823..034c5bc 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -296,3 +296,14 @@ async def test_process_tool_result_unwraps_only_single_key_result_wrapper(transp # No structuredContent: fall back to text content. text_only = SimpleNamespace(structuredContent=None, content=[SimpleNamespace(text="7")]) assert transport._process_tool_result(text_only, "t") == 7 + + +@pytest.mark.asyncio +async def test_mcp_client_and_session_are_reused_across_calls(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + """Repeated calls with the same configuration reuse one client and one session + instead of spawning a new server process per call.""" + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "one"}, mcp_manual) + client_after_first = transport._mcp_client + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "two"}, mcp_manual) + assert transport._mcp_client is client_after_first + assert list(transport._mcp_client.sessions.keys()) == [SERVER_NAME] From 8d7142191bc182863f343ab664fa9fd9d7813c94 Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:17:09 +0200 Subject: [PATCH 2/9] Pre-release review fixes, second batch - sse: an event block without an `event:` field has the type "message" (spec), so event_type="message" now matches it. An empty `id:` resets the last event ID and no Last-Event-ID header is sent for it; ids containing NUL are ignored. A 200 whose Content-Type is not text/event-stream raises SseProtocolError instead of parsing into zero events. Calls that send a request body are never reconnected: a re-issued POST could re-execute a non-idempotent tool. - _errors: a body nested deeper than the JSON parser's recursion limit raised RecursionError past the HTTP error handling; it is caught and treated as text. Control characters are collapsed so a server cannot forge log records or terminal escapes through an error message. - mcp: close() referenced a _session_locks attribute that was never assigned and always raised AttributeError after cleanup. A child that starts but fails the MCP handshake is now disconnected and removed from active_sessions instead of lingering. Tests for each. Co-Authored-By: Claude Fable 5.1 --- .../http/src/utcp_http/_errors.py | 22 ++++-- .../utcp_http/sse_communication_protocol.py | 28 ++++++-- .../tests/test_http_communication_protocol.py | 21 ++++++ .../tests/test_sse_communication_protocol.py | 72 +++++++++++++++++++ .../utcp_mcp/mcp_communication_protocol.py | 12 +++- .../mcp/tests/test_mcp_transport.py | 8 +++ 6 files changed, 150 insertions(+), 13 deletions(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/_errors.py b/plugins/communication_protocols/http/src/utcp_http/_errors.py index 71ddb7e..a3ac17d 100644 --- a/plugins/communication_protocols/http/src/utcp_http/_errors.py +++ b/plugins/communication_protocols/http/src/utcp_http/_errors.py @@ -7,6 +7,7 @@ status code. Mirrors the TypeScript SDK's ``_normalizeToolError``. """ import json +import re from typing import Optional import aiohttp @@ -23,6 +24,15 @@ _DETAIL_KEYS = ("error", "message", "detail") +# Control characters (newlines, ANSI escape introducers, NUL) are collapsed so +# server-controlled text folded into an exception message or a log line cannot +# forge extra log records or terminal escape sequences. +_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]+") + + +def _clean(text: str) -> str: + return _CONTROL_CHARS.sub(" ", text).strip()[:MAX_DETAIL_CHARS] + def error_detail_from_body(text: str) -> Optional[str]: """Extract a human-readable reason from an error response body. @@ -39,8 +49,10 @@ def error_detail_from_body(text: str) -> Optional[str]: return None try: data = json.loads(body) - except ValueError: - return body[:MAX_DETAIL_CHARS] + except (ValueError, RecursionError): + # RecursionError: a deeply nested body ("[[[[...") within the read cap + # can exceed the parser's recursion limit; it is still just text. + return _clean(body) if isinstance(data, dict): for key in _DETAIL_KEYS: if key not in data or data[key] is None: @@ -48,11 +60,11 @@ def error_detail_from_body(text: str) -> Optional[str]: value = data[key] if isinstance(value, str): if value.strip(): - return value.strip()[:MAX_DETAIL_CHARS] + return _clean(value) continue # Structured error: show it rather than a later generic string. - return body[:MAX_DETAIL_CHARS] - return body[:MAX_DETAIL_CHARS] + return _clean(body) + return _clean(body) async def _read_body_bounded(response: aiohttp.ClientResponse, limit: int) -> str: diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index f10c936..5ef191b 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -252,7 +252,11 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, data = body_content if "application/json" not in content_type else None json_data = body_content if "application/json" in content_type else None - reconnect = bool(tool_call_template.reconnect) + # Never re-send a request body: a reconnect re-issues the request, and for + # a POST that would re-execute a possibly non-idempotent tool. + reconnect = bool(tool_call_template.reconnect) and body_content is None + if tool_call_template.reconnect and body_content is not None: + logger.info(f"Reconnection is disabled for '{tool_call_template.name}' because the call sends a request body.") retry_delay_ms = tool_call_template.retry_timeout last_event_id: Optional[str] = None reconnect_attempts = 0 @@ -260,8 +264,9 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, while True: attempt_headers = dict(request_headers) - if last_event_id is not None: - # Let the server resume from where we left off (SSE spec). + if last_event_id: + # Let the server resume from where we left off (SSE spec). An empty + # last event ID means "none": the header is not sent. attempt_headers["Last-Event-ID"] = last_event_id session = aiohttp.ClientSession() @@ -293,6 +298,16 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, f"the final URL directly." ) await raise_for_status_with_body(response) + # Anything but an event stream would be parsed into silence: a + # JSON error document, say, yields zero events and a "successful" call. + content_type = response.headers.get("Content-Type", "") + if "text/event-stream" not in content_type.lower(): + response.release() + raise SseProtocolError( + f"Expected a text/event-stream response but got {content_type or 'no Content-Type'!r}" + ) + except SseProtocolError: + raise except Exception as e: if reconnect_attempts == 0: # The initial handshake failing (refused, timed out, non-2xx) is a @@ -315,13 +330,16 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, try: async for event in self._iter_sse_events(response): - if event.get("id") is not None: + # Per the SSE spec an id containing NUL is ignored, and an empty id + # resets the last event ID. + if event.get("id") is not None and "\x00" not in event["id"]: last_event_id = event["id"] if event.get("retry") is not None: retry_delay_ms = event["retry"] if "data" not in event: continue - if tool_call_template.event_type and event.get("event") != tool_call_template.event_type: + # An event block without an ``event:`` field has the type "message". + if tool_call_template.event_type and (event.get("event") or "message") != tool_call_template.event_type: continue yield self._parse_event_data(event["data"]) # The server ended the stream cleanly: the tool call is complete. diff --git a/plugins/communication_protocols/http/tests/test_http_communication_protocol.py b/plugins/communication_protocols/http/tests/test_http_communication_protocol.py index 0dd25e5..4504e2c 100644 --- a/plugins/communication_protocols/http/tests/test_http_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_http_communication_protocol.py @@ -174,6 +174,12 @@ async def forbidden_bad_charset_handler(request): return web.Response(status=403, body=b'{"error": "odd charset"}', content_type="application/json", charset="x-unknown-charset") app.router.add_route('*', '/forbidden-no-charset', forbidden_no_charset_handler) + + # A body nested deeper than the JSON parser's recursion limit. + async def forbidden_deep_handler(request): + return web.Response(status=503, body=b"[" * 20000, content_type="application/json") + + app.router.add_route('*', '/forbidden-deep', forbidden_deep_handler) app.router.add_route('*', '/forbidden-bad-charset', forbidden_bad_charset_handler) return app @@ -866,3 +872,18 @@ async def test_error_body_is_surfaced_without_a_usable_charset(http_transport, a with pytest.raises(aiohttp.ClientResponseError) as excinfo: await http_transport.call_tool(None, "t.tool", {}, call_template) assert expected in excinfo.value.message + + +@pytest.mark.asyncio +async def test_deeply_nested_error_body_still_raises_client_response_error(http_transport, aiohttp_client, app): + """A body that overflows the JSON parser's recursion limit must not escape as RecursionError.""" + client = await aiohttp_client(app) + call_template = HttpCallTemplate(name="t", url=f"http://localhost:{client.port}/forbidden-deep", http_method="POST") + with pytest.raises(aiohttp.ClientResponseError) as excinfo: + await http_transport.call_tool(None, "t.tool", {}, call_template) + assert excinfo.value.status == 503 + + +def test_error_detail_collapses_control_characters(): + from utcp_http._errors import error_detail_from_body + assert error_detail_from_body('{"error": "line one\\nline two\\u001b[31m"}') == "line one line two [31m" diff --git a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py index 96d5351..2e00b11 100644 --- a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py @@ -169,6 +169,27 @@ async def huge_retry_events_handler(request): await response.write(b'id: 2\ndata: {"seq": 2}\n\n') return response +async def json_not_sse_handler(request): + """A 200 that is not an event stream at all.""" + return web.json_response({"error": "not a stream"}) + + +async def empty_id_events_handler(request): + """Sets an id, then resets it with an empty id, then drops the connection.""" + state = request.app["empty_id"] + state["connections"] += 1 + state["last_event_ids"].append(request.headers.get("Last-Event-ID")) + response = web.StreamResponse(status=200, headers={'Content-Type': 'text/event-stream'}) + await response.prepare(request) + if state["connections"] == 1: + await response.write(b'id: 1\ndata: {"seq": 1}\n\nid\ndata: {"seq": 2}\n\n') + await asyncio.sleep(0.01) + request.transport.close() + return response + await response.write(b'data: {"seq": 3}\n\n') + return response + + async def flaky_events_handler(request): """Serves the first event then drops the TCP connection on the first connection (or on every connection when ``always_drop`` is set). A reconnecting client is @@ -203,6 +224,10 @@ def app(): app = web.Application() app.router.add_get("/tools", tools_handler) app.router.add_route('*', '/events', events_handler) + app.router.add_post("/flaky_events", flaky_events_handler) + app.router.add_get("/json_not_sse", json_not_sse_handler) + app.router.add_get("/empty_id_events", empty_id_events_handler) + app["empty_id"] = {"connections": 0, "last_event_ids": []} app.router.add_post("/token", token_handler) app.router.add_post("/token_header_auth", token_header_auth_handler) app.router.add_get("/error", error_handler) @@ -596,3 +621,50 @@ async def test_register_manual_surfaces_server_error_body(sse_transport, aiohttp assert result.success is False assert "discovery refused: tenant is not provisioned for streaming" in result.errors[0] assert "403" in result.errors[0] + + +# --- Spec conformance follow-ups --- + +@pytest.mark.asyncio +async def test_event_type_message_matches_events_without_an_event_field(sse_transport, aiohttp_client, app): + """Per the SSE spec an event block without `event:` has the type "message".""" + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/events")), event_type="message") + results = [e async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template)] + assert results == [{"message": "First part"}] + + +@pytest.mark.asyncio +async def test_empty_id_resets_last_event_id(sse_transport, aiohttp_client, app): + """An empty `id` line resets the last event ID, so no Last-Event-ID header is sent on reconnect.""" + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/empty_id_events")), reconnect=True, retry_timeout=10) + results = [e async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template)] + assert results == [{"seq": 1}, {"seq": 2}, {"seq": 3}] + assert app["empty_id"]["last_event_ids"] == [None, None] + + +@pytest.mark.asyncio +async def test_non_event_stream_response_raises(sse_transport, aiohttp_client, app): + """A 200 that is not text/event-stream fails instead of yielding zero events.""" + from utcp_http.sse_communication_protocol import SseProtocolError + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/json_not_sse"))) + with pytest.raises(SseProtocolError): + async for _ in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template): + pass + + +@pytest.mark.asyncio +async def test_post_stream_is_not_reconnected(sse_transport, aiohttp_client, app): + """A dropped POST stream is not re-issued: that could re-execute a non-idempotent tool.""" + client = await aiohttp_client(app) + call_template = SseCallTemplate( + name="test-sse", url=str(client.make_url("/flaky_events")), reconnect=True, retry_timeout=10, body_field="payload" + ) + received = [] + with pytest.raises(aiohttp.ClientError): + async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {"payload": {"n": 1}}, call_template): + received.append(e) + assert received == [{"message": "First part"}] + assert app["flaky"]["connections"] == 1 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 b488b2a..e5ee9e5 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 @@ -67,9 +67,16 @@ async def create_session(self, server_name: str, auto_initialize: bool = True): try: await session.initialize() except Exception: - # Mirror the base class: a session that failed to initialize must - # not stay cached, or the next lookup would hand back a dead one. + # The base class only registers a session after a successful + # initialize; undo the early registration, and disconnect so a + # child that started but failed the MCP handshake does not linger. self.sessions.pop(server_name, None) + if server_name in self.active_sessions: + self.active_sessions.remove(server_name) + try: + await session.disconnect() + except Exception as disconnect_error: + logger.warning(f"Failed to disconnect '{server_name}' after a failed initialize: {disconnect_error}") raise return session @@ -569,7 +576,6 @@ 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() - self._session_locks.clear() self._log_info("MCP communication protocol closed successfully") async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str: diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index 034c5bc..38977d6 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -307,3 +307,11 @@ async def test_mcp_client_and_session_are_reused_across_calls(transport: McpComm await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "two"}, mcp_manual) assert transport._mcp_client is client_after_first assert list(transport._mcp_client.sessions.keys()) == [SERVER_NAME] + + +@pytest.mark.asyncio +async def test_close_after_use_does_not_raise(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + """close() used to raise AttributeError after cleaning up; it must complete.""" + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "one"}, mcp_manual) + await transport.close() + assert transport._mcp_client.sessions == {} From b83d77c7aa65155821675855b8af0193f0ed88dc Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:31:47 +0200 Subject: [PATCH 3/9] Pre-release review fixes, third batch - sse: `retry:` is honoured only when made of ASCII digits (spec); "-1" or "20ms" no longer change the reconnect delay. A stream that ends in the middle of an event no longer dispatches the incomplete event (spec: pending data is discarded at end of file). - streamable_http: Content-Type is matched case-insensitively. - udp: remove the stale comment block describing the bug this release fixed. - tests: the slow-handshake handler no longer outlives the test by seconds. Co-Authored-By: Claude Fable 5.1 --- .../utcp_http/sse_communication_protocol.py | 15 ++++-------- .../streamable_http_communication_protocol.py | 2 +- .../tests/test_sse_communication_protocol.py | 23 +++++++++++++++++-- .../utcp_socket/udp_communication_protocol.py | 4 ---- 4 files changed, 27 insertions(+), 17 deletions(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index 5ef191b..f7b1236 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -393,10 +393,9 @@ def flush(event_string: str): elif field == 'id': current_event['id'] = value elif field == 'retry': - try: + # Spec: only a value made of ASCII digits sets the reconnection time. + if value.isascii() and value.isdigit(): current_event['retry'] = int(value) - except ValueError: - pass if data_lines: current_event['data'] = '\n'.join(data_lines) return current_event or None @@ -431,13 +430,9 @@ def normalise(text: str) -> str: f"SSE event exceeded {self.MAX_EVENT_BUFFER_CHARS} characters without a blank-line delimiter" ) - # Flush a trailing event that was not terminated by a blank line. - buffer += normalise(decoder.decode(b"", final=True)) - if pending_cr: - buffer += "\n" - event = flush(buffer) - if event is not None: - yield event + # Spec: if the stream ends in the middle of an event, before the final + # blank line, the incomplete event is not dispatched. + decoder.decode(b"", final=True) @staticmethod def _parse_event_data(data: str) -> Any: diff --git a/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py index 0b93639..0208605 100644 --- a/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py @@ -310,7 +310,7 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, async def _process_http_stream(self, response: ClientResponse, chunk_size: Optional[int], provider_name: str) -> AsyncIterator[Any]: """Process the HTTP stream and yield chunks based on content type.""" try: - content_type = response.headers.get('Content-Type', '') + content_type = response.headers.get('Content-Type', '').lower() if 'application/x-ndjson' in content_type: async for line in response.content: diff --git a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py index 2e00b11..2045fee 100644 --- a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py @@ -150,8 +150,9 @@ async def flaky_503_events_handler(request): async def slow_handshake_handler(request): - """Accepts the connection but does not send response headers for a long time.""" - await asyncio.sleep(5) + """Accepts the connection but does not send response headers until well past + the (patched) handshake timeout.""" + await asyncio.sleep(1) return web.Response(status=204) @@ -169,6 +170,15 @@ async def huge_retry_events_handler(request): await response.write(b'id: 2\ndata: {"seq": 2}\n\n') return response +async def bad_retry_events_handler(request): + """A retry field that is not made of digits must be ignored; the stream then + ends in the middle of an event, which must not be dispatched.""" + response = web.StreamResponse(status=200, headers={'Content-Type': 'text/event-stream'}) + await response.prepare(request) + await response.write(b'retry: -1\ndata: {"seq": 1}\n\nretry: 20ms\n\ndata: {"seq": 2}') + return response + + async def json_not_sse_handler(request): """A 200 that is not an event stream at all.""" return web.json_response({"error": "not a stream"}) @@ -226,6 +236,7 @@ def app(): app.router.add_route('*', '/events', events_handler) app.router.add_post("/flaky_events", flaky_events_handler) app.router.add_get("/json_not_sse", json_not_sse_handler) + app.router.add_get("/bad_retry_events", bad_retry_events_handler) app.router.add_get("/empty_id_events", empty_id_events_handler) app["empty_id"] = {"connections": 0, "last_event_ids": []} app.router.add_post("/token", token_handler) @@ -668,3 +679,11 @@ async def test_post_stream_is_not_reconnected(sse_transport, aiohttp_client, app received.append(e) assert received == [{"message": "First part"}] assert app["flaky"]["connections"] == 1 + + +@pytest.mark.asyncio +async def test_malformed_retry_is_ignored_and_unterminated_trailing_event_is_dropped(sse_transport, aiohttp_client, app): + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/bad_retry_events"))) + results = [e async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template)] + assert results == [{"seq": 1}] diff --git a/plugins/communication_protocols/socket/src/utcp_socket/udp_communication_protocol.py b/plugins/communication_protocols/socket/src/utcp_socket/udp_communication_protocol.py index fa1f98e..dec60c7 100644 --- a/plugins/communication_protocols/socket/src/utcp_socket/udp_communication_protocol.py +++ b/plugins/communication_protocols/socket/src/utcp_socket/udp_communication_protocol.py @@ -327,10 +327,6 @@ async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], too raise # Copilot AI (5 days ago): - # The call_tool_streaming method wraps a generator function but doesn't use the async def syntax for the method itself. - # While this works, it's inconsistent with the other implementation in tcp_communication_protocol.py (lines 384-387) which properly uses async def with an inner generator. - # For consistency and clarity, this should also use async def directly: - # async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> AsyncGenerator[Any, None]: """REQUIRED Streaming variant: the UDP protocol does not natively stream, so the full result is yielded as a single chunk.""" From 2533abd9dc38fe7c13f0dc6ac501c465858a37ce Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:41:16 +0200 Subject: [PATCH 4/9] sse: compare the response media type exactly A substring check let a Content-Type such as text/event-stream-invalid through; compare the media-type portion exactly, parameters allowed. Co-Authored-By: Claude Fable 5.1 --- .../http/src/utcp_http/sse_communication_protocol.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index f7b1236..9abcaa8 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -301,7 +301,10 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, # Anything but an event stream would be parsed into silence: a # JSON error document, say, yields zero events and a "successful" call. content_type = response.headers.get("Content-Type", "") - if "text/event-stream" not in content_type.lower(): + # Compare the media type exactly (parameters such as charset allowed), + # so "text/event-stream-invalid" does not pass a substring check. + media_type = content_type.split(";", 1)[0].strip().lower() + if media_type != "text/event-stream": response.release() raise SseProtocolError( f"Expected a text/event-stream response but got {content_type or 'no Content-Type'!r}" From b0849f370002d936d3884d02f50dfaa26b724412 Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:52:01 +0200 Subject: [PATCH 5/9] Pre-release review fixes, fourth batch - mcp: one MCPClient per distinct server configuration instead of a single shared client. The protocol object is registered once per process and shared by every manual, so a single client made manuals with different configurations evict each other's sessions, including ones still in use by a concurrent call. Clients are keyed by the canonical configuration, created under a lock so two concurrent first calls cannot each spawn a server, and never evicted by another manual's activity. close() closes every client's sessions and keeps a client whose shutdown failed so a later close() can retry it. - sse: the error-body read on a refused stream is bounded by the handshake timeout, so a server that answers 4xx/5xx and then stalls cannot hang the call. A final blank line ending in a lone CR still completes the last event; only genuinely incomplete events are discarded. Absurdly long retry digit strings are ignored. - _errors: control-character collapsing covers the C1 range too. - tests: streaming calls against a 5xx assert the body is surfaced for both SSE and Streamable HTTP; separate-clients-per-configuration test; the malformed-retry test is named for what it verifies. Co-Authored-By: Claude Fable 5.1 --- .../http/src/utcp_http/_errors.py | 3 +- .../utcp_http/sse_communication_protocol.py | 26 ++++-- .../tests/test_sse_communication_protocol.py | 31 ++++++- ..._streamable_http_communication_protocol.py | 12 +++ .../utcp_mcp/mcp_communication_protocol.py | 86 +++++++++++-------- .../mcp/tests/test_mcp_transport.py | 31 +++++-- 6 files changed, 142 insertions(+), 47 deletions(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/_errors.py b/plugins/communication_protocols/http/src/utcp_http/_errors.py index a3ac17d..9d4fa96 100644 --- a/plugins/communication_protocols/http/src/utcp_http/_errors.py +++ b/plugins/communication_protocols/http/src/utcp_http/_errors.py @@ -27,7 +27,8 @@ # Control characters (newlines, ANSI escape introducers, NUL) are collapsed so # server-controlled text folded into an exception message or a log line cannot # forge extra log records or terminal escape sequences. -_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]+") +# C0 and C1 control ranges: C1 (U+0080..U+009F) carries escape introducers too. +_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f-\x9f]+") def _clean(text: str) -> str: diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index 9abcaa8..03e6d88 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -297,7 +297,9 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, f"handshakes; update the call template to point at " f"the final URL directly." ) - await raise_for_status_with_body(response) + # The error-body read is bounded like the handshake, so a server that + # answers 4xx/5xx and then stalls cannot hang the call either. + await asyncio.wait_for(raise_for_status_with_body(response), timeout=self.HANDSHAKE_TIMEOUT_SECONDS) # Anything but an event stream would be parsed into silence: a # JSON error document, say, yields zero events and a "successful" call. content_type = response.headers.get("Content-Type", "") @@ -397,8 +399,12 @@ def flush(event_string: str): current_event['id'] = value elif field == 'retry': # Spec: only a value made of ASCII digits sets the reconnection time. + # int() refuses absurdly long digit strings; ignore those too. if value.isascii() and value.isdigit(): - current_event['retry'] = int(value) + try: + current_event['retry'] = int(value) + except ValueError: + pass if data_lines: current_event['data'] = '\n'.join(data_lines) return current_event or None @@ -433,9 +439,19 @@ def normalise(text: str) -> str: f"SSE event exceeded {self.MAX_EVENT_BUFFER_CHARS} characters without a blank-line delimiter" ) - # Spec: if the stream ends in the middle of an event, before the final - # blank line, the incomplete event is not dispatched. - decoder.decode(b"", final=True) + # At end of stream, a held-back CR is a real line terminator and may + # complete the closing blank line of the last event. Dispatch whatever is + # fully delimited; per spec, an event still incomplete after that (no + # final blank line) is discarded. + buffer += normalise(decoder.decode(b"", final=True)) + if pending_cr: + buffer += "\n" + pending_cr = False + while "\n\n" in buffer: + event_string, buffer = buffer.split("\n\n", 1) + event = flush(event_string) + if event is not None: + yield event @staticmethod def _parse_event_data(data: str) -> Any: diff --git a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py index 2045fee..c6cfedc 100644 --- a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py @@ -179,6 +179,14 @@ async def bad_retry_events_handler(request): return response +async def cr_eof_events_handler(request): + """A complete event whose closing blank line ends in a lone CR at end of stream.""" + response = web.StreamResponse(status=200, headers={'Content-Type': 'text/event-stream'}) + await response.prepare(request) + await response.write(b'data: {"seq": 1}\n\r') + return response + + async def json_not_sse_handler(request): """A 200 that is not an event stream at all.""" return web.json_response({"error": "not a stream"}) @@ -237,6 +245,7 @@ def app(): app.router.add_post("/flaky_events", flaky_events_handler) app.router.add_get("/json_not_sse", json_not_sse_handler) app.router.add_get("/bad_retry_events", bad_retry_events_handler) + app.router.add_get("/cr_eof_events", cr_eof_events_handler) app.router.add_get("/empty_id_events", empty_id_events_handler) app["empty_id"] = {"connections": 0, "last_event_ids": []} app.router.add_post("/token", token_handler) @@ -682,8 +691,28 @@ async def test_post_stream_is_not_reconnected(sse_transport, aiohttp_client, app @pytest.mark.asyncio -async def test_malformed_retry_is_ignored_and_unterminated_trailing_event_is_dropped(sse_transport, aiohttp_client, app): +async def test_malformed_retry_does_not_abort_and_unterminated_trailing_event_is_dropped(sse_transport, aiohttp_client, app): client = await aiohttp_client(app) call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/bad_retry_events"))) results = [e async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template)] assert results == [{"seq": 1}] + + +@pytest.mark.asyncio +async def test_final_blank_line_ending_in_lone_cr_completes_last_event(sse_transport, aiohttp_client, app): + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/cr_eof_events"))) + results = [e async for e in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template)] + assert results == [{"seq": 1}] + + +@pytest.mark.asyncio +async def test_streaming_call_error_surfaces_server_body(sse_transport, aiohttp_client, app): + """A refused stream carries the server's body, like discovery does.""" + client = await aiohttp_client(app) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/error"))) + with pytest.raises(aiohttp.ClientResponseError) as excinfo: + async for _ in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template): + pass + assert excinfo.value.status == 500 + assert "Internal Server Error" in excinfo.value.message diff --git a/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py b/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py index e026f45..4315952 100644 --- a/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py @@ -356,3 +356,15 @@ async def test_register_manual_surfaces_server_error_body(streamable_http_transp assert result.success is False assert "discovery refused: tenant is not provisioned for streaming" in result.errors[0] assert "403" in result.errors[0] + + +@pytest.mark.asyncio +async def test_streaming_call_error_surfaces_server_body(streamable_http_transport, aiohttp_client, app): + """A refused stream carries the server's body, like discovery does.""" + client = await aiohttp_client(app) + call_template = StreamableHttpCallTemplate(name="test-provider", url=f"{client.make_url('/error')}") + with pytest.raises(aiohttp.ClientResponseError) as excinfo: + async for _ in streamable_http_transport.call_tool_streaming(None, "test-provider.t", {}, call_template): + pass + assert excinfo.value.status == 500 + assert "Internal Server Error" in excinfo.value.message 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 e5ee9e5..c158119 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 @@ -1,3 +1,4 @@ +import asyncio import os import sys from typing import Any, Dict, Optional, AsyncGenerator, TYPE_CHECKING, Tuple, TextIO @@ -91,7 +92,12 @@ class McpCommunicationProtocol(CommunicationProtocol): def __init__(self): self._oauth_tokens: Dict[str, Dict[str, Any]] = {} - self._mcp_client: Optional[MCPClient] = None + # 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. + self._mcp_clients: Dict[str, MCPClient] = {} + self._clients_lock = asyncio.Lock() def _log_info(self, message: str): """Log informational messages.""" @@ -105,40 +111,45 @@ def _log_error(self, message: str): """Log error messages.""" logger.error(f"[McpCommunicationProtocol] {message}") - async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate'): - """Ensure MCPClient is initialized with the current configuration.""" - # ``MCPClient.config`` is the whole ``{"mcpServers": ...}`` dict, so it must - # be compared against its ``mcpServers`` entry. Comparing the whole dict to - # the servers mapping was always unequal, which rebuilt the client and - # spawned a fresh server process on every call without ever closing the - # previous ones. - current_servers = self._mcp_client.config.get("mcpServers") if self._mcp_client is not None else None - if self._mcp_client is None or current_servers != manual_call_template.config.mcpServers: - if self._mcp_client is not None: - # The configuration changed: release the previous client's sessions - # so their child processes do not outlive it. - try: - await self._mcp_client.close_all_sessions() - except Exception as e: - self._log_warning(f"Failed to close sessions of the previous MCP client: {e}") - # Create a new MCPClient with the server configuration - config = {"mcpServers": manual_call_template.config.mcpServers} - self._mcp_client = _QuietStdioMCPClient.from_dict(config) + @staticmethod + def _config_key(manual_call_template: 'McpCallTemplate') -> str: + """Canonical key for a manual's server configuration.""" + return json.dumps(manual_call_template.config.mcpServers, sort_keys=True, default=str) + + async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> MCPClient: + """Return the MCPClient for this manual's configuration, creating it once. + + Clients are keyed by configuration and never evicted by another manual's + activity, so sessions are reused across calls and a call in flight on one + configuration is never torn down by a call on another. Creation is + serialised so two concurrent first calls cannot each spawn a client. + """ + key = self._config_key(manual_call_template) + client = self._mcp_clients.get(key) + if client is not None: + return client + async with self._clients_lock: + client = self._mcp_clients.get(key) + if client is None: + config = {"mcpServers": manual_call_template.config.mcpServers} + client = _QuietStdioMCPClient.from_dict(config) + self._mcp_clients[key] = client + return client async def _get_or_create_session(self, server_name: str, manual_call_template: 'McpCallTemplate'): """Get an existing session or create a new one using MCPClient.""" - await self._ensure_mcp_client(manual_call_template) + client = await self._ensure_mcp_client(manual_call_template) try: # Try to get existing session - session = self._mcp_client.get_session(server_name) + session = client.get_session(server_name) self._log_info(f"Reusing existing session for server: {server_name}") return session except ValueError: # Session doesn't exist, create a new one self._log_info(f"Creating new session for server: {server_name}") try: - session = await self._mcp_client.create_session(server_name, auto_initialize=True) + session = await client.create_session(server_name, auto_initialize=True) except Exception as e: server_config = manual_call_template.config.mcpServers.get(server_name) is_stdio = isinstance(server_config, dict) and "command" in server_config @@ -150,16 +161,23 @@ async def _get_or_create_session(self, server_name: str, manual_call_template: ' raise return session - async def _cleanup_session(self, server_name: str): - """Clean up a specific session.""" - if self._mcp_client: - await self._mcp_client.close_session(server_name) + async def _cleanup_session(self, server_name: str, manual_call_template: 'McpCallTemplate'): + """Clean up a specific session of the client serving this manual.""" + client = self._mcp_clients.get(self._config_key(manual_call_template)) + if client is not None and server_name in client.sessions: + await client.close_session(server_name) self._log_info(f"Cleaned up session for server: {server_name}") async def _cleanup_all_sessions(self): - """Clean up all active sessions.""" - if self._mcp_client: - await self._mcp_client.close_all_sessions() + """Close every session of every client. A client whose shutdown fails is + kept so a later close() can retry it instead of leaking its processes.""" + for key, client in list(self._mcp_clients.items()): + try: + await client.close_all_sessions() + del self._mcp_clients[key] + except Exception as e: + self._log_warning(f"Failed to close sessions of an MCP client: {e}") + if not self._mcp_clients: self._log_info("Cleaned up all sessions") def _add_server_to_tool_name(self, tools, server_name: str): @@ -192,7 +210,7 @@ async def _list_tools_with_session(self, server_name: str, manual_call_template: if is_session_error: # Only restart session for connection/transport level issues - await self._cleanup_session(server_name) + await self._cleanup_session(server_name, manual_call_template) self._log_warning(f"Session-level error for list_tools, retrying with fresh session: {e}") # Retry with a fresh session @@ -220,7 +238,7 @@ async def _list_resources_with_session(self, server_name: str, manual_call_templ return resources_response except Exception as e: # If there's an error, clean up the potentially bad session and try once more - await self._cleanup_session(server_name) + await self._cleanup_session(server_name, manual_call_template) self._log_warning(f"Session failed for list_resources, retrying: {e}") # Retry with a fresh session @@ -240,7 +258,7 @@ async def _read_resource_with_session(self, server_name: str, manual_call_templa return result except Exception as e: # If there's an error, clean up the potentially bad session and try once more - await self._cleanup_session(server_name) + await self._cleanup_session(server_name, manual_call_template) self._log_warning(f"Session failed for read_resource '{resource_uri}', retrying: {e}") # Retry with a fresh session @@ -569,7 +587,7 @@ async def deregister_manual(self, caller: 'UtcpClient', manual_call_template: Ca # Clean up sessions for all servers in this manual if manual_call_template.config and manual_call_template.config.mcpServers: for server_name, server_config in manual_call_template.config.mcpServers.items(): - await self._cleanup_session(server_name) + await self._cleanup_session(server_name, manual_call_template) self._log_info(f"Cleaned up session for server '{server_name}'") async def close(self) -> None: diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index 38977d6..65b4da9 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -264,7 +264,7 @@ async def test_stdio_child_stderr_suppressed_by_default(transport: McpCommunicat assert session.connector.errlog is not sys.stderr assert session.connector.errlog.name == os.devnull finally: - await transport._cleanup_session(SERVER_NAME) + await transport._cleanup_session(SERVER_NAME, mcp_manual) @pytest.mark.asyncio @@ -275,7 +275,7 @@ async def test_stdio_child_stderr_inherit_opt_in(transport: McpCommunicationProt try: assert session.connector.errlog is sys.stderr finally: - await transport._cleanup_session(SERVER_NAME) + await transport._cleanup_session(SERVER_NAME, mcp_manual) @pytest.mark.asyncio @@ -303,10 +303,29 @@ async def test_mcp_client_and_session_are_reused_across_calls(transport: McpComm """Repeated calls with the same configuration reuse one client and one session instead of spawning a new server process per call.""" await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "one"}, mcp_manual) - client_after_first = transport._mcp_client + assert len(transport._mcp_clients) == 1 + client_after_first = next(iter(transport._mcp_clients.values())) await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "two"}, mcp_manual) - assert transport._mcp_client is client_after_first - assert list(transport._mcp_client.sessions.keys()) == [SERVER_NAME] + assert len(transport._mcp_clients) == 1 + assert next(iter(transport._mcp_clients.values())) is client_after_first + assert list(client_after_first.sessions.keys()) == [SERVER_NAME] + + +@pytest.mark.asyncio +async def test_manuals_with_different_configurations_get_separate_clients(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + """The protocol object is shared by every manual; one manual's calls must not + evict another manual's sessions.""" + other_manual = McpCallTemplate( + name="other_manual", + call_template_type="mcp", + config=McpConfig(mcpServers={"other_server": dict(mcp_manual.config.mcpServers[SERVER_NAME])}), + ) + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "a"}, mcp_manual) + await transport.call_tool(None, "other_server.echo", {"message": "b"}, other_manual) + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "c"}, mcp_manual) + assert len(transport._mcp_clients) == 2 + sessions = sorted(name for c in transport._mcp_clients.values() for name in c.sessions) + assert sessions == sorted([SERVER_NAME, "other_server"]) @pytest.mark.asyncio @@ -314,4 +333,4 @@ async def test_close_after_use_does_not_raise(transport: McpCommunicationProtoco """close() used to raise AttributeError after cleaning up; it must complete.""" await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "one"}, mcp_manual) await transport.close() - assert transport._mcp_client.sessions == {} + assert transport._mcp_clients == {} From 7e00e6fbe24f9d5a6b2d551bba01d8ff1c3279a8 Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:59:47 +0200 Subject: [PATCH 6/9] Pre-release review fixes, fifth batch - mcp: when a manual's configuration changes and no other manual uses the old one, the old client's sessions are closed instead of lingering until close(). Ownership is tracked per manual name. - sse: the residual buffer is checked against the event cap at end of stream too; retry digit strings are bounded to 18 digits before conversion. - tests: the connection-drop handlers wait 100 ms after writing the first event before closing the socket. On Windows CI the event and the close otherwise arrived together and aiohttp raised before delivering the event, failing six tests that pass elsewhere. Co-Authored-By: Claude Fable 5.1 --- .../utcp_http/sse_communication_protocol.py | 16 +++++++++------ .../tests/test_sse_communication_protocol.py | 20 +++++++++++++++---- .../utcp_mcp/mcp_communication_protocol.py | 20 ++++++++++++++++++- .../mcp/tests/test_mcp_transport.py | 15 ++++++++++++++ 4 files changed, 60 insertions(+), 11 deletions(-) diff --git a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py index 03e6d88..1c2ad6b 100644 --- a/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py +++ b/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py @@ -399,12 +399,10 @@ def flush(event_string: str): current_event['id'] = value elif field == 'retry': # Spec: only a value made of ASCII digits sets the reconnection time. - # int() refuses absurdly long digit strings; ignore those too. - if value.isascii() and value.isdigit(): - try: - current_event['retry'] = int(value) - except ValueError: - pass + # Anything longer than 18 digits is absurd (and would be capped anyway); + # bounding the length keeps the conversion cheap whatever the interpreter. + if value.isascii() and value.isdigit() and len(value) <= 18: + current_event['retry'] = int(value) if data_lines: current_event['data'] = '\n'.join(data_lines) return current_event or None @@ -452,6 +450,12 @@ def normalise(text: str) -> str: event = flush(event_string) if event is not None: yield event + # The residual (discarded) buffer is still subject to the cap, so an + # over-limit malformed stream fails the same way at end of stream. + if len(buffer) > self.MAX_EVENT_BUFFER_CHARS: + raise SseProtocolError( + f"SSE event exceeded {self.MAX_EVENT_BUFFER_CHARS} characters without a blank-line delimiter" + ) @staticmethod def _parse_event_data(data: str) -> Any: diff --git a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py index c6cfedc..6c7df54 100644 --- a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py @@ -141,7 +141,10 @@ async def flaky_503_events_handler(request): await response.prepare(request) if state["connections"] == 1: await response.write(SAMPLE_SSE_EVENTS[0].encode('utf-8')) - await asyncio.sleep(0.01) + # Make sure the event has left the socket before dropping it: on Windows the + # data and the close otherwise arrive together and aiohttp raises before + # delivering the event. + await asyncio.sleep(0.1) request.transport.close() return response for event in SAMPLE_SSE_EVENTS[1:]: @@ -164,7 +167,10 @@ async def huge_retry_events_handler(request): await response.prepare(request) if state["connections"] == 1: await response.write(b'id: 1\nretry: 100000\ndata: {"seq": 1}\n\n') - await asyncio.sleep(0.01) + # Make sure the event has left the socket before dropping it: on Windows the + # data and the close otherwise arrive together and aiohttp raises before + # delivering the event. + await asyncio.sleep(0.1) request.transport.close() return response await response.write(b'id: 2\ndata: {"seq": 2}\n\n') @@ -201,7 +207,10 @@ async def empty_id_events_handler(request): await response.prepare(request) if state["connections"] == 1: await response.write(b'id: 1\ndata: {"seq": 1}\n\nid\ndata: {"seq": 2}\n\n') - await asyncio.sleep(0.01) + # Make sure the event has left the socket before dropping it: on Windows the + # data and the close otherwise arrive together and aiohttp raises before + # delivering the event. + await asyncio.sleep(0.1) request.transport.close() return response await response.write(b'data: {"seq": 3}\n\n') @@ -221,7 +230,10 @@ async def flaky_events_handler(request): if state["always_drop"] or state["connections"] == 1: await response.write(SAMPLE_SSE_EVENTS[0].encode('utf-8')) - await asyncio.sleep(0.01) + # Make sure the event has left the socket before dropping it: on Windows the + # data and the close otherwise arrive together and aiohttp raises before + # delivering the event. + await asyncio.sleep(0.1) request.transport.close() return response 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 c158119..c2d54ba 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 @@ -97,6 +97,10 @@ def __init__(self): # 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 manual (by name) currently uses, so a client + # nothing references any more can be closed when a manual's + # configuration changes. + self._manual_config_keys: Dict[str, str] = {} self._clients_lock = asyncio.Lock() def _log_info(self, message: str): @@ -125,8 +129,9 @@ async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> M serialised so two concurrent first calls cannot each spawn a client. """ key = self._config_key(manual_call_template) + manual_name = manual_call_template.name or key client = self._mcp_clients.get(key) - if client is not None: + if client is not None and self._manual_config_keys.get(manual_name) == key: return client async with self._clients_lock: client = self._mcp_clients.get(key) @@ -134,6 +139,19 @@ async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> M config = {"mcpServers": manual_call_template.config.mcpServers} client = _QuietStdioMCPClient.from_dict(config) self._mcp_clients[key] = client + previous_key = self._manual_config_keys.get(manual_name) + self._manual_config_keys[manual_name] = key + if previous_key is not None and previous_key != key and previous_key not in self._manual_config_keys.values(): + # This manual's configuration changed and no other manual uses the + # old one: release the old client's sessions and processes. + stale = self._mcp_clients.pop(previous_key, None) + if stale is not None: + try: + await stale.close_all_sessions() + except Exception as e: + # Keep it so close() can retry rather than leaking its processes. + self._mcp_clients[previous_key] = stale + self._log_warning(f"Failed to close sessions of a stale MCP client: {e}") return client async def _get_or_create_session(self, server_name: str, manual_call_template: 'McpCallTemplate'): diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index 65b4da9..bdf412e 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -334,3 +334,18 @@ async def test_close_after_use_does_not_raise(transport: McpCommunicationProtoco await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "one"}, mcp_manual) await transport.close() assert transport._mcp_clients == {} + + +@pytest.mark.asyncio +async def test_changed_configuration_releases_the_stale_client(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + """When a manual's configuration changes and nothing else uses the old one, + the old client's sessions are closed instead of lingering until close().""" + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "a"}, mcp_manual) + changed = McpCallTemplate( + name=mcp_manual.name, + call_template_type="mcp", + config=McpConfig(mcpServers={SERVER_NAME: {**mcp_manual.config.mcpServers[SERVER_NAME], "env": {"CHANGED": "1"}}}), + ) + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "b"}, changed) + assert len(transport._mcp_clients) == 1 + assert next(iter(transport._mcp_clients.values())).config["mcpServers"][SERVER_NAME]["env"] == {"CHANGED": "1"} From b9ce1562128d879ff8c5da3a34c30d52da23ceeb Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:09:04 +0200 Subject: [PATCH 7/9] Pre-release review fixes, sixth batch - mcp: deregistering a manual drops only that manual's claim on its client; the client's sessions are closed when no manual references the configuration any more. Two manuals with identical configurations share one client, and deregistering one no longer tears down the other's sessions or leaves a ghost ownership entry. - tests: the streaming error-body tests use a body distinct from the reason phrase so they can only pass when the body is surfaced; deregistration test for shared configurations. Co-Authored-By: Claude Fable 5.1 --- .../tests/test_sse_communication_protocol.py | 12 ++++++-- ..._streamable_http_communication_protocol.py | 11 +++++-- .../utcp_mcp/mcp_communication_protocol.py | 30 ++++++++++++++++--- .../mcp/tests/test_mcp_transport.py | 20 +++++++++++++ 4 files changed, 63 insertions(+), 10 deletions(-) diff --git a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py index 6c7df54..6e99a26 100644 --- a/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_sse_communication_protocol.py @@ -106,6 +106,10 @@ async def error_handler(request): return web.Response(status=500, text="Internal Server Error") +async def refused_handler(request): + return web.Response(status=503, text="streaming refused: backend down") + + async def forbidden_discovery_handler(request): return web.Response(status=403, text="discovery refused: tenant is not provisioned for streaming") @@ -263,6 +267,7 @@ def app(): app.router.add_post("/token", token_handler) app.router.add_post("/token_header_auth", token_header_auth_handler) app.router.add_get("/error", error_handler) + app.router.add_get("/refused", refused_handler) app.router.add_get("/forbidden-discovery", forbidden_discovery_handler) app.router.add_get("/flaky_events", flaky_events_handler) app["flaky"] = {"connections": 0, "last_event_ids": [], "always_drop": False} @@ -722,9 +727,10 @@ async def test_final_blank_line_ending_in_lone_cr_completes_last_event(sse_trans async def test_streaming_call_error_surfaces_server_body(sse_transport, aiohttp_client, app): """A refused stream carries the server's body, like discovery does.""" client = await aiohttp_client(app) - call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/error"))) + call_template = SseCallTemplate(name="test-sse", url=str(client.make_url("/refused"))) with pytest.raises(aiohttp.ClientResponseError) as excinfo: async for _ in sse_transport.call_tool_streaming(None, "test-sse.t", {}, call_template): pass - assert excinfo.value.status == 500 - assert "Internal Server Error" in excinfo.value.message + assert excinfo.value.status == 503 + # Distinct from the reason phrase, so only a surfaced body satisfies this. + assert "streaming refused: backend down" in excinfo.value.message diff --git a/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py b/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py index 4315952..c76ed9c 100644 --- a/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py +++ b/plugins/communication_protocols/http/tests/test_streamable_http_communication_protocol.py @@ -109,6 +109,9 @@ async def check_oauth(request): async def error_endpoint(request): return web.Response(status=500, text="Internal Server Error") + async def refused_endpoint(request): + return web.Response(status=503, text="streaming refused: backend down") + async def forbidden_discovery(request): return web.Response(status=403, text="discovery refused: tenant is not provisioned for streaming") @@ -123,6 +126,7 @@ async def forbidden_discovery(request): web.post('/token', oauth_token_handler), web.post('/token-header', oauth_token_header_handler), web.get('/error', error_endpoint), + web.get('/refused', refused_endpoint), web.get('/forbidden-discovery', forbidden_discovery), ]) return app @@ -362,9 +366,10 @@ async def test_register_manual_surfaces_server_error_body(streamable_http_transp async def test_streaming_call_error_surfaces_server_body(streamable_http_transport, aiohttp_client, app): """A refused stream carries the server's body, like discovery does.""" client = await aiohttp_client(app) - call_template = StreamableHttpCallTemplate(name="test-provider", url=f"{client.make_url('/error')}") + call_template = StreamableHttpCallTemplate(name="test-provider", url=f"{client.make_url('/refused')}") with pytest.raises(aiohttp.ClientResponseError) as excinfo: async for _ in streamable_http_transport.call_tool_streaming(None, "test-provider.t", {}, call_template): pass - assert excinfo.value.status == 500 - assert "Internal Server Error" in excinfo.value.message + assert excinfo.value.status == 503 + # Distinct from the reason phrase, so only a surfaced body satisfies this. + assert "streaming refused: backend down" in excinfo.value.message 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 c2d54ba..3566dd3 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 @@ -179,6 +179,29 @@ async def _get_or_create_session(self, server_name: str, manual_call_template: ' raise return session + async def _release_manual_client(self, manual_call_template: 'McpCallTemplate') -> None: + """Drop this manual's claim on its client. The client's sessions are closed + only when no manual references that configuration any more; two manuals + with identical configurations share one client, and deregistering one + must not tear down the other's sessions.""" + key = self._config_key(manual_call_template) + manual_name = manual_call_template.name or key + async with self._clients_lock: + if self._manual_config_keys.get(manual_name) == key: + del self._manual_config_keys[manual_name] + if key in self._manual_config_keys.values(): + return + client = self._mcp_clients.pop(key, None) + if client is None: + return + try: + await client.close_all_sessions() + self._log_info(f"Closed the MCP client of manual '{manual_call_template.name}'") + except Exception as e: + # Keep it so close() can retry rather than leaking its processes. + self._mcp_clients[key] = client + self._log_warning(f"Failed to close sessions of the MCP client of manual '{manual_call_template.name}': {e}") + async def _cleanup_session(self, server_name: str, manual_call_template: 'McpCallTemplate'): """Clean up a specific session of the client serving this manual.""" client = self._mcp_clients.get(self._config_key(manual_call_template)) @@ -602,11 +625,10 @@ async def deregister_manual(self, caller: 'UtcpClient', manual_call_template: Ca self._log_info(f"Deregistering manual '{manual_call_template.name}' and cleaning up sessions") - # Clean up sessions for all servers in this manual + # Release this manual's claim on its client; the client's sessions are + # closed only when no other manual shares that configuration. if manual_call_template.config and manual_call_template.config.mcpServers: - for server_name, server_config in manual_call_template.config.mcpServers.items(): - await self._cleanup_session(server_name, manual_call_template) - self._log_info(f"Cleaned up session for server '{server_name}'") + await self._release_manual_client(manual_call_template) async def close(self) -> None: """Close all active sessions and clean up resources.""" diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index bdf412e..d5f31a3 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -349,3 +349,23 @@ async def test_changed_configuration_releases_the_stale_client(transport: McpCom await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "b"}, changed) assert len(transport._mcp_clients) == 1 assert next(iter(transport._mcp_clients.values())).config["mcpServers"][SERVER_NAME]["env"] == {"CHANGED": "1"} + + +@pytest.mark.asyncio +async def test_deregistering_one_of_two_manuals_sharing_a_configuration_keeps_the_client(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + twin = McpCallTemplate( + name="twin_manual", + call_template_type="mcp", + config=McpConfig(mcpServers=dict(mcp_manual.config.mcpServers)), + ) + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "a"}, mcp_manual) + await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "b"}, twin) + assert len(transport._mcp_clients) == 1 + + await transport.deregister_manual(None, mcp_manual) + # The twin still owns the configuration: its client and session survive. + assert len(transport._mcp_clients) == 1 + assert await transport.call_tool(None, f"{SERVER_NAME}.echo", {"message": "c"}, twin) == {"reply": "you said: c"} + + await transport.deregister_manual(None, twin) + assert transport._mcp_clients == {} From 5a3f9c6866292c4b44b16d596b3b23463fb6b879 Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:30:44 +0200 Subject: [PATCH 8/9] mcp: track client ownership per calling UtcpClient, not per manual name alone The protocol object is a process-wide singleton, so two UtcpClient instances may register a manual of the same name with different configurations; keyed by name alone, the second registration overwrote the first's ownership entry and closed its client. The calling client's identity is now part of the owner key, carried through the public entry points with a context variable rather than threading `caller` through every helper. Test added. Co-Authored-By: Claude Fable 5.1 --- .../utcp_mcp/mcp_communication_protocol.py | 42 ++++++++++++++++--- .../mcp/tests/test_mcp_transport.py | 23 ++++++++++ 2 files changed, 60 insertions(+), 5 deletions(-) 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 3566dd3..15e1f12 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 @@ -1,4 +1,6 @@ import asyncio +import contextvars +import functools import os import sys from typing import Any, Dict, Optional, AsyncGenerator, TYPE_CHECKING, Tuple, TextIO @@ -25,6 +27,28 @@ logger = logging.getLogger(__name__) +# Identity of the UtcpClient behind the current call. This protocol object is a +# process-wide singleton, so manual names alone cannot identify an owner: two +# UtcpClient instances may register a manual of the same name with different +# configurations. Set by the public entry points and read where client +# ownership is tracked, without threading ``caller`` through every helper. +_CURRENT_OWNER: contextvars.ContextVar[Optional[int]] = contextvars.ContextVar("utcp_mcp_owner", default=None) + + +def _with_owner(method): + """Run an entry point with ``_CURRENT_OWNER`` bound to its ``caller``.""" + + @functools.wraps(method) + async def wrapper(self, caller, *args, **kwargs): + token = _CURRENT_OWNER.set(id(caller) if caller is not None else None) + try: + return await method(self, caller, *args, **kwargs) + finally: + _CURRENT_OWNER.reset(token) + + return wrapper + + # Environment variable that opts stdio MCP children back into writing to the # host's stderr. Same name and semantics as the TypeScript SDK. CHILD_STDERR_ENV_VAR = "UTCP_MCP_CHILD_STDERR" @@ -97,9 +121,9 @@ def __init__(self): # 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 manual (by name) currently uses, so a client - # nothing references any more can be closed when a manual's - # configuration changes. + # Which configuration each owner (calling UtcpClient plus manual name) + # currently uses, so a client nothing references any more can be closed + # when a manual's configuration changes. self._manual_config_keys: Dict[str, str] = {} self._clients_lock = asyncio.Lock() @@ -120,6 +144,11 @@ def _config_key(manual_call_template: 'McpCallTemplate') -> str: """Canonical key for a manual's server configuration.""" return json.dumps(manual_call_template.config.mcpServers, sort_keys=True, default=str) + @staticmethod + def _owner_key(manual_call_template: 'McpCallTemplate', config_key: str) -> str: + """Identifies who holds a configuration: the calling UtcpClient plus the manual name.""" + return f"{_CURRENT_OWNER.get()}:{manual_call_template.name or config_key}" + async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> MCPClient: """Return the MCPClient for this manual's configuration, creating it once. @@ -129,7 +158,7 @@ async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> M serialised so two concurrent first calls cannot each spawn a client. """ key = self._config_key(manual_call_template) - manual_name = manual_call_template.name or key + manual_name = self._owner_key(manual_call_template, key) client = self._mcp_clients.get(key) if client is not None and self._manual_config_keys.get(manual_name) == key: return client @@ -185,7 +214,7 @@ async def _release_manual_client(self, manual_call_template: 'McpCallTemplate') with identical configurations share one client, and deregistering one must not tear down the other's sessions.""" key = self._config_key(manual_call_template) - manual_name = manual_call_template.name or key + manual_name = self._owner_key(manual_call_template, key) async with self._clients_lock: if self._manual_config_keys.get(manual_name) == key: del self._manual_config_keys[manual_name] @@ -313,6 +342,7 @@ async def _call_tool_with_session(self, server_name: str, manual_call_template: result = await session.call_tool(tool_name, arguments=inputs) return result + @_with_owner async def register_manual(self, caller: 'UtcpClient', manual_call_template: CallTemplate) -> RegisterManualResult: """REQUIRED Register a manual with the communication protocol. @@ -386,6 +416,7 @@ async def register_manual(self, caller: 'UtcpClient', manual_call_template: Call errors=errors ) + @_with_owner async def call_tool(self, caller: 'UtcpClient', tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> Any: """REQUIRED Call a tool using the model context protocol. @@ -617,6 +648,7 @@ def _parse_text_content(self, text: str) -> Any: # Return as string return text + @_with_owner async def deregister_manual(self, caller: 'UtcpClient', manual_call_template: CallTemplate) -> None: """Deregister an MCP manual and clean up associated sessions.""" if not isinstance(manual_call_template, McpCallTemplate): diff --git a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py index d5f31a3..df75aed 100644 --- a/plugins/communication_protocols/mcp/tests/test_mcp_transport.py +++ b/plugins/communication_protocols/mcp/tests/test_mcp_transport.py @@ -369,3 +369,26 @@ async def test_deregistering_one_of_two_manuals_sharing_a_configuration_keeps_th await transport.deregister_manual(None, twin) assert transport._mcp_clients == {} + + +@pytest.mark.asyncio +async def test_same_manual_name_from_two_clients_with_different_configurations(transport: McpCommunicationProtocol, mcp_manual: McpCallTemplate): + """The protocol is a process-wide singleton: two UtcpClient instances may register + a manual of the same name with different configurations, and one must not + close the other's client.""" + client_a, client_b = object(), object() + manual_b = McpCallTemplate( + name=mcp_manual.name, + call_template_type="mcp", + config=McpConfig(mcpServers={SERVER_NAME: {**mcp_manual.config.mcpServers[SERVER_NAME], "env": {"OWNER": "b"}}}), + ) + await transport.call_tool(client_a, f"{SERVER_NAME}.echo", {"message": "a"}, mcp_manual) + await transport.call_tool(client_b, f"{SERVER_NAME}.echo", {"message": "b"}, manual_b) + assert len(transport._mcp_clients) == 2 + # Client a's session is still alive and reused. + await transport.call_tool(client_a, f"{SERVER_NAME}.echo", {"message": "a2"}, mcp_manual) + assert len(transport._mcp_clients) == 2 + + await transport.deregister_manual(client_b, manual_b) + assert len(transport._mcp_clients) == 1 + assert await transport.call_tool(client_a, f"{SERVER_NAME}.echo", {"message": "a3"}, mcp_manual) == {"reply": "you said: a3"} From 319046a9c52967ad10532826c93b0f5b5076e1c6 Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:43:55 +0200 Subject: [PATCH 9/9] mcp: keep clients whose shutdown failed in a separate retry list Restoring a failed client into the live map happened outside the lock and could overwrite a newer client for the same configuration. Failed clients now go to a separate list that close() retries. Co-Authored-By: Claude Fable 5.1 --- .../utcp_mcp/mcp_communication_protocol.py | 24 ++++++++++++++----- 1 file changed, 18 insertions(+), 6 deletions(-) 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 15e1f12..6464642 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 @@ -3,7 +3,7 @@ import functools import os import sys -from typing import Any, Dict, Optional, AsyncGenerator, TYPE_CHECKING, Tuple, TextIO +from typing import Any, Dict, List, Optional, AsyncGenerator, TYPE_CHECKING, Tuple, TextIO import json from mcp_use import MCPClient @@ -126,6 +126,10 @@ def __init__(self): # when a manual's configuration changes. self._manual_config_keys: Dict[str, str] = {} self._clients_lock = asyncio.Lock() + # Clients whose shutdown failed. Kept apart from the live map so a retry + # on close() is possible without ever overwriting a newer live client + # for the same configuration. + self._failed_clients: List[MCPClient] = [] def _log_info(self, message: str): """Log informational messages.""" @@ -178,8 +182,8 @@ async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> M try: await stale.close_all_sessions() except Exception as e: - # Keep it so close() can retry rather than leaking its processes. - self._mcp_clients[previous_key] = stale + # Keep it aside so close() can retry rather than leaking its processes. + self._failed_clients.append(stale) self._log_warning(f"Failed to close sessions of a stale MCP client: {e}") return client @@ -227,8 +231,10 @@ async def _release_manual_client(self, manual_call_template: 'McpCallTemplate') await client.close_all_sessions() self._log_info(f"Closed the MCP client of manual '{manual_call_template.name}'") except Exception as e: - # Keep it so close() can retry rather than leaking its processes. - self._mcp_clients[key] = client + # Keep it aside so close() can retry rather than leaking its processes; + # never back into the live map, where a newer client for the same + # configuration may already live. + self._failed_clients.append(client) self._log_warning(f"Failed to close sessions of the MCP client of manual '{manual_call_template.name}': {e}") async def _cleanup_session(self, server_name: str, manual_call_template: 'McpCallTemplate'): @@ -247,7 +253,13 @@ async def _cleanup_all_sessions(self): del self._mcp_clients[key] except Exception as e: self._log_warning(f"Failed to close sessions of an MCP client: {e}") - if not self._mcp_clients: + for client in list(self._failed_clients): + try: + await client.close_all_sessions() + self._failed_clients.remove(client) + except Exception as e: + self._log_warning(f"Failed to close sessions of a previously failed MCP client: {e}") + if not self._mcp_clients and not self._failed_clients: self._log_info("Cleaned up all sessions") def _add_server_to_tool_name(self, tools, server_name: str):