diff --git a/docs/advanced/events.md b/docs/advanced/events.md index 590ad6ea..d96beacb 100644 --- a/docs/advanced/events.md +++ b/docs/advanced/events.md @@ -28,11 +28,13 @@ $server = Server::builder() ## Protocol Events -The SDK dispatches 4 broad event types at the protocol level, allowing you to observe and modify all server operations: +`RequestEvent`, `ResponseEvent` and `ErrorEvent` fire on both the handshake era and the modern (`2026-07-28`) era. + +On the modern era the session carried by these events is per-request and discarded after the response. ### RequestEvent -**Dispatched**: When any request is received from the client, before it's processed by handlers. +**Dispatched**: When any request is received from the client, before it's processed by handlers. On the modern era this includes a multi round-trip retry MRTR : if the client sent `inputResponses`, they are already lifted onto the session as `InputContext` (`$event->getSession()->get(InputContext::class)`). They are not on the typed `Request` (for example `CallToolRequest` only carries `name` and `arguments`). **Properties**: - `getRequest(): Request` - The incoming request @@ -43,6 +45,9 @@ The SDK dispatches 4 broad event types at the protocol level, allowing you to ob ### ResponseEvent **Dispatched**: When a successful response is ready to be sent to the client, after handler execution. +Also dispatched when a suspended Fiber completes (e.g. after elicitation or sampling on a handshake connection). + +On the modern era an elicitation is a successful result of the original method (`resultType: input_required`). Listen for `ResponseEvent` whose `$event->getResponse()->result` is an `InputRequiredResult`. **Properties**: - `getResponse(): Response` - The response being sent @@ -53,7 +58,7 @@ The SDK dispatches 4 broad event types at the protocol level, allowing you to ob ### ErrorEvent -**Dispatched**: When an error occurs during request processing. +**Dispatched**: When an error occurs during request processing. Also dispatched when a suspended Fiber completes with an error. **Properties**: - `getError(): Error` - The error being sent @@ -62,6 +67,10 @@ The SDK dispatches 4 broad event types at the protocol level, allowing you to ob - `getThrowable(): ?\Throwable` - The exception that caused the error (if any) - `getSession(): SessionInterface` - The current session +## Handshake era only, before MCP 2026-07-28 + +These events are dispatched only on handshake-era connections (protocol revisions before `2026-07-28`). The modern revision has no client-to-server notification handlers over HTTP, and no server-initiated JSON-RPC requests. + ### NotificationEvent **Dispatched**: When a notification is received from the client, before it's processed by handlers. @@ -72,6 +81,26 @@ The SDK dispatches 4 broad event types at the protocol level, allowing you to ob - `getSession(): SessionInterface` - The current session - `getMethod(): string` - Convenience method to get the notification method +### ServerRequestEvent + +**Dispatched**: When the server sends a request to the client (e.g. `elicitation/create`, `sampling/create`). + +**Properties**: +- `getRequest(): Request` - The outgoing request (with server-assigned ID) +- `getSession(): SessionInterface` - The current session +- `getTimeout(): int` - Maximum time to wait for the client response (seconds) +- `getMethod(): string` - Convenience method to get the request method + +### ClientResponseEvent + +**Dispatched**: When the server receives a client response to a prior outgoing request. + +**Properties**: +- `getResponse(): Response|Error` - The client's reply +- `getSession(): SessionInterface` - The current session +- `getId(): string|int` - The JSON-RPC message ID +- `isError(): bool` - Whether the client returned a JSON-RPC error + ## List Change Events These events are dispatched when the lists of available capabilities change: diff --git a/src/Event/ClientResponseEvent.php b/src/Event/ClientResponseEvent.php new file mode 100644 index 00000000..84b34fff --- /dev/null +++ b/src/Event/ClientResponseEvent.php @@ -0,0 +1,56 @@ + + */ +final class ClientResponseEvent +{ + /** + * @param Response|Error $response + */ + public function __construct( + private readonly Response|Error $response, + private readonly SessionInterface $session, + ) { + } + + /** + * @return Response|Error + */ + public function getResponse(): Response|Error + { + return $this->response; + } + + public function getSession(): SessionInterface + { + return $this->session; + } + + public function getId(): string|int + { + return $this->response->getId(); + } + + public function isError(): bool + { + return $this->response instanceof Error; + } +} diff --git a/src/Event/ServerRequestEvent.php b/src/Event/ServerRequestEvent.php new file mode 100644 index 00000000..fff38cd5 --- /dev/null +++ b/src/Event/ServerRequestEvent.php @@ -0,0 +1,50 @@ + + */ +final class ServerRequestEvent +{ + public function __construct( + private readonly Request $request, + private readonly int $timeout, + private readonly SessionInterface $session, + ) { + } + + public function getRequest(): Request + { + return $this->request; + } + + public function getSession(): SessionInterface + { + return $this->session; + } + + public function getTimeout(): int + { + return $this->timeout; + } + + public function getMethod(): string + { + return $this->request::getMethod(); + } +} diff --git a/src/JsonRpc/MessageFactory.php b/src/JsonRpc/MessageFactory.php index 27860a2c..e2b0b8e5 100644 --- a/src/JsonRpc/MessageFactory.php +++ b/src/JsonRpc/MessageFactory.php @@ -148,7 +148,7 @@ public function create(string $input): array throw new InvalidInputMessageException('A JSON-RPC message must be a JSON object.'); } - $messages[] = $this->createMessage($message); + $messages[] = $this->createFromArray($message); } catch (InvalidInputMessageException $e) { // Recover the id only when it's a valid JSON-RPC scalar; // a null or malformed id is left at the exception's null default. @@ -169,7 +169,7 @@ public function create(string $input): array * * @throws InvalidInputMessageException */ - private function createMessage(array $data): MessageInterface + public function createFromArray(array $data): MessageInterface { try { if (isset($data['error'])) { diff --git a/src/Server/Builder.php b/src/Server/Builder.php index 1a5dd73b..9e500bfb 100644 --- a/src/Server/Builder.php +++ b/src/Server/Builder.php @@ -982,6 +982,7 @@ public function buildStateless(array $supportedVersions = [ProtocolVersion::V202 cachePolicy: $this->cachePolicy, notificationBus: $this->notificationBus, extensionMethods: $this->extensionMethods, + eventDispatcher: $parts['eventDispatcher'], ); } diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 4d6e5f5c..f4d7e570 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -11,10 +11,12 @@ namespace Mcp\Server; +use Mcp\Event\ClientResponseEvent; use Mcp\Event\ErrorEvent; use Mcp\Event\NotificationEvent; use Mcp\Event\RequestEvent; use Mcp\Event\ResponseEvent; +use Mcp\Event\ServerRequestEvent; use Mcp\Exception\InvalidInputMessageException; use Mcp\JsonRpc\MessageFactory; use Mcp\Schema\JsonRpc\Error; @@ -58,6 +60,9 @@ class Protocol /** Session key for outgoing message queue */ private const SESSION_OUTGOING_QUEUE = '_mcp.outgoing_queue'; + /** Session key for the client request that started a suspended Fiber */ + private const SESSION_FIBER_PARENT_REQUEST = '_mcp.fiber_parent_request'; + /** Session key for active request meta */ public const SESSION_ACTIVE_REQUEST_META = '_mcp.active_request_meta'; @@ -106,6 +111,8 @@ public function connect(TransportInterface $transport): void $transport->setFiberYieldHandler($this->handleFiberYield(...)); + $transport->setFiberTerminationHandler($this->handleFiberTermination(...)); + $this->logger->info('Protocol connected to transport', ['transport' => $transport::class]); } @@ -298,6 +305,8 @@ private function handleRequest(TransportInterface $transport, Request $request, $result = $fiber->start(); if ($fiber->isSuspended()) { + $session->set(self::SESSION_FIBER_PARENT_REQUEST, $request->jsonSerialize()); + if (\is_array($result) && isset($result['type'])) { if ('notification' === $result['type']) { $notification = $result['notification']; @@ -361,6 +370,8 @@ private function handleResponse(Response|Error $response, SessionInterface $sess { $this->logger->info('Handling response from client.', ['response' => $response]); + $this->dispatchEvent(new ClientResponseEvent($response, $session)); + $messageId = $response->getId(); if (null === $messageId) { @@ -408,6 +419,8 @@ public function sendRequest(Request $request, int $timeout, SessionInterface $se $requestWithId = $request->withId($requestId); + $this->dispatchEvent(new ServerRequestEvent($requestWithId, $timeout, $session)); + $this->logger->info('Queueing server request to client', [ 'request_id' => $requestId, 'method' => $request::getMethod(), @@ -645,6 +658,49 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): void } } + /** + * Handle the final result of a suspended Fiber when it completes. + * + * Dispatches ResponseEvent or ErrorEvent for the original client request that + * started the Fiber, allowing listeners to observe deferred responses. + * + * @phpstan-param Response|Error $finalResult + * + * @phpstan-return Response|Error + */ + public function handleFiberTermination(Response|Error $finalResult, Uuid $sessionId): Response|Error + { + $session = $this->sessionManager->createWithId($sessionId); + $parentRequest = $this->resolveFiberParentRequest( + $session->pull(self::SESSION_FIBER_PARENT_REQUEST) + ); + + if (null !== $parentRequest) { + if ($finalResult instanceof Response) { + $responseEvent = $this->dispatchEvent(new ResponseEvent($finalResult, $parentRequest, $session)); + $finalResult = $responseEvent->getResponse(); + } else { + $errorEvent = $this->dispatchEvent(new ErrorEvent($finalResult, $parentRequest, $session, null)); + $finalResult = $errorEvent->getError(); + } + } + + $session->save(); + + return $finalResult; + } + + private function resolveFiberParentRequest(mixed $data): ?Request + { + if (!\is_array($data)) { + return null; + } + + $message = $this->messageFactory->createFromArray($data); + + return $message instanceof Request ? $message : null; + } + /** * @param array $messages */ diff --git a/src/Server/Stateless/StatelessProtocol.php b/src/Server/Stateless/StatelessProtocol.php index a179a217..fb8017a1 100644 --- a/src/Server/Stateless/StatelessProtocol.php +++ b/src/Server/Stateless/StatelessProtocol.php @@ -11,6 +11,9 @@ namespace Mcp\Server\Stateless; +use Mcp\Event\ErrorEvent; +use Mcp\Event\RequestEvent; +use Mcp\Event\ResponseEvent; use Mcp\Exception\InvalidInputMessageException; use Mcp\Exception\LogicException; use Mcp\Exception\MissingRequestMetaException; @@ -32,11 +35,13 @@ use Mcp\Server\Protocol; use Mcp\Server\Session\InMemorySessionStore; use Mcp\Server\Session\Session; +use Mcp\Server\Session\SessionInterface; use Mcp\Server\Subscription\NotificationBusInterface; use Mcp\Server\Wire\CachePolicy; use Mcp\Server\Wire\InboundClassifier; use Mcp\Server\Wire\Rev2026Codec; use Mcp\Server\Wire\WireCodecInterface; +use Psr\EventDispatcher\EventDispatcherInterface; use Psr\Log\LoggerInterface; use Psr\Log\NullLogger; @@ -106,6 +111,7 @@ public function __construct( ?CachePolicy $cachePolicy = null, private readonly ?NotificationBusInterface $notificationBus = null, private readonly array $extensionMethods = [], + private readonly ?EventDispatcherInterface $eventDispatcher = null, ) { $this->codec = $codec ?? new Rev2026Codec($configuration->serverInfo, $cachePolicy); @@ -118,6 +124,18 @@ public function __construct( } } + /** + * @template T of object + * + * @param T $event + * + * @return T + */ + private function dispatchEvent(object $event): object + { + return $this->eventDispatcher?->dispatch($event) ?? $event; + } + /** * The modern revisions this dispatcher answers for. * @@ -470,6 +488,11 @@ private function dispatch(string $method, array $decoded, RequestMeta $meta, str // the handshake era sets under the same key. $session->set(Protocol::SESSION_ACTIVE_REQUEST_META, $request->getMeta()); + $event = $this->dispatchEvent(new RequestEvent($request, $session)); + $request = $event->getRequest(); + $id = $request->getId(); + $method = $request::getMethod(); + foreach ($this->requestHandlers as $handler) { if (!$handler->supports($request)) { continue; @@ -485,11 +508,11 @@ private function dispatch(string $method, array $decoded, RequestMeta $meta, str // with 400 rather than an error frame under a 200. $run->rewind(); } catch (\Throwable $e) { - return $this->toErrorResult($method, $id, $e); + return $this->toErrorResult($request, $session, $e); } if ($run->valid() && $wantsStream) { - return StatelessResult::stream(fn (): \Generator => $this->streamFrames($run, $meta, $method, $id, null === $input)); + return StatelessResult::stream(fn (): \Generator => $this->streamFrames($run, $request, $session, $meta, $method, $id, null === $input)); } try { @@ -506,21 +529,16 @@ private function dispatch(string $method, array $decoded, RequestMeta $meta, str $result = $run->getReturn(); } catch (\Throwable $e) { - return $this->toErrorResult($method, $id, $e); - } - - if ($result instanceof Error) { - return StatelessResult::error($result, 400); - } - - if (null !== $capabilityError = $this->checkInputRequests($result->result, $meta, $method, $id)) { - return $capabilityError; + return $this->toErrorResult($request, $session, $e); } - return $this->encode($method, $id, $result->result, null === $input); + return $this->finalize($result, $request, $session, $meta, $method, $id, null === $input); } - return StatelessResult::error($this->unknownMethod($method, $id), 404); + $error = $this->unknownMethod($method, $id); + $errorEvent = $this->dispatchEvent(new ErrorEvent($error, $request, $session, null)); + + return StatelessResult::error($errorEvent->getError(), 404); } /** @@ -671,6 +689,46 @@ private static function readElicitation(mixed $suspended): ?array return [\is_string($key) ? $key : null, $request]; } + /** + * Applies protocol events to a handler's answer, then encodes it. + * + * Shared by the JSON and streaming paths so a listener cannot see one + * shape of result on a stream and another on a single response. + * + * @param Response|Error $result + */ + private function finalize( + Response|Error $result, + Request $request, + SessionInterface $session, + RequestMeta $meta, + string $method, + string|int $id, + bool $cacheable, + ): StatelessResult { + if ($result instanceof Error) { + $errorEvent = $this->dispatchEvent(new ErrorEvent($result, $request, $session, null)); + + return StatelessResult::error($errorEvent->getError(), 400); + } + + if (null !== $capabilityError = $this->checkInputRequests($result->result, $meta, $method, $id)) { + $error = $capabilityError->message; + if ($error instanceof Error) { + $errorEvent = $this->dispatchEvent(new ErrorEvent($error, $request, $session, null)); + + return StatelessResult::error($errorEvent->getError(), 400); + } + + return $capabilityError; + } + + $responseEvent = $this->dispatchEvent(new ResponseEvent($result, $request, $session)); + $result = $responseEvent->getResponse(); + + return $this->encode($method, $result->getId(), $result->result, $cacheable); + } + /** * The frames of a request-scoped response stream: the notifications the * handler emits, then the response that ends it. @@ -679,7 +737,7 @@ private static function readElicitation(mixed $suspended): ?array * * @return \Generator */ - private function streamFrames(\Generator $run, RequestMeta $meta, string $method, string|int $id, bool $cacheable): \Generator + private function streamFrames(\Generator $run, Request $request, SessionInterface $session, RequestMeta $meta, string $method, string|int $id, bool $cacheable): \Generator { try { while ($run->valid()) { @@ -692,20 +750,12 @@ private function streamFrames(\Generator $run, RequestMeta $meta, string $method } catch (\Throwable $e) { // Headers left long ago, so the status is already 200 and the only // way left to report this is a frame. - yield $this->toErrorResult($method, $id, $e)->message?->jsonSerialize(); + yield $this->toErrorResult($request, $session, $e)->message?->jsonSerialize(); return; } - if (!$result instanceof Error && null !== $capabilityError = $this->checkInputRequests($result->result, $meta, $method, $id)) { - yield $capabilityError->message?->jsonSerialize(); - - return; - } - - yield $result instanceof Error - ? $result->jsonSerialize() - : ['jsonrpc' => '2.0', 'id' => $id, 'result' => $this->codec->encodeResult($method, (array) $result->result->jsonSerialize(), $cacheable)]; + yield json_decode($this->finalize($result, $request, $session, $meta, $method, $id, $cacheable)->toJson(), true, flags: \JSON_THROW_ON_ERROR); } /** @@ -772,28 +822,32 @@ private static function withTraceContext(array $frame, array $traceContext): arr * The one place a handler's exception becomes an answer, so the streaming * and non-streaming paths cannot disagree about which code it earns. */ - private function toErrorResult(string $method, string|int $id, \Throwable $e): StatelessResult + private function toErrorResult(Request $request, SessionInterface $session, \Throwable $e): StatelessResult { - if ($e instanceof MissingRequiredClientCapabilityException) { - return StatelessResult::error( - Error::forMissingRequiredClientCapability($e->getMessage(), $e->requiredCapabilities, $id), - 400, - ); - } - - if ($e instanceof \InvalidArgumentException) { - return StatelessResult::error(Error::forInvalidParams($e->getMessage(), $id), 400); - } + $id = $request->getId(); + $method = $request::getMethod(); - if ($e instanceof LogicException) { + if ($e instanceof MissingRequiredClientCapabilityException) { + $error = Error::forMissingRequiredClientCapability($e->getMessage(), $e->requiredCapabilities, $id); + $status = 400; + } elseif ($e instanceof \InvalidArgumentException) { + $error = Error::forInvalidParams($e->getMessage(), $id); + $status = 400; + } elseif ($e instanceof LogicException) { // Guidance for the tool author, not a detail leaked from their // code or a dependency's — safe to echo back verbatim. - return StatelessResult::error(Error::forInternalError($e->getMessage(), $id), 500); + $error = Error::forInternalError($e->getMessage(), $id); + $status = 500; + } else { + $this->logger->error('Uncaught exception handling a modern-era request.', ['method' => $method, 'exception' => $e]); + + $error = Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id); + $status = 500; } - $this->logger->error('Uncaught exception handling a modern-era request.', ['method' => $method, 'exception' => $e]); + $errorEvent = $this->dispatchEvent(new ErrorEvent($error, $request, $session, $e)); - return StatelessResult::error(Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id), 500); + return StatelessResult::error($errorEvent->getError(), $status); } /** diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php index 58172352..07f9486d 100644 --- a/src/Server/Transport/BaseTransport.php +++ b/src/Server/Transport/BaseTransport.php @@ -127,6 +127,27 @@ protected function handleFiberYield(mixed $yielded, ?Uuid $sessionId): void } } + /** + * @phpstan-param FiberReturn $finalResult + * + * @phpstan-return FiberReturn + */ + protected function handleFiberTerminationResult(Response|Error $finalResult): Response|Error + { + if ($this->sessionId && \is_callable($this->fiberTerminationHandler)) { + try { + return ($this->fiberTerminationHandler)($finalResult, $this->sessionId); + } catch (\Throwable $e) { + $this->logger->error('Fiber termination handler failed.', [ + 'exception' => $e, + 'sessionId' => $this->sessionId->toRfc4122(), + ]); + } + } + + return $finalResult; + } + protected function handleMessage(string $payload, ?Uuid $sessionId): void { if (\is_callable($this->messageListener)) { diff --git a/src/Server/Transport/ManagesTransportCallbacks.php b/src/Server/Transport/ManagesTransportCallbacks.php index 072d3f0e..69f934f3 100644 --- a/src/Server/Transport/ManagesTransportCallbacks.php +++ b/src/Server/Transport/ManagesTransportCallbacks.php @@ -44,6 +44,9 @@ trait ManagesTransportCallbacks /** @var callable(FiberSuspend|null, ?Uuid): void */ protected $fiberYieldHandler; + /** @var callable(FiberReturn, Uuid): FiberReturn */ + protected $fiberTerminationHandler; + public function onMessage(callable $listener): void { $this->messageListener = $listener; @@ -79,4 +82,12 @@ public function setFiberYieldHandler(callable $handler): void { $this->fiberYieldHandler = $handler; } + + /** + * @param callable(FiberReturn, Uuid): FiberReturn $handler + */ + public function setFiberTerminationHandler(callable $handler): void + { + $this->fiberTerminationHandler = $handler; + } } diff --git a/src/Server/Transport/StdioTransport.php b/src/Server/Transport/StdioTransport.php index 565f7da7..096bae25 100644 --- a/src/Server/Transport/StdioTransport.php +++ b/src/Server/Transport/StdioTransport.php @@ -176,6 +176,8 @@ private function handleFiberTermination(): void $finalResult = $this->sessionFiber->getReturn(); if (null !== $finalResult) { + $finalResult = $this->handleFiberTerminationResult($finalResult); + try { $encoded = json_encode($finalResult, \JSON_THROW_ON_ERROR); $this->writeLine($encoded); diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php index 2bf088bc..00c3f685 100644 --- a/src/Server/Transport/StreamableHttpTransport.php +++ b/src/Server/Transport/StreamableHttpTransport.php @@ -308,6 +308,8 @@ protected function handleFiberTermination(): void $finalResult = $this->sessionFiber->getReturn(); if (null !== $finalResult) { + $finalResult = $this->handleFiberTerminationResult($finalResult); + try { $encoded = json_encode($finalResult, \JSON_THROW_ON_ERROR); echo "event: message\n"; diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php index 58d09789..a8636d4a 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/src/Server/Transport/TransportInterface.php @@ -118,6 +118,15 @@ public function setResponseFinder(callable $finder): void; */ public function setFiberYieldHandler(callable $handler): void; + /** + * Set a handler invoked when a suspended Fiber completes. + * + * The transport calls this before sending the Fiber's final result to the client. + * + * @param callable(FiberReturn, Uuid): FiberReturn $handler + */ + public function setFiberTerminationHandler(callable $handler): void; + /** * @param McpFiber $fiber */ diff --git a/tests/Unit/JsonRpc/MessageFactoryTest.php b/tests/Unit/JsonRpc/MessageFactoryTest.php index fe2c0d2a..33f02ecd 100644 --- a/tests/Unit/JsonRpc/MessageFactoryTest.php +++ b/tests/Unit/JsonRpc/MessageFactoryTest.php @@ -38,6 +38,18 @@ protected function setUp(): void ]); } + public function testCreateFromArrayRequest(): void + { + $message = $this->factory->createFromArray([ + 'jsonrpc' => '2.0', + 'method' => 'ping', + 'id' => 1, + ]); + + $this->assertInstanceOf(PingRequest::class, $message); + $this->assertSame(1, $message->getId()); + } + public function testCreateRequestWithIntegerId(): void { $json = '{"jsonrpc": "2.0", "method": "prompts/get", "params": {"name": "create_story"}, "id": 123}'; diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index 997c39d0..434d3cbf 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -11,10 +11,12 @@ namespace Mcp\Tests\Unit\Server; +use Mcp\Event\ClientResponseEvent; use Mcp\Event\ErrorEvent; use Mcp\Event\NotificationEvent; use Mcp\Event\RequestEvent; use Mcp\Event\ResponseEvent; +use Mcp\Event\ServerRequestEvent; use Mcp\JsonRpc\MessageFactory; use Mcp\Schema\Enum\LoggingLevel; use Mcp\Schema\JsonRpc\Error; @@ -25,6 +27,8 @@ use Mcp\Server\Handler\Notification\NotificationHandlerInterface; use Mcp\Server\Handler\Request\RequestHandlerInterface; use Mcp\Server\Protocol; +use Mcp\Server\Session\InMemorySessionStore; +use Mcp\Server\Session\Session; use Mcp\Server\Session\SessionInterface; use Mcp\Server\Session\SessionManagerInterface; use Mcp\Server\Transport\TransportInterface; @@ -1589,4 +1593,346 @@ public function testNotificationEventWithNullDispatcher(): void $sessionId ); } + + #[TestDox('ServerRequestEvent is dispatched when server sends a request to the client')] + public function testServerRequestEventIsDispatched(): void + { + $capturedEvent = null; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->expects($this->once()) + ->method('dispatch') + ->with($this->callback(static function ($event) use (&$capturedEvent) { + $capturedEvent = $event; + + return $event instanceof ServerRequestEvent; + })) + ->willReturnArgument(0); + + $session = $this->createMock(SessionInterface::class); + $session->method('get')->willReturnCallback(static function ($key, $default = null) { + if ('_mcp.request_id_counter' === $key) { + return 1000; + } + + return $default; + }); + $session->method('getId')->willReturn(Uuid::v4()); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $request = PingRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 0, + 'method' => 'ping', + ]); + + $protocol->sendRequest($request, 60, $session); + + $this->assertInstanceOf(ServerRequestEvent::class, $capturedEvent); + $this->assertSame($session, $capturedEvent->getSession()); + $this->assertSame(60, $capturedEvent->getTimeout()); + $this->assertSame('ping', $capturedEvent->getMethod()); + $this->assertSame(1000, $capturedEvent->getRequest()->getId()); + } + + #[TestDox('ClientResponseEvent is dispatched when a client response is received')] + public function testClientResponseEventIsDispatched(): void + { + $capturedEvent = null; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->expects($this->once()) + ->method('dispatch') + ->with($this->callback(static function ($event) use (&$capturedEvent) { + $capturedEvent = $event; + + return $event instanceof ClientResponseEvent; + })) + ->willReturnArgument(0); + + $session = $this->createMock(SessionInterface::class); + + $this->sessionManager->method('createWithId')->willReturn($session); + $this->sessionManager->method('exists')->willReturn(true); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $sessionId = Uuid::v4(); + $protocol->processInput( + $this->transport, + '{"jsonrpc": "2.0", "id": 1000, "result": {"action": "accept"}}', + $sessionId + ); + + $this->assertInstanceOf(ClientResponseEvent::class, $capturedEvent); + $this->assertSame($session, $capturedEvent->getSession()); + $this->assertSame(1000, $capturedEvent->getId()); + $this->assertFalse($capturedEvent->isError()); + } + + #[TestDox('ClientResponseEvent reports errors via isError()')] + public function testClientResponseEventIsError(): void + { + $capturedEvent = null; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function ($event) use (&$capturedEvent) { + if ($event instanceof ClientResponseEvent) { + $capturedEvent = $event; + } + + return $event; + }); + + $session = $this->createMock(SessionInterface::class); + + $this->sessionManager->method('createWithId')->willReturn($session); + $this->sessionManager->method('exists')->willReturn(true); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $sessionId = Uuid::v4(); + $protocol->processInput( + $this->transport, + '{"jsonrpc": "2.0", "id": 1000, "error": {"code": -32603, "message": "Client error"}}', + $sessionId + ); + + $this->assertInstanceOf(ClientResponseEvent::class, $capturedEvent); + $this->assertTrue($capturedEvent->isError()); + } + + #[TestDox('ResponseEvent is dispatched when a suspended Fiber completes')] + public function testResponseEventIsDispatchedOnFiberTermination(): void + { + $capturedEvents = []; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function ($event) use (&$capturedEvents) { + $capturedEvents[] = $event; + + return $event; + }); + + $sessionId = Uuid::v4(); + $session = $this->createMock(SessionInterface::class); + $session->method('getId')->willReturn($sessionId); + + $parentRequest = PingRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 1, + 'method' => 'ping', + ]); + + $session->method('pull') + ->with('_mcp.fiber_parent_request') + ->willReturn($parentRequest->jsonSerialize()); + + $this->sessionManager->method('createWithId')->willReturn($session); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $finalResult = Response::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 1, + 'result' => ['status' => 'ok'], + ]); + $result = $protocol->handleFiberTermination($finalResult, $sessionId); + + $this->assertSame(['status' => 'ok'], $result->result); + $this->assertCount(1, $capturedEvents); + $this->assertInstanceOf(ResponseEvent::class, $capturedEvents[0]); + $this->assertSame('ping', $capturedEvents[0]->getMethod()); + $this->assertSame($session, $capturedEvents[0]->getSession()); + } + + #[TestDox('ErrorEvent is dispatched when a suspended Fiber completes with an error')] + public function testErrorEventIsDispatchedOnFiberTermination(): void + { + $capturedEvents = []; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function ($event) use (&$capturedEvents) { + $capturedEvents[] = $event; + + return $event; + }); + + $sessionId = Uuid::v4(); + $session = $this->createMock(SessionInterface::class); + $session->method('getId')->willReturn($sessionId); + + $parentRequest = PingRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 1, + 'method' => 'ping', + ]); + + $session->method('pull') + ->with('_mcp.fiber_parent_request') + ->willReturn($parentRequest->jsonSerialize()); + + $this->sessionManager->method('createWithId')->willReturn($session); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $finalResult = Error::forInternalError('Fiber failed', 1); + $result = $protocol->handleFiberTermination($finalResult, $sessionId); + + $this->assertInstanceOf(Error::class, $result); + $this->assertCount(1, $capturedEvents); + $this->assertInstanceOf(ErrorEvent::class, $capturedEvents[0]); + $this->assertSame('ping', $capturedEvents[0]->getRequest()::getMethod()); + } + + #[TestDox('Fiber parent request is stored when handler suspends')] + public function testFiberParentRequestIsStoredOnSuspend(): void + { + $storedParentRequest = null; + + $handler = $this->createMock(RequestHandlerInterface::class); + $handler->method('supports')->willReturn(true); + $handler->method('handle')->willReturnCallback(static function () { + \Fiber::suspend([ + 'type' => 'request', + 'request' => PingRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 0, + 'method' => 'ping', + ]), + 'timeout' => 60, + ]); + + return new Response(1, []); + }); + + $session = $this->createMock(SessionInterface::class); + $session->method('getId')->willReturn(Uuid::v4()); + $session->method('get')->willReturnCallback(static function ($key, $default = null) { + if ('_mcp.request_id_counter' === $key) { + return 1000; + } + + return $default; + }); + $session->method('set')->willReturnCallback(static function ($key, $value) use (&$storedParentRequest) { + if ('_mcp.fiber_parent_request' === $key) { + $storedParentRequest = $value; + } + }); + + $this->sessionManager->method('createWithId')->willReturn($session); + $this->sessionManager->method('exists')->willReturn(true); + + $this->transport->expects($this->once())->method('attachFiberToSession'); + + $protocol = new Protocol( + requestHandlers: [$handler], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $this->sessionManager, + ); + + $sessionId = Uuid::v4(); + $protocol->processInput( + $this->transport, + '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', + $sessionId + ); + + $this->assertIsArray($storedParentRequest); + $this->assertSame('ping', $storedParentRequest['method']); + $this->assertSame(1, $storedParentRequest['id']); + } + + #[TestDox('ResponseEvent is dispatched after session reload when Fiber completes')] + public function testResponseEventIsDispatchedOnFiberTerminationAfterSessionSave(): void + { + $capturedEvents = []; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function ($event) use (&$capturedEvents) { + $capturedEvents[] = $event; + + return $event; + }); + + $store = new InMemorySessionStore(); + $sessionId = Uuid::v4(); + $session = new Session($store, $sessionId); + + $parentRequest = PingRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 1, + 'method' => 'ping', + ]); + $session->set('_mcp.fiber_parent_request', $parentRequest->jsonSerialize()); + $session->save(); + + $sessionManager = $this->createMock(SessionManagerInterface::class); + $sessionManager->method('createWithId')->willReturnCallback( + static fn (Uuid $id) => new Session($store, $id) + ); + + $protocol = new Protocol( + requestHandlers: [], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $sessionManager, + eventDispatcher: $eventDispatcher, + ); + + $finalResult = Response::fromArray([ + 'jsonrpc' => '2.0', + 'id' => 1, + 'result' => ['status' => 'ok'], + ]); + $result = $protocol->handleFiberTermination($finalResult, $sessionId); + + $this->assertSame(['status' => 'ok'], $result->result); + $this->assertCount(1, $capturedEvents); + $this->assertInstanceOf(ResponseEvent::class, $capturedEvents[0]); + $this->assertSame('ping', $capturedEvents[0]->getMethod()); + } } diff --git a/tests/Unit/Server/Stateless/StatelessProtocolTest.php b/tests/Unit/Server/Stateless/StatelessProtocolTest.php index 62c7eb79..7e1ae91f 100644 --- a/tests/Unit/Server/Stateless/StatelessProtocolTest.php +++ b/tests/Unit/Server/Stateless/StatelessProtocolTest.php @@ -11,8 +11,14 @@ namespace Mcp\Tests\Unit\Server\Stateless; +use Mcp\Event\ClientResponseEvent; +use Mcp\Event\ErrorEvent; +use Mcp\Event\RequestEvent; +use Mcp\Event\ResponseEvent; +use Mcp\Event\ServerRequestEvent; use Mcp\Exception\MissingRequiredClientCapabilityException; use Mcp\Schema\ClientCapabilities; +use Mcp\Schema\Content\TextContent; use Mcp\Schema\Content\TextResourceContents; use Mcp\Schema\Elicitation\ElicitationSchema; use Mcp\Schema\Elicitation\StringSchemaDefinition; @@ -20,16 +26,20 @@ use Mcp\Schema\Enum\LoggingLevel; use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\JsonRpc\Error; +use Mcp\Schema\JsonRpc\Response; use Mcp\Schema\Notification\PromptListChangedNotification; use Mcp\Schema\Notification\ResourceUpdatedNotification; use Mcp\Schema\Notification\ToolListChangedNotification; +use Mcp\Schema\Request\CallToolRequest; use Mcp\Schema\Request\ElicitRequest; use Mcp\Schema\Request\ListRootsRequest; +use Mcp\Schema\Result\CallToolResult; use Mcp\Schema\Result\InputRequiredResult; use Mcp\Schema\Result\ReadResourceResult; use Mcp\Schema\ServerCapabilities; use Mcp\Server; use Mcp\Server\RequestContext; +use Mcp\Server\Stateless\InputContext; use Mcp\Server\Stateless\RequestMeta; use Mcp\Server\Stateless\StatelessProtocol; use Mcp\Server\Stateless\StatelessResult; @@ -40,15 +50,16 @@ use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\TestDox; use PHPUnit\Framework\TestCase; +use Psr\EventDispatcher\EventDispatcherInterface; class StatelessProtocolTest extends TestCase { /** * @param array $capabilities */ - private static function protocol(array $capabilities = []): StatelessProtocol + private static function protocol(array $capabilities = [], ?EventDispatcherInterface $eventDispatcher = null): StatelessProtocol { - return Server::builder() + $builder = Server::builder() ->setServerInfo('test-server', '1.0.0') ->addTool(static fn (): string => 'ok', name: 'plain_tool', description: 'Returns a fixed string') ->addTool( @@ -174,8 +185,13 @@ static function (RequestContext $context): string|InputRequiredResult { 'test://gated', 'gated', 'A resource that asks who is reading before it answers', - ) - ->buildStateless([ProtocolVersion::V2026_07_28]); + ); + + if (null !== $eventDispatcher) { + $builder->setEventDispatcher($eventDispatcher); + } + + return $builder->buildStateless([ProtocolVersion::V2026_07_28]); } /** @@ -1203,4 +1219,212 @@ public function testElicitationDefaultsToFormMode(): void $this->assertSame('elicitation', $answer['body']['result']['content'][0]['text']); } + + /** + * @param list $captured + */ + private function capturingDispatcher(array &$captured): EventDispatcherInterface + { + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function (object $event) use (&$captured): object { + $captured[] = $event; + + return $event; + }); + + return $eventDispatcher; + } + + #[TestDox('RequestEvent and ResponseEvent are dispatched on a modern-era tool call')] + public function testRequestAndResponseEventsAreDispatched(): void + { + $captured = []; + $answer = self::call( + self::protocol([], $this->capturingDispatcher($captured)), + 'tools/call', + ['name' => 'plain_tool', 'arguments' => []], + ['Mcp-Name' => 'plain_tool'], + ); + + $this->assertSame(200, $answer['status']); + $this->assertCount(2, $captured); + $this->assertInstanceOf(RequestEvent::class, $captured[0]); + $this->assertSame('tools/call', $captured[0]->getMethod()); + $this->assertInstanceOf(ResponseEvent::class, $captured[1]); + $this->assertSame('tools/call', $captured[1]->getMethod()); + $this->assertInstanceOf(CallToolResult::class, $captured[1]->getResponse()->result); + } + + #[TestDox('RequestEvent setRequest() is used by the handler')] + public function testRequestEventModificationIsUsed(): void + { + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function (object $event): object { + if ($event instanceof RequestEvent) { + $event->setRequest(CallToolRequest::fromArray([ + 'jsonrpc' => '2.0', + 'id' => $event->getRequest()->getId(), + 'method' => 'tools/call', + 'params' => [ + 'name' => 'probe_capabilities', + 'arguments' => [], + ], + ])); + } + + return $event; + }); + + $answer = self::call( + self::protocol([], $eventDispatcher), + 'tools/call', + ['name' => 'plain_tool', 'arguments' => []], + ['Mcp-Name' => 'plain_tool'], + ); + + $this->assertSame(200, $answer['status']); + $this->assertSame('none', $answer['body']['result']['content'][0]['text']); + } + + #[TestDox('ResponseEvent setResponse() is encoded')] + public function testResponseEventModificationIsUsed(): void + { + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function (object $event): object { + if ($event instanceof ResponseEvent) { + $event->setResponse(new Response( + $event->getResponse()->getId(), + new CallToolResult([new TextContent('modified')]), + )); + } + + return $event; + }); + + $answer = self::call( + self::protocol([], $eventDispatcher), + 'tools/call', + ['name' => 'plain_tool', 'arguments' => []], + ['Mcp-Name' => 'plain_tool'], + ); + + $this->assertSame(200, $answer['status']); + $this->assertSame('modified', $answer['body']['result']['content'][0]['text']); + } + + #[TestDox('a gateway elicitation is observed as ResponseEvent with InputRequiredResult')] + public function testGatewayElicitationDispatchesResponseEvent(): void + { + $captured = []; + $answer = self::call( + self::protocol([], $this->capturingDispatcher($captured)), + 'tools/call', + ['name' => 'elicits_directly', 'arguments' => []], + ['Mcp-Name' => 'elicits_directly'], + ['elicitation' => new \stdClass()], + ); + + $this->assertSame('input_required', $answer['body']['result']['resultType']); + $this->assertInstanceOf(RequestEvent::class, $captured[0]); + $this->assertInstanceOf(ResponseEvent::class, $captured[1]); + $this->assertInstanceOf(InputRequiredResult::class, $captured[1]->getResponse()->result); + $this->assertSame([], array_filter($captured, static fn (object $event): bool => $event instanceof ServerRequestEvent || $event instanceof ClientResponseEvent)); + } + + #[TestDox('an explicit InputRequiredResult is observed as ResponseEvent')] + public function testExplicitAskDispatchesResponseEvent(): void + { + $captured = []; + $answer = self::call( + self::protocol([], $this->capturingDispatcher($captured)), + 'tools/call', + ['name' => 'asks_by_url', 'arguments' => []], + ['Mcp-Name' => 'asks_by_url'], + ['elicitation' => ['url' => new \stdClass()]], + ); + + $this->assertSame('input_required', $answer['body']['result']['resultType']); + $this->assertInstanceOf(ResponseEvent::class, $captured[1]); + $this->assertInstanceOf(InputRequiredResult::class, $captured[1]->getResponse()->result); + $this->assertArrayHasKey('consent', $captured[1]->getResponse()->result->inputRequests); + } + + #[TestDox('an elicitation retry exposes InputContext on RequestEvent')] + public function testElicitationRetryExposesInputContextOnRequestEvent(): void + { + $captured = []; + $input = null; + + $eventDispatcher = $this->createMock(EventDispatcherInterface::class); + $eventDispatcher + ->method('dispatch') + ->willReturnCallback(static function (object $event) use (&$captured, &$input): object { + $captured[] = $event; + + if ($event instanceof RequestEvent) { + $context = $event->getSession()->get(InputContext::class); + $input = $context instanceof InputContext ? $context : null; + } + + return $event; + }); + + $answer = self::call( + self::protocol([], $eventDispatcher), + 'tools/call', + [ + 'name' => 'elicits_directly', + 'arguments' => [], + 'inputResponses' => ['elicitation_1' => ['action' => 'accept', 'content' => ['n' => 'ada']]], + ], + ['Mcp-Name' => 'elicits_directly'], + ['elicitation' => new \stdClass()], + ); + + $this->assertSame('hello ada', $answer['body']['result']['content'][0]['text']); + $this->assertInstanceOf(RequestEvent::class, $captured[0]); + $this->assertInstanceOf(InputContext::class, $input); + $this->assertTrue($input->has('elicitation_1')); + $this->assertInstanceOf(ResponseEvent::class, $captured[1]); + $this->assertInstanceOf(CallToolResult::class, $captured[1]->getResponse()->result); + $this->assertSame([], array_filter($captured, static fn (object $event): bool => $event instanceof ServerRequestEvent || $event instanceof ClientResponseEvent)); + } + + #[TestDox('ErrorEvent is dispatched when a handler throws')] + public function testErrorEventIsDispatchedOnHandlerException(): void + { + $captured = []; + $answer = self::call( + self::protocol([], $this->capturingDispatcher($captured)), + 'tools/call', + ['name' => 'capability_tool', 'arguments' => []], + ['Mcp-Name' => 'capability_tool'], + ); + + $this->assertSame(400, $answer['status']); + $this->assertInstanceOf(RequestEvent::class, $captured[0]); + $this->assertInstanceOf(ErrorEvent::class, $captured[1]); + $this->assertSame(Error::MISSING_REQUIRED_CLIENT_CAPABILITY, $captured[1]->getError()->code); + $this->assertInstanceOf(MissingRequiredClientCapabilityException::class, $captured[1]->getThrowable()); + } + + #[TestDox('parse errors are rejected before RequestEvent')] + public function testParseErrorDoesNotDispatchEvents(): void + { + $captured = []; + $eventDispatcher = $this->capturingDispatcher($captured); + + $result = self::protocol([], $eventDispatcher)->handle('not json', [ + 'MCP-Protocol-Version' => ProtocolVersion::V2026_07_28->value, + ]); + + $this->assertSame(400, $result->httpStatus); + $this->assertSame([], $captured); + } }