From 0145476a7da7b5121631fd89ee6df0ed4fec5c9c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Webagil=20-=20K=C3=A9vin=20PAULMIER?= <9626689+webagil-kevin@users.noreply.github.com> Date: Sun, 25 Oct 2026 17:47:50 +0100 Subject: [PATCH 1/2] [Server] Let an event loop drive the fibers request handlers run in --- CHANGELOG.md | 4 + docs/run/server-builder.md | 33 ++++ src/Server/Builder.php | 20 +++ src/Server/ClientGateway.php | 5 +- src/Server/Fiber/FiberSchedulerInterface.php | 35 ++++ src/Server/Fiber/FiberSettlement.php | 64 +++++++ src/Server/Fiber/HandlerFiberState.php | 101 +++++++++++ src/Server/Fiber/HandlerFibers.php | 138 +++++++++++++++ src/Server/Protocol.php | 9 +- src/Server/Stateless/StatelessProtocol.php | 11 +- src/Server/Transport/BaseTransport.php | 3 +- src/Server/Transport/StdioTransport.php | 5 +- .../Transport/StreamableHttpTransport.php | 5 +- .../Server/Fiber/EventLoopToolCallTest.php | 164 ++++++++++++++++++ tests/Unit/Server/Fiber/FakeEventLoop.php | 50 ++++++ tests/Unit/Server/Fiber/HandlerFibersTest.php | 124 +++++++++++++ 16 files changed, 757 insertions(+), 14 deletions(-) create mode 100644 src/Server/Fiber/FiberSchedulerInterface.php create mode 100644 src/Server/Fiber/FiberSettlement.php create mode 100644 src/Server/Fiber/HandlerFiberState.php create mode 100644 src/Server/Fiber/HandlerFibers.php create mode 100644 tests/Unit/Server/Fiber/EventLoopToolCallTest.php create mode 100644 tests/Unit/Server/Fiber/FakeEventLoop.php create mode 100644 tests/Unit/Server/Fiber/HandlerFibersTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 83668bcc..4549729d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ All notable changes to `mcp/sdk` will be documented in this file. * [BC Break] `DocBlockParser::parseDocBlock()` no longer accepts `null`; pass `ReflectionMethod::getDocComment()` as is * Fix tools with an array shape parameter whose keys are nullable or unions (e.g. `array{name: string, nickname: string|null}`) rejecting valid arguments with `-32602`: a `null` inside the shape no longer turns the parameter into `array|null`, and such keys get their own types instead of `object`, including lists and nested shapes (`string[]|null`, `array{x: int}|null`). * Fix `array` and other generics with a union element getting `items` of type `object`. A `null` inside a generic no longer makes the parameter nullable, and neither does a DocBlock class whose name contains `null`. +* Event loops + * Add `Builder::setFiberScheduler()` and `Mcp\Server\Fiber\FiberSchedulerInterface`: under an event loop — Revolt (AMPHP), ReactPHP's async — a request handler awaiting I/O suspends its fiber for the loop, and the server now waits through the scheduler until the handler suspends for the server again or returns. The guide shows a Revolt scheduler in ten lines. + * Fix the server resuming a request handler's fiber that the runtime had suspended: the handler's await returned early or threw (`Must call resume() or throw() before calling suspend() again`), and the runtime later resumed a fiber that had moved on — under Revolt, terminating the event loop. Without a scheduler, such a handler now fails the request with a `LogicException` naming `Builder::setFiberScheduler()`. `Protocol::__construct()` and `StatelessProtocol::__construct()` take the scheduler as optional last argument. + * What a handler throws while the runtime drives its fiber reaches the server, as it does when the server drives it, never the runtime. ### Client diff --git a/docs/run/server-builder.md b/docs/run/server-builder.md index b1fb6a16..256a9ebd 100644 --- a/docs/run/server-builder.md +++ b/docs/run/server-builder.md @@ -265,6 +265,38 @@ $server = Server::builder() ->setEventDispatcher($eventDispatcher); ``` +### Fiber Scheduler + +The server runs every request handler in a fiber, so that the handler can wait on the client — for an +elicitation, a sampling request, or to send a notification — through the +[`ClientGateway`](../handlers/client-communication.md). Under an event loop (Revolt, which AMPHP builds on, or +ReactPHP's async), the handler's own I/O suspends that same fiber, for the loop. Tell the server how to wait on +your runtime, and it leaves the fiber to the loop until the handler suspends for the server again or returns: + +```php +use Mcp\Server\Fiber\FiberSchedulerInterface; +use Mcp\Server\Fiber\FiberSettlement; +use Revolt\EventLoop; + +final class RevoltFiberScheduler implements FiberSchedulerInterface +{ + public function wait(FiberSettlement $settlement): void + { + $suspension = EventLoop::getSuspension(); + $settlement->then($suspension->resume(...)); + $suspension->suspend(); + } +} + +$server = Server::builder() + ->setFiberScheduler(new RevoltFiberScheduler()); +``` + +Without a scheduler, a handler whose fiber is suspended by anything but the server fails the request with a +`LogicException`, rather than being resumed with nothing. Under an event loop, wait on the client through the +gateway: a suspension payload a handler hands to `\Fiber::suspend()` itself, after the loop resumed it, reaches +the loop, not the server. + ## Complete Example Here's a comprehensive example showing all major configuration options: @@ -344,6 +376,7 @@ $server = Server::builder() | `setLogger()` | logger | Set PSR-3 logger | | `setContainer()` | container | Set PSR-11 container | | `setEventDispatcher()` | eventDispatcher | Set PSR-14 event dispatcher, see [Events](../advanced/events.md) | +| `setFiberScheduler()` | scheduler | Wait on an event loop (Revolt, ReactPHP) while a handler awaits I/O, see [Fiber Scheduler](#fiber-scheduler) | | `setSchemaGenerator()` | schemaGenerator | Replace the generator that builds input schemas from PHP types, see [Schema generation](../servers/schemas.md) | | `setSchemaValidator()` | schemaValidator | Replace the validator for tool arguments and structured output, see [Schema generation](../servers/schemas.md) | | `setReferenceHandler()` | referenceHandler | Replace the class that calls the handlers of tools, resources and prompts | diff --git a/src/Server/Builder.php b/src/Server/Builder.php index 13f1eb59..0a0dddd5 100644 --- a/src/Server/Builder.php +++ b/src/Server/Builder.php @@ -44,6 +44,7 @@ use Mcp\Schema\Tool; use Mcp\Schema\ToolAnnotations; use Mcp\Server; +use Mcp\Server\Fiber\FiberSchedulerInterface; use Mcp\Server\Handler\ElementHandlerInterface; use Mcp\Server\Handler\Notification\NotificationHandlerInterface; use Mcp\Server\Handler\PromptHandlerInterface; @@ -135,6 +136,8 @@ final class Builder private float $subscriptionLifetime = 30.0; + private ?FiberSchedulerInterface $fiberScheduler = null; + /** @var array RPC method to the extension identifier defining it */ private array $extensionMethods = []; @@ -344,6 +347,21 @@ public function setSubscriptionLifetime(float $seconds): self return $this; } + /** + * Sets the runtime that may suspend a request handler's fiber besides the server: an event loop. + * + * The server runs every request handler in a fiber, to let it wait on the client. Under an event + * loop — Revolt (AMPHP), ReactPHP's async — a handler awaiting I/O suspends that fiber for the loop: + * the server then waits through this scheduler until the handler suspends for the server again or + * returns, instead of resuming the fiber itself. Without one, such a handler fails the request. + */ + public function setFiberScheduler(FiberSchedulerInterface $scheduler): self + { + $this->fiberScheduler = $scheduler; + + return $this; + } + /** * Sets how long, and to whom, this server's answers may be cached (SEP-2549). * @@ -979,6 +997,7 @@ public function build(): Server : null, requestStateCodec: $this->requestStateCodec(), sessionLock: $this->sessionLock, + fiberScheduler: $this->fiberScheduler, ); $modernVersions = $this->modernVersions ?? ProtocolVersion::modernVersions(); @@ -1018,6 +1037,7 @@ public function buildStateless(array $supportedVersions = [ProtocolVersion::V202 notificationBus: $this->notificationBus, extensionMethods: $this->extensionMethods, eventDispatcher: $parts['eventDispatcher'], + fiberScheduler: $this->fiberScheduler, ); } diff --git a/src/Server/ClientGateway.php b/src/Server/ClientGateway.php index 01ebfc5b..af410f73 100644 --- a/src/Server/ClientGateway.php +++ b/src/Server/ClientGateway.php @@ -39,6 +39,7 @@ use Mcp\Schema\Result\ListRootsResult; use Mcp\Schema\Tool; use Mcp\Schema\ToolChoice; +use Mcp\Server\Fiber\HandlerFibers; use Mcp\Server\Session\SessionInterface; use Mcp\Server\Suspension\NotificationSuspension; use Mcp\Server\Suspension\RequestSuspension; @@ -99,7 +100,7 @@ public function __construct( */ public function notify(Notification $notification): void { - $result = \Fiber::suspend(new NotificationSuspension($notification, $this->session->getId()->toRfc4122())); + $result = HandlerFibers::suspend(new NotificationSuspension($notification, $this->session->getId()->toRfc4122())); if ($result instanceof Error) { throw new ClientException($result); @@ -470,7 +471,7 @@ public function request(Request $request, int $timeout = 120): Response|Error */ private function suspend(Request $request, int $timeout, ?string $key = null): Response|Error { - $response = \Fiber::suspend(new RequestSuspension($request, $this->session->getId()->toRfc4122(), $timeout, $key)); + $response = HandlerFibers::suspend(new RequestSuspension($request, $this->session->getId()->toRfc4122(), $timeout, $key)); if (!$response instanceof Response && !$response instanceof Error) { throw new RuntimeException('Transport returned an unexpected payload; expected a Response or Error message.'); diff --git a/src/Server/Fiber/FiberSchedulerInterface.php b/src/Server/Fiber/FiberSchedulerInterface.php new file mode 100644 index 00000000..5cd8cc40 --- /dev/null +++ b/src/Server/Fiber/FiberSchedulerInterface.php @@ -0,0 +1,35 @@ + */ + private array $callbacks = []; + + public function isSettled(): bool + { + return $this->settled; + } + + /** + * Calls $callback once settled, right away when it already is. + * + * @param \Closure(): void $callback + */ + public function then(\Closure $callback): void + { + if ($this->settled) { + $callback(); + + return; + } + + $this->callbacks[] = $callback; + } + + /** + * @internal called by the server, from the handler fiber + */ + public function settle(): void + { + if ($this->settled) { + return; + } + + $this->settled = true; + $callbacks = $this->callbacks; + $this->callbacks = []; + + foreach ($callbacks as $callback) { + $callback(); + } + } +} diff --git a/src/Server/Fiber/HandlerFiberState.php b/src/Server/Fiber/HandlerFiberState.php new file mode 100644 index 00000000..a5cd6b60 --- /dev/null +++ b/src/Server/Fiber/HandlerFiberState.php @@ -0,0 +1,101 @@ +parked; + } + + public function suspension(): ?object + { + return $this->suspension; + } + + /** + * The handler suspended for the server, on $suspension. + */ + public function parked(object $suspension): void + { + $this->parked = true; + $this->suspension = $suspension; + $this->settlement?->settle(); + } + + /** + * The server resumes the fiber: until it suspends for the server again, it is not. + */ + public function unparked(): void + { + $this->parked = false; + $this->suspension = null; + } + + public function threw(\Throwable $throwable): void + { + $this->throwable = $throwable; + } + + public function ended(): void + { + $this->settlement?->settle(); + } + + /** + * Rethrows, once, what the handler threw. + */ + public function rethrow(): void + { + $throwable = $this->throwable; + $this->throwable = null; + + if (null !== $throwable) { + throw $throwable; + } + } + + public function waitForTheRuntime(): void + { + if (null === $this->scheduler) { + throw new LogicException('A request handler suspended its fiber for something other than the server, an event loop awaiting I/O for instance: set the runtime\'s fiber scheduler with Builder::setFiberScheduler().'); + } + + $this->settlement = new FiberSettlement(); + try { + $this->scheduler->wait($this->settlement); + } finally { + $this->settlement = null; + } + } +} diff --git a/src/Server/Fiber/HandlerFibers.php b/src/Server/Fiber/HandlerFibers.php new file mode 100644 index 00000000..ef38239f --- /dev/null +++ b/src/Server/Fiber/HandlerFibers.php @@ -0,0 +1,138 @@ +, HandlerFiberState>|null */ + private static ?\WeakMap $states = null; + + /** + * A fiber running $handler, driven through this class. + * + * @template TReturn + * + * @param \Closure(): TReturn $handler + * + * @return \Fiber + */ + public static function create(\Closure $handler, ?FiberSchedulerInterface $scheduler = null): \Fiber + { + $state = new HandlerFiberState($scheduler); + $fiber = new \Fiber(static function () use ($handler, $state): mixed { + try { + return $handler(); + } catch (\Throwable $e) { + // Thrown while the runtime drives the fiber, it would escape into the runtime: the server rethrows it. + $state->threw($e); + + return null; + } finally { + $state->ended(); + } + }); + self::states()[$fiber] = $state; + + return $fiber; + } + + /** + * Starts $fiber; returns what it suspended with for the server, or null once it returned. + * + * @param \Fiber $fiber + */ + public static function start(\Fiber $fiber): mixed + { + return self::step($fiber, static fn (): mixed => $fiber->start()); + } + + /** + * Resumes $fiber with $value; returns what it suspended with for the server next, or null once it returned. + * + * @param \Fiber $fiber + */ + public static function resume(\Fiber $fiber, mixed $value = null): mixed + { + return self::step($fiber, static fn (): mixed => $fiber->resume($value)); + } + + /** + * Suspends the current handler fiber for the server, until the server resumes it: how a handler waits on the client. + */ + public static function suspend(object $suspension): mixed + { + $fiber = \Fiber::getCurrent(); + if (null !== $fiber) { + (self::states()[$fiber] ?? null)?->parked($suspension); + } + + return \Fiber::suspend($suspension); + } + + /** + * @param \Fiber $fiber + * @param \Closure(): mixed $step + */ + private static function step(\Fiber $fiber, \Closure $step): mixed + { + $state = self::states()[$fiber] ?? null; + if (null === $state) { + return $step(); + } + + $state->unparked(); + $suspended = $step(); + + // Suspended straight with a server payload, without the gateway: for the server too. + if ($fiber->isSuspended() && !$state->isParked() && ($suspended instanceof NotificationSuspension || $suspended instanceof RequestSuspension)) { + $state->parked($suspended); + } + + while ($fiber->isSuspended() && !$state->isParked()) { + $state->waitForTheRuntime(); + } + + if ($fiber->isTerminated()) { + $state->rethrow(); + + return null; + } + + return $state->suspension() ?? $suspended; + } + + /** + * @return \WeakMap<\Fiber, HandlerFiberState> + */ + private static function states(): \WeakMap + { + return self::$states ??= new \WeakMap(); + } +} diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 1d89caca..80a257e1 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -30,6 +30,8 @@ use Mcp\Schema\JsonRpc\ResultInterface; use Mcp\Schema\Request\InitializeRequest; use Mcp\Server\Authorization\AccessToken; +use Mcp\Server\Fiber\FiberSchedulerInterface; +use Mcp\Server\Fiber\HandlerFibers; use Mcp\Server\Handler\Notification\NotificationHandlerInterface; use Mcp\Server\Handler\Request\RequestHandlerInterface; use Mcp\Server\Session\SessionInterface; @@ -123,6 +125,7 @@ public function __construct( private readonly ?InputRequiredShim $inputRequiredShim = null, private readonly ?RequestStateCodec $requestStateCodec = null, private readonly ?SessionLockInterface $sessionLock = null, + private readonly ?FiberSchedulerInterface $fiberScheduler = null, ) { $this->awaitedRequestIds = new \WeakMap(); $this->fiberOutgoingMessages = new \WeakMap(); @@ -412,7 +415,7 @@ private function handleRequest(TransportInterface $transport, Request $request, // re-enters inside it each round rather than needing a new one. // Handled in the fiber, as the transport resumes it outside the catch below. /** @var McpFiber $fiber */ - $fiber = new \Fiber(static function () use ($handler, $request, $session, $shim, $codec, $dispatchResult, $errorForThrowable, $saveResumedSession, &$suspended): Response|Error { + $fiber = HandlerFibers::create(static function () use ($handler, $request, $session, $shim, $codec, $dispatchResult, $errorForThrowable, $saveResumedSession, &$suspended): Response|Error { try { $result = $handler->handle($request, $session); $result = $shim?->fulfill($result, $handler, $request, $session, $codec) ?? $result; @@ -427,9 +430,9 @@ private function handleRequest(TransportInterface $transport, Request $request, } return $result; - }); + }, $this->fiberScheduler); - $result = $fiber->start(); + $result = HandlerFibers::start($fiber); if ($fiber->isSuspended()) { $beforeSuspension = $session->all(); diff --git a/src/Server/Stateless/StatelessProtocol.php b/src/Server/Stateless/StatelessProtocol.php index 53e4c3b4..b1f72a41 100644 --- a/src/Server/Stateless/StatelessProtocol.php +++ b/src/Server/Stateless/StatelessProtocol.php @@ -32,6 +32,8 @@ use Mcp\Schema\Result\InputRequiredResult; use Mcp\Server\Authorization\AccessToken; use Mcp\Server\Configuration; +use Mcp\Server\Fiber\FiberSchedulerInterface; +use Mcp\Server\Fiber\HandlerFibers; use Mcp\Server\Handler\Request\RequestHandlerInterface; use Mcp\Server\Protocol; use Mcp\Server\Session\InMemorySessionStore; @@ -114,6 +116,7 @@ public function __construct( private readonly ?NotificationBusInterface $notificationBus = null, private readonly array $extensionMethods = [], private readonly ?EventDispatcherInterface $eventDispatcher = null, + private readonly ?FiberSchedulerInterface $fiberScheduler = null, ) { $this->codec = $codec ?? new Rev2026Codec($configuration->serverInfo, $cachePolicy); @@ -672,11 +675,11 @@ private function unknownMethod(string $method, string|int|null $id): Error */ private function run(RequestHandlerInterface $handler, Request $request, Session $session, RequestMeta $meta): \Generator { - $fiber = new \Fiber(static fn (): mixed => $handler->handle($request, $session)); + $fiber = HandlerFibers::create(static fn (): mixed => $handler->handle($request, $session), $this->fiberScheduler); $input = $session->get(InputContext::class); $replay = new ElicitationReplay($input instanceof InputContext ? $input : null, $this->requestStateCodec); - $suspended = $fiber->start(); + $suspended = HandlerFibers::start($fiber); while (!$fiber->isTerminated()) { if (null !== $elicitation = self::readElicitation($suspended)) { @@ -693,7 +696,7 @@ private function run(RequestHandlerInterface $handler, Request $request, Session } } - $suspended = $fiber->resume(new Response($request->getId(), $answer)); + $suspended = HandlerFibers::resume($fiber, new Response($request->getId(), $answer)); continue; } @@ -704,7 +707,7 @@ private function run(RequestHandlerInterface $handler, Request $request, Session yield $notification; } - $suspended = $fiber->resume(null); + $suspended = HandlerFibers::resume($fiber); } /** @var Response|Error $return */ diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php index db496f5f..65f445ad 100644 --- a/src/Server/Transport/BaseTransport.php +++ b/src/Server/Transport/BaseTransport.php @@ -14,6 +14,7 @@ use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\JsonRpc\Response; use Mcp\Server\Authorization\AccessToken; +use Mcp\Server\Fiber\HandlerFibers; use Psr\Log\LoggerInterface; use Psr\Log\NullLogger; use Symfony\Component\Uid\Uuid; @@ -141,7 +142,7 @@ protected function handleFiberYield(mixed $yielded, ?Uuid $sessionId, \Fiber $fi return; } - $yielded = $fiber->resume(Error::forInternalError(self::YIELD_FAILED_MESSAGE)); + $yielded = HandlerFibers::resume($fiber, Error::forInternalError(self::YIELD_FAILED_MESSAGE)); } } diff --git a/src/Server/Transport/StdioTransport.php b/src/Server/Transport/StdioTransport.php index 54ecd2cb..b8f44de3 100644 --- a/src/Server/Transport/StdioTransport.php +++ b/src/Server/Transport/StdioTransport.php @@ -16,6 +16,7 @@ use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\Notification\CancelledNotification; +use Mcp\Server\Fiber\HandlerFibers; use Mcp\Server\Stateless\StatelessProtocol; use Mcp\Server\Transport\Stdio\RunnerControl; use Mcp\Server\Transport\Stdio\RunnerControlInterface; @@ -398,7 +399,7 @@ private function resumeFiber(\Fiber $fiber): void $pendingRequests = $this->getPendingRequests($this->sessionId, $fiber); if (empty($pendingRequests)) { - $yielded = $fiber->resume(); + $yielded = HandlerFibers::resume($fiber); $this->handleFiberYield($yielded, $this->sessionId, $fiber); return; @@ -410,7 +411,7 @@ private function resumeFiber(\Fiber $fiber): void $response = $this->checkForResponse($requestId, $this->sessionId); if (null !== $response) { - $yielded = $fiber->resume($response); + $yielded = HandlerFibers::resume($fiber, $response); $this->handleFiberYield($yielded, $this->sessionId, $fiber); return; diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php index aee34971..c5270296 100644 --- a/src/Server/Transport/StreamableHttpTransport.php +++ b/src/Server/Transport/StreamableHttpTransport.php @@ -16,6 +16,7 @@ use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\JsonRpc\Error; use Mcp\Server\Authorization\AccessToken; +use Mcp\Server\Fiber\HandlerFibers; use Mcp\Server\Stateless\StatelessProtocol; use Mcp\Server\Transport\Http\Middleware\CorsMiddleware; use Mcp\Server\Transport\Http\Middleware\DnsRebindingProtectionMiddleware; @@ -313,7 +314,7 @@ protected function createStreamedResponse(): ResponseInterface $pendingRequests = $this->getPendingRequests($this->sessionId, $fiber); if (empty($pendingRequests)) { - $yielded = $fiber->resume(); + $yielded = HandlerFibers::resume($fiber); $this->handleFiberYield($yielded, $this->sessionId, $fiber); continue; } @@ -325,7 +326,7 @@ protected function createStreamedResponse(): ResponseInterface $response = $this->checkForResponse($requestId, $this->sessionId); if (null !== $response) { - $yielded = $fiber->resume($response); + $yielded = HandlerFibers::resume($fiber, $response); $this->handleFiberYield($yielded, $this->sessionId, $fiber); $resumed = true; break; diff --git a/tests/Unit/Server/Fiber/EventLoopToolCallTest.php b/tests/Unit/Server/Fiber/EventLoopToolCallTest.php new file mode 100644 index 00000000..cda71e7d --- /dev/null +++ b/tests/Unit/Server/Fiber/EventLoopToolCallTest.php @@ -0,0 +1,164 @@ +factory = new Psr17Factory(); + $this->loop = new FakeEventLoop(); + } + + /** + * @return iterable + */ + public static function eras(): iterable + { + yield 'the 2026-07-28 era' => [true]; + yield 'the handshake era' => [false]; + } + + #[DataProvider('eras')] + #[TestDox('a tool awaiting I/O answers with what it awaited, the server waiting on the loop')] + public function testAToolAwaitingIoAnswersWithWhatItAwaited(bool $modern): void + { + $answer = $this->call($this->server(scheduled: true), $modern); + + $this->assertSame('lookup:found', $answer['result']['content'][0]['text'] ?? null); + $this->assertTrue($this->tookTheResult); + $this->assertSame(1, $this->loop->waits); + } + + #[DataProvider('eras')] + #[TestDox('without a scheduler, a tool awaiting I/O fails the call instead of being resumed with nothing')] + public function testWithoutASchedulerTheCallFails(bool $modern): void + { + $answer = $this->call($this->server(scheduled: false), $modern); + + $this->assertSame(-32603, $answer['error']['code'] ?? null); + $this->assertFalse($this->tookTheResult); + } + + private function server(bool $scheduled): Server + { + $loop = $this->loop; + $tookTheResult = &$this->tookTheResult; + $builder = Server::builder() + ->setServerInfo('event-loop-server', '1.0.0') + ->addTool(static function () use ($loop, &$tookTheResult): string { + $found = $loop->await('found'); + $tookTheResult = true; + + return 'lookup:'.$found; + }, name: 'lookup', description: 'Looks something up'); + + if ($scheduled) { + $builder->setFiberScheduler($loop); + } + + return $builder->build(); + } + + /** + * @return array the JSON-RPC answer to a tools/call of the lookup tool + */ + private function call(Server $server, bool $modern): array + { + if ($modern) { + return $this->post($server, [ + 'jsonrpc' => '2.0', + 'id' => 1, + 'method' => 'tools/call', + 'params' => ['name' => 'lookup', 'arguments' => new \stdClass(), '_meta' => [ + RequestMeta::PROTOCOL_VERSION => ProtocolVersion::V2026_07_28->value, + RequestMeta::CLIENT_CAPABILITIES => new \stdClass(), + ]], + ], [ + 'MCP-Protocol-Version' => ProtocolVersion::V2026_07_28->value, + 'Mcp-Method' => 'tools/call', + 'Mcp-Name' => 'lookup', + ])['body']; + } + + $session = $this->post($server, [ + 'jsonrpc' => '2.0', + 'id' => 1, + 'method' => 'initialize', + 'params' => [ + 'protocolVersion' => ProtocolVersion::V2025_11_25->value, + 'capabilities' => new \stdClass(), + 'clientInfo' => ['name' => 'event-loop-client', 'version' => '1.0.0'], + ], + ])['session']; + $headers = ['Mcp-Session-Id' => $session, 'MCP-Protocol-Version' => ProtocolVersion::V2025_11_25->value]; + $this->post($server, ['jsonrpc' => '2.0', 'method' => 'notifications/initialized'], $headers); + + return $this->post($server, [ + 'jsonrpc' => '2.0', + 'id' => 2, + 'method' => 'tools/call', + 'params' => ['name' => 'lookup', 'arguments' => new \stdClass()], + ], $headers)['body']; + } + + /** + * @param array $message + * @param array $headers + * + * @return array{session: string, body: array} + */ + private function post(Server $server, array $message, array $headers = []): array + { + $request = $this->factory->createServerRequest('POST', 'http://localhost/') + ->withHeader('Host', 'localhost') + ->withHeader('Content-Type', 'application/json') + ->withHeader('Accept', 'application/json, text/event-stream') + ->withBody($this->factory->createStream(json_encode($message, \JSON_THROW_ON_ERROR))); + + foreach ($headers as $name => $value) { + $request = $request->withHeader($name, $value); + } + + $response = $server->run(new StreamableHttpTransport($request, $this->factory, $this->factory)); + $payload = (string) $response->getBody(); + $body = '' === $payload ? [] : json_decode($payload, true, 512, \JSON_THROW_ON_ERROR); + + return [ + 'session' => $response->getHeaderLine('Mcp-Session-Id'), + 'body' => \is_array($body) ? $body : [], + ]; + } +} diff --git a/tests/Unit/Server/Fiber/FakeEventLoop.php b/tests/Unit/Server/Fiber/FakeEventLoop.php new file mode 100644 index 00000000..0938cdeb --- /dev/null +++ b/tests/Unit/Server/Fiber/FakeEventLoop.php @@ -0,0 +1,50 @@ + */ + private array $ready = []; + + public int $waits = 0; + + /** + * Awaits $result like I/O: the current fiber is suspended for the loop until a turn resumes it. + */ + public function await(mixed $result): mixed + { + $fiber = \Fiber::getCurrent() ?? throw new \LogicException('Awaiting needs a fiber.'); + $this->ready[] = static function () use ($fiber, $result): void { + $fiber->resume($result); + }; + + return \Fiber::suspend(); + } + + public function wait(FiberSettlement $settlement): void + { + ++$this->waits; + + while (!$settlement->isSettled()) { + $turn = array_shift($this->ready) ?? throw new \LogicException('Nothing left to run: the fiber never settles.'); + $turn(); + } + } +} diff --git a/tests/Unit/Server/Fiber/HandlerFibersTest.php b/tests/Unit/Server/Fiber/HandlerFibersTest.php new file mode 100644 index 00000000..cdbd46bd --- /dev/null +++ b/tests/Unit/Server/Fiber/HandlerFibersTest.php @@ -0,0 +1,124 @@ + 'answered with '.HandlerFibers::suspend($asked)); + + $this->assertSame($asked, HandlerFibers::start($fiber)); + $this->assertNull(HandlerFibers::resume($fiber, 'pong')); + $this->assertSame('answered with pong', $fiber->getReturn()); + } + + #[TestDox('a handler suspending straight with a server payload, without the gateway, still suspends for the server')] + public function testAPayloadSuspendedWithoutTheGatewayIsForTheServer(): void + { + $notification = new NotificationSuspension(new ToolListChangedNotification(), 'session'); + $fiber = HandlerFibers::create(static fn (): mixed => \Fiber::suspend($notification)); + + $this->assertSame($notification, HandlerFibers::start($fiber)); + } + + #[TestDox('without a scheduler, a handler fiber suspended for something else fails, instead of being resumed with nothing')] + public function testAFiberSuspendedForTheRuntimeFailsWithoutAScheduler(): void + { + $resumed = false; + $fiber = HandlerFibers::create(static function () use (&$resumed): void { + \Fiber::suspend(); + $resumed = true; + }); + + try { + HandlerFibers::start($fiber); + $this->fail('The server resumed a fiber it did not suspend.'); + } catch (LogicException $e) { + $this->assertStringContainsString('Builder::setFiberScheduler()', $e->getMessage()); + } + + $this->assertFalse($resumed); + $this->assertTrue($fiber->isSuspended()); + } + + #[TestDox('the server waits on the runtime while the handler awaits I/O, and gets what it returns')] + public function testTheServerWaitsOnTheRuntimeUntilTheHandlerReturns(): void + { + $loop = new FakeEventLoop(); + $fiber = HandlerFibers::create(static fn (): int => $loop->await(40) + $loop->await(2), $loop); + + $this->assertNull(HandlerFibers::start($fiber)); + $this->assertSame(42, $fiber->getReturn()); + $this->assertSame(1, $loop->waits); + } + + #[TestDox('a handler the runtime resumed reaches the server when it waits on the client, and gets the answer')] + public function testASuspensionAfterTheRuntimeResumedTheHandlerReachesTheServer(): void + { + $loop = new FakeEventLoop(); + $asked = new RequestSuspension(new PingRequest(), 'session'); + $fiber = HandlerFibers::create(static function () use ($loop, $asked): string { + $before = $loop->await('read'); + $answer = HandlerFibers::suspend($asked); + + return \sprintf('%s, %s, %s', $before, $answer, $loop->await('written')); + }, $loop); + + $this->assertSame($asked, HandlerFibers::start($fiber)); + $this->assertNull(HandlerFibers::resume($fiber, 'pong')); + $this->assertSame('read, pong, written', $fiber->getReturn()); + $this->assertSame(2, $loop->waits); + } + + #[TestDox('what a handler throws while the runtime drives it reaches the server, not the runtime')] + public function testWhatTheHandlerThrowsWhileTheRuntimeDrivesItReachesTheServer(): void + { + $loop = new FakeEventLoop(); + $fiber = HandlerFibers::create(static function () use ($loop): never { + $loop->await(null); + + throw new \DomainException(LoggingLevel::Error->value); + }, $loop); + + try { + HandlerFibers::start($fiber); + $this->fail('The handler\'s exception was lost.'); + } catch (\DomainException $e) { + $this->assertSame('error', $e->getMessage()); + } + + $this->assertTrue($fiber->isTerminated()); + } + + #[TestDox('a fiber the server did not create is started and resumed as is')] + public function testAFiberNotCreatedHereIsDrivenAsIs(): void + { + $fiber = new \Fiber(static fn (): mixed => \Fiber::suspend('anything')); + + $this->assertSame('anything', HandlerFibers::start($fiber)); + $this->assertNull(HandlerFibers::resume($fiber)); + $this->assertTrue($fiber->isTerminated()); + } +} From 3cba941be798c644209118826bef8a62aae847ac Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Tue, 27 Oct 2026 22:48:58 +0100 Subject: [PATCH 2/2] [Server] Add a Revolt event loop example --- composer.json | 4 ++ examples/server/README.md | 13 ++++ .../server/revolt/RevoltFiberScheduler.php | 29 ++++++++ examples/server/revolt/SlowTools.php | 70 +++++++++++++++++++ examples/server/revolt/server.php | 69 ++++++++++++++++++ .../Http/HttpInspectorSnapshotTestCase.php | 15 ++-- tests/Interop/Server/Http/HttpRevoltTest.php | 48 +++++++++++++ .../HttpRevoltTest-prompts_list.json | 4 ++ .../HttpRevoltTest-resources_list.json | 4 ++ ...tpRevoltTest-resources_templates_list.json | 4 ++ ...pRevoltTest-tools_call-check_services.json | 15 ++++ .../HttpRevoltTest-tools_call-wait.json | 10 +++ .../snapshots/HttpRevoltTest-tools_list.json | 29 ++++++++ 13 files changed, 308 insertions(+), 6 deletions(-) create mode 100644 examples/server/revolt/RevoltFiberScheduler.php create mode 100644 examples/server/revolt/SlowTools.php create mode 100644 examples/server/revolt/server.php create mode 100644 tests/Interop/Server/Http/HttpRevoltTest.php create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-prompts_list.json create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_list.json create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_templates_list.json create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-check_services.json create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-wait.json create mode 100644 tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_list.json diff --git a/composer.json b/composer.json index be233eee..f7bdc336 100644 --- a/composer.json +++ b/composer.json @@ -43,6 +43,8 @@ }, "require-dev": { "ext-openssl": "*", + "amphp/amp": "^3.0", + "amphp/http-server": "^3.3", "composer/semver": "^3.0", "firebase/php-jwt": "^6.10 || ^7.0", "laminas/laminas-httphandlerrunner": "^2.12", @@ -55,6 +57,7 @@ "phpunit/phpunit": "^10.5", "psr/cache": "^1.0 || ^2.0 || ^3.0", "psr/simple-cache": "^2.0 || ^3.0", + "revolt/event-loop": "^1.0", "symfony/cache": "^5.4 || ^6.4 || ^7.3 || ^8.0", "symfony/console": "^5.4 || ^6.4 || ^7.3 || ^8.0", "symfony/finder": "^5.4 || ^6.4 || ^7.3 || ^8.0", @@ -88,6 +91,7 @@ "Mcp\\Example\\Server\\McpApps\\": "examples/server/mcp-apps/", "Mcp\\Example\\Server\\OAuthKeycloak\\": "examples/server/oauth-keycloak/", "Mcp\\Example\\Server\\OAuthMicrosoft\\": "examples/server/oauth-microsoft/", + "Mcp\\Example\\Server\\Revolt\\": "examples/server/revolt/", "Mcp\\Example\\Server\\SchemaShowcase\\": "examples/server/schema-showcase/", "Mcp\\Tests\\": "tests/" }, diff --git a/examples/server/README.md b/examples/server/README.md index 4fa3e88e..57328617 100644 --- a/examples/server/README.md +++ b/examples/server/README.md @@ -49,6 +49,19 @@ curl -sS http://127.0.0.1:8000/ \ `tests/Integration/StatelessLifecycleTest.php` drives this example end to end — discovery, a tool call, the multi round-trip flow, and the response stream carrying progress and log notifications. +## Event loop + +`revolt/server.php` runs the server on the Revolt event loop with `amphp/http-server`. It is a +long-running process, so you start it directly instead of with `php -S`: + +```bash +php examples/server/revolt/server.php 127.0.0.1:8000 +``` + +Its tools wait on the loop with `Amp\delay()`, and other requests are served in the meantime. The +`RevoltFiberScheduler` set with `Builder::setFiberScheduler()` makes this work, see +[Fiber Scheduler](../../docs/run/server-builder.md#fiber-scheduler). + ## Debugging You can enable debug output by setting the `DEBUG` environment variable to `1`, and additionally log to a file by diff --git a/examples/server/revolt/RevoltFiberScheduler.php b/examples/server/revolt/RevoltFiberScheduler.php new file mode 100644 index 00000000..ad7a90de --- /dev/null +++ b/examples/server/revolt/RevoltFiberScheduler.php @@ -0,0 +1,29 @@ +then($suspension->resume(...)); + $suspension->suspend(); + } +} diff --git a/examples/server/revolt/SlowTools.php b/examples/server/revolt/SlowTools.php new file mode 100644 index 00000000..226224f4 --- /dev/null +++ b/examples/server/revolt/SlowTools.php @@ -0,0 +1,70 @@ + latency of each simulated service, in seconds */ + private const SERVICES = [ + 'database' => 0.3, + 'cache' => 0.1, + 'search' => 0.2, + ]; + + public function __construct( + private readonly LoggerInterface $logger, + ) { + } + + /** + * Waits on the event loop without blocking other requests. + * + * @param float $seconds how long to wait, at most 5 seconds + */ + public function wait(float $seconds): string + { + $seconds = max(0.0, min($seconds, 5.0)); + $this->logger->info('Waiting on the event loop.', ['seconds' => $seconds]); + + delay($seconds); + + return \sprintf('Waited %s seconds on the event loop.', $seconds); + } + + /** + * Checks all services concurrently, in the time of the slowest one. + * + * @return array the status of each service + */ + public function checkServices(): array + { + $checks = []; + foreach (self::SERVICES as $service => $latency) { + $checks[$service] = async(static function () use ($latency): string { + delay($latency); + + return 'up'; + }); + } + + return array_map(static fn (Future $check): string => $check->await(), $checks); + } +} diff --git a/examples/server/revolt/server.php b/examples/server/revolt/server.php new file mode 100644 index 00000000..f3642acb --- /dev/null +++ b/examples/server/revolt/server.php @@ -0,0 +1,69 @@ +setServerInfo('Revolt Server', '1.0.0') + ->setLogger(logger()) + ->setContainer(container()) + ->setFiberScheduler(new RevoltFiberScheduler()) + ->addTool([SlowTools::class, 'wait'], 'wait') + ->addTool([SlowTools::class, 'checkServices'], 'check_services') + ->build(); + +$factory = new Psr17Factory(); + +$http = SocketHttpServer::createForDirectAccess(logger()); +$http->expose($address); +$http->start(new ClosureRequestHandler(static function (Request $request) use ($server, $factory): Response { + $psrRequest = $factory->createServerRequest($request->getMethod(), (string) $request->getUri()) + ->withBody($factory->createStream($request->getBody()->buffer())); + foreach ($request->getHeaders() as $name => $values) { + $psrRequest = $psrRequest->withHeader($name, $values); + } + + $response = $server->run(new StreamableHttpTransport($psrRequest, $factory, $factory, logger())); + + return new Response($response->getStatusCode(), $response->getHeaders(), (string) $response->getBody()); +}), new DefaultErrorHandler()); + +logger()->info(sprintf('Revolt MCP server listening on http://%s', $address)); + +EventLoop::run(); diff --git a/tests/Interop/Server/Http/HttpInspectorSnapshotTestCase.php b/tests/Interop/Server/Http/HttpInspectorSnapshotTestCase.php index 525d2ec7..4410af7c 100644 --- a/tests/Interop/Server/Http/HttpInspectorSnapshotTestCase.php +++ b/tests/Interop/Server/Http/HttpInspectorSnapshotTestCase.php @@ -31,6 +31,14 @@ protected function tearDown(): void abstract protected function getServerScript(): string; + /** + * @return list + */ + protected function getServerCommand(string $address): array + { + return ['php', '-S', $address, $this->getServerScript()]; + } + protected function getServerConnectionArgs(): array { return [\sprintf('http://127.0.0.1:%d', $this->serverPort)]; @@ -45,12 +53,7 @@ private function startServer(): void { $this->serverPort = 8000 + (getmypid() % 1000); - $this->serverProcess = new Process([ - 'php', - '-S', - \sprintf('127.0.0.1:%d', $this->serverPort), - $this->getServerScript(), - ]); + $this->serverProcess = new Process($this->getServerCommand(\sprintf('127.0.0.1:%d', $this->serverPort))); $this->serverProcess->start(); diff --git a/tests/Interop/Server/Http/HttpRevoltTest.php b/tests/Interop/Server/Http/HttpRevoltTest.php new file mode 100644 index 00000000..b38af2f4 --- /dev/null +++ b/tests/Interop/Server/Http/HttpRevoltTest.php @@ -0,0 +1,48 @@ + [ + 'method' => 'tools/call', + 'options' => [ + 'toolName' => 'wait', + 'toolArgs' => ['seconds' => '0.2'], + ], + 'testName' => 'wait', + ], + 'Check Services Tool' => [ + 'method' => 'tools/call', + 'options' => [ + 'toolName' => 'check_services', + 'toolArgs' => [], + ], + 'testName' => 'check_services', + ], + ]; + } + + protected function getServerScript(): string + { + return \dirname(__DIR__, 4).'/examples/server/revolt/server.php'; + } + + protected function getServerCommand(string $address): array + { + return ['php', $this->getServerScript(), $address]; + } +} diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-prompts_list.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-prompts_list.json new file mode 100644 index 00000000..911451f2 --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-prompts_list.json @@ -0,0 +1,4 @@ +{ + "prompts": [] +} + diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_list.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_list.json new file mode 100644 index 00000000..f146635d --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_list.json @@ -0,0 +1,4 @@ +{ + "resources": [] +} + diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_templates_list.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_templates_list.json new file mode 100644 index 00000000..b921e802 --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-resources_templates_list.json @@ -0,0 +1,4 @@ +{ + "resourceTemplates": [] +} + diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-check_services.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-check_services.json new file mode 100644 index 00000000..3b3a88d1 --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-check_services.json @@ -0,0 +1,15 @@ +{ + "content": [ + { + "type": "text", + "text": "{\n \"database\": \"up\",\n \"cache\": \"up\",\n \"search\": \"up\"\n}" + } + ], + "structuredContent": { + "database": "up", + "cache": "up", + "search": "up" + }, + "isError": false +} + diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-wait.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-wait.json new file mode 100644 index 00000000..e418640d --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_call-wait.json @@ -0,0 +1,10 @@ +{ + "content": [ + { + "type": "text", + "text": "Waited 0.2 seconds on the event loop." + } + ], + "isError": false +} + diff --git a/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_list.json b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_list.json new file mode 100644 index 00000000..640c2c65 --- /dev/null +++ b/tests/Interop/Server/Http/snapshots/HttpRevoltTest-tools_list.json @@ -0,0 +1,29 @@ +{ + "tools": [ + { + "name": "wait", + "description": "Waits on the event loop without blocking other requests.", + "inputSchema": { + "type": "object", + "properties": { + "seconds": { + "type": "number", + "description": "how long to wait, at most 5 seconds" + } + }, + "required": [ + "seconds" + ] + } + }, + { + "name": "check_services", + "description": "Checks all services concurrently, in the time of the slowest one.", + "inputSchema": { + "type": "object", + "properties": {} + } + } + ] +} +