modelcontextprotocol/php-sdk
Watch
1
0
Fork
You've already forked php-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Decouple Server from TransportInterface for reusability (#160)

Browse source
This commit is contained in:
落月 2025-11-30 18:21:21 +08:00 • committed by GitHub
commit 280e8b62be
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 77 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

95
src/Server/Protocol.php
View file

@ -32,8 +32,8 @@ use Symfony\Component\Uid\Uuid;
/**
* @final
*
* @phpstan-import-type McpFiber from \Mcp\Server\Transport\TransportInterface
* @phpstan-import-type FiberSuspend from \Mcp\Server\Transport\TransportInterface
* @phpstan-import-type McpFiber from TransportInterface
* @phpstan-import-type FiberSuspend from TransportInterface
*
* @author Christopher Hertel <mail@christopher-hertel.de>
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
@ -55,9 +55,6 @@ class Protocol
/** Session key for active request meta */
public const SESSION_ACTIVE_REQUEST_META = '_mcp.active_request_meta';
/** @var TransportInterface<mixed>|null */
private ?TransportInterface $transport = null;
/**
* @param array<int, RequestHandlerInterface<ResultInterface|array<string, mixed>>> $requestHandlers
* @param array<int, NotificationHandlerInterface> $notificationHandlers
@ -73,15 +70,7 @@ class Protocol
}
/**
* @return TransportInterface<mixed>
*/
public function getTransport(): TransportInterface
{
return $this->transport;
}
/**
* Connect this protocol to a transport.
* Connect this protocol to transport.
*
* The protocol takes ownership of the transport and sets up all callbacks.
*
@ -89,23 +78,17 @@ class Protocol
*/
public function connect(TransportInterface $transport): void
{
if ($this->transport) {
throw new \RuntimeException('Protocol already connected to a transport');
}
$transport->onMessage($this->processInput(...));
$this->transport = $transport;
$transport->onSessionEnd($this->destroySession(...));
$this->transport->onMessage([$this, 'processInput']);
$transport->setOutgoingMessagesProvider($this->consumeOutgoingMessages(...));
$this->transport->onSessionEnd([$this, 'destroySession']);
$transport->setPendingRequestsProvider($this->getPendingRequests(...));
$this->transport->setOutgoingMessagesProvider([$this, 'consumeOutgoingMessages']);
$transport->setResponseFinder($this->checkResponse(...));
$this->transport->setPendingRequestsProvider([$this, 'getPendingRequests']);
$this->transport->setResponseFinder([$this, 'checkResponse']);
$this->transport->setFiberYieldHandler([$this, 'handleFiberYield']);
$transport->setFiberYieldHandler($this->handleFiberYield(...));
$this->logger->info('Protocol connected to transport', ['transport' => $transport::class]);
}
@ -114,8 +97,10 @@ class Protocol
* Handle an incoming message from the transport.
*
* This is called by the transport whenever ANY message arrives.
*
* @param TransportInterface<mixed> $transport
*/
public function processInput(string $input, ?Uuid $sessionId): void
public function processInput(TransportInterface $transport, string $input, ?Uuid $sessionId): void
{
$this->logger->info('Received message to process.', ['message' => $input]);
@ -126,21 +111,21 @@ class Protocol
} catch (\JsonException $e) {
$this->logger->warning('Failed to decode json message.', ['exception' => $e]);
$error = Error::forParseError($e->getMessage());
$this->sendResponse($error, null);
$this->sendResponse($transport, $error, null);
return;
}
$session = $this->resolveSession($sessionId, $messages);
$session = $this->resolveSession($transport, $sessionId, $messages);
if (null === $session) {
return;
}
foreach ($messages as $message) {
if ($message instanceof InvalidInputMessageException) {
$this->handleInvalidMessage($message, $session);
$this->handleInvalidMessage($transport, $message, $session);
} elseif ($message instanceof Request) {
$this->handleRequest($message, $session);
$this->handleRequest($transport, $message, $session);
} elseif ($message instanceof Response || $message instanceof Error) {
$this->handleResponse($message, $session);
} elseif ($message instanceof Notification) {
@ -151,15 +136,25 @@ class Protocol
$session->save();
}
private function handleInvalidMessage(InvalidInputMessageException $exception, SessionInterface $session): void
/**
* Handle an invalid message from the transport.
*
* @param TransportInterface<mixed> $transport
*/
private function handleInvalidMessage(TransportInterface $transport, InvalidInputMessageException $exception, SessionInterface $session): void
{
$this->logger->warning('Failed to create message.', ['exception' => $exception]);
$error = Error::forInvalidRequest($exception->getMessage());
$this->sendResponse($error, $session);
$this->sendResponse($transport, $error, $session);
}
private function handleRequest(Request $request, SessionInterface $session): void
/**
* Handle a request from the transport.
*
* @param TransportInterface<mixed> $transport
*/
private function handleRequest(TransportInterface $transport, Request $request, SessionInterface $session): void
{
$this->logger->info('Handling request.', ['request' => $request]);
@ -192,24 +187,24 @@ class Protocol
}
}
$this->transport->attachFiberToSession($fiber, $session->getId());
$transport->attachFiberToSession($fiber, $session->getId());
return;
} else {
$finalResult = $fiber->getReturn();
$this->sendResponse($finalResult, $session);
$this->sendResponse($transport, $finalResult, $session);
}
} catch (\InvalidArgumentException $e) {
$this->logger->warning(\sprintf('Invalid argument: %s', $e->getMessage()), ['exception' => $e]);
$error = Error::forInvalidParams($e->getMessage(), $request->getId());
$this->sendResponse($error, $session);
$this->sendResponse($transport, $error, $session);
} catch (\Throwable $e) {
$this->logger->error(\sprintf('Uncaught exception: %s', $e->getMessage()), ['exception' => $e]);
$error = Error::forInternalError($e->getMessage(), $request->getId());
$this->sendResponse($error, $session);
$this->sendResponse($transport, $error, $session);
}
break;
@ -217,7 +212,7 @@ class Protocol
if (!$handlerFound) {
$error = Error::forMethodNotFound(\sprintf('No handler found for method "%s".', $request::getMethod()), $request->getId());
$this->sendResponse($error, $session);
$this->sendResponse($transport, $error, $session);
}
}
@ -299,10 +294,11 @@ class Protocol
/**
* Sends a response either immediately or queued for later delivery.
*
* @param TransportInterface<mixed> $transport
* @param Response<ResultInterface|array<string, mixed>>|Error $response
* @param array<string, mixed> $context
*/
private function sendResponse(Response|Error $response, ?SessionInterface $session, array $context = []): void
private function sendResponse(TransportInterface $transport, Response|Error $response, ?SessionInterface $session, array $context = []): void
{
if (null === $session) {
$this->logger->info('Sending immediate response', [
@ -327,7 +323,7 @@ class Protocol
}
$context['type'] = 'response';
$this->transport->send($encoded, $context);
$transport->send($encoded, $context);
} else {
$this->logger->info('Queueing server response', [
'response_id' => $response->getId(),
@ -519,16 +515,17 @@ class Protocol
/**
* Resolves and validates the session based on the request context.
*
* @param Uuid|null $sessionId The session ID from the transport
* @param array<int,mixed> $messages The parsed messages
* @param TransportInterface<mixed> $transport
* @param Uuid|null $sessionId The session ID from the transport
* @param array<int,mixed> $messages The parsed messages
*/
private function resolveSession(?Uuid $sessionId, array $messages): ?SessionInterface
private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, array $messages): ?SessionInterface
{
if ($this->hasInitializeRequest($messages)) {
// Spec: An initialize request must not be part of a batch.
if (\count($messages) > 1) {
$error = Error::forInvalidRequest('The "initialize" request MUST NOT be part of a batch.');
$this->sendResponse($error, null);
$this->sendResponse($transport, $error, null);
return null;
}
@ -536,7 +533,7 @@ class Protocol
// Spec: An initialize request must not have a session ID.
if ($sessionId) {
$error = Error::forInvalidRequest('A session ID MUST NOT be sent with an "initialize" request.');
$this->sendResponse($error, null);
$this->sendResponse($transport, $error, null);
return null;
}
@ -546,21 +543,21 @@ class Protocol
'session_id' => $session->getId()->toRfc4122(),
]);
$this->transport->setSessionId($session->getId());
$transport->setSessionId($session->getId());
return $session;
}
if (!$sessionId) {
$error = Error::forInvalidRequest('A valid session id is REQUIRED for non-initialize requests.');
$this->sendResponse($error, null, ['status_code' => 400]);
$this->sendResponse($transport, $error, null, ['status_code' => 400]);
return null;
}
if (!$this->sessionStore->exists($sessionId)) {
$error = Error::forInvalidRequest('Session not found or has expired.');
$this->sendResponse($error, null, ['status_code' => 404]);
$this->sendResponse($transport, $error, null, ['status_code' => 404]);
return null;
}

8
src/Server/Transport/BaseTransport.php
View file

@ -26,9 +26,13 @@ use Symfony\Component\Uid\Uuid;
* @phpstan-import-type FiberSuspend from TransportInterface
* @phpstan-import-type McpFiber from TransportInterface
*
* @template TResult
*
* @implements TransportInterface<TResult>
*
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
*/
abstract class BaseTransport
abstract class BaseTransport implements TransportInterface
{
use ManagesTransportCallbacks;
@ -126,7 +130,7 @@ abstract class BaseTransport
protected function handleMessage(string $payload, ?Uuid $sessionId): void
{
if (\is_callable($this->messageListener)) {
($this->messageListener)($payload, $sessionId);
($this->messageListener)($this, $payload, $sessionId);
}
}

4
src/Server/Transport/InMemoryTransport.php
View file

@ -15,11 +15,11 @@ use Psr\Log\LoggerInterface;
use Symfony\Component\Uid\Uuid;
/**
* @implements TransportInterface<null>
* @extends BaseTransport<null>
*
* @author Tobias Nyholm <tobias.nyholm@gmail.com>
*/
class InMemoryTransport extends BaseTransport implements TransportInterface
class InMemoryTransport extends BaseTransport
{
/**
* @param list<string> $messages

2
src/Server/Transport/ManagesTransportCallbacks.php
View file

@ -26,7 +26,7 @@ use Symfony\Component\Uid\Uuid;
* */
trait ManagesTransportCallbacks
{
/** @var callable(string, ?Uuid): void */
/** @var callable(TransportInterface<mixed>, string, ?Uuid): void */
protected $messageListener;
/** @var callable(Uuid): void */

6
src/Server/Transport/StdioTransport.php
View file

@ -15,11 +15,11 @@ use Mcp\Schema\JsonRpc\Error;
use Psr\Log\LoggerInterface;
/**
* @implements TransportInterface<int>
* @extends BaseTransport<int>
*
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
* */
class StdioTransport extends BaseTransport implements TransportInterface
*/
class StdioTransport extends BaseTransport
{
/**
* @param resource $input

6
src/Server/Transport/StreamableHttpTransport.php
View file

@ -21,11 +21,11 @@ use Psr\Log\LoggerInterface;
use Symfony\Component\Uid\Uuid;
/**
* @implements TransportInterface<ResponseInterface>
* @extends BaseTransport<ResponseInterface>
*
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
* */
class StreamableHttpTransport extends BaseTransport implements TransportInterface
*/
class StreamableHttpTransport extends BaseTransport
{
private ResponseFactoryInterface $responseFactory;
private StreamFactoryInterface $streamFactory;

2
src/Server/Transport/TransportInterface.php
View file

@ -70,7 +70,7 @@ interface TransportInterface
*
* The transport calls this whenever ANY message arrives, regardless of source.
*
* @param callable(string $message, ?Uuid $sessionId): void $listener
* @param callable(TransportInterface<TResult> $transport, string $message, ?Uuid $sessionId): void $listener
*/
public function onMessage(callable $listener): void;

45
tests/Unit/Server/ProtocolTest.php
View file

@ -68,10 +68,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "method": "notifications/initialized"}',
$sessionId
);
@ -127,10 +126,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
$sessionId
);
@ -166,10 +164,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "1.0"}}}',
$sessionId
);
@ -198,9 +195,8 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$protocol->processInput(
$this->transport,
'[{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "1.0"}}}, {"jsonrpc": "2.0", "method": "ping", "id": 2}]',
null
);
@ -231,9 +227,8 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
null
);
@ -266,10 +261,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
$sessionId
);
@ -298,9 +292,8 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$protocol->processInput(
$this->transport,
'invalid json',
null
);
@ -343,10 +336,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "params": {}}',
$sessionId
);
@ -397,10 +389,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "ping"}',
$sessionId
);
@ -456,10 +447,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "test"}}',
$sessionId
);
@ -515,10 +505,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "test"}}',
$sessionId
);
@ -553,10 +542,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "method": "notifications/initialized"}',
$sessionId
);
@ -607,9 +595,8 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
$sessionId
);
@ -670,10 +657,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'[{"jsonrpc": "2.0", "method": "tools/list", "id": 1}, {"jsonrpc": "2.0", "method": "prompts/list", "id": 2}]',
$sessionId
);
@ -706,10 +692,9 @@ final class ProtocolTest extends TestCase
sessionStore: $this->sessionStore,
);
$protocol->connect($this->transport);
$sessionId = Uuid::v4();
$protocol->processInput(
$this->transport,
'{"jsonrpc": "2.0", "method": "notifications/initialized"}',
$sessionId
);