Max Isbey
200a6070db
_handle_request is now the single exception-to-wire boundary: - MCPError -> JSONRPCError(e.error) - pydantic ValidationError -> INVALID_PARAMS - Exception -> INTERNAL_ERROR(str(e)), logged, optionally re-raised - outer-cancel (run() TG shutdown) -> shielded REQUEST_CANCELLED write, re-raise - peer-cancel (notifications/cancelled) -> scope swallows, no response written dctx.close() runs in an inner finally so the back-channel shuts the moment the handler exits. _write_result/_write_error swallow Broken/ClosedResourceError so a dropped connection during the response write doesn't crash the dispatcher. All 22 contract tests now pass against both DirectDispatcher and JSONRPCDispatcher; chunk-c xfail markers removed.
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""Shared fixtures for `Dispatcher` contract tests.
|
|
|
|
The `pair_factory` fixture parametrizes contract tests over every `Dispatcher`
|
|
implementation, so the same behavioral assertions run against `DirectDispatcher`
|
|
(in-memory) and `JSONRPCDispatcher` (over crossed anyio memory streams).
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
|
|
from mcp.shared.dispatcher import Dispatcher
|
|
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
|
from mcp.shared.message import SessionMessage
|
|
from mcp.shared.transport_context import TransportContext
|
|
|
|
DispatcherTriple = tuple[Dispatcher[TransportContext], Dispatcher[TransportContext], Callable[[], None]]
|
|
PairFactory = Callable[..., DispatcherTriple]
|
|
|
|
|
|
def direct_pair(*, can_send_request: bool = True) -> DispatcherTriple:
|
|
client, server = create_direct_dispatcher_pair(can_send_request=can_send_request)
|
|
|
|
def close() -> None:
|
|
client.close()
|
|
server.close()
|
|
|
|
return client, server, close
|
|
|
|
|
|
def jsonrpc_pair(*, can_send_request: bool = True) -> DispatcherTriple:
|
|
"""Two `JSONRPCDispatcher`s wired over crossed in-memory streams."""
|
|
c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
|
|
s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
|
|
|
|
def builder(_rid: object, _meta: object) -> TransportContext:
|
|
return TransportContext(kind="jsonrpc", can_send_request=can_send_request)
|
|
|
|
client: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(s2c_recv, c2s_send, transport_builder=builder)
|
|
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send, transport_builder=builder)
|
|
|
|
def close() -> None:
|
|
for s in (c2s_send, c2s_recv, s2c_send, s2c_recv):
|
|
s.close()
|
|
|
|
return client, server, close
|
|
|
|
|
|
@pytest.fixture(
|
|
params=[
|
|
pytest.param(direct_pair, id="direct"),
|
|
pytest.param(jsonrpc_pair, id="jsonrpc"),
|
|
]
|
|
)
|
|
def pair_factory(request: pytest.FixtureRequest) -> PairFactory:
|
|
return request.param
|
|
|
|
|
|
__all__ = ["PairFactory", "direct_pair", "jsonrpc_pair"]
|