refactor: rename Dispatcher.call to send_request, replace RequestSender with Outbound
The design doc's `send_request = call` alias only makes the concrete class satisfy RequestSender, not the abstract Dispatcher Protocol — so any consumer typed against `Dispatcher[TT]` (Connection, ServerRunner) couldn't pass it to something expecting a RequestSender without a cast or hand-written bridge. RequestSender was also half a contract: every implementor (Dispatcher, DispatchContext, Connection, Context) has `notify` too, and PeerMixin needs both for its typed sugar (elicit/sample are requests, log is a notification). Outbound(Protocol) declares both methods; Dispatcher and DispatchContext extend it. PeerMixin will wrap an Outbound. One verb everywhere, no aliases, no extra Protocols. - Dispatcher.call -> send_request - OnCall -> OnRequest, on_call -> on_request - RequestSender -> Outbound (now also declares notify) - Dispatcher(Outbound, Protocol[TT]), DispatchContext(Outbound, Protocol[TT])
This commit is contained in:
Max Isbey
@@ -1,7 +1,7 @@
|
||||
"""In-memory `Dispatcher` that wires two peers together with no transport.
|
||||
|
||||
`DirectDispatcher` is the simplest possible `Dispatcher` implementation: a call
|
||||
on one side directly invokes the other side's `on_call`. There is no
|
||||
`DirectDispatcher` is the simplest possible `Dispatcher` implementation: a
|
||||
request on one side directly invokes the other side's `on_request`. There is no
|
||||
serialization, no JSON-RPC framing, and no streams. It exists to:
|
||||
|
||||
* prove the `Dispatcher` Protocol is implementable without JSON-RPC
|
||||
@@ -21,7 +21,7 @@ from typing import Any
|
||||
|
||||
import anyio
|
||||
|
||||
from mcp.shared.dispatcher import CallOptions, OnCall, OnNotify, ProgressFnT
|
||||
from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest, ProgressFnT
|
||||
from mcp.shared.exceptions import MCPError, NoBackChannelError
|
||||
from mcp.shared.transport_context import TransportContext
|
||||
from mcp.types import INTERNAL_ERROR, REQUEST_TIMEOUT
|
||||
@@ -31,20 +31,20 @@ __all__ = ["DirectDispatcher", "create_direct_dispatcher_pair"]
|
||||
DIRECT_TRANSPORT_KIND = "direct"
|
||||
|
||||
|
||||
_Call = Callable[[str, Mapping[str, Any] | None, CallOptions | None], Awaitable[dict[str, Any]]]
|
||||
_Request = Callable[[str, Mapping[str, Any] | None, CallOptions | None], Awaitable[dict[str, Any]]]
|
||||
_Notify = Callable[[str, Mapping[str, Any] | None], Awaitable[None]]
|
||||
|
||||
|
||||
@dataclass
|
||||
class _DirectDispatchContext:
|
||||
"""`DispatchContext` for an inbound call on a `DirectDispatcher`.
|
||||
"""`DispatchContext` for an inbound request on a `DirectDispatcher`.
|
||||
|
||||
The back-channel callables target the *originating* side, so a handler's
|
||||
`send_request` reaches the peer that made the inbound call.
|
||||
`send_request` reaches the peer that made the inbound request.
|
||||
"""
|
||||
|
||||
transport: TransportContext
|
||||
_back_call: _Call
|
||||
_back_request: _Request
|
||||
_back_notify: _Notify
|
||||
_on_progress: ProgressFnT | None = None
|
||||
cancel_requested: anyio.Event = field(default_factory=anyio.Event)
|
||||
@@ -60,7 +60,7 @@ class _DirectDispatchContext:
|
||||
) -> dict[str, Any]:
|
||||
if not self.transport.can_send_request:
|
||||
raise NoBackChannelError(method)
|
||||
return await self._back_call(method, params, opts)
|
||||
return await self._back_request(method, params, opts)
|
||||
|
||||
async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
|
||||
if self._on_progress is not None:
|
||||
@@ -71,14 +71,14 @@ class DirectDispatcher:
|
||||
"""A `Dispatcher` that calls a peer's handlers directly, in-process.
|
||||
|
||||
Two instances are wired together with `create_direct_dispatcher_pair`; each
|
||||
holds a reference to the other. `call` on one awaits the peer's `on_call`.
|
||||
`run` parks until `close` is called.
|
||||
holds a reference to the other. `send_request` on one awaits the peer's
|
||||
`on_request`. `run` parks until `close` is called.
|
||||
"""
|
||||
|
||||
def __init__(self, transport_ctx: TransportContext):
|
||||
self._transport_ctx = transport_ctx
|
||||
self._peer: DirectDispatcher | None = None
|
||||
self._on_call: OnCall | None = None
|
||||
self._on_request: OnRequest | None = None
|
||||
self._on_notify: OnNotify | None = None
|
||||
self._ready = anyio.Event()
|
||||
self._closed = anyio.Event()
|
||||
@@ -86,7 +86,7 @@ class DirectDispatcher:
|
||||
def connect_to(self, peer: DirectDispatcher) -> None:
|
||||
self._peer = peer
|
||||
|
||||
async def call(
|
||||
async def send_request(
|
||||
self,
|
||||
method: str,
|
||||
params: Mapping[str, Any] | None,
|
||||
@@ -94,15 +94,15 @@ class DirectDispatcher:
|
||||
) -> dict[str, Any]:
|
||||
if self._peer is None:
|
||||
raise RuntimeError("DirectDispatcher has no peer; use create_direct_dispatcher_pair()")
|
||||
return await self._peer._dispatch_call(method, params, opts)
|
||||
return await self._peer._dispatch_request(method, params, opts)
|
||||
|
||||
async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
|
||||
if self._peer is None:
|
||||
raise RuntimeError("DirectDispatcher has no peer; use create_direct_dispatcher_pair()")
|
||||
await self._peer._dispatch_notify(method, params)
|
||||
|
||||
async def run(self, on_call: OnCall, on_notify: OnNotify) -> None:
|
||||
self._on_call = on_call
|
||||
async def run(self, on_request: OnRequest, on_notify: OnNotify) -> None:
|
||||
self._on_request = on_request
|
||||
self._on_notify = on_notify
|
||||
self._ready.set()
|
||||
await self._closed.wait()
|
||||
@@ -115,25 +115,25 @@ class DirectDispatcher:
|
||||
peer = self._peer
|
||||
return _DirectDispatchContext(
|
||||
transport=self._transport_ctx,
|
||||
_back_call=lambda m, p, o: peer._dispatch_call(m, p, o),
|
||||
_back_request=lambda m, p, o: peer._dispatch_request(m, p, o),
|
||||
_back_notify=lambda m, p: peer._dispatch_notify(m, p),
|
||||
_on_progress=on_progress,
|
||||
)
|
||||
|
||||
async def _dispatch_call(
|
||||
async def _dispatch_request(
|
||||
self,
|
||||
method: str,
|
||||
params: Mapping[str, Any] | None,
|
||||
opts: CallOptions | None,
|
||||
) -> dict[str, Any]:
|
||||
await self._ready.wait()
|
||||
assert self._on_call is not None
|
||||
assert self._on_request is not None
|
||||
opts = opts or {}
|
||||
dctx = self._make_context(on_progress=opts.get("on_progress"))
|
||||
try:
|
||||
with anyio.fail_after(opts.get("timeout")):
|
||||
try:
|
||||
return await self._on_call(dctx, method, params)
|
||||
return await self._on_request(dctx, method, params)
|
||||
except MCPError:
|
||||
raise
|
||||
except Exception as e:
|
||||
|
||||
@@ -2,9 +2,9 @@
|
||||
|
||||
A Dispatcher turns a duplex message channel into two things:
|
||||
|
||||
* an outbound API: ``call(method, params)`` and ``notify(method, params)``
|
||||
* an inbound pump: ``run(on_call, on_notify)`` that drives the receive loop and
|
||||
invokes the supplied handlers for each incoming request/notification
|
||||
* an outbound API: ``send_request(method, params)`` and ``notify(method, params)``
|
||||
* an inbound pump: ``run(on_request, on_notify)`` that drives the receive loop
|
||||
and invokes the supplied handlers for each incoming request/notification
|
||||
|
||||
It is deliberately *not* MCP-aware. Method names are strings, params and
|
||||
results are ``dict[str, Any]``. The MCP type layer (request/result models,
|
||||
@@ -28,23 +28,23 @@ __all__ = [
|
||||
"DispatchContext",
|
||||
"DispatchMiddleware",
|
||||
"Dispatcher",
|
||||
"OnCall",
|
||||
"OnNotify",
|
||||
"OnRequest",
|
||||
"Outbound",
|
||||
"ProgressFnT",
|
||||
"RequestSender",
|
||||
]
|
||||
|
||||
TransportT_co = TypeVar("TransportT_co", bound=TransportContext, covariant=True)
|
||||
|
||||
|
||||
class ProgressFnT(Protocol):
|
||||
"""Callback invoked when a progress notification arrives for a pending call."""
|
||||
"""Callback invoked when a progress notification arrives for a pending request."""
|
||||
|
||||
async def __call__(self, progress: float, total: float | None, message: str | None) -> None: ...
|
||||
|
||||
|
||||
class CallOptions(TypedDict, total=False):
|
||||
"""Per-call options for `RequestSender.send_request` / `Dispatcher.call`.
|
||||
"""Per-call options for `Outbound.send_request`.
|
||||
|
||||
All keys are optional. Dispatchers ignore keys they do not understand.
|
||||
"""
|
||||
@@ -53,21 +53,22 @@ class CallOptions(TypedDict, total=False):
|
||||
"""Seconds to wait for a result before raising and sending ``notifications/cancelled``."""
|
||||
|
||||
on_progress: ProgressFnT
|
||||
"""Receive ``notifications/progress`` updates for this call."""
|
||||
"""Receive ``notifications/progress`` updates for this request."""
|
||||
|
||||
resumption_token: str
|
||||
"""Opaque token to resume a previously interrupted call (transport-dependent)."""
|
||||
"""Opaque token to resume a previously interrupted request (transport-dependent)."""
|
||||
|
||||
on_resumption_token: Callable[[str], Awaitable[None]]
|
||||
"""Receive a resumption token when the transport issues one."""
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class RequestSender(Protocol):
|
||||
"""Anything that can send a request and await its result.
|
||||
class Outbound(Protocol):
|
||||
"""Anything that can send requests and notifications to the peer.
|
||||
|
||||
`DispatchContext` satisfies this; `PeerMixin` (and `Connection`/`Peer`) wrap
|
||||
a `RequestSender` to provide typed request methods.
|
||||
Both `Dispatcher` (top-level outbound) and `DispatchContext` (back-channel
|
||||
during an inbound request) extend this. `PeerMixin` wraps an `Outbound` to
|
||||
provide typed MCP request/notification methods.
|
||||
"""
|
||||
|
||||
async def send_request(
|
||||
@@ -75,74 +76,6 @@ class RequestSender(Protocol):
|
||||
method: str,
|
||||
params: Mapping[str, Any] | None,
|
||||
opts: CallOptions | None = None,
|
||||
) -> dict[str, Any]: ...
|
||||
|
||||
|
||||
class DispatchContext(Protocol[TransportT_co]):
|
||||
"""Per-request context handed to ``on_call`` / ``on_notify``.
|
||||
|
||||
Carries the transport metadata for the inbound message and provides the
|
||||
back-channel for sending requests/notifications to the peer while handling
|
||||
it.
|
||||
"""
|
||||
|
||||
@property
|
||||
def transport(self) -> TransportT_co:
|
||||
"""Transport-specific metadata for this inbound message."""
|
||||
...
|
||||
|
||||
@property
|
||||
def cancel_requested(self) -> anyio.Event:
|
||||
"""Set when the peer sends ``notifications/cancelled`` for this request."""
|
||||
...
|
||||
|
||||
async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
|
||||
"""Send a notification to the peer."""
|
||||
...
|
||||
|
||||
async def send_request(
|
||||
self,
|
||||
method: str,
|
||||
params: Mapping[str, Any] | None,
|
||||
opts: CallOptions | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Send a request to the peer on the back-channel and await its result.
|
||||
|
||||
Raises:
|
||||
NoBackChannelError: if ``transport.can_send_request`` is ``False``.
|
||||
"""
|
||||
...
|
||||
|
||||
async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
|
||||
"""Report progress for the inbound request, if the peer supplied a progress token.
|
||||
|
||||
A no-op when no token was supplied.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
OnCall = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[dict[str, Any]]]
|
||||
"""Handler for inbound requests: ``(ctx, method, params) -> result``. Raise ``MCPError`` to send an error response."""
|
||||
|
||||
OnNotify = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[None]]
|
||||
"""Handler for inbound notifications: ``(ctx, method, params)``."""
|
||||
|
||||
DispatchMiddleware = Callable[[OnCall], OnCall]
|
||||
"""Wraps an ``OnCall`` to produce another ``OnCall``. Applied outermost-first."""
|
||||
|
||||
|
||||
class Dispatcher(Protocol[TransportT_co]):
|
||||
"""A duplex request/notification channel with call-return semantics.
|
||||
|
||||
Implementations own correlation of outbound calls to inbound results, the
|
||||
receive loop, per-request concurrency, and cancellation/progress wiring.
|
||||
"""
|
||||
|
||||
async def call(
|
||||
self,
|
||||
method: str,
|
||||
params: Mapping[str, Any] | None,
|
||||
opts: CallOptions | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Send a request and await its result.
|
||||
|
||||
@@ -157,11 +90,56 @@ class Dispatcher(Protocol[TransportT_co]):
|
||||
"""Send a fire-and-forget notification."""
|
||||
...
|
||||
|
||||
async def run(self, on_call: OnCall, on_notify: OnNotify) -> None:
|
||||
|
||||
class DispatchContext(Outbound, Protocol[TransportT_co]):
|
||||
"""Per-request context handed to ``on_request`` / ``on_notify``.
|
||||
|
||||
Carries the transport metadata for the inbound message and provides the
|
||||
back-channel for sending requests/notifications to the peer while handling
|
||||
it. `send_request` raises `NoBackChannelError` if
|
||||
``transport.can_send_request`` is ``False``.
|
||||
"""
|
||||
|
||||
@property
|
||||
def transport(self) -> TransportT_co:
|
||||
"""Transport-specific metadata for this inbound message."""
|
||||
...
|
||||
|
||||
@property
|
||||
def cancel_requested(self) -> anyio.Event:
|
||||
"""Set when the peer sends ``notifications/cancelled`` for this request."""
|
||||
...
|
||||
|
||||
async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
|
||||
"""Report progress for the inbound request, if the peer supplied a progress token.
|
||||
|
||||
A no-op when no token was supplied.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
OnRequest = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[dict[str, Any]]]
|
||||
"""Handler for inbound requests: ``(ctx, method, params) -> result``. Raise ``MCPError`` to send an error response."""
|
||||
|
||||
OnNotify = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[None]]
|
||||
"""Handler for inbound notifications: ``(ctx, method, params)``."""
|
||||
|
||||
DispatchMiddleware = Callable[[OnRequest], OnRequest]
|
||||
"""Wraps an ``OnRequest`` to produce another ``OnRequest``. Applied outermost-first."""
|
||||
|
||||
|
||||
class Dispatcher(Outbound, Protocol[TransportT_co]):
|
||||
"""A duplex request/notification channel with call-return semantics.
|
||||
|
||||
Implementations own correlation of outbound requests to inbound results, the
|
||||
receive loop, per-request concurrency, and cancellation/progress wiring.
|
||||
"""
|
||||
|
||||
async def run(self, on_request: OnRequest, on_notify: OnNotify) -> None:
|
||||
"""Drive the receive loop until the underlying channel closes.
|
||||
|
||||
Each inbound request is dispatched to ``on_call`` in its own task; the
|
||||
returned dict (or raised ``MCPError``) is sent back as the response.
|
||||
Each inbound request is dispatched to ``on_request`` in its own task;
|
||||
the returned dict (or raised ``MCPError``) is sent back as the response.
|
||||
Inbound notifications go to ``on_notify``.
|
||||
"""
|
||||
...
|
||||
|
||||
@@ -13,7 +13,7 @@ import anyio
|
||||
import pytest
|
||||
|
||||
from mcp.shared.direct_dispatcher import DirectDispatcher, create_direct_dispatcher_pair
|
||||
from mcp.shared.dispatcher import DispatchContext, Dispatcher, OnCall, OnNotify
|
||||
from mcp.shared.dispatcher import DispatchContext, Dispatcher, OnNotify, OnRequest, Outbound
|
||||
from mcp.shared.exceptions import MCPError, NoBackChannelError
|
||||
from mcp.shared.transport_context import TransportContext
|
||||
from mcp.types import INTERNAL_ERROR, INVALID_PARAMS, INVALID_REQUEST, REQUEST_TIMEOUT
|
||||
@@ -21,17 +21,17 @@ from mcp.types import INTERNAL_ERROR, INVALID_PARAMS, INVALID_REQUEST, REQUEST_T
|
||||
|
||||
class Recorder:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, Mapping[str, Any] | None]] = []
|
||||
self.requests: list[tuple[str, Mapping[str, Any] | None]] = []
|
||||
self.notifications: list[tuple[str, Mapping[str, Any] | None]] = []
|
||||
self.contexts: list[DispatchContext[TransportContext]] = []
|
||||
self.notified = anyio.Event()
|
||||
|
||||
|
||||
def echo_handlers(recorder: Recorder) -> tuple[OnCall, OnNotify]:
|
||||
async def on_call(
|
||||
def echo_handlers(recorder: Recorder) -> tuple[OnRequest, OnNotify]:
|
||||
async def on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
recorder.calls.append((method, params))
|
||||
recorder.requests.append((method, params))
|
||||
recorder.contexts.append(ctx)
|
||||
return {"echoed": method, "params": dict(params or {})}
|
||||
|
||||
@@ -39,26 +39,26 @@ def echo_handlers(recorder: Recorder) -> tuple[OnCall, OnNotify]:
|
||||
recorder.notifications.append((method, params))
|
||||
recorder.notified.set()
|
||||
|
||||
return on_call, on_notify
|
||||
return on_request, on_notify
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def running_pair(
|
||||
*,
|
||||
server_on_call: OnCall | None = None,
|
||||
server_on_request: OnRequest | None = None,
|
||||
server_on_notify: OnNotify | None = None,
|
||||
client_on_call: OnCall | None = None,
|
||||
client_on_request: OnRequest | None = None,
|
||||
client_on_notify: OnNotify | None = None,
|
||||
can_send_request: bool = True,
|
||||
) -> AsyncIterator[tuple[DirectDispatcher, DirectDispatcher, Recorder, Recorder]]:
|
||||
"""Yield ``(client, server, client_recorder, server_recorder)`` with both ``run()`` loops live."""
|
||||
client, server = create_direct_dispatcher_pair(can_send_request=can_send_request)
|
||||
client_rec, server_rec = Recorder(), Recorder()
|
||||
c_call, c_notify = echo_handlers(client_rec)
|
||||
s_call, s_notify = echo_handlers(server_rec)
|
||||
c_req, c_notify = echo_handlers(client_rec)
|
||||
s_req, s_notify = echo_handlers(server_rec)
|
||||
async with anyio.create_task_group() as tg:
|
||||
tg.start_soon(client.run, client_on_call or c_call, client_on_notify or c_notify)
|
||||
tg.start_soon(server.run, server_on_call or s_call, server_on_notify or s_notify)
|
||||
tg.start_soon(client.run, client_on_request or c_req, client_on_notify or c_notify)
|
||||
tg.start_soon(server.run, server_on_request or s_req, server_on_notify or s_notify)
|
||||
try:
|
||||
yield client, server, client_rec, server_rec
|
||||
finally:
|
||||
@@ -67,53 +67,53 @@ async def running_pair(
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_returns_result_from_peer_on_call():
|
||||
async def test_send_request_returns_result_from_peer_on_request():
|
||||
async with running_pair() as (client, _server, _crec, srec):
|
||||
with anyio.fail_after(5):
|
||||
result = await client.call("tools/list", {"cursor": "abc"})
|
||||
result = await client.send_request("tools/list", {"cursor": "abc"})
|
||||
assert result == {"echoed": "tools/list", "params": {"cursor": "abc"}}
|
||||
assert srec.calls == [("tools/list", {"cursor": "abc"})]
|
||||
assert srec.requests == [("tools/list", {"cursor": "abc"})]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_reraises_mcperror_from_handler_unchanged():
|
||||
async def on_call(
|
||||
async def test_send_request_reraises_mcperror_from_handler_unchanged():
|
||||
async def on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
raise MCPError(code=INVALID_PARAMS, message="bad cursor")
|
||||
|
||||
async with running_pair(server_on_call=on_call) as (client, *_):
|
||||
async with running_pair(server_on_request=on_request) as (client, *_):
|
||||
with anyio.fail_after(5), pytest.raises(MCPError) as exc:
|
||||
await client.call("tools/list", {})
|
||||
await client.send_request("tools/list", {})
|
||||
assert exc.value.error.code == INVALID_PARAMS
|
||||
assert exc.value.error.message == "bad cursor"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_wraps_non_mcperror_exception_as_internal_error():
|
||||
async def on_call(
|
||||
async def test_send_request_wraps_non_mcperror_exception_as_internal_error():
|
||||
async def on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
raise ValueError("oops")
|
||||
|
||||
async with running_pair(server_on_call=on_call) as (client, *_):
|
||||
async with running_pair(server_on_request=on_request) as (client, *_):
|
||||
with anyio.fail_after(5), pytest.raises(MCPError) as exc:
|
||||
await client.call("tools/list", {})
|
||||
await client.send_request("tools/list", {})
|
||||
assert exc.value.error.code == INTERNAL_ERROR
|
||||
assert isinstance(exc.value.__cause__, ValueError)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_with_timeout_raises_mcperror_request_timeout():
|
||||
async def on_call(
|
||||
async def test_send_request_with_timeout_raises_mcperror_request_timeout():
|
||||
async def on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
await anyio.sleep_forever()
|
||||
raise NotImplementedError
|
||||
|
||||
async with running_pair(server_on_call=on_call) as (client, *_):
|
||||
async with running_pair(server_on_request=on_request) as (client, *_):
|
||||
with anyio.fail_after(5), pytest.raises(MCPError) as exc:
|
||||
await client.call("slow", None, {"timeout": 0})
|
||||
await client.send_request("slow", None, {"timeout": 0})
|
||||
assert exc.value.error.code == REQUEST_TIMEOUT
|
||||
|
||||
|
||||
@@ -128,53 +128,53 @@ async def test_notify_invokes_peer_on_notify():
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ctx_send_request_round_trips_to_calling_side():
|
||||
"""A handler's ctx.send_request reaches the side that made the inbound call."""
|
||||
"""A handler's ctx.send_request reaches the side that made the inbound request."""
|
||||
|
||||
async def server_on_call(
|
||||
async def server_on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
sample = await ctx.send_request("sampling/createMessage", {"prompt": "hi"})
|
||||
return {"sampled": sample}
|
||||
|
||||
async with running_pair(server_on_call=server_on_call) as (client, _server, crec, _srec):
|
||||
async with running_pair(server_on_request=server_on_request) as (client, _server, crec, _srec):
|
||||
with anyio.fail_after(5):
|
||||
result = await client.call("tools/call", None)
|
||||
assert crec.calls == [("sampling/createMessage", {"prompt": "hi"})]
|
||||
result = await client.send_request("tools/call", None)
|
||||
assert crec.requests == [("sampling/createMessage", {"prompt": "hi"})]
|
||||
assert result == {"sampled": {"echoed": "sampling/createMessage", "params": {"prompt": "hi"}}}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ctx_send_request_raises_nobackchannelerror_when_transport_disallows():
|
||||
async def server_on_call(
|
||||
async def server_on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
return await ctx.send_request("sampling/createMessage", None)
|
||||
|
||||
async with running_pair(server_on_call=server_on_call, can_send_request=False) as (client, *_):
|
||||
async with running_pair(server_on_request=server_on_request, can_send_request=False) as (client, *_):
|
||||
with anyio.fail_after(5), pytest.raises(NoBackChannelError) as exc:
|
||||
await client.call("tools/call", None)
|
||||
await client.send_request("tools/call", None)
|
||||
assert exc.value.method == "sampling/createMessage"
|
||||
assert exc.value.error.code == INVALID_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ctx_notify_invokes_calling_side_on_notify():
|
||||
async def server_on_call(
|
||||
async def server_on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
await ctx.notify("notifications/message", {"level": "info"})
|
||||
return {}
|
||||
|
||||
async with running_pair(server_on_call=server_on_call) as (client, _server, crec, _srec):
|
||||
async with running_pair(server_on_request=server_on_request) as (client, _server, crec, _srec):
|
||||
with anyio.fail_after(5):
|
||||
await client.call("tools/call", None)
|
||||
await client.send_request("tools/call", None)
|
||||
await crec.notified.wait()
|
||||
assert crec.notifications == [("notifications/message", {"level": "info"})]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ctx_progress_invokes_caller_on_progress_callback():
|
||||
async def server_on_call(
|
||||
async def server_on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
await ctx.progress(0.5, total=1.0, message="halfway")
|
||||
@@ -185,27 +185,27 @@ async def test_ctx_progress_invokes_caller_on_progress_callback():
|
||||
async def on_progress(progress: float, total: float | None, message: str | None) -> None:
|
||||
received.append((progress, total, message))
|
||||
|
||||
async with running_pair(server_on_call=server_on_call) as (client, *_):
|
||||
async with running_pair(server_on_request=server_on_request) as (client, *_):
|
||||
with anyio.fail_after(5):
|
||||
await client.call("tools/call", None, {"on_progress": on_progress})
|
||||
await client.send_request("tools/call", None, {"on_progress": on_progress})
|
||||
assert received == [(0.5, 1.0, "halfway")]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_issued_before_peer_run_blocks_until_peer_ready():
|
||||
async def test_send_request_issued_before_peer_run_blocks_until_peer_ready():
|
||||
client, server = create_direct_dispatcher_pair()
|
||||
s_call, s_notify = echo_handlers(Recorder())
|
||||
c_call, c_notify = echo_handlers(Recorder())
|
||||
s_req, s_notify = echo_handlers(Recorder())
|
||||
c_req, c_notify = echo_handlers(Recorder())
|
||||
|
||||
async def late_start():
|
||||
await anyio.sleep(0)
|
||||
await server.run(s_call, s_notify)
|
||||
await server.run(s_req, s_notify)
|
||||
|
||||
async with anyio.create_task_group() as tg:
|
||||
tg.start_soon(client.run, c_call, c_notify)
|
||||
tg.start_soon(client.run, c_req, c_notify)
|
||||
tg.start_soon(late_start)
|
||||
with anyio.fail_after(5):
|
||||
result = await client.call("ping", None)
|
||||
result = await client.send_request("ping", None)
|
||||
assert result == {"echoed": "ping", "params": {}}
|
||||
client.close()
|
||||
server.close()
|
||||
@@ -213,23 +213,23 @@ async def test_call_issued_before_peer_run_blocks_until_peer_ready():
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ctx_progress_is_noop_when_caller_supplied_no_callback():
|
||||
async def server_on_call(
|
||||
async def server_on_request(
|
||||
ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
await ctx.progress(0.5)
|
||||
return {"ok": True}
|
||||
|
||||
async with running_pair(server_on_call=server_on_call) as (client, *_):
|
||||
async with running_pair(server_on_request=server_on_request) as (client, *_):
|
||||
with anyio.fail_after(5):
|
||||
result = await client.call("tools/call", None)
|
||||
result = await client.send_request("tools/call", None)
|
||||
assert result == {"ok": True}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_call_and_notify_raise_runtimeerror_when_no_peer_connected():
|
||||
async def test_send_request_and_notify_raise_runtimeerror_when_no_peer_connected():
|
||||
d = DirectDispatcher(TransportContext(kind="direct", can_send_request=True))
|
||||
with pytest.raises(RuntimeError, match="no peer"):
|
||||
await d.call("ping", None)
|
||||
await d.send_request("ping", None)
|
||||
with pytest.raises(RuntimeError, match="no peer"):
|
||||
await d.notify("ping", None)
|
||||
|
||||
@@ -237,16 +237,15 @@ async def test_call_and_notify_raise_runtimeerror_when_no_peer_connected():
|
||||
@pytest.mark.anyio
|
||||
async def test_close_makes_run_return():
|
||||
client, server = create_direct_dispatcher_pair()
|
||||
on_call, on_notify = echo_handlers(Recorder())
|
||||
on_request, on_notify = echo_handlers(Recorder())
|
||||
with anyio.fail_after(5):
|
||||
async with anyio.create_task_group() as tg:
|
||||
tg.start_soon(server.run, on_call, on_notify)
|
||||
tg.start_soon(client.run, on_call, on_notify)
|
||||
tg.start_soon(server.run, on_request, on_notify)
|
||||
tg.start_soon(client.run, on_request, on_notify)
|
||||
client.close()
|
||||
server.close()
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
_dispatcher_check: Dispatcher[TransportContext] = DirectDispatcher(
|
||||
TransportContext(kind="direct", can_send_request=True)
|
||||
)
|
||||
_d: Dispatcher[TransportContext] = DirectDispatcher(TransportContext(kind="direct", can_send_request=True))
|
||||
_o: Outbound = _d
|
||||
|
||||
Reference in New Issue
Block a user