error handling pro max
This commit is contained in:
@@ -8,9 +8,14 @@ from dataclasses import dataclass
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Generic, TypeVar
|
from typing import Any, Generic, TypeVar
|
||||||
|
|
||||||
|
import anyio
|
||||||
|
import httpx
|
||||||
from fastmcp import FastMCP
|
from fastmcp import FastMCP
|
||||||
from fastmcp.client.transports.config import MCPConfigTransport
|
from fastmcp.client.transports.config import MCPConfigTransport
|
||||||
from fastmcp.server.providers.proxy import FastMCPProxy, StatefulProxyClient
|
from fastmcp.server.providers.proxy import FastMCPProxy, StatefulProxyClient
|
||||||
|
from mcp.client.streamable_http import StreamableHTTPError
|
||||||
|
from mcp.shared.exceptions import McpError
|
||||||
|
|
||||||
from ..models import BrokerConfig, ConnectionConfig
|
from ..models import BrokerConfig, ConnectionConfig
|
||||||
from ..proxy_results import ResourceLinkNamespace
|
from ..proxy_results import ResourceLinkNamespace
|
||||||
from ..proxy_config import broker_config_to_fastmcp_config
|
from ..proxy_config import broker_config_to_fastmcp_config
|
||||||
@@ -22,6 +27,16 @@ ProxyMountFactory = Callable[[ConnectionConfig, Path], "ProxyMount[ProxyT]"]
|
|||||||
# whole broker. Eight seconds is intentionally longer than normal local stdio
|
# whole broker. Eight seconds is intentionally longer than normal local stdio
|
||||||
# startup/handshake time, but short enough to make a broken source visible.
|
# startup/handshake time, but short enough to make a broken source visible.
|
||||||
_PROXY_LIST_TIMEOUT_SECONDS = 8.0
|
_PROXY_LIST_TIMEOUT_SECONDS = 8.0
|
||||||
|
_PROXY_LIST_FAILURES = (
|
||||||
|
TimeoutError,
|
||||||
|
OSError,
|
||||||
|
anyio.ClosedResourceError,
|
||||||
|
anyio.EndOfStream,
|
||||||
|
anyio.BrokenResourceError,
|
||||||
|
httpx.HTTPError,
|
||||||
|
McpError,
|
||||||
|
StreamableHTTPError,
|
||||||
|
)
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@@ -183,7 +198,7 @@ async def _bounded_proxy_list(
|
|||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
return await asyncio.wait_for(listing, timeout=timeout_seconds)
|
return await asyncio.wait_for(listing, timeout=timeout_seconds)
|
||||||
except (TimeoutError, OSError, ConnectionError) as exc:
|
except _PROXY_LIST_FAILURES as exc:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Skipping %s for connection %s after %s listing failure: %s",
|
"Skipping %s for connection %s after %s listing failure: %s",
|
||||||
operation,
|
operation,
|
||||||
|
|||||||
@@ -6,8 +6,11 @@ import logging
|
|||||||
import sys
|
import sys
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
import anyio
|
||||||
|
import httpx
|
||||||
import mcp.types as mcp_types
|
import mcp.types as mcp_types
|
||||||
import pytest
|
import pytest
|
||||||
|
from mcp.shared.exceptions import McpError
|
||||||
|
|
||||||
from wf_mcp.broker import load_broker_config
|
from wf_mcp.broker import load_broker_config
|
||||||
from wf_mcp.events import EventBus, InMemoryEventSink
|
from wf_mcp.events import EventBus, InMemoryEventSink
|
||||||
@@ -166,6 +169,47 @@ def test_proxy_listing_degrades_when_one_source_has_connection_error(
|
|||||||
assert "tools/list" in caplog.text
|
assert "tools/list" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("exc", "expected_log_name"),
|
||||||
|
[
|
||||||
|
(
|
||||||
|
McpError(
|
||||||
|
mcp_types.ErrorData(
|
||||||
|
code=mcp_types.INTERNAL_ERROR,
|
||||||
|
message="connection closed",
|
||||||
|
)
|
||||||
|
),
|
||||||
|
"McpError",
|
||||||
|
),
|
||||||
|
(anyio.ClosedResourceError(), "ClosedResourceError"),
|
||||||
|
(anyio.EndOfStream(), "EndOfStream"),
|
||||||
|
(httpx.ConnectError("connection refused"), "ConnectError"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_proxy_listing_degrades_when_session_transport_closes(
|
||||||
|
exc: Exception,
|
||||||
|
expected_log_name: str,
|
||||||
|
caplog: pytest.LogCaptureFixture,
|
||||||
|
) -> None:
|
||||||
|
async def failed_listing() -> list[Any]:
|
||||||
|
raise exc
|
||||||
|
|
||||||
|
async def run_failure() -> None:
|
||||||
|
result = await _bounded_proxy_list(
|
||||||
|
failed_listing(),
|
||||||
|
connection_id="remote.default",
|
||||||
|
operation="tools/list",
|
||||||
|
timeout_seconds=1,
|
||||||
|
)
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
with caplog.at_level(logging.WARNING, logger="wf_mcp.proxy.mounts"):
|
||||||
|
asyncio.run(run_failure())
|
||||||
|
|
||||||
|
assert expected_log_name in caplog.text
|
||||||
|
assert "remote.default" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
def test_proxy_registers_admin_tools_on_local_provider() -> None:
|
def test_proxy_registers_admin_tools_on_local_provider() -> None:
|
||||||
config = BrokerConfig(
|
config = BrokerConfig(
|
||||||
store_root=local_temp_root() / "proxy_local_admin_store",
|
store_root=local_temp_root() / "proxy_local_admin_store",
|
||||||
|
|||||||
Reference in New Issue
Block a user