添加MCP客户端超时配置和连接管理改进

添加了MCP客户端的超时配置功能，包括启动超时和工具调用超时参数。
改进了HTTP客户端和标准IO客户端的超时处理机制，确保请求在指定时间内完成或取消。
增加了对MCP服务器数量的限制，防止配置过多服务器导致系统不稳定。
增强了错误处理机制，当连接异常时能够正确清理资源并移除桥接主机。
添加了对大型MCP消息的大小验证，防止过大的请求导致系统问题。
优化了密钥更改后的处理流程，确保在修改密钥时停用服务器并要求重新测试。

backend/app/extensions/mcp.py

@@ -146,7 +146,7 @@ class McpStdioClient:
         timeout_code: str,
         response_error_code: str = "MCP_TOOL_CALL_FAILED",
     ) -> dict[str, Any]:
-        request_id, pending = self.begin_request(method, params)
+        request_id, pending = self.begin_request(method, params, timeout=timeout)
         return self.wait_response(
             request_id,
             pending,
@@ -156,7 +156,7 @@ class McpStdioClient:
         )
 
     def begin_request(
-        self, method: str, params: dict[str, Any]
+        self, method: str, params: dict[str, Any], *, timeout: float | None = None
     ) -> tuple[int, _PendingRequest]:
         self._ensure_running()
         with self._pending_lock:
@@ -388,6 +388,7 @@ class McpHttpClient:
         url: str,
         *,
         headers: dict[str, str],
+        startup_timeout_seconds: float = 15,
         on_seen: Callable[[], None],
         on_broken: Callable[[str], None],
         on_tools_changed: Callable[[], None],
@@ -407,6 +408,7 @@ class McpHttpClient:
         self._stream_started = False
         self._last_event_id: str | None = None
         self._stop_event = threading.Event()
+        self._startup_timeout_seconds = startup_timeout_seconds
 
     def start(self) -> None:
         return
@@ -429,7 +431,7 @@ class McpHttpClient:
         timeout_code: str,
         response_error_code: str = "MCP_TOOL_CALL_FAILED",
     ) -> dict[str, Any]:
-        request_id, pending = self.begin_request(method, params)
+        request_id, pending = self.begin_request(method, params, timeout=timeout)
         return self.wait_response(
             request_id,
             pending,
@@ -439,7 +441,7 @@ class McpHttpClient:
         )
 
     def begin_request(
-        self, method: str, params: dict[str, Any]
+        self, method: str, params: dict[str, Any], *, timeout: float | None = None
     ) -> tuple[int, _PendingRequest]:
         with self._pending_lock:
             request_id = self._next_id
@@ -454,7 +456,7 @@ class McpHttpClient:
         }
         threading.Thread(
             target=self._dispatch_request,
-            args=(request_id, message),
+            args=(request_id, message, timeout),
             daemon=True,
         ).start()
         return request_id, pending
@@ -526,7 +528,10 @@ class McpHttpClient:
         if self._session_id:
             try:
                 request = self._client.build_request(
-                    "DELETE", self.url, headers=self._request_headers()
+                    "DELETE",
+                    self.url,
+                    headers=self._request_headers(),
+                    timeout=min(self._startup_timeout_seconds, 5),
                 )
                 response = self._client.send(request, stream=True)
                 response.close()
@@ -539,9 +544,14 @@ class McpHttpClient:
             )
         )
 
-    def _dispatch_request(self, request_id: int, message: dict[str, Any]) -> None:
+    def _dispatch_request(
+        self,
+        request_id: int,
+        message: dict[str, Any],
+        timeout: float | None,
+    ) -> None:
         try:
-            response = self._post(message, timeout=None)
+            response = self._post(message, timeout=timeout)
             try:
                 self._capture_session(response)
                 content_type = response.headers.get("content-type", "").lower()
@@ -588,7 +598,12 @@ class McpHttpClient:
 
     def _post_notification(self, message: dict[str, Any]) -> None:
         try:
-            response = self._post(message, timeout=10)
+            timeout = (
+                self._startup_timeout_seconds
+                if message.get("method") == "notifications/initialized"
+                else 10
+            )
+            response = self._post(message, timeout=timeout)
         except httpx.HTTPError as exc:
             raise McpBridgeError(
                 "MCP_HTTP_REQUEST_FAILED",
@@ -616,6 +631,7 @@ class McpHttpClient:
             self.url,
             content=encoded.encode("utf-8"),
             headers=self._request_headers(),
+            timeout=timeout,
         )
         return self._client.send(request, stream=True)
 
@@ -714,7 +730,7 @@ class McpLegacySseClient(McpHttpClient):
     def start(self) -> None:
         threading.Thread(target=self._event_loop, daemon=True).start()
         try:
-            endpoint = self._endpoint_ready.get(timeout=15)
+            endpoint = self._endpoint_ready.get(timeout=self._startup_timeout_seconds)
         except queue.Empty as exc:
             raise McpBridgeError(
                 "MCP_INITIALIZE_FAILED",
@@ -730,9 +746,14 @@ class McpLegacySseClient(McpHttpClient):
 
         return
 
-    def _dispatch_request(self, request_id: int, message: dict[str, Any]) -> None:
+    def _dispatch_request(
+        self,
+        request_id: int,
+        message: dict[str, Any],
+        timeout: float | None,
+    ) -> None:
         try:
-            response = self._post(message, timeout=10)
+            response = self._post(message, timeout=timeout)
             try:
                 if response.status_code not in {200, 202, 204}:
                     raise McpBridgeError(
@@ -761,6 +782,8 @@ class McpLegacySseClient(McpHttpClient):
                 "MCP_INITIALIZE_FAILED", "Legacy MCP endpoint is not ready."
             )
         encoded = json.dumps(message, ensure_ascii=False, separators=(",", ":"))
+        if len(encoded.encode("utf-8")) > MAX_MCP_MESSAGE_BYTES:
+            raise McpBridgeError("MCP_TOOL_CALL_FAILED", "MCP request is too large.")
         request = self._client.build_request(
             "POST",
             self._endpoint,
@@ -770,6 +793,7 @@ class McpLegacySseClient(McpHttpClient):
                 "Accept": "application/json, text/event-stream",
                 "Content-Type": "application/json",
             },
+            timeout=timeout,
         )
         return self._client.send(request, stream=True)
 
@@ -801,6 +825,8 @@ class McpLegacySseClient(McpHttpClient):
                         self._endpoint = endpoint
                         continue
                     self._handle_message(_json_rpc_message(data))
+            if not self._stopping:
+                self.on_broken("Legacy MCP SSE stream ended unexpectedly.")
         except (McpBridgeError, httpx.HTTPError) as exc:
             if self._endpoint is None:
                 self._endpoint_ready.put(exc)
@@ -827,7 +853,7 @@ class _McpClient(Protocol):
         response_error_code: str = "MCP_TOOL_CALL_FAILED",
     ) -> dict[str, Any]: ...
     def begin_request(
-        self, method: str, params: dict[str, Any]
+        self, method: str, params: dict[str, Any], *, timeout: float | None = None
     ) -> tuple[int, _PendingRequest]: ...
     def wait_response(
         self,
@@ -931,6 +957,7 @@ class McpBridge:
             client = client_type(
                 url,
                 headers=headers or {},
+                startup_timeout_seconds=backend.startup_timeout_seconds,
                 on_seen=seen,
                 on_broken=broken,
                 on_tools_changed=tools_changed,
@@ -989,6 +1016,12 @@ class McpBridge:
             discovered = self._discover_tools(
                 plugin_id, client, backend, declared_permissions, tool_source
             )
+            if status.status == PluginHostState.unhealthy:
+                raise McpBridgeError(
+                    "PLUGIN_HOST_UNAVAILABLE",
+                    status.error or "MCP event stream became unavailable during startup.",
+                    status_code=503,
+                )
             status.status = PluginHostState.ready
             status.tools_count = len(discovered)
             status.last_seen_at = datetime.now(UTC)
@@ -1019,7 +1052,9 @@ class McpBridge:
     ) -> Any:
         host = self._host(plugin_id)
         rpc_id, pending = host.client.begin_request(
-            "tools/call", {"name": remote_name, "arguments": arguments}
+            "tools/call",
+            {"name": remote_name, "arguments": arguments},
+            timeout=host.backend.tool_timeout_seconds,
         )
         call_key = (plugin_id, request_id)
         with self._lock:

backend/app/extensions/mcp_registry.py

@@ -13,12 +13,13 @@ from typing import Any
 from urllib.parse import urlsplit
 from uuid import uuid4
 
-from pydantic import BaseModel, ConfigDict, create_model
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, create_model
 
 from app.agent.permissions import KNOWN_PERMISSIONS
 from app.agent.tools import ToolExecutionContext, ToolRegistry
 from app.contracts import (
     McpServer,
+    McpServerConfig,
     McpServerCreateRequest,
     McpServerSecretStatus,
     McpServerTransport,
@@ -40,6 +41,23 @@ _RESERVED_HEADERS = {
     "mcp-protocol-version",
     "mcp-session-id",
 }
+_SERVER_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
+_MAX_MCP_SERVERS = 256
+
+
+class _McpServerRecord(McpServerConfig):
+    """Validated on-disk representation with defaults for older C.1 records."""
+
+    version: int = Field(default=1, ge=1)
+    enabled: bool = False
+    approved_digest: str | None = Field(
+        default=None, min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$"
+    )
+    tested_digest: str | None = Field(
+        default=None, min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$"
+    )
+    last_tested_at: datetime | None = None
+    last_test_succeeded: bool | None = None
 
 
 class McpRegistryError(RuntimeError):
@@ -105,6 +123,13 @@ class McpServerRegistry:
     @_serialized_lifecycle
     def create(self, request: McpServerCreateRequest) -> McpServer:
         self._validate(request)
+        with self._lock:
+            if len(self._records) >= _MAX_MCP_SERVERS:
+                raise McpRegistryError(
+                    "MCP_SERVER_LIMIT_REACHED",
+                    f"At most {_MAX_MCP_SERVERS} MCP servers can be configured.",
+                    status_code=409,
+                )
         server_id = uuid4().hex[:12]
         record = request.model_dump(mode="json")
         record["name"] = request.name.strip()
@@ -137,8 +162,8 @@ class McpServerRegistry:
         self.disable(server_id)
         with self._lock:
             previous = self._record(server_id)
-            removed = [
-                (kind, key)
+            removed_secret_ids = [
+                self._secret_id(server_id, key, kind)
                 for kind, old_keys, new_keys in (
                     (
                         "environment",
@@ -153,6 +178,13 @@ class McpServerRegistry:
                 )
                 for key in set(old_keys) - set(new_keys)
             ]
+        try:
+            self.credentials.delete_many(removed_secret_ids)
+        except CredentialStoreError as exc:
+            raise McpRegistryError(
+                "MCP_SECRET_STORE_ERROR", str(exc), status_code=500
+            ) from exc
+        with self._lock:
             record = request.model_dump(mode="json", exclude={"version"})
             record["name"] = request.name.strip()
             record["command"] = request.command.strip() if request.command else None
@@ -170,13 +202,6 @@ class McpServerRegistry:
             self._records = updated
             self._last_status.pop(server_id, None)
             self._summaries.pop(server_id, None)
-        for kind, key in removed:
-            try:
-                self.credentials.delete(self._secret_id(server_id, key, kind))
-            except CredentialStoreError as exc:
-                raise McpRegistryError(
-                    "MCP_SECRET_STORE_ERROR", str(exc), status_code=500
-                ) from exc
         return self.get(server_id)
 
     @_serialized_lifecycle
@@ -192,18 +217,19 @@ class McpServerRegistry:
                 )
                 for key in keys
             ]
-            updated = dict(self._records)
-            del updated[server_id]
-            self._write(updated)
-            self._records = updated
-            self._last_status.pop(server_id, None)
-            self._summaries.pop(server_id, None)
         try:
             self.credentials.delete_many(secret_ids)
         except CredentialStoreError as exc:
             raise McpRegistryError(
                 "MCP_SECRET_STORE_ERROR", str(exc), status_code=500
             ) from exc
+        with self._lock:
+            updated = dict(self._records)
+            del updated[server_id]
+            self._write(updated)
+            self._records = updated
+            self._last_status.pop(server_id, None)
+            self._summaries.pop(server_id, None)
         self.bridge.remove(self._host_id(server_id))
 
     @_serialized_lifecycle
@@ -236,6 +262,9 @@ class McpServerRegistry:
                     "MCP_SECRET_NOT_DECLARED",
                     "Secret environment key is not declared in this server configuration.",
                 )
+        if record.get("enabled"):
+            self.disable(server_id)
+        self._invalidate_test(server_id)
         try:
             self.credentials.put(self._secret_id(server_id, key, kind), secret)
         except CredentialStoreError as exc:
@@ -254,6 +283,9 @@ class McpServerRegistry:
                 "MCP_SECRET_NOT_DECLARED",
                 "Secret environment key is not declared in this server configuration.",
             )
+        if record.get("enabled"):
+            self.disable(server_id)
+        self._invalidate_test(server_id)
         try:
             self.credentials.delete(self._secret_id(server_id, key, kind))
         except CredentialStoreError as exc:
@@ -465,17 +497,27 @@ class McpServerRegistry:
         self.tools.register(definition, arguments_model, executor)
 
     def _unavailable(self, server_id: str, message: str) -> None:
-        with self._lock:
-            for name in self._registered.pop(server_id, []):
-                self.tools.unregister(name)
-            record = self._records.get(server_id)
-            if record is not None:
-                self._records[server_id] = {**record, "enabled": False}
-                self._last_status[server_id] = {
-                    "status": PluginHostState.unhealthy,
-                    "error": message,
-                }
-                self._write()
+        # A failure may race with enable(). Waiting for the lifecycle mutation makes
+        # sure tools registered immediately before the callback are also removed.
+        with self._lifecycle_lock:
+            try:
+                with self._lock:
+                    record = self._records.get(server_id)
+                    registered = self._registered.pop(server_id, [])
+                    for name in registered:
+                        self.tools.unregister(name)
+                    if record is not None and (record.get("enabled") or registered):
+                        self._records[server_id] = {**record, "enabled": False}
+                        self._last_status[server_id] = {
+                            "status": PluginHostState.unhealthy,
+                            "error": message,
+                        }
+                        self._write()
+            finally:
+                # broken() can run on the client's reader/event thread. stop() does
+                # not join that thread, and setting _stopping before closing the
+                # transport prevents the close itself from reporting another failure.
+                self.bridge.remove(self._host_id(server_id))
 
     def _require_launch_allowed(
         self, record: dict[str, Any], *, require_test: bool
@@ -767,6 +809,23 @@ class McpServerRegistry:
                 status_code=404,
             ) from exc
 
+    def _invalidate_test(self, server_id: str) -> None:
+        """Make credential changes safe before touching the encrypted store."""
+
+        with self._lock:
+            record = self._record(server_id)
+            invalidated = {
+                **record,
+                "tested_digest": None,
+                "last_tested_at": None,
+                "last_test_succeeded": None,
+            }
+            updated = {**self._records, server_id: invalidated}
+            self._write(updated)
+            self._records = updated
+            self._last_status.pop(server_id, None)
+            self._summaries.pop(server_id, None)
+
     @property
     def _path(self) -> Path:
         return self.data_dir / "mcp" / "servers.json"
@@ -788,7 +847,29 @@ class McpServerRegistry:
                 "MCP server registry has an invalid format.",
                 status_code=500,
             )
-        return value
+        if len(value) > _MAX_MCP_SERVERS:
+            raise McpRegistryError(
+                "MCP_REGISTRY_INVALID",
+                "MCP server registry contains too many records.",
+                status_code=500,
+            )
+        normalized: dict[str, dict[str, Any]] = {}
+        config_fields = set(McpServerConfig.model_fields)
+        try:
+            for server_id, raw in value.items():
+                if not isinstance(server_id, str) or not _SERVER_ID.fullmatch(server_id):
+                    raise ValueError("invalid server id")
+                record = _McpServerRecord.model_validate(raw)
+                config = record.model_dump(mode="json", include=config_fields)
+                self._validate(McpServerCreateRequest.model_validate(config))
+                normalized[server_id] = record.model_dump(mode="json")
+        except (McpRegistryError, ValidationError, ValueError, TypeError) as exc:
+            raise McpRegistryError(
+                "MCP_REGISTRY_INVALID",
+                "MCP server registry contains an invalid record.",
+                status_code=500,
+            ) from exc
+        return normalized
 
     def _write(self, records: dict[str, dict[str, Any]] | None = None) -> None:
         temporary = self._path.with_suffix(".tmp")

backend/app/routes.py

@@ -607,7 +607,7 @@ async def put_mcp_server_secret(
     request: McpServerSecretWriteRequest,
     kind: str = Query(default="environment", pattern="^(environment|header)$"),
 ) -> McpServerSecretStatus:
-    return mcp_call(
+    return await mcp_call_async(
         lambda: container.mcp_servers.put_secret(
             server_id, key, request.secret.get_secret_value(), kind=kind
         )
@@ -624,7 +624,7 @@ async def delete_mcp_server_secret(
     key: str,
     kind: str = Query(default="environment", pattern="^(environment|header)$"),
 ) -> McpServerSecretStatus:
-    return mcp_call(
+    return await mcp_call_async(
         lambda: container.mcp_servers.delete_secret(server_id, key, kind=kind)
     )
 

backend/tests/test_api.py

@@ -1,26 +1,10 @@
 import asyncio
+import threading
+from types import SimpleNamespace
 
-from app.main import health, service_status
-from app.routes import (
-    get_index_status,
-    list_notes,
-    list_plugins,
-    list_provider_presets,
-    list_providers,
-    list_skills,
-)
-from app.routes import (
-    create_provider,
-    create_task,
-    delete_provider,
-    delete_task,
-    get_provider,
-    get_task,
-    list_tasks,
-    update_provider,
-    update_task,
-)
 from app.contracts import (
+    McpServerSecretStatus,
+    McpServerSecretWriteRequest,
     ProviderCreateRequest,
     ProviderType,
     ProviderUpdateRequest,
@@ -28,6 +12,63 @@ from app.contracts import (
     TaskStatus,
     TaskUpdateRequest,
 )
+from app.main import health, service_status
+from app.routes import (
+    create_provider,
+    create_task,
+    delete_provider,
+    delete_task,
+    get_index_status,
+    get_provider,
+    get_task,
+    list_notes,
+    list_plugins,
+    list_provider_presets,
+    list_providers,
+    list_skills,
+    list_tasks,
+    update_provider,
+    update_task,
+)
+
+
+def test_mcp_secret_routes_offload_blocking_lifecycle_work(monkeypatch) -> None:
+    from app import routes
+
+    caller_thread = threading.get_ident()
+    worker_threads: list[int] = []
+
+    class FakeMcpRegistry:
+        def put_secret(self, server_id, key, secret, *, kind):
+            worker_threads.append(threading.get_ident())
+            return McpServerSecretStatus(key=key, configured=True)
+
+        def delete_secret(self, server_id, key, *, kind):
+            worker_threads.append(threading.get_ident())
+            return McpServerSecretStatus(key=key, configured=False)
+
+    monkeypatch.setattr(
+        routes,
+        "container",
+        SimpleNamespace(mcp_servers=FakeMcpRegistry()),
+    )
+    written = asyncio.run(
+        routes.put_mcp_server_secret(
+            "server-1",
+            "TOKEN",
+            McpServerSecretWriteRequest(secret="hidden"),
+            kind="environment",
+        )
+    )
+    deleted = asyncio.run(
+        routes.delete_mcp_server_secret(
+            "server-1", "TOKEN", kind="environment"
+        )
+    )
+
+    assert written.configured is True
+    assert deleted.configured is False
+    assert worker_threads and all(item != caller_thread for item in worker_threads)
 
 
 def test_health() -> None:
@@ -101,6 +142,14 @@ def test_openapi_contains_documented_frontend_interfaces() -> None:
         "/api/plugins/{plugin_id}/settings/{key}/secret",
         "/api/plugins/{plugin_id}/enable",
         "/api/plugins/{plugin_id}/disable",
+        "/api/mcp/servers",
+        "/api/mcp/servers/{server_id}",
+        "/api/mcp/servers/{server_id}/tools",
+        "/api/mcp/servers/{server_id}/trust",
+        "/api/mcp/servers/{server_id}/test",
+        "/api/mcp/servers/{server_id}/enable",
+        "/api/mcp/servers/{server_id}/disable",
+        "/api/mcp/servers/{server_id}/secrets/{key}",
         "/api/providers/test",
         "/api/providers/presets",
         "/api/credentials/{credential_id}",

backend/tests/test_mcp_registry.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import sys
+import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
 
@@ -10,8 +11,9 @@ import pytest
 from app.agent.tools import ToolExecutionContext, ToolRegistry
 from app.config import BACKEND_DIR, get_settings
 from app.contracts import McpServerCreateRequest, McpServerUpdateRequest, ToolCall
+from app.extensions.mcp import McpLegacySseClient
 from app.extensions.mcp_registry import McpRegistryError, McpServerRegistry
-from app.providers.credentials import EncryptedCredentialStore
+from app.providers.credentials import CredentialStoreError, EncryptedCredentialStore
 
 SERVER = BACKEND_DIR / "extensions" / "fixtures" / "mcp-echo" / "server.py"
 
@@ -59,6 +61,28 @@ def test_registry_requires_current_trust_and_never_returns_secret() -> None:
     service.shutdown()
 
 
+def test_secret_change_disables_server_and_requires_a_new_connection_test() -> None:
+    service = registry()
+    created = service.create(request())
+    service.put_secret(created.server_id, "TEST_MCP_SECRET", "first")
+    service.trust(created.server_id, created.command_digest)
+    service.test(created.server_id)
+    service.enable(created.server_id)
+
+    service.put_secret(created.server_id, "TEST_MCP_SECRET", "second")
+    current = service.get(created.server_id)
+    assert current.enabled is False
+    assert current.last_test_succeeded is None
+    assert not any(
+        item.name.startswith(f"mcp.{created.server_id}.")
+        for item in service.tools.definitions()
+    )
+    with pytest.raises(McpRegistryError) as error:
+        service.enable(created.server_id)
+    assert error.value.code == "MCP_CONNECTION_TEST_REQUIRED"
+    service.shutdown()
+
+
 def test_update_disables_server_and_revokes_command_trust() -> None:
     service = registry()
     created = service.create(request(secret_environment_keys=[]))
@@ -87,6 +111,70 @@ def test_update_disables_server_and_revokes_command_trust() -> None:
     service.shutdown()
 
 
+def test_update_remains_retryable_when_removed_secret_cleanup_fails(monkeypatch) -> None:
+    service = registry()
+    created = service.create(request())
+    service.put_secret(created.server_id, "TEST_MCP_SECRET", "keep-until-retry")
+
+    def fail_delete_many(_secret_ids: list[str]) -> set[str]:
+        raise CredentialStoreError("credential store unavailable")
+
+    monkeypatch.setattr(service.credentials, "delete_many", fail_delete_many)
+    with pytest.raises(McpRegistryError) as error:
+        service.update(
+            created.server_id,
+            McpServerUpdateRequest(
+                **request(secret_environment_keys=[]).model_dump(),
+                version=created.version,
+            ),
+        )
+
+    current = service.get(created.server_id)
+    assert error.value.code == "MCP_SECRET_STORE_ERROR"
+    assert current.version == created.version
+    assert current.secret_environment == {"TEST_MCP_SECRET": True}
+    service.shutdown()
+
+
+def test_delete_keeps_server_retryable_when_secret_cleanup_fails(monkeypatch) -> None:
+    service = registry()
+    created = service.create(request())
+    service.put_secret(created.server_id, "TEST_MCP_SECRET", "keep-until-retry")
+
+    def fail_delete_many(_secret_ids: list[str]) -> set[str]:
+        raise CredentialStoreError("credential store unavailable")
+
+    monkeypatch.setattr(service.credentials, "delete_many", fail_delete_many)
+    with pytest.raises(McpRegistryError) as error:
+        service.delete(created.server_id)
+
+    current = service.get(created.server_id)
+    assert error.value.code == "MCP_SECRET_STORE_ERROR"
+    assert current.server_id == created.server_id
+    assert current.secret_environment == {"TEST_MCP_SECRET": True}
+    service.shutdown()
+
+
+def test_unavailable_server_removes_bridge_host(monkeypatch) -> None:
+    service = registry()
+    created = service.create(request(secret_environment_keys=[]))
+    with service._lock:
+        service._records[created.server_id] = {
+            **service._records[created.server_id],
+            "enabled": True,
+        }
+    removed: list[str] = []
+    monkeypatch.setattr(service.bridge, "remove", removed.append)
+
+    service._unavailable(created.server_id, "connection lost")
+
+    current = service.get(created.server_id)
+    assert removed == [f"mcp.{created.server_id}"]
+    assert current.enabled is False
+    assert current.status == "unhealthy"
+    service.shutdown()
+
+
 def test_production_rejects_process_launch_even_after_approval() -> None:
     service = registry(launch=False)
     created = service.create(request(secret_environment_keys=[]))
@@ -141,6 +229,36 @@ def test_registry_rejects_corrupt_persisted_json(tmp_path) -> None:
     assert error.value.code == "MCP_REGISTRY_INVALID"
 
 
+def test_registry_rejects_structurally_invalid_record(tmp_path) -> None:
+    path = tmp_path / "mcp"
+    path.mkdir()
+    (path / "servers.json").write_text(
+        json.dumps({"server-1": {"name": "Broken", "transport": "stdio"}}),
+        encoding="utf-8",
+    )
+    with pytest.raises(McpRegistryError) as error:
+        McpServerRegistry(
+            ToolRegistry(),
+            EncryptedCredentialStore(),
+            tmp_path,
+            allow_process_launch=True,
+        )
+    assert error.value.code == "MCP_REGISTRY_INVALID"
+
+
+def test_registry_rejects_create_before_exceeding_persisted_limit(
+    monkeypatch,
+) -> None:
+    service = registry()
+    service.create(request(name="Only server"))
+    monkeypatch.setattr("app.extensions.mcp_registry._MAX_MCP_SERVERS", 1)
+    with pytest.raises(McpRegistryError) as error:
+        service.create(request(name="One too many"))
+    assert error.value.code == "MCP_SERVER_LIMIT_REACHED"
+    assert len(service.list()) == 1
+    service.shutdown()
+
+
 def test_stdio_command_is_not_parsed_as_a_shell_string() -> None:
     service = registry()
     created = service.create(
@@ -217,6 +335,7 @@ def test_streamable_http_supports_session_headers_secrets_and_tool_summary(
     monkeypatch,
 ) -> None:
     requests: list[httpx.Request] = []
+    request_timeouts: dict[str, float] = {}
 
     def handler(request_value: httpx.Request) -> httpx.Response:
         requests.append(request_value)
@@ -225,6 +344,9 @@ def test_streamable_http_supports_session_headers_secrets_and_tool_summary(
         if request_value.method == "DELETE":
             return httpx.Response(405)
         payload = json.loads(request_value.content)
+        timeout = request_value.extensions.get("timeout", {}).get("read")
+        if isinstance(timeout, (int, float)):
+            request_timeouts[payload.get("method", "notification")] = float(timeout)
         if payload.get("method") == "initialize":
             response = _http_result(
                 payload["id"],
@@ -290,6 +412,9 @@ def test_streamable_http_supports_session_headers_secrets_and_tool_summary(
     assert all(
         request.headers.get("authorization") == "Bearer hidden" for request in requests
     )
+    assert request_timeouts["initialize"] == 15
+    assert request_timeouts["notifications/initialized"] == 15
+    assert request_timeouts["tools/list"] == 15
     enabled = service.enable(created.server_id)
     tool_name = service.list_tools(created.server_id)[0].name
     result = asyncio.run(
@@ -301,11 +426,15 @@ def test_streamable_http_supports_session_headers_secrets_and_tool_summary(
     assert enabled.enabled is True
     assert result.success is True
     assert result.output == {"transport": "http"}
+    assert request_timeouts["tools/call"] == 30
     service.disable(created.server_id)
     service.shutdown()
 
 
 class _LegacyEventStream(httpx.SyncByteStream):
+    def __init__(self) -> None:
+        self.closed = threading.Event()
+
     def __iter__(self):
         yield b"event: endpoint\ndata: /messages\n\n"
         time.sleep(0.1)
@@ -326,17 +455,22 @@ class _LegacyEventStream(httpx.SyncByteStream):
             "result": {"tools": []},
         }
         yield f"data: {json.dumps(tools)}\n\n".encode()
+        self.closed.wait()
+
+    def close(self) -> None:
+        self.closed.set()
 
 
 def test_legacy_sse_uses_same_origin_endpoint(monkeypatch) -> None:
     posted_urls: list[str] = []
+    event_stream = _LegacyEventStream()
 
     def handler(request_value: httpx.Request) -> httpx.Response:
         if request_value.method == "GET":
             return httpx.Response(
                 200,
                 headers={"content-type": "text/event-stream"},
-                stream=_LegacyEventStream(),
+                stream=event_stream,
             )
         posted_urls.append(str(request_value.url))
         return httpx.Response(202)
@@ -361,6 +495,39 @@ def test_legacy_sse_uses_same_origin_endpoint(monkeypatch) -> None:
         url == "https://legacy.example.test/messages" for url in posted_urls
     )
     service.shutdown()
+    event_stream.close()
+
+
+class _EndingLegacyEventStream(httpx.SyncByteStream):
+    def __iter__(self):
+        yield b"event: endpoint\ndata: /messages\n\n"
+
+
+def test_legacy_sse_eof_marks_client_unavailable(monkeypatch) -> None:
+    def handler(_request_value: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            stream=_EndingLegacyEventStream(),
+        )
+
+    real_client = httpx.Client
+    monkeypatch.setattr(
+        "app.extensions.mcp.httpx.Client",
+        lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs),
+    )
+    broken = threading.Event()
+    client = McpLegacySseClient(
+        "https://legacy.example.test/sse",
+        headers={},
+        startup_timeout_seconds=1,
+        on_seen=lambda: None,
+        on_broken=lambda _message: broken.set(),
+        on_tools_changed=lambda: None,
+    )
+    client.start()
+    assert broken.wait(timeout=1)
+    client.stop()
 
 
 class _CrossOriginLegacyEventStream(httpx.SyncByteStream):