mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
refactor(mcp): remove periodic refresh, add manual refresh/reconnect controls
Deletes the _periodic_refresh task and its supporting state
(_refresh_task, _refresh_failures, _refresh_backoff_until,
_REFRESH_BACKOFF_BASE/MAX, _DEFAULT_REFRESH_INTERVAL, refresh_interval
kwarg) from MCPClientManager. Push notifications and operator-driven
manual refresh now cover all catalog-update needs; the long-running
4-hour timer was dead complexity that obscured the per-user pool
work to come.
Catalog freshness on auto-reconnect is preserved by scheduling an
unblocking _refresh_server task on the mcp-loop after _connect_one
succeeds; the calling thread returns immediately so half-open
recovery latency does not double. Adds MCPClientManager.reconnect_sync
(clears the circuit, closes any existing session, calls _connect_one,
clears stale catalog on failure).
Wires a new pair of operator endpoints —
POST /v1/api/admin/mcp-servers/{name}/refresh and
/v1/api/admin/mcp-servers/{name}/reconnect — that fan out to all
nodes through the existing _internal route family, with per-row
"Refresh" and "Reconnect" buttons in the MCP Servers admin tab.
The new node-internal paths /api/_internal/mcp-{refresh,reconnect}/
are gated to the approve scope to prevent direct unprivileged
reconnects bypassing the console's admin.mcp gate. Internal
endpoints return generic error messages and a filtered status
payload (no command/url) to keep transport details admin-gated.
Drops the [mcp] refresh_interval setting, the
--mcp-refresh-interval CLI flag, and the matching config-mapping
entry; updates docs/architecture.md, docs/tools.md,
docs/settings.md, and the three PlantUML diagrams that referenced
the periodic loop.
Tradeoffs (intentional):
- Idle nodes will not auto-rejoin a recovered MCP server until
traffic arrives or an operator clicks Reconnect. The previous
background reconnection loop is gone by design — push
notifications + operator controls replace it.
- Console fan-out blocks on the slowest node (existing pattern);
not changed here.
This is Phase 1 of the OAuth-MCP series — feature subtraction
ahead of per-user state.
This commit is contained in:
Patrick Buckley
@@ -546,11 +546,9 @@ adds, removes, or reconnects servers as needed.
|
||||
6. `_exec_mcp_tool()` calls `call_tool_sync()` which dispatches to the async loop
|
||||
via `asyncio.run_coroutine_threadsafe()`
|
||||
|
||||
**Tool refresh:** Three mechanisms keep tools up-to-date without restart:
|
||||
**Tool refresh:** Two mechanisms keep tools up-to-date without restart:
|
||||
- **Push:** Servers declaring `tools.listChanged` send `ToolListChangedNotification`;
|
||||
the registered `message_handler` triggers immediate single-server refresh.
|
||||
- **Periodic:** Servers without push support are polled on a staggered interval
|
||||
(default 4 h, configurable via `[mcp] refresh_interval` or `--mcp-refresh-interval`).
|
||||
- **Manual:** `/mcp refresh [server]` calls `refresh_sync()` for on-demand refresh
|
||||
(also attempts reconnection for disconnected servers).
|
||||
|
||||
@@ -572,10 +570,11 @@ from a healthy connection do not trip the breaker. When the cooldown expires
|
||||
(`call_tool_sync`, `read_resource_sync`, `get_prompt_sync`, `refresh_sync`)
|
||||
cancel orphaned futures on timeout to prevent coroutine accumulation on the
|
||||
background event loop. Push notification refreshes are debounced (5 s per
|
||||
server) to protect against notification storms. The periodic refresh loop
|
||||
attempts reconnection for disconnected servers with exponential backoff
|
||||
(60 s–1 h). Transport stream references are pre-closed before stack teardown to
|
||||
work around the MCP SDK's anyio cancel-scope CPU busy-loop (SDK #2147).
|
||||
server) to protect against notification storms. Operators can force a
|
||||
catalog refresh or full reconnect from the admin panel; reconnects clear
|
||||
the circuit breaker and run a fresh handshake. Transport stream references
|
||||
are pre-closed before stack teardown to work around the MCP SDK's anyio
|
||||
cancel-scope CPU busy-loop (SDK #2147).
|
||||
|
||||
**Error isolation:** Per-server connection/refresh failures are caught and logged; other
|
||||
servers are unaffected. Tool execution errors return error strings to the LLM
|
||||
|
||||
@@ -40,7 +40,7 @@ package "turnstone/core/" <<Rectangle>> {
|
||||
component [auth.py\nAuthentication] as auth <<core>>
|
||||
component [healthcheck.py\nBackendHealthMonitor] as healthcheck <<core>>
|
||||
component [ratelimit.py\nRateLimiter] as ratelimit <<core>>
|
||||
component [mcp_client.py\nMCPClientManager\n(push + periodic refresh)] as mcp <<core>>
|
||||
component [mcp_client.py\nMCPClientManager\n(push + manual refresh)] as mcp <<core>>
|
||||
component [tool_search.py\nToolSearchManager, BM25] as toolsearch <<core>>
|
||||
component [model_registry.py\nModelRegistry] as registry <<core>>
|
||||
}
|
||||
|
||||
@@ -253,7 +253,7 @@ class "MCPClientManager" as MCPMgr {
|
||||
Background asyncio event loop
|
||||
bridges async MCP SDK to
|
||||
sync ChatSession dispatch.
|
||||
Push + periodic + manual refresh.
|
||||
Push + manual refresh.
|
||||
Resources + prompts discovered
|
||||
alongside tools at startup.
|
||||
--
|
||||
|
||||
@@ -190,21 +190,25 @@ group Push Notifications (debounced 5s per server)
|
||||
MCPMgr -> Storage : sync_prompts_to_storage()
|
||||
end
|
||||
|
||||
group Periodic Polling (default 4h)
|
||||
MCPMgr -> MCPMgr : _periodic_refresh()
|
||||
group Manual Refresh
|
||||
Session -> MCPMgr : refresh_sync()
|
||||
note right
|
||||
Only polls capabilities
|
||||
without push support.
|
||||
Staggered per-server.
|
||||
Disconnected servers get
|
||||
reconnect attempts with
|
||||
exponential backoff (60s-1h).
|
||||
/mcp refresh [server] —
|
||||
re-fetches catalog and
|
||||
attempts reconnect for
|
||||
disconnected servers.
|
||||
end note
|
||||
end
|
||||
|
||||
group Manual Refresh
|
||||
Session -> MCPMgr : refresh_sync()
|
||||
note right: /mcp refresh [server]
|
||||
group Manual Reconnect
|
||||
Session -> MCPMgr : reconnect_sync(name)
|
||||
note right
|
||||
Operator-driven via the
|
||||
console admin panel —
|
||||
tears down session, clears
|
||||
circuit breaker, runs a
|
||||
fresh handshake.
|
||||
end note
|
||||
end
|
||||
|
||||
== Policy Evaluation ==
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
version https://git-lfs.github.com/spec/v1
|
||||
oid sha256:a3b5c59403a6febd81667fc8fd2a7d22bc59da6130eba0dea5449c42668d0ede
|
||||
size 387044
|
||||
oid sha256:95dd5ebc899a1261d516686a5aa3319a7f45015d411302825fa28afbfc82e1ce
|
||||
size 326766
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
version https://git-lfs.github.com/spec/v1
|
||||
oid sha256:474b900448ec04d1117b48a2b55614524721b2f04ac4bda66170bd0a06aae0f2
|
||||
size 624573
|
||||
oid sha256:25b5448bbb7da8ddafe4f65c6c5e6cbcaa9cb9f31746ca46d3a2241bc47b1956
|
||||
size 259687
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
version https://git-lfs.github.com/spec/v1
|
||||
oid sha256:7623df33be9baf7647ca1c2450640df57e1cd73e8be1f8168aae16e546ad683c
|
||||
size 459941
|
||||
oid sha256:d6aff446a062aa08f316985d00c2183148694f786d7f22172bc50b30046c728b
|
||||
size 379259
|
||||
|
||||
+1
-1
@@ -100,7 +100,7 @@ initialization:
|
||||
| `tools` | timeout, truncation, agent_max_turns, skip_permissions, search, search_threshold, search_max_results |
|
||||
| `server` | workstream_idle_timeout, max_workstreams |
|
||||
| `cluster` | node_fan_out_limit, mcp_max_servers |
|
||||
| `mcp` | config_path, refresh_interval, registry_url |
|
||||
| `mcp` | config_path, registry_url |
|
||||
| `ratelimit` | enabled, requests_per_second, burst, trusted_proxies |
|
||||
| `health` | backend_probe_interval, backend_probe_timeout, circuit_breaker_threshold, circuit_breaker_cooldown |
|
||||
| `judge` | enabled, model, provider, base_url, api_key, confidence_threshold, max_context_ratio, timeout, read_only_tools, output_guard, redact_secrets, cancel_on_approval |
|
||||
|
||||
+5
-14
@@ -758,22 +758,18 @@ MCP tools (3):
|
||||
|
||||
### Dynamic tool refresh
|
||||
|
||||
MCP tool lists stay up-to-date without restart through three mechanisms:
|
||||
MCP tool lists stay up-to-date without restart through two mechanisms:
|
||||
|
||||
1. **Push notifications** -- MCP servers that declare `tools.listChanged: true` in
|
||||
their capabilities send `notifications/tools/list_changed` when their tool list
|
||||
changes. `MCPClientManager` registers a `message_handler` on each `ClientSession`
|
||||
that triggers an immediate refresh for that server.
|
||||
|
||||
2. **Periodic timer** -- Servers that do *not* support push notifications are polled
|
||||
on a configurable interval (default 4 hours). The timer is staggered using a
|
||||
launch-time seed (`monotonic_ns ^ pid`) so cluster nodes don't all hit MCP
|
||||
servers simultaneously. Configure via `[mcp] refresh_interval` in `config.toml`
|
||||
or `--mcp-refresh-interval SECONDS` on the CLI. Set to `0` to disable.
|
||||
|
||||
3. **Manual** -- `/mcp refresh` re-fetches tools from all servers immediately.
|
||||
2. **Manual** -- `/mcp refresh` re-fetches tools from all servers immediately.
|
||||
`/mcp refresh <server>` targets a single server. If a server has disconnected,
|
||||
manual refresh attempts reconnection.
|
||||
manual refresh attempts reconnection. The console admin panel exposes the
|
||||
same controls (refresh / reconnect buttons per server) for cluster-wide
|
||||
fan-out.
|
||||
|
||||
When tools change, `MCPClientManager` rebuilds its merged tool list using copy-on-write
|
||||
(new list/dict objects assigned atomically) and notifies all active `ChatSession`
|
||||
@@ -781,11 +777,6 @@ instances via registered listener callbacks. Each session rebuilds its `_tools`,
|
||||
`_task_tools`, `_agent_tools`, and reconstructs its `ToolSearchManager` (if active),
|
||||
preserving the set of previously expanded (discovered) tools.
|
||||
|
||||
```toml
|
||||
[mcp]
|
||||
refresh_interval = 14400 # seconds (default 4h), 0 to disable
|
||||
```
|
||||
|
||||
```
|
||||
/mcp refresh
|
||||
MCP refresh complete:
|
||||
|
||||
@@ -207,6 +207,30 @@ class TestRequiredScope:
|
||||
"""Only POST is elevated — GET falls through to read."""
|
||||
assert required_scope("GET", "/api/_internal/mcp-reload") == "read"
|
||||
|
||||
def test_internal_mcp_refresh_one_needs_approve(self):
|
||||
assert required_scope("POST", "/api/_internal/mcp-refresh/srv") == "approve"
|
||||
|
||||
def test_v1_internal_mcp_refresh_one_needs_approve(self):
|
||||
assert required_scope("POST", "/v1/api/_internal/mcp-refresh/srv") == "approve"
|
||||
|
||||
def test_proxy_internal_mcp_refresh_one_needs_approve(self):
|
||||
assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-refresh/srv") == "approve"
|
||||
|
||||
def test_proxy_no_v1_internal_mcp_refresh_one_needs_approve(self):
|
||||
assert required_scope("POST", "/node/n1/api/_internal/mcp-refresh/srv") == "approve"
|
||||
|
||||
def test_internal_mcp_reconnect_one_needs_approve(self):
|
||||
assert required_scope("POST", "/api/_internal/mcp-reconnect/srv") == "approve"
|
||||
|
||||
def test_v1_internal_mcp_reconnect_one_needs_approve(self):
|
||||
assert required_scope("POST", "/v1/api/_internal/mcp-reconnect/srv") == "approve"
|
||||
|
||||
def test_proxy_internal_mcp_reconnect_one_needs_approve(self):
|
||||
assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-reconnect/srv") == "approve"
|
||||
|
||||
def test_proxy_no_v1_internal_mcp_reconnect_one_needs_approve(self):
|
||||
assert required_scope("POST", "/node/n1/api/_internal/mcp-reconnect/srv") == "approve"
|
||||
|
||||
# Workstream sub-resource mutations (parametric paths)
|
||||
def test_ws_delete_needs_write(self):
|
||||
assert required_scope("POST", "/api/workstreams/abc123/delete") == "write"
|
||||
|
||||
+441
-2
@@ -20,12 +20,16 @@ if TYPE_CHECKING:
|
||||
|
||||
from turnstone.console.server import (
|
||||
_collect_mcp_status,
|
||||
_notify_nodes_mcp_reconnect_one,
|
||||
_notify_nodes_mcp_refresh_one,
|
||||
_notify_nodes_mcp_reload,
|
||||
admin_create_mcp_server,
|
||||
admin_delete_mcp_server,
|
||||
admin_get_mcp_server,
|
||||
admin_import_mcp_config,
|
||||
admin_list_mcp_servers,
|
||||
admin_mcp_reconnect_one,
|
||||
admin_mcp_refresh_one,
|
||||
admin_mcp_reload,
|
||||
admin_update_mcp_server,
|
||||
)
|
||||
@@ -102,6 +106,16 @@ _ROUTES = [
|
||||
admin_mcp_reload,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{name}/refresh",
|
||||
admin_mcp_refresh_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{name}/reconnect",
|
||||
admin_mcp_reconnect_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{server_id}",
|
||||
admin_get_mcp_server,
|
||||
@@ -122,8 +136,12 @@ _ROUTES = [
|
||||
|
||||
|
||||
def _routes_with_internal() -> list[Mount]:
|
||||
"""Routes including the node-side internal endpoint (lazy-imported)."""
|
||||
from turnstone.server import internal_mcp_reload
|
||||
"""Routes including the node-side internal endpoints (lazy-imported)."""
|
||||
from turnstone.server import (
|
||||
internal_mcp_reconnect_one,
|
||||
internal_mcp_refresh_one,
|
||||
internal_mcp_reload,
|
||||
)
|
||||
|
||||
return [
|
||||
Mount(
|
||||
@@ -131,6 +149,16 @@ def _routes_with_internal() -> list[Mount]:
|
||||
routes=[
|
||||
*_ROUTES[0].routes, # type: ignore[union-attr]
|
||||
Route("/api/_internal/mcp-reload", internal_mcp_reload, methods=["POST"]),
|
||||
Route(
|
||||
"/api/_internal/mcp-refresh/{name}",
|
||||
internal_mcp_refresh_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/_internal/mcp-reconnect/{name}",
|
||||
internal_mcp_reconnect_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -889,3 +917,414 @@ class TestInternalMcpReloadEndpoint:
|
||||
assert data["added"] == ["a"]
|
||||
assert data["removed"] == ["b"]
|
||||
assert data["updated"] == ["c"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _notify_nodes_mcp_refresh_one / _notify_nodes_mcp_reconnect_one
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNotifyNodesMcpRefreshOne:
|
||||
@pytest.mark.anyio
|
||||
async def test_returns_json_on_success(self):
|
||||
client = AsyncMock()
|
||||
client.post.return_value = _mock_resp(200, {"status": "ok"})
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_refresh_one(req, "srv")
|
||||
assert result == {"n1": {"status": "ok"}}
|
||||
# Verify the URL used the safe-encoded name segment
|
||||
call_args = client.post.call_args
|
||||
assert call_args[0][0].endswith("/v1/api/_internal/mcp-refresh/srv")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_skips_nodes_without_url(self):
|
||||
client = AsyncMock()
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": ""},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_refresh_one(req, "srv")
|
||||
assert result == {}
|
||||
client.post.assert_not_called()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_records_error_on_exception(self):
|
||||
client = AsyncMock()
|
||||
client.post.side_effect = ConnectionError("refused")
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_refresh_one(req, "srv")
|
||||
assert "n1" in result
|
||||
assert "error" in result["n1"]
|
||||
assert "refused" in result["n1"]["error"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multiple_nodes_mixed(self):
|
||||
client = AsyncMock()
|
||||
client.post.side_effect = [
|
||||
_mock_resp(200, {"status": "ok"}),
|
||||
TimeoutError("timeout"),
|
||||
]
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
{"node_id": "n2", "server_url": "http://n2:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_refresh_one(req, "srv")
|
||||
assert result["n1"] == {"status": "ok"}
|
||||
assert "error" in result["n2"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_empty_cluster(self):
|
||||
req = _fake_request()
|
||||
result = await _notify_nodes_mcp_refresh_one(req, "srv")
|
||||
assert result == {}
|
||||
|
||||
|
||||
class TestNotifyNodesMcpReconnectOne:
|
||||
@pytest.mark.anyio
|
||||
async def test_returns_json_on_success(self):
|
||||
client = AsyncMock()
|
||||
client.post.return_value = _mock_resp(200, {"status": "ok"})
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_reconnect_one(req, "srv")
|
||||
assert result == {"n1": {"status": "ok"}}
|
||||
call_args = client.post.call_args
|
||||
assert call_args[0][0].endswith("/v1/api/_internal/mcp-reconnect/srv")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_skips_nodes_without_url(self):
|
||||
client = AsyncMock()
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": ""},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_reconnect_one(req, "srv")
|
||||
assert result == {}
|
||||
client.post.assert_not_called()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_records_error_on_exception(self):
|
||||
client = AsyncMock()
|
||||
client.post.side_effect = ConnectionError("refused")
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_reconnect_one(req, "srv")
|
||||
assert "n1" in result
|
||||
assert "error" in result["n1"]
|
||||
assert "refused" in result["n1"]["error"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multiple_nodes_mixed(self):
|
||||
client = AsyncMock()
|
||||
client.post.side_effect = [
|
||||
_mock_resp(200, {"status": "ok"}),
|
||||
TimeoutError("timeout"),
|
||||
]
|
||||
req = _fake_request(
|
||||
{"node_id": "n1", "server_url": "http://n1:8000"},
|
||||
{"node_id": "n2", "server_url": "http://n2:8000"},
|
||||
proxy_client=client,
|
||||
)
|
||||
result = await _notify_nodes_mcp_reconnect_one(req, "srv")
|
||||
assert result["n1"] == {"status": "ok"}
|
||||
assert "error" in result["n2"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_empty_cluster(self):
|
||||
req = _fake_request()
|
||||
result = await _notify_nodes_mcp_reconnect_one(req, "srv")
|
||||
assert result == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Console refresh / reconnect endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAdminMcpRefreshOneEndpoint:
|
||||
"""HTTP-level tests for the console refresh-one endpoint."""
|
||||
|
||||
def test_refresh_one_success(self, client: TestClient) -> None:
|
||||
with patch(
|
||||
"turnstone.console.server._notify_nodes_mcp_action",
|
||||
new_callable=AsyncMock,
|
||||
return_value={"n1": {"status": "ok"}},
|
||||
) as mock_notify:
|
||||
r = client.post("/v1/api/admin/mcp-servers/srv/refresh")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["results"] == {"n1": {"status": "ok"}}
|
||||
# The shared helper is called with the action verb.
|
||||
mock_notify.assert_awaited_once()
|
||||
args = mock_notify.await_args.args
|
||||
assert args[1] == "refresh"
|
||||
assert args[2] == "srv"
|
||||
|
||||
def test_refresh_one_permission_denied(self, client_no_perm: TestClient) -> None:
|
||||
r = client_no_perm.post("/v1/api/admin/mcp-servers/srv/refresh")
|
||||
assert r.status_code == 403
|
||||
assert "admin.mcp" in r.json()["error"]
|
||||
|
||||
def test_refresh_one_invalid_name(self, client: TestClient) -> None:
|
||||
# Names with '__' (reserved delimiter) are rejected.
|
||||
r = client.post("/v1/api/admin/mcp-servers/bad__name/refresh")
|
||||
assert r.status_code == 400
|
||||
assert "invalid" in r.json()["error"].lower()
|
||||
|
||||
|
||||
class TestAdminMcpReconnectOneEndpoint:
|
||||
"""HTTP-level tests for the console reconnect-one endpoint."""
|
||||
|
||||
def test_reconnect_one_success(self, client: TestClient) -> None:
|
||||
with patch(
|
||||
"turnstone.console.server._notify_nodes_mcp_action",
|
||||
new_callable=AsyncMock,
|
||||
return_value={"n1": {"status": "ok"}},
|
||||
) as mock_notify:
|
||||
r = client.post("/v1/api/admin/mcp-servers/srv/reconnect")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["results"] == {"n1": {"status": "ok"}}
|
||||
mock_notify.assert_awaited_once()
|
||||
args = mock_notify.await_args.args
|
||||
assert args[1] == "reconnect"
|
||||
assert args[2] == "srv"
|
||||
|
||||
def test_reconnect_one_permission_denied(self, client_no_perm: TestClient) -> None:
|
||||
r = client_no_perm.post("/v1/api/admin/mcp-servers/srv/reconnect")
|
||||
assert r.status_code == 403
|
||||
assert "admin.mcp" in r.json()["error"]
|
||||
|
||||
def test_reconnect_one_invalid_name(self, client: TestClient) -> None:
|
||||
r = client.post("/v1/api/admin/mcp-servers/bad__name/reconnect")
|
||||
assert r.status_code == 400
|
||||
assert "invalid" in r.json()["error"].lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Node refresh-one endpoint: POST /v1/api/_internal/mcp-refresh/{name}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestInternalMcpRefreshOneEndpoint:
|
||||
"""HTTP-level tests for the node-side per-server refresh endpoint."""
|
||||
|
||||
@pytest.fixture()
|
||||
def node_app_factory(self, storage: SQLiteBackend):
|
||||
"""Build a TestClient with an MCP client manager on app.state."""
|
||||
|
||||
def _make(mgr: Any) -> TestClient:
|
||||
app = Starlette(
|
||||
routes=_routes_with_internal(),
|
||||
middleware=[Middleware(_InjectAuthMiddleware)],
|
||||
)
|
||||
app.state.auth_storage = storage
|
||||
if mgr is not None:
|
||||
app.state.mcp_client = mgr
|
||||
return TestClient(app, raise_server_exceptions=False)
|
||||
|
||||
return _make
|
||||
|
||||
def test_refresh_one_success(self, node_app_factory) -> None:
|
||||
mgr = MagicMock()
|
||||
mgr.refresh_sync.return_value = None
|
||||
mgr.get_server_status.return_value = {
|
||||
"connected": True,
|
||||
"tools": 3,
|
||||
"resources": 0,
|
||||
"prompts": 1,
|
||||
"error": "",
|
||||
"transport": "stdio",
|
||||
"command": "/usr/bin/secret-stdio",
|
||||
"url": "",
|
||||
"circuit_open": False,
|
||||
"consecutive_failures": 0,
|
||||
}
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-refresh/srv")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["status"] == "ok"
|
||||
# sec-3: command/url stripped from response.
|
||||
assert "command" not in data["server"]
|
||||
assert "url" not in data["server"]
|
||||
assert data["server"]["tools"] == 3
|
||||
mgr.refresh_sync.assert_called_once_with(server_name="srv")
|
||||
|
||||
def test_refresh_one_no_mcp_client_returns_503(self, node_app_factory) -> None:
|
||||
c = node_app_factory(None)
|
||||
r = c.post("/v1/api/_internal/mcp-refresh/srv")
|
||||
assert r.status_code == 503
|
||||
assert r.json()["status"] == "error"
|
||||
|
||||
def test_refresh_one_raises_returns_500(self, node_app_factory) -> None:
|
||||
mgr = MagicMock()
|
||||
mgr.refresh_sync.side_effect = RuntimeError("internal stdio path /etc/shadow blew up")
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-refresh/srv")
|
||||
assert r.status_code == 500
|
||||
# sec-2: raw exception detail must not leak to the caller.
|
||||
body = r.json()
|
||||
assert body["error"] == "refresh failed"
|
||||
assert "shadow" not in body["error"]
|
||||
|
||||
def test_refresh_one_per_server_error_returns_500(self, node_app_factory) -> None:
|
||||
# q-3 / bug-3: refresh_sync swallows per-server errors into _last_error,
|
||||
# so a 200 from refresh_sync is not enough — get_server_status reports.
|
||||
mgr = MagicMock()
|
||||
mgr.refresh_sync.return_value = None
|
||||
mgr.get_server_status.return_value = {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "Refresh failed: connection refused",
|
||||
"transport": "stdio",
|
||||
"command": "secret",
|
||||
"url": "",
|
||||
"circuit_open": True,
|
||||
"consecutive_failures": 5,
|
||||
}
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-refresh/srv")
|
||||
assert r.status_code == 500
|
||||
data = r.json()
|
||||
assert data["status"] == "error"
|
||||
assert data["error"] == "refresh failed"
|
||||
# Public status echoed but command/url stripped.
|
||||
assert "command" not in data["server"]
|
||||
assert "url" not in data["server"]
|
||||
assert data["server"]["circuit_open"] is True
|
||||
|
||||
def test_refresh_one_invalid_name_returns_400(self, node_app_factory) -> None:
|
||||
# sec-4: name validation symmetric with console side.
|
||||
mgr = MagicMock()
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-refresh/bad__name")
|
||||
assert r.status_code == 400
|
||||
assert "invalid" in r.json()["error"].lower()
|
||||
mgr.refresh_sync.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Node reconnect-one endpoint: POST /v1/api/_internal/mcp-reconnect/{name}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestInternalMcpReconnectOneEndpoint:
|
||||
"""HTTP-level tests for the node-side per-server reconnect endpoint."""
|
||||
|
||||
@pytest.fixture()
|
||||
def node_app_factory(self, storage: SQLiteBackend):
|
||||
"""Build a TestClient with an MCP client manager on app.state."""
|
||||
|
||||
def _make(mgr: Any) -> TestClient:
|
||||
app = Starlette(
|
||||
routes=_routes_with_internal(),
|
||||
middleware=[Middleware(_InjectAuthMiddleware)],
|
||||
)
|
||||
app.state.auth_storage = storage
|
||||
if mgr is not None:
|
||||
app.state.mcp_client = mgr
|
||||
return TestClient(app, raise_server_exceptions=False)
|
||||
|
||||
return _make
|
||||
|
||||
def test_reconnect_one_success(self, node_app_factory) -> None:
|
||||
mgr = MagicMock()
|
||||
mgr.reconnect_sync.return_value = {
|
||||
"connected": True,
|
||||
"tools": 2,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "",
|
||||
}
|
||||
mgr.get_server_status.return_value = {
|
||||
"connected": True,
|
||||
"tools": 2,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "",
|
||||
"transport": "stdio",
|
||||
"command": "secret-cmd",
|
||||
"url": "",
|
||||
"circuit_open": False,
|
||||
"consecutive_failures": 0,
|
||||
}
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-reconnect/srv")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["status"] == "ok"
|
||||
# sec-3: command/url stripped from response.
|
||||
assert "command" not in data["server"]
|
||||
assert "url" not in data["server"]
|
||||
mgr.reconnect_sync.assert_called_once_with("srv")
|
||||
|
||||
def test_reconnect_one_no_mcp_client_returns_503(self, node_app_factory) -> None:
|
||||
c = node_app_factory(None)
|
||||
r = c.post("/v1/api/_internal/mcp-reconnect/srv")
|
||||
assert r.status_code == 503
|
||||
|
||||
def test_reconnect_one_returns_error_dict_500(self, node_app_factory) -> None:
|
||||
mgr = MagicMock()
|
||||
mgr.reconnect_sync.return_value = {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "secret stdio at /etc/shadow timed out",
|
||||
}
|
||||
mgr.get_server_status.return_value = {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "secret stdio at /etc/shadow timed out",
|
||||
"transport": "stdio",
|
||||
"command": "secret",
|
||||
"url": "",
|
||||
"circuit_open": False,
|
||||
"consecutive_failures": 1,
|
||||
}
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-reconnect/srv")
|
||||
assert r.status_code == 500
|
||||
body = r.json()
|
||||
# The top-level `error` is generic; the inner `server.error` echoes
|
||||
# whatever ``get_server_status`` returned (still admin-facing).
|
||||
assert body["error"] == "reconnect failed"
|
||||
assert "command" not in body["server"]
|
||||
assert "url" not in body["server"]
|
||||
|
||||
def test_reconnect_one_raises_returns_500(self, node_app_factory) -> None:
|
||||
mgr = MagicMock()
|
||||
mgr.reconnect_sync.side_effect = RuntimeError("internal stdio /etc/shadow blew up")
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-reconnect/srv")
|
||||
assert r.status_code == 500
|
||||
body = r.json()
|
||||
assert body["error"] == "reconnect failed"
|
||||
assert "shadow" not in body["error"]
|
||||
|
||||
def test_reconnect_one_invalid_name_returns_400(self, node_app_factory) -> None:
|
||||
# sec-4: name validation symmetric with console side.
|
||||
mgr = MagicMock()
|
||||
c = node_app_factory(mgr)
|
||||
r = c.post("/v1/api/_internal/mcp-reconnect/bad__name")
|
||||
assert r.status_code == 400
|
||||
assert "invalid" in r.json()["error"].lower()
|
||||
mgr.reconnect_sync.assert_not_called()
|
||||
|
||||
+191
-47
@@ -130,6 +130,31 @@ def _fake_prompt_dict(
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def running_loop_mgr():
|
||||
"""Yield a (mgr, loop, thread) triple with a background loop already running.
|
||||
|
||||
Spawns an MCPClientManager with a default config of {"srv": stdio echo}
|
||||
and a fresh asyncio loop driven by a daemon thread. Tests that need a
|
||||
different config can mutate ``mgr._server_configs`` directly. The
|
||||
fixture stops the loop and joins the thread on teardown so each test
|
||||
leaves a clean slate.
|
||||
"""
|
||||
import threading as _threading
|
||||
|
||||
cfg = {"srv": {"type": "stdio", "command": "echo"}}
|
||||
mgr = MCPClientManager(cfg)
|
||||
loop = asyncio.new_event_loop()
|
||||
thread = _threading.Thread(target=loop.run_forever, daemon=True)
|
||||
thread.start()
|
||||
mgr._loop = loop
|
||||
try:
|
||||
yield mgr, loop, thread
|
||||
finally:
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
thread.join(timeout=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Schema conversion
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -1903,60 +1928,179 @@ class TestNotificationDebounce:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fix 5: Periodic refresh backoff
|
||||
# reconnect_sync — operator-driven full reconnect
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPeriodicRefreshBackoff:
|
||||
"""Verify periodic refresh backoff and auto-reconnect."""
|
||||
class TestReconnectSync:
|
||||
"""Verify reconnect_sync tears down old session, clears CB, calls _connect_one."""
|
||||
|
||||
def test_backoff_set_on_failure(self):
|
||||
def test_reconnect_unknown_server_returns_error(self):
|
||||
mgr = MCPClientManager({})
|
||||
mgr._refresh_failures["srv"] = 1
|
||||
# Simulate what _periodic_refresh does on failure
|
||||
failures = mgr._refresh_failures.get("srv", 0) + 1
|
||||
mgr._refresh_failures["srv"] = failures
|
||||
backoff = min(mgr._REFRESH_BACKOFF_BASE * (2 ** (failures - 1)), mgr._REFRESH_BACKOFF_MAX)
|
||||
mgr._refresh_backoff_until["srv"] = time.monotonic() + backoff
|
||||
assert mgr._refresh_backoff_until["srv"] > time.monotonic()
|
||||
assert failures == 2
|
||||
result = mgr.reconnect_sync("missing")
|
||||
assert result == {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "unknown server",
|
||||
}
|
||||
|
||||
def test_backoff_doubles(self):
|
||||
mgr = MCPClientManager({})
|
||||
b1 = min(mgr._REFRESH_BACKOFF_BASE * (2**0), mgr._REFRESH_BACKOFF_MAX)
|
||||
b2 = min(mgr._REFRESH_BACKOFF_BASE * (2**1), mgr._REFRESH_BACKOFF_MAX)
|
||||
b3 = min(mgr._REFRESH_BACKOFF_BASE * (2**2), mgr._REFRESH_BACKOFF_MAX)
|
||||
assert b1 == 60
|
||||
assert b2 == 120
|
||||
assert b3 == 240
|
||||
def test_reconnect_clears_circuit_breaker(self, running_loop_mgr):
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
def test_backoff_capped(self):
|
||||
mgr = MCPClientManager({})
|
||||
b = min(mgr._REFRESH_BACKOFF_BASE * (2**20), mgr._REFRESH_BACKOFF_MAX)
|
||||
assert b == mgr._REFRESH_BACKOFF_MAX
|
||||
async def _fake_connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
mgr._sessions[name] = MagicMock()
|
||||
|
||||
def test_backoff_clears_on_success(self):
|
||||
mgr = MCPClientManager({})
|
||||
mgr._refresh_failures["srv"] = 3
|
||||
mgr._refresh_backoff_until["srv"] = time.monotonic() + 1000
|
||||
# Simulate success
|
||||
mgr._refresh_failures.pop("srv", None)
|
||||
mgr._refresh_backoff_until.pop("srv", None)
|
||||
assert "srv" not in mgr._refresh_failures
|
||||
assert "srv" not in mgr._refresh_backoff_until
|
||||
|
||||
def test_server_status_includes_circuit_info(self):
|
||||
mgr = MCPClientManager({"srv": {"type": "stdio", "command": "echo"}})
|
||||
status = mgr.get_server_status("srv")
|
||||
assert "circuit_open" in status
|
||||
assert "consecutive_failures" in status
|
||||
assert status["circuit_open"] is False
|
||||
assert status["consecutive_failures"] == 0
|
||||
|
||||
def test_server_status_shows_open_circuit(self):
|
||||
mgr = MCPClientManager({"srv": {"type": "stdio", "command": "echo"}})
|
||||
# Pre-trip the breaker
|
||||
for _ in range(3):
|
||||
mgr._cb_record_failure("srv")
|
||||
status = mgr.get_server_status("srv")
|
||||
assert status["circuit_open"] is True
|
||||
assert status["consecutive_failures"] == 3
|
||||
assert "srv" in mgr._circuit_open_until
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_connect_one", side_effect=_fake_connect_one),
|
||||
patch.object(mgr, "_pre_close_streams", new=AsyncMock()),
|
||||
):
|
||||
result = mgr.reconnect_sync("srv")
|
||||
assert result["connected"] is True
|
||||
assert result["error"] == ""
|
||||
assert "srv" not in mgr._circuit_open_until
|
||||
assert "srv" not in mgr._consecutive_failures
|
||||
|
||||
def test_reconnect_closes_old_session_then_calls_connect_one(self, running_loop_mgr):
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
order: list[str] = []
|
||||
old_stack = MagicMock(spec=AsyncExitStack)
|
||||
|
||||
async def _pre_close(name: str) -> None:
|
||||
order.append("pre_close")
|
||||
|
||||
async def _safe_close(stack: Any) -> None:
|
||||
order.append("safe_close")
|
||||
assert stack is old_stack
|
||||
|
||||
async def _connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
order.append("connect_one")
|
||||
mgr._sessions[name] = MagicMock()
|
||||
|
||||
mgr._sessions["srv"] = MagicMock() # populated old session
|
||||
mgr._per_server_stacks["srv"] = old_stack
|
||||
mgr._server_streams["srv"] = (MagicMock(), MagicMock())
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_pre_close_streams", side_effect=_pre_close),
|
||||
patch.object(mgr, "_safe_close_stack", side_effect=_safe_close),
|
||||
patch.object(mgr, "_connect_one", side_effect=_connect_one),
|
||||
):
|
||||
result = mgr.reconnect_sync("srv")
|
||||
assert result["connected"] is True
|
||||
assert order == ["pre_close", "safe_close", "connect_one"]
|
||||
assert "srv" not in mgr._per_server_stacks
|
||||
|
||||
def test_reconnect_failure_returns_error_dict(self, running_loop_mgr):
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
async def _connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
raise RuntimeError("handshake failed")
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_connect_one", side_effect=_connect_one),
|
||||
patch.object(mgr, "_pre_close_streams", new=AsyncMock()),
|
||||
):
|
||||
result = mgr.reconnect_sync("srv")
|
||||
assert result["connected"] is False
|
||||
assert "handshake failed" in result["error"]
|
||||
|
||||
def test_reconnect_failure_clears_stale_catalog(self, running_loop_mgr):
|
||||
# bug-2: when _connect_one fails mid-reconnect, the per-server
|
||||
# catalog must be dropped so the merged tool/resource/prompt maps
|
||||
# don't keep advertising entries with no live session.
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
# Seed catalog state from a previous successful connect.
|
||||
mgr._per_server_tools["srv"] = [_fake_openai_tool("mcp__srv__t")]
|
||||
mgr._per_server_resources["srv"] = [_fake_resource_dict(server="srv")]
|
||||
mgr._per_server_prompts["srv"] = [_fake_prompt_dict(server="srv")]
|
||||
mgr._rebuild_tools()
|
||||
mgr._rebuild_resources()
|
||||
mgr._rebuild_prompts()
|
||||
|
||||
async def _connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
raise RuntimeError("handshake failed")
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_connect_one", side_effect=_connect_one),
|
||||
patch.object(mgr, "_pre_close_streams", new=AsyncMock()),
|
||||
):
|
||||
result = mgr.reconnect_sync("srv")
|
||||
assert result["connected"] is False
|
||||
# Per-server catalog and merged maps should both be empty for srv.
|
||||
assert "srv" not in mgr._per_server_tools
|
||||
assert "srv" not in mgr._per_server_resources
|
||||
assert "srv" not in mgr._per_server_prompts
|
||||
assert "mcp__srv__t" not in mgr._tool_map
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _cb_auto_reconnect — refresh-on-reconnect
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCBAutoReconnectRefresh:
|
||||
"""Verify _cb_auto_reconnect schedules catalog refresh after a successful reconnect.
|
||||
|
||||
The refresh runs as a fire-and-forget background task on the loop so it
|
||||
doesn't block the caller (perf-1). Tests wait briefly for the scheduled
|
||||
task to run and observe its effect.
|
||||
"""
|
||||
|
||||
def test_auto_reconnect_schedules_refresh_server_on_success(self, running_loop_mgr):
|
||||
import threading as _threading
|
||||
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
new_session = MagicMock()
|
||||
refresh_event = _threading.Event()
|
||||
|
||||
async def _connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
mgr._sessions[name] = new_session
|
||||
|
||||
async def _refresh(name: str) -> tuple[list[str], list[str]]:
|
||||
refresh_event.set()
|
||||
return [], []
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_connect_one", side_effect=_connect_one),
|
||||
patch.object(mgr, "_refresh_server", side_effect=_refresh),
|
||||
):
|
||||
session = mgr._cb_auto_reconnect("srv")
|
||||
# Wait for the scheduled refresh task to actually run on the loop.
|
||||
assert refresh_event.wait(timeout=5), "refresh task was not scheduled"
|
||||
assert session is new_session
|
||||
|
||||
def test_auto_reconnect_swallows_refresh_failure(self, running_loop_mgr):
|
||||
import threading as _threading
|
||||
|
||||
mgr, _loop, _thread = running_loop_mgr
|
||||
|
||||
new_session = MagicMock()
|
||||
refresh_started = _threading.Event()
|
||||
|
||||
async def _connect_one(name: str, _cfg: dict[str, Any]) -> None:
|
||||
mgr._sessions[name] = new_session
|
||||
|
||||
async def _refresh_failing(name: str) -> tuple[list[str], list[str]]:
|
||||
refresh_started.set()
|
||||
raise RuntimeError("catalog fetch broke")
|
||||
|
||||
with (
|
||||
patch.object(mgr, "_connect_one", side_effect=_connect_one),
|
||||
patch.object(mgr, "_refresh_server", side_effect=_refresh_failing),
|
||||
):
|
||||
# Must not raise — refresh failures are non-fatal.
|
||||
session = mgr._cb_auto_reconnect("srv")
|
||||
# Background refresh actually started and exception was swallowed
|
||||
# by the task without affecting the synchronous caller.
|
||||
assert refresh_started.wait(timeout=5), "refresh task was not scheduled"
|
||||
assert session is new_session
|
||||
|
||||
@@ -117,7 +117,6 @@
|
||||
|
||||
[mcp]
|
||||
# config_path = "" # Path to MCP servers config file (JSON)
|
||||
# refresh_interval = 14400 # Refresh interval in seconds (default: 4h)
|
||||
|
||||
# --- Server (node, console) ---
|
||||
|
||||
|
||||
@@ -1015,15 +1015,6 @@ def main() -> None:
|
||||
help="Path to MCP server config file (standard mcpServers JSON format)",
|
||||
)
|
||||
|
||||
from turnstone.core.config import nonneg_float
|
||||
|
||||
parser.add_argument(
|
||||
"--mcp-refresh-interval",
|
||||
type=nonneg_float,
|
||||
default=14400,
|
||||
metavar="SECONDS",
|
||||
help="Periodic MCP tool refresh interval for servers without push notifications (default: 14400 = 4h, 0 to disable)",
|
||||
)
|
||||
judge_group = parser.add_argument_group("Judge options")
|
||||
judge_group.add_argument(
|
||||
"--judge",
|
||||
@@ -1136,7 +1127,6 @@ def main() -> None:
|
||||
|
||||
mcp_client = create_mcp_client(
|
||||
getattr(args, "mcp_config", None),
|
||||
refresh_interval=getattr(args, "mcp_refresh_interval", 14400),
|
||||
storage=_get_storage(),
|
||||
)
|
||||
|
||||
|
||||
@@ -8388,6 +8388,57 @@ async def _notify_nodes_mcp_reload(request: Request) -> dict[str, Any]:
|
||||
return {nid: data for nid, data in results if data is not None}
|
||||
|
||||
|
||||
async def _notify_nodes_mcp_action(request: Request, action: str, name: str) -> dict[str, Any]:
|
||||
"""Tell all nodes to perform a per-server MCP action.
|
||||
|
||||
*action* is the suffix of the internal endpoint — currently
|
||||
``"refresh"`` or ``"reconnect"``. Each node is hit at
|
||||
``/v1/api/_internal/mcp-{action}/{name}``.
|
||||
"""
|
||||
collector: ClusterCollector = request.app.state.collector
|
||||
nodes = collector.get_all_nodes()
|
||||
client: httpx.AsyncClient = request.app.state.proxy_client
|
||||
headers = _proxy_auth_headers(request)
|
||||
sem = asyncio.Semaphore(_get_fan_out_limit(request))
|
||||
safe_name = urllib.parse.quote(name, safe="")
|
||||
|
||||
async def _notify(node: dict[str, Any]) -> tuple[str, Any]:
|
||||
node_id = node.get("node_id", "")
|
||||
url = node.get("server_url", "")
|
||||
if not url:
|
||||
return node_id, None
|
||||
async with sem:
|
||||
try:
|
||||
resp = await client.post(
|
||||
f"{url.rstrip('/')}/v1/api/_internal/mcp-{action}/{safe_name}",
|
||||
headers=headers,
|
||||
timeout=30,
|
||||
)
|
||||
return node_id, resp.json()
|
||||
except Exception as exc:
|
||||
log.debug(
|
||||
"Failed to notify node %s for MCP %s of %s",
|
||||
node_id,
|
||||
action,
|
||||
name,
|
||||
exc_info=True,
|
||||
)
|
||||
return node_id, {"error": str(exc)}
|
||||
|
||||
results = await asyncio.gather(*[_notify(n) for n in nodes])
|
||||
return {nid: data for nid, data in results if data is not None}
|
||||
|
||||
|
||||
async def _notify_nodes_mcp_refresh_one(request: Request, name: str) -> dict[str, Any]:
|
||||
"""Tell all nodes to refresh a single MCP server's catalog."""
|
||||
return await _notify_nodes_mcp_action(request, "refresh", name)
|
||||
|
||||
|
||||
async def _notify_nodes_mcp_reconnect_one(request: Request, name: str) -> dict[str, Any]:
|
||||
"""Tell all nodes to force-reconnect a single MCP server."""
|
||||
return await _notify_nodes_mcp_action(request, "reconnect", name)
|
||||
|
||||
|
||||
async def admin_mcp_reload(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/admin/mcp-servers/reload — tell nodes to re-read DB."""
|
||||
from turnstone.core.auth import require_permission
|
||||
@@ -8404,6 +8455,57 @@ async def admin_mcp_reload(request: Request) -> JSONResponse:
|
||||
return JSONResponse({"status": "ok", "results": results})
|
||||
|
||||
|
||||
async def _admin_mcp_action(request: Request, action: str) -> JSONResponse:
|
||||
"""Shared body for the per-server MCP admin actions.
|
||||
|
||||
*action* is the verb suffix — ``"refresh"`` or ``"reconnect"``.
|
||||
Auth, name validation, audit, and per-node fan-out are identical
|
||||
across the two; only the audit action string and the notify call
|
||||
vary, both keyed on *action*.
|
||||
"""
|
||||
from turnstone.core.audit import record_audit
|
||||
from turnstone.core.auth import require_permission
|
||||
from turnstone.core.web_helpers import require_storage_or_503
|
||||
|
||||
storage, err = require_storage_or_503(request)
|
||||
if err:
|
||||
return err
|
||||
err = require_permission(request, "admin.mcp")
|
||||
if err:
|
||||
return err
|
||||
|
||||
name = request.path_params["name"]
|
||||
if "__" in name:
|
||||
return JSONResponse({"error": "invalid server name"}, status_code=400)
|
||||
|
||||
existing = storage.get_mcp_server_by_name(name)
|
||||
target_id = existing.get("id", name) if existing else name
|
||||
|
||||
audit_uid, ip = _audit_context(request)
|
||||
record_audit(
|
||||
storage,
|
||||
audit_uid,
|
||||
f"mcp_server.{action}",
|
||||
"mcp_server",
|
||||
target_id,
|
||||
{"name": name},
|
||||
ip,
|
||||
)
|
||||
|
||||
results = await _notify_nodes_mcp_action(request, action, name)
|
||||
return JSONResponse({"status": "ok", "results": results})
|
||||
|
||||
|
||||
async def admin_mcp_refresh_one(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/admin/mcp-servers/{name}/refresh — refresh one server's catalog."""
|
||||
return await _admin_mcp_action(request, "refresh")
|
||||
|
||||
|
||||
async def admin_mcp_reconnect_one(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/admin/mcp-servers/{name}/reconnect — force-reconnect one server."""
|
||||
return await _admin_mcp_action(request, "reconnect")
|
||||
|
||||
|
||||
async def admin_import_mcp_config(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/admin/mcp-servers/import — import from pasted JSON config."""
|
||||
import uuid
|
||||
@@ -11198,6 +11300,16 @@ def create_app(
|
||||
admin_mcp_reload,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{name}/refresh",
|
||||
admin_mcp_refresh_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{name}/reconnect",
|
||||
admin_mcp_reconnect_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/admin/mcp-servers/{server_id}",
|
||||
admin_get_mcp_server,
|
||||
|
||||
@@ -3316,9 +3316,17 @@ function _renderMcpServers(items) {
|
||||
var detailAttr = isConfig
|
||||
? 'data-mcp-detail-name="' + escapeHtml(s.name) + '"'
|
||||
: 'data-mcp-detail="' + escapeHtml(s.server_id) + '"';
|
||||
var actionBtns =
|
||||
'<button class="admin-btn-action" data-mcp-refresh="' +
|
||||
escapeHtml(s.name) +
|
||||
'">refresh</button>' +
|
||||
'<button class="admin-btn-action" data-mcp-reconnect="' +
|
||||
escapeHtml(s.name) +
|
||||
'">reconnect</button>';
|
||||
var actions = isConfig
|
||||
? ""
|
||||
: '<button class="admin-btn-action" data-mcp-edit="' +
|
||||
? actionBtns
|
||||
: actionBtns +
|
||||
'<button class="admin-btn-action" data-mcp-edit="' +
|
||||
escapeHtml(s.server_id) +
|
||||
'">edit</button>' +
|
||||
'<button class="admin-btn-danger" data-mcp-delete="' +
|
||||
@@ -3383,6 +3391,46 @@ function _renderMcpServers(items) {
|
||||
showEditMcpModal(this.getAttribute("data-mcp-edit"));
|
||||
});
|
||||
});
|
||||
el.querySelectorAll("[data-mcp-refresh]").forEach(function (btn) {
|
||||
btn.addEventListener("click", function () {
|
||||
var name = this.getAttribute("data-mcp-refresh");
|
||||
authFetch(
|
||||
"/v1/api/admin/mcp-servers/" + encodeURIComponent(name) + "/refresh",
|
||||
{ method: "POST" },
|
||||
)
|
||||
.then(function (r) {
|
||||
if (!r.ok) throw new Error();
|
||||
return r.json();
|
||||
})
|
||||
.then(function () {
|
||||
showToast("Refreshed " + name);
|
||||
loadAdminMcp();
|
||||
})
|
||||
.catch(function () {
|
||||
showToast("Failed to refresh " + name);
|
||||
});
|
||||
});
|
||||
});
|
||||
el.querySelectorAll("[data-mcp-reconnect]").forEach(function (btn) {
|
||||
btn.addEventListener("click", function () {
|
||||
var name = this.getAttribute("data-mcp-reconnect");
|
||||
authFetch(
|
||||
"/v1/api/admin/mcp-servers/" + encodeURIComponent(name) + "/reconnect",
|
||||
{ method: "POST" },
|
||||
)
|
||||
.then(function (r) {
|
||||
if (!r.ok) throw new Error();
|
||||
return r.json();
|
||||
})
|
||||
.then(function () {
|
||||
showToast("Reconnected " + name);
|
||||
loadAdminMcp();
|
||||
})
|
||||
.catch(function () {
|
||||
showToast("Failed to reconnect " + name);
|
||||
});
|
||||
});
|
||||
});
|
||||
el.querySelectorAll("[data-mcp-delete]").forEach(function (btn) {
|
||||
btn.addEventListener("click", function () {
|
||||
var sid = this.getAttribute("data-mcp-delete");
|
||||
|
||||
@@ -533,6 +533,13 @@ def required_scope(method: str, path: str) -> str:
|
||||
if method == "POST" and normalized in APPROVE_PATHS:
|
||||
return "approve"
|
||||
|
||||
# Path-keyed internal admin actions: /api/_internal/mcp-{refresh,reconnect}/{name}
|
||||
if method == "POST" and (
|
||||
normalized.startswith("/api/_internal/mcp-refresh/")
|
||||
or normalized.startswith("/api/_internal/mcp-reconnect/")
|
||||
):
|
||||
return "approve"
|
||||
|
||||
# Write endpoints
|
||||
if method == "POST" and normalized in WRITE_PATHS:
|
||||
return "write"
|
||||
@@ -594,6 +601,10 @@ def required_scope(method: str, path: str) -> str:
|
||||
if proxied:
|
||||
if proxied in APPROVE_PATHS:
|
||||
return "approve"
|
||||
if proxied.startswith("/api/_internal/mcp-refresh/") or proxied.startswith(
|
||||
"/api/_internal/mcp-reconnect/"
|
||||
):
|
||||
return "approve"
|
||||
if proxied in WRITE_PATHS:
|
||||
return "write"
|
||||
# Parametric workstream sub-resource mutations
|
||||
|
||||
@@ -124,7 +124,6 @@ _CONFIG_MAP: dict[str, dict[str, str]] = {
|
||||
},
|
||||
"mcp": {
|
||||
"config_path": "mcp_config",
|
||||
"refresh_interval": "mcp_refresh_interval",
|
||||
},
|
||||
"ratelimit": {
|
||||
"enabled": "ratelimit_enabled",
|
||||
|
||||
+95
-133
@@ -8,12 +8,10 @@ synchronous. We bridge the two by running a dedicated asyncio event loop
|
||||
in a daemon thread. ``call_tool_sync`` dispatches coroutines onto that loop
|
||||
via ``asyncio.run_coroutine_threadsafe``.
|
||||
|
||||
Refresh: three mechanisms keep tool/resource/prompt lists up-to-date:
|
||||
Refresh: two mechanisms keep tool/resource/prompt lists up-to-date:
|
||||
1. Push notifications — servers declaring ``listChanged`` on the
|
||||
respective capability trigger immediate refresh.
|
||||
2. Periodic timer — servers *without* push support are polled on a
|
||||
staggered interval (configurable, default 4 h, seeded at launch).
|
||||
3. Manual — ``/mcp refresh [server]`` triggers ``refresh_sync()``.
|
||||
2. Manual — ``/mcp refresh [server]`` triggers ``refresh_sync()``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -22,7 +20,6 @@ import asyncio
|
||||
import concurrent.futures
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import threading
|
||||
import time
|
||||
@@ -44,8 +41,6 @@ from turnstone.core.log import get_logger
|
||||
|
||||
log = get_logger("turnstone.mcp")
|
||||
|
||||
_DEFAULT_REFRESH_INTERVAL: float = 14400 # 4 hours
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCP ↔ OpenAI schema conversion
|
||||
@@ -88,12 +83,8 @@ class MCPClientManager:
|
||||
def __init__(
|
||||
self,
|
||||
server_configs: dict[str, dict[str, Any]],
|
||||
*,
|
||||
refresh_interval: float = _DEFAULT_REFRESH_INTERVAL,
|
||||
) -> None:
|
||||
self._server_configs = server_configs
|
||||
if refresh_interval < 0:
|
||||
refresh_interval = 0.0
|
||||
self._loop: asyncio.AbstractEventLoop | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self._exit_stack: AsyncExitStack | None = None
|
||||
@@ -147,10 +138,6 @@ class MCPClientManager:
|
||||
self._storage: Any = None
|
||||
self._sync_lock = threading.Lock()
|
||||
|
||||
# Periodic refresh for servers without push notifications
|
||||
self._refresh_interval = refresh_interval
|
||||
self._refresh_task: asyncio.Task[None] | None = None
|
||||
|
||||
# Circuit breaker (per-server) — prevents repeated calls to broken servers
|
||||
self._consecutive_failures: dict[str, int] = {}
|
||||
self._circuit_open_until: dict[str, float] = {} # monotonic timestamp
|
||||
@@ -163,10 +150,6 @@ class MCPClientManager:
|
||||
# Notification debounce (per-server)
|
||||
self._last_notification_refresh: dict[str, float] = {}
|
||||
|
||||
# Periodic refresh backoff (per-server)
|
||||
self._refresh_failures: dict[str, int] = {}
|
||||
self._refresh_backoff_until: dict[str, float] = {} # monotonic timestamp
|
||||
|
||||
# -- lifecycle -----------------------------------------------------------
|
||||
|
||||
def start(self) -> None:
|
||||
@@ -201,22 +184,6 @@ class MCPClientManager:
|
||||
|
||||
self._connected.set()
|
||||
|
||||
# Start periodic refresh for servers without push notifications
|
||||
needs_periodic = any(
|
||||
not self._supports_list_changed.get(name, False)
|
||||
or (
|
||||
self._supports_resources.get(name, False)
|
||||
and not self._supports_resource_list_changed.get(name, False)
|
||||
)
|
||||
or (
|
||||
self._supports_prompts.get(name, False)
|
||||
and not self._supports_prompt_list_changed.get(name, False)
|
||||
)
|
||||
for name in self._sessions
|
||||
)
|
||||
if needs_periodic and self._refresh_interval > 0:
|
||||
self._refresh_task = asyncio.get_running_loop().create_task(self._periodic_refresh())
|
||||
|
||||
_CONNECT_TIMEOUT = 30 # seconds — prevents hung connections on broken remotes
|
||||
_TCP_PROBE_TIMEOUT = 5 # seconds — fast TCP pre-flight for HTTP transports
|
||||
|
||||
@@ -228,10 +195,6 @@ class MCPClientManager:
|
||||
# Notification debounce
|
||||
_NOTIFICATION_DEBOUNCE = 5.0 # seconds between refreshes per server
|
||||
|
||||
# Periodic refresh backoff
|
||||
_REFRESH_BACKOFF_BASE = 60.0 # seconds
|
||||
_REFRESH_BACKOFF_MAX = 3600.0 # 1 hour
|
||||
|
||||
# -- circuit breaker (per-server) -----------------------------------------
|
||||
|
||||
def _cb_check(self, name: str) -> tuple[bool, bool]:
|
||||
@@ -666,9 +629,12 @@ class MCPClientManager:
|
||||
Returns ``(added_tools, removed_tools)`` names (tool diff only,
|
||||
for backward compatibility with ``/mcp refresh`` output).
|
||||
"""
|
||||
added, removed = await self._refresh_server_tools(name)
|
||||
await self._refresh_server_resources(name)
|
||||
await self._refresh_server_prompts(name)
|
||||
tool_diff, _, _ = await asyncio.gather(
|
||||
self._refresh_server_tools(name),
|
||||
self._refresh_server_resources(name),
|
||||
self._refresh_server_prompts(name),
|
||||
)
|
||||
added, removed = tool_diff
|
||||
self._last_error.pop(name, None)
|
||||
return added, removed
|
||||
|
||||
@@ -728,85 +694,6 @@ class MCPClientManager:
|
||||
future.cancel()
|
||||
raise TimeoutError(f"MCP refresh timed out after {timeout}s") from None
|
||||
|
||||
async def _periodic_refresh(self) -> None:
|
||||
"""Periodically refresh servers that lack push notifications.
|
||||
|
||||
Applies per-server exponential backoff on failure and attempts
|
||||
reconnection for disconnected servers.
|
||||
"""
|
||||
# Stagger start using a launch-time seed so cluster nodes don't
|
||||
# all hit MCP servers simultaneously.
|
||||
seed = random.Random(time.monotonic_ns() ^ os.getpid()).random()
|
||||
initial_delay = seed * self._refresh_interval
|
||||
await asyncio.sleep(initial_delay)
|
||||
while True:
|
||||
for name in list(self._server_configs):
|
||||
now = time.monotonic()
|
||||
|
||||
# Check per-server backoff
|
||||
backoff_until = self._refresh_backoff_until.get(name, 0.0)
|
||||
if now < backoff_until:
|
||||
continue # still in backoff
|
||||
|
||||
if name not in self._sessions:
|
||||
# Attempt reconnection for disconnected servers
|
||||
cfg = self._server_configs.get(name)
|
||||
if cfg:
|
||||
try:
|
||||
log.info("Periodic reconnect attempt for '%s'", name)
|
||||
await self._connect_one(name, cfg)
|
||||
self._refresh_failures.pop(name, None)
|
||||
self._refresh_backoff_until.pop(name, None)
|
||||
self._cb_record_success(name)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
failures = self._refresh_failures.get(name, 0) + 1
|
||||
self._refresh_failures[name] = failures
|
||||
backoff = min(
|
||||
self._REFRESH_BACKOFF_BASE * (2 ** (failures - 1)),
|
||||
self._REFRESH_BACKOFF_MAX,
|
||||
)
|
||||
self._refresh_backoff_until[name] = time.monotonic() + backoff
|
||||
log.warning(
|
||||
"Periodic reconnect failed for '%s' (attempt %d, backoff %.0fs)",
|
||||
name,
|
||||
failures,
|
||||
backoff,
|
||||
)
|
||||
self._set_error(name, f"Reconnect failed: {exc}")
|
||||
continue
|
||||
|
||||
try:
|
||||
if not self._supports_list_changed.get(name, False):
|
||||
await self._refresh_server_tools(name)
|
||||
if not self._supports_resource_list_changed.get(name, False):
|
||||
await self._refresh_server_resources(name)
|
||||
if not self._supports_prompt_list_changed.get(name, False):
|
||||
await self._refresh_server_prompts(name)
|
||||
self._last_error.pop(name, None)
|
||||
self._refresh_failures.pop(name, None)
|
||||
self._refresh_backoff_until.pop(name, None)
|
||||
except Exception as exc:
|
||||
failures = self._refresh_failures.get(name, 0) + 1
|
||||
self._refresh_failures[name] = failures
|
||||
backoff = min(
|
||||
self._REFRESH_BACKOFF_BASE * (2 ** (failures - 1)),
|
||||
self._REFRESH_BACKOFF_MAX,
|
||||
)
|
||||
self._refresh_backoff_until[name] = time.monotonic() + backoff
|
||||
log.warning(
|
||||
"Periodic refresh failed for '%s' (attempt %d, backoff %.0fs)",
|
||||
name,
|
||||
failures,
|
||||
backoff,
|
||||
)
|
||||
self._set_error(name, f"Periodic refresh failed: {exc}")
|
||||
# Note: per-server backoff (max 1h) is only meaningful when
|
||||
# refresh_interval is shorter than _REFRESH_BACKOFF_MAX. With
|
||||
# the default 4h interval this sleep already bounds retry frequency.
|
||||
await asyncio.sleep(self._refresh_interval)
|
||||
|
||||
# -- resource refresh ----------------------------------------------------
|
||||
|
||||
def _rebuild_resources(self) -> None:
|
||||
@@ -1142,10 +1029,6 @@ class MCPClientManager:
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""Close all MCP sessions and stop the background loop."""
|
||||
# Cancel periodic refresh
|
||||
if self._refresh_task and self._loop:
|
||||
self._loop.call_soon_threadsafe(self._refresh_task.cancel)
|
||||
|
||||
# Close all per-server stacks (transports + sessions)
|
||||
if self._loop and self._per_server_stacks:
|
||||
|
||||
@@ -1204,8 +1087,6 @@ class MCPClientManager:
|
||||
self._circuit_trip_count.clear()
|
||||
self._server_streams.clear()
|
||||
self._last_notification_refresh.clear()
|
||||
self._refresh_failures.clear()
|
||||
self._refresh_backoff_until.clear()
|
||||
|
||||
log.info("MCP client shut down")
|
||||
|
||||
@@ -1252,6 +1133,77 @@ class MCPClientManager:
|
||||
"error": "",
|
||||
}
|
||||
|
||||
def reconnect_sync(self, name: str, timeout: int = 30) -> dict[str, Any]:
|
||||
"""Force a fresh connection to an MCP server (blocks the calling thread).
|
||||
|
||||
Tears down the current session/transport (if any), clears the circuit
|
||||
breaker, and runs a new ``_connect_one``. Returns status dict with
|
||||
keys: connected, tools, resources, prompts, error (parity with
|
||||
``add_server_sync``).
|
||||
"""
|
||||
if name not in self._server_configs:
|
||||
return {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "unknown server",
|
||||
}
|
||||
if self._loop is None:
|
||||
return {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": "MCP event loop not running",
|
||||
}
|
||||
|
||||
cfg = self._server_configs[name]
|
||||
|
||||
async def _reconnect() -> None:
|
||||
self._cb_clear(name)
|
||||
self._sessions.pop(name, None)
|
||||
await self._pre_close_streams(name)
|
||||
stack = self._per_server_stacks.pop(name, None)
|
||||
if stack is not None:
|
||||
await self._safe_close_stack(stack)
|
||||
try:
|
||||
await self._connect_one(name, cfg)
|
||||
except Exception:
|
||||
# Connect failed mid-reconnect — drop the stale per-server
|
||||
# catalog so the merged tool/resource/prompt maps don't keep
|
||||
# advertising entries with no live session behind them.
|
||||
self._per_server_tools.pop(name, None)
|
||||
self._per_server_resources.pop(name, None)
|
||||
self._per_server_prompts.pop(name, None)
|
||||
self._rebuild_tools()
|
||||
self._rebuild_resources()
|
||||
self._rebuild_prompts()
|
||||
raise
|
||||
|
||||
future = asyncio.run_coroutine_threadsafe(_reconnect(), self._loop)
|
||||
try:
|
||||
future.result(timeout=timeout)
|
||||
except concurrent.futures.TimeoutError:
|
||||
future.cancel()
|
||||
return {
|
||||
"connected": False,
|
||||
"tools": 0,
|
||||
"resources": 0,
|
||||
"prompts": 0,
|
||||
"error": f"MCP server '{name}' reconnect timed out",
|
||||
}
|
||||
except Exception as exc:
|
||||
return {"connected": False, "tools": 0, "resources": 0, "prompts": 0, "error": str(exc)}
|
||||
|
||||
return {
|
||||
"connected": name in self._sessions,
|
||||
"tools": len(self._per_server_tools.get(name, [])),
|
||||
"resources": len(self._per_server_resources.get(name, [])),
|
||||
"prompts": len(self._per_server_prompts.get(name, [])),
|
||||
"error": "",
|
||||
}
|
||||
|
||||
def remove_server_sync(self, name: str, timeout: int = 15) -> bool:
|
||||
"""Disconnect and remove an MCP server at runtime (blocks the calling thread).
|
||||
|
||||
@@ -1285,8 +1237,6 @@ class MCPClientManager:
|
||||
self._supports_prompt_list_changed.pop(name, None)
|
||||
self._last_error.pop(name, None)
|
||||
self._last_notification_refresh.pop(name, None)
|
||||
self._refresh_failures.pop(name, None)
|
||||
self._refresh_backoff_until.pop(name, None)
|
||||
self._cb_clear(name)
|
||||
# Rebuild merged state (serialized with notification handlers)
|
||||
self._rebuild_tools()
|
||||
@@ -1312,8 +1262,6 @@ class MCPClientManager:
|
||||
self._supports_prompt_list_changed.pop(name, None)
|
||||
self._last_error.pop(name, None)
|
||||
self._last_notification_refresh.pop(name, None)
|
||||
self._refresh_failures.pop(name, None)
|
||||
self._refresh_backoff_until.pop(name, None)
|
||||
self._cb_clear(name)
|
||||
self._rebuild_tools()
|
||||
self._rebuild_resources()
|
||||
@@ -1527,6 +1475,21 @@ class MCPClientManager:
|
||||
if session is None:
|
||||
self._cb_record_failure(server_name)
|
||||
raise RuntimeError(f"MCP server '{server_name}' reconnect produced no session")
|
||||
|
||||
# Schedule catalog refresh on the loop without blocking the caller.
|
||||
# The reconnected session is valid for the imminent dispatch; catalog
|
||||
# drift will be reconciled on the loop in the background.
|
||||
def _schedule_refresh() -> None:
|
||||
try:
|
||||
asyncio.create_task(self._refresh_server(server_name))
|
||||
except Exception:
|
||||
log.warning(
|
||||
"Catalog refresh after reconnect failed for '%s'",
|
||||
server_name,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
self._loop.call_soon_threadsafe(_schedule_refresh)
|
||||
return session
|
||||
|
||||
def call_tool_sync(
|
||||
@@ -1794,7 +1757,6 @@ def load_mcp_config(
|
||||
def create_mcp_client(
|
||||
config_path: str | None = None,
|
||||
*,
|
||||
refresh_interval: float = _DEFAULT_REFRESH_INTERVAL,
|
||||
storage: Any = None,
|
||||
) -> MCPClientManager | None:
|
||||
"""Create and start an MCP client manager.
|
||||
@@ -1815,7 +1777,7 @@ def create_mcp_client(
|
||||
if not servers:
|
||||
return None
|
||||
|
||||
mgr = MCPClientManager(servers, refresh_interval=refresh_interval)
|
||||
mgr = MCPClientManager(servers)
|
||||
# Mark DB-sourced servers so reconcile_sync won't remove config-file servers
|
||||
mgr._db_managed = {name for name in servers if name in db_names}
|
||||
mgr.start()
|
||||
|
||||
@@ -340,14 +340,6 @@ def _build_registry() -> dict[str, SettingDef]:
|
||||
"Tip: use the MCP Servers tab to manage servers via the database instead.",
|
||||
reference_url="https://modelcontextprotocol.io",
|
||||
),
|
||||
SettingDef(
|
||||
"mcp.refresh_interval",
|
||||
"int",
|
||||
14400,
|
||||
"MCP resource/prompt refresh interval in seconds (0 = disabled, default 4h)",
|
||||
"mcp",
|
||||
min_value=0,
|
||||
),
|
||||
SettingDef(
|
||||
"mcp.registry_url",
|
||||
"str",
|
||||
|
||||
+95
-1
@@ -2850,6 +2850,91 @@ def internal_mcp_status(request: Request) -> JSONResponse:
|
||||
return JSONResponse({"servers": mcp_mgr.get_all_server_status()})
|
||||
|
||||
|
||||
_SERVER_STATUS_PUBLIC_KEYS: tuple[str, ...] = (
|
||||
"connected",
|
||||
"tools",
|
||||
"resources",
|
||||
"prompts",
|
||||
"error",
|
||||
"transport",
|
||||
"circuit_open",
|
||||
"consecutive_failures",
|
||||
)
|
||||
|
||||
|
||||
def _public_server_status(mcp_mgr: Any, name: str) -> dict[str, Any]:
|
||||
"""Strip ``command``/``url`` from ``get_server_status`` before returning over the wire.
|
||||
|
||||
The full status dict embeds stdio argv and remote URLs that are
|
||||
admin-only context. Internal node endpoints surface only the
|
||||
operational fields callers need to reflect to the operator.
|
||||
"""
|
||||
full = mcp_mgr.get_server_status(name)
|
||||
return {k: full[k] for k in _SERVER_STATUS_PUBLIC_KEYS if k in full}
|
||||
|
||||
|
||||
def internal_mcp_refresh_one(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/_internal/mcp-refresh/{name} — refresh a single MCP server's catalog."""
|
||||
name = request.path_params["name"]
|
||||
if "__" in name:
|
||||
return JSONResponse({"status": "error", "error": "invalid name"}, status_code=400)
|
||||
mcp_mgr = getattr(request.app.state, "mcp_client", None)
|
||||
if mcp_mgr is None:
|
||||
return JSONResponse({"status": "error", "error": "MCP client not running"}, status_code=503)
|
||||
|
||||
try:
|
||||
mcp_mgr.refresh_sync(server_name=name)
|
||||
except Exception as exc:
|
||||
log.warning("internal_mcp_refresh_one failed for %s: %s", name, exc)
|
||||
return JSONResponse({"status": "error", "error": "refresh failed"}, status_code=500)
|
||||
|
||||
# _refresh_all swallows per-server errors into _last_error rather than
|
||||
# raising, so a 200-OK from refresh_sync isn't enough — re-check status
|
||||
# and surface 500 if the refresh actually failed for this server.
|
||||
status = _public_server_status(mcp_mgr, name)
|
||||
if status.get("error"):
|
||||
log.warning(
|
||||
"internal_mcp_refresh_one: refresh reported error for %s: %s", name, status["error"]
|
||||
)
|
||||
return JSONResponse(
|
||||
{"status": "error", "error": "refresh failed", "server": status},
|
||||
status_code=500,
|
||||
)
|
||||
return JSONResponse({"status": "ok", "server": status})
|
||||
|
||||
|
||||
def internal_mcp_reconnect_one(request: Request) -> JSONResponse:
|
||||
"""POST /v1/api/_internal/mcp-reconnect/{name} — force-reconnect a single MCP server."""
|
||||
name = request.path_params["name"]
|
||||
if "__" in name:
|
||||
return JSONResponse({"status": "error", "error": "invalid name"}, status_code=400)
|
||||
mcp_mgr = getattr(request.app.state, "mcp_client", None)
|
||||
if mcp_mgr is None:
|
||||
return JSONResponse({"status": "error", "error": "MCP client not running"}, status_code=503)
|
||||
|
||||
try:
|
||||
result = mcp_mgr.reconnect_sync(name)
|
||||
except Exception as exc:
|
||||
log.warning("internal_mcp_reconnect_one failed for %s: %s", name, exc)
|
||||
return JSONResponse({"status": "error", "error": "reconnect failed"}, status_code=500)
|
||||
|
||||
if result.get("error"):
|
||||
log.warning(
|
||||
"internal_mcp_reconnect_one: reconnect reported error for %s: %s",
|
||||
name,
|
||||
result.get("error", ""),
|
||||
)
|
||||
return JSONResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"error": "reconnect failed",
|
||||
"server": _public_server_status(mcp_mgr, name),
|
||||
},
|
||||
status_code=500,
|
||||
)
|
||||
return JSONResponse({"status": "ok", "server": _public_server_status(mcp_mgr, name)})
|
||||
|
||||
|
||||
# -- internal model management -----------------------------------------------
|
||||
|
||||
|
||||
@@ -3790,6 +3875,16 @@ def create_app(
|
||||
Route("/api/_internal/config-reload", config_reload, methods=["POST"]),
|
||||
Route("/api/_internal/mcp-reload", internal_mcp_reload, methods=["POST"]),
|
||||
Route("/api/_internal/mcp-status", internal_mcp_status),
|
||||
Route(
|
||||
"/api/_internal/mcp-refresh/{name}",
|
||||
internal_mcp_refresh_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/_internal/mcp-reconnect/{name}",
|
||||
internal_mcp_reconnect_one,
|
||||
methods=["POST"],
|
||||
),
|
||||
Route(
|
||||
"/api/_internal/model-reload",
|
||||
internal_model_reload,
|
||||
@@ -4052,7 +4147,6 @@ def main() -> None:
|
||||
mcp_config_cli = args.mcp_config # CLI-only (no config.toml for this)
|
||||
mcp_client = create_mcp_client(
|
||||
mcp_config_cli or config_store.get("mcp.config_path") or None,
|
||||
refresh_interval=config_store.get("mcp.refresh_interval"),
|
||||
storage=_get_storage(),
|
||||
)
|
||||
# Mutable ref so session_factory always sees the latest MCP client,
|
||||
|
||||
Reference in New Issue
Block a user