feat(server): Rework transport architecture and add StreamableHttpTransport (#49)
* feat(server): Rework transport architecture and add StreamableHttpTransport - `Server::connect()` no longer contains a processing loop. - `TransportInterface` is updated with `setMessageHandler()` and `listen()`. - `StdioTransport` is updated to implement the new interface. - A new, minimal `StreamableHttpTransport` is added for stateless HTTP. * refactor(server): use event-driven message handling * feat(server): Introduce a formal session management system * fix: ensure session elements are preserved when building server (regression from #46) * refactor: consolidate HTTP example to use shared dependencies * feat(server): Enhance message handling with session support - Updated `TransportInterface` to use `onMessage` for handling incoming messages with session IDs. - Refactored `Server`, `Handler`, and transport classes to accommodate session management using `Uuid`. - Introduced methods for creating sessions with auto-generated and specific UUIDs in `SessionFactory` and `SessionFactoryInterface`. * feat(server): Integrate session management in message handling - Added session support to the `Server` and `Handler` classes, allowing for session data to be managed during message processing. - Updated `TransportInterface` to include session context in the `send` method. - Refactored various request handlers to utilize session information, ensuring proper session handling for incoming requests. - Introduced a file-based session store for persistent session data management * feat: Update session builder method to use set* instead of with* * feat: Fix test compatibility with new session management architecture * chore: apply code style fixes * fix: remove deprecated SSE transport and resolve PHPStan issues
This commit is contained in:
Kyrian Obikwelu
GitHub
parent
b1e54f1f86
commit
719262b15d
6 changed files with 1614 additions and 406 deletions
|
|
@ -12,13 +12,20 @@
|
|||
namespace Mcp\Server\Transport;
|
||||
|
||||
use Mcp\Server\TransportInterface;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* @author Tobias Nyholm <tobias.nyholm@gmail.com>
|
||||
*/
|
||||
class InMemoryTransport implements TransportInterface
|
||||
{
|
||||
private bool $connected = true;
|
||||
/** @var callable(string, ?Uuid): void */
|
||||
private $messageListener;
|
||||
|
||||
/** @var callable(Uuid): void */
|
||||
private $sessionDestroyListener;
|
||||
|
||||
private ?Uuid $sessionId = null;
|
||||
|
||||
/**
|
||||
* @param list<string> $messages
|
||||
|
|
@ -32,22 +39,42 @@ class InMemoryTransport implements TransportInterface
|
|||
{
|
||||
}
|
||||
|
||||
public function isConnected(): bool
|
||||
public function onMessage(callable $listener): void
|
||||
{
|
||||
return $this->connected;
|
||||
$this->messageListener = $listener;
|
||||
}
|
||||
|
||||
public function receive(): \Generator
|
||||
public function send(string $data, array $context): void
|
||||
{
|
||||
yield from $this->messages;
|
||||
$this->connected = false;
|
||||
if (isset($context['session_id'])) {
|
||||
$this->sessionId = $context['session_id'];
|
||||
}
|
||||
}
|
||||
|
||||
public function send(string $data): void
|
||||
public function listen(): mixed
|
||||
{
|
||||
foreach ($this->messages as $message) {
|
||||
if (\is_callable($this->messageListener)) {
|
||||
\call_user_func($this->messageListener, $message, $this->sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
if (\is_callable($this->sessionDestroyListener) && null !== $this->sessionId) {
|
||||
\call_user_func($this->sessionDestroyListener, $this->sessionId);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
public function onSessionEnd(callable $listener): void
|
||||
{
|
||||
$this->sessionDestroyListener = $listener;
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
if (\is_callable($this->sessionDestroyListener) && null !== $this->sessionId) {
|
||||
\call_user_func($this->sessionDestroyListener, $this->sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,65 +0,0 @@
|
|||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the official PHP MCP SDK.
|
||||
*
|
||||
* A collaboration between Symfony and the PHP Foundation.
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Mcp\Server\Transport\Sse\Store;
|
||||
|
||||
use Mcp\Server\Transport\Sse\StoreInterface;
|
||||
use Psr\Cache\CacheItemPoolInterface;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* @author Christopher Hertel <mail@christopher-hertel.de>
|
||||
*/
|
||||
final class CachePoolStore implements StoreInterface
|
||||
{
|
||||
public function __construct(
|
||||
private readonly CacheItemPoolInterface $cachePool,
|
||||
) {
|
||||
}
|
||||
|
||||
public function push(Uuid $id, string $message): void
|
||||
{
|
||||
$item = $this->cachePool->getItem($this->getCacheKey($id));
|
||||
|
||||
$messages = $item->isHit() ? $item->get() : [];
|
||||
$messages[] = $message;
|
||||
$item->set($messages);
|
||||
|
||||
$this->cachePool->save($item);
|
||||
}
|
||||
|
||||
public function pop(Uuid $id): ?string
|
||||
{
|
||||
$item = $this->cachePool->getItem($this->getCacheKey($id));
|
||||
|
||||
if (!$item->isHit()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$messages = $item->get();
|
||||
$message = array_shift($messages);
|
||||
|
||||
$item->set($messages);
|
||||
$this->cachePool->save($item);
|
||||
|
||||
return $message;
|
||||
}
|
||||
|
||||
public function remove(Uuid $id): void
|
||||
{
|
||||
$this->cachePool->deleteItem($this->getCacheKey($id));
|
||||
}
|
||||
|
||||
private function getCacheKey(Uuid $id): string
|
||||
{
|
||||
return 'message_'.$id->toRfc4122();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the official PHP MCP SDK.
|
||||
*
|
||||
* A collaboration between Symfony and the PHP Foundation.
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Mcp\Server\Transport\Sse;
|
||||
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* @author Christopher Hertel <mail@christopher-hertel.de>
|
||||
*/
|
||||
interface StoreInterface
|
||||
{
|
||||
public function push(Uuid $id, string $message): void;
|
||||
|
||||
public function pop(Uuid $id): ?string;
|
||||
|
||||
public function remove(Uuid $id): void;
|
||||
}
|
||||
|
|
@ -1,65 +0,0 @@
|
|||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the official PHP MCP SDK.
|
||||
*
|
||||
* A collaboration between Symfony and the PHP Foundation.
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Mcp\Server\Transport\Sse;
|
||||
|
||||
use Mcp\Server\TransportInterface;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* @author Christopher Hertel <mail@christopher-hertel.de>
|
||||
*/
|
||||
final class StreamTransport implements TransportInterface
|
||||
{
|
||||
public function __construct(
|
||||
private readonly string $messageEndpoint,
|
||||
private readonly StoreInterface $store,
|
||||
private readonly Uuid $id,
|
||||
) {
|
||||
}
|
||||
|
||||
public function initialize(): void
|
||||
{
|
||||
ignore_user_abort(true);
|
||||
$this->flushEvent('endpoint', $this->messageEndpoint);
|
||||
}
|
||||
|
||||
public function isConnected(): bool
|
||||
{
|
||||
return 0 === connection_aborted();
|
||||
}
|
||||
|
||||
public function receive(): \Generator
|
||||
{
|
||||
yield $this->store->pop($this->id);
|
||||
}
|
||||
|
||||
public function send(string $data): void
|
||||
{
|
||||
$this->flushEvent('message', $data);
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
$this->store->remove($this->id);
|
||||
}
|
||||
|
||||
private function flushEvent(string $event, string $data): void
|
||||
{
|
||||
echo \sprintf('event: %s', $event).\PHP_EOL;
|
||||
echo \sprintf('data: %s', $data).\PHP_EOL;
|
||||
echo \PHP_EOL;
|
||||
if (false !== ob_get_length()) {
|
||||
ob_flush();
|
||||
}
|
||||
flush();
|
||||
}
|
||||
}
|
||||
|
|
@ -14,13 +14,20 @@ namespace Mcp\Server\Transport;
|
|||
use Mcp\Server\TransportInterface;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Psr\Log\NullLogger;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* Heavily inspired by https://jolicode.com/blog/mcp-the-open-protocol-that-turns-llm-chatbots-into-intelligent-agents.
|
||||
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
|
||||
*/
|
||||
class StdioTransport implements TransportInterface
|
||||
{
|
||||
private string $buffer = '';
|
||||
/** @var callable(string, ?Uuid): void */
|
||||
private $messageListener;
|
||||
|
||||
/** @var callable(Uuid): void */
|
||||
private $sessionEndListener;
|
||||
|
||||
private ?Uuid $sessionId = null;
|
||||
|
||||
/**
|
||||
* @param resource $input
|
||||
|
|
@ -37,39 +44,67 @@ class StdioTransport implements TransportInterface
|
|||
{
|
||||
}
|
||||
|
||||
public function isConnected(): bool
|
||||
public function onMessage(callable $listener): void
|
||||
{
|
||||
return true;
|
||||
$this->messageListener = $listener;
|
||||
}
|
||||
|
||||
public function receive(): \Generator
|
||||
{
|
||||
$line = fgets($this->input);
|
||||
|
||||
$this->logger->debug('Received message on StdioTransport.', [
|
||||
'line' => $line,
|
||||
]);
|
||||
|
||||
if (false === $line) {
|
||||
return;
|
||||
}
|
||||
$this->buffer .= rtrim($line).\PHP_EOL;
|
||||
if (str_contains($this->buffer, \PHP_EOL)) {
|
||||
$lines = explode(\PHP_EOL, $this->buffer);
|
||||
$this->buffer = array_pop($lines);
|
||||
|
||||
yield from $lines;
|
||||
}
|
||||
}
|
||||
|
||||
public function send(string $data): void
|
||||
public function send(string $data, array $context): void
|
||||
{
|
||||
$this->logger->debug('Sending data to client via StdioTransport.', ['data' => $data]);
|
||||
|
||||
if (isset($context['session_id'])) {
|
||||
$this->sessionId = $context['session_id'];
|
||||
}
|
||||
|
||||
fwrite($this->output, $data.\PHP_EOL);
|
||||
}
|
||||
|
||||
public function listen(): mixed
|
||||
{
|
||||
$this->logger->info('StdioTransport is listening for messages on STDIN...');
|
||||
|
||||
while (!feof($this->input)) {
|
||||
$line = fgets($this->input);
|
||||
if (false === $line) {
|
||||
break;
|
||||
}
|
||||
|
||||
$trimmedLine = trim($line);
|
||||
if (!empty($trimmedLine)) {
|
||||
$this->logger->debug('Received message on StdioTransport.', ['line' => $trimmedLine]);
|
||||
if (\is_callable($this->messageListener)) {
|
||||
\call_user_func($this->messageListener, $trimmedLine, $this->sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$this->logger->info('StdioTransport finished listening.');
|
||||
|
||||
if (\is_callable($this->sessionEndListener) && null !== $this->sessionId) {
|
||||
\call_user_func($this->sessionEndListener, $this->sessionId);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
public function onSessionEnd(callable $listener): void
|
||||
{
|
||||
$this->sessionEndListener = $listener;
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
if (\is_callable($this->sessionEndListener) && null !== $this->sessionId) {
|
||||
\call_user_func($this->sessionEndListener, $this->sessionId);
|
||||
}
|
||||
|
||||
if (\is_resource($this->input)) {
|
||||
fclose($this->input);
|
||||
}
|
||||
|
||||
if (\is_resource($this->output)) {
|
||||
fclose($this->output);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
218
src/Server/Transport/StreamableHttpTransport.php
Normal file
218
src/Server/Transport/StreamableHttpTransport.php
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the official PHP MCP SDK.
|
||||
*
|
||||
* A collaboration between Symfony and the PHP Foundation.
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Mcp\Server\Transport;
|
||||
|
||||
use Mcp\Schema\JsonRpc\Error;
|
||||
use Mcp\Server\TransportInterface;
|
||||
use Psr\Http\Message\ResponseFactoryInterface;
|
||||
use Psr\Http\Message\ResponseInterface;
|
||||
use Psr\Http\Message\ServerRequestInterface;
|
||||
use Psr\Http\Message\StreamFactoryInterface;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Psr\Log\NullLogger;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* @author Kyrian Obikwelu <koshnawaza@gmail.com>
|
||||
*/
|
||||
class StreamableHttpTransport implements TransportInterface
|
||||
{
|
||||
/** @var callable(string, ?Uuid): void */
|
||||
private $messageListener;
|
||||
|
||||
/** @var callable(Uuid): void */
|
||||
private $sessionEndListener;
|
||||
|
||||
private ?Uuid $sessionId = null;
|
||||
|
||||
/** @var string[] */
|
||||
private array $outgoingMessages = [];
|
||||
private ?Uuid $outgoingSessionId = null;
|
||||
private ?int $outgoingStatusCode = null;
|
||||
|
||||
/** @var array<string, string> */
|
||||
private array $corsHeaders = [
|
||||
'Access-Control-Allow-Origin' => '*',
|
||||
'Access-Control-Allow-Methods' => 'GET, POST, DELETE, OPTIONS',
|
||||
'Access-Control-Allow-Headers' => 'Content-Type, Mcp-Session-Id, Last-Event-ID, Authorization, Accept',
|
||||
];
|
||||
|
||||
public function __construct(
|
||||
private readonly ServerRequestInterface $request,
|
||||
private readonly ResponseFactoryInterface $responseFactory,
|
||||
private readonly StreamFactoryInterface $streamFactory,
|
||||
private readonly LoggerInterface $logger = new NullLogger(),
|
||||
) {
|
||||
$sessionIdString = $this->request->getHeaderLine('Mcp-Session-Id');
|
||||
$this->sessionId = $sessionIdString ? Uuid::fromString($sessionIdString) : null;
|
||||
}
|
||||
|
||||
public function initialize(): void
|
||||
{
|
||||
}
|
||||
|
||||
public function send(string $data, array $context): void
|
||||
{
|
||||
$this->outgoingMessages[] = $data;
|
||||
|
||||
if (isset($context['session_id'])) {
|
||||
$this->outgoingSessionId = $context['session_id'];
|
||||
}
|
||||
|
||||
if (isset($context['status_code']) && \is_int($context['status_code'])) {
|
||||
$this->outgoingStatusCode = $context['status_code'];
|
||||
}
|
||||
|
||||
$this->logger->debug('Sending data to client via StreamableHttpTransport.', [
|
||||
'data' => $data,
|
||||
'session_id' => $this->outgoingSessionId?->toRfc4122(),
|
||||
'status_code' => $this->outgoingStatusCode,
|
||||
]);
|
||||
}
|
||||
|
||||
public function listen(): mixed
|
||||
{
|
||||
return match ($this->request->getMethod()) {
|
||||
'OPTIONS' => $this->handleOptionsRequest(),
|
||||
'GET' => $this->handleGetRequest(),
|
||||
'POST' => $this->handlePostRequest(),
|
||||
'DELETE' => $this->handleDeleteRequest(),
|
||||
default => $this->handleUnsupportedRequest(),
|
||||
};
|
||||
}
|
||||
|
||||
public function onMessage(callable $listener): void
|
||||
{
|
||||
$this->messageListener = $listener;
|
||||
}
|
||||
|
||||
public function onSessionEnd(callable $listener): void
|
||||
{
|
||||
$this->sessionEndListener = $listener;
|
||||
}
|
||||
|
||||
protected function handleOptionsRequest(): ResponseInterface
|
||||
{
|
||||
return $this->withCorsHeaders($this->responseFactory->createResponse(204));
|
||||
}
|
||||
|
||||
protected function handlePostRequest(): ResponseInterface
|
||||
{
|
||||
$acceptHeader = $this->request->getHeaderLine('Accept');
|
||||
if (!str_contains($acceptHeader, 'application/json') || !str_contains($acceptHeader, 'text/event-stream')) {
|
||||
$error = Error::forInvalidRequest('Not Acceptable: Client must accept both application/json and text/event-stream.');
|
||||
$this->logger->warning('Client does not accept required content types.', ['accept' => $acceptHeader]);
|
||||
|
||||
return $this->createErrorResponse($error, 406);
|
||||
}
|
||||
|
||||
if (!str_contains($this->request->getHeaderLine('Content-Type'), 'application/json')) {
|
||||
$error = Error::forInvalidRequest('Unsupported Media Type: Content-Type must be application/json.');
|
||||
$this->logger->warning('Client sent unsupported content type.', ['content_type' => $this->request->getHeaderLine('Content-Type')]);
|
||||
|
||||
return $this->createErrorResponse($error, 415);
|
||||
}
|
||||
|
||||
$body = $this->request->getBody()->getContents();
|
||||
if (empty($body)) {
|
||||
$error = Error::forInvalidRequest('Bad Request: Empty request body.');
|
||||
$this->logger->warning('Client sent empty request body.');
|
||||
|
||||
return $this->createErrorResponse($error, 400);
|
||||
}
|
||||
|
||||
$this->logger->debug('Received message on StreamableHttpTransport.', [
|
||||
'body' => $body,
|
||||
'session_id' => $this->sessionId?->toRfc4122(),
|
||||
]);
|
||||
|
||||
if (\is_callable($this->messageListener)) {
|
||||
\call_user_func($this->messageListener, $body, $this->sessionId);
|
||||
}
|
||||
|
||||
if (empty($this->outgoingMessages)) {
|
||||
return $this->withCorsHeaders($this->responseFactory->createResponse(202));
|
||||
}
|
||||
|
||||
$responseBody = 1 === \count($this->outgoingMessages)
|
||||
? $this->outgoingMessages[0]
|
||||
: '['.implode(',', $this->outgoingMessages).']';
|
||||
|
||||
$status = $this->outgoingStatusCode ?? 200;
|
||||
|
||||
$response = $this->responseFactory->createResponse($status)
|
||||
->withHeader('Content-Type', 'application/json')
|
||||
->withBody($this->streamFactory->createStream($responseBody));
|
||||
|
||||
if ($this->outgoingSessionId) {
|
||||
$response = $response->withHeader('Mcp-Session-Id', $this->outgoingSessionId->toRfc4122());
|
||||
}
|
||||
|
||||
return $this->withCorsHeaders($response);
|
||||
}
|
||||
|
||||
protected function handleGetRequest(): ResponseInterface
|
||||
{
|
||||
$response = $this->createErrorResponse(Error::forInvalidRequest('Not Yet Implemented'), 405);
|
||||
|
||||
return $this->withCorsHeaders($response);
|
||||
}
|
||||
|
||||
protected function handleDeleteRequest(): ResponseInterface
|
||||
{
|
||||
if (!$this->sessionId) {
|
||||
$error = Error::forInvalidRequest('Bad Request: Mcp-Session-Id header is required for DELETE requests.');
|
||||
$this->logger->warning('DELETE request received without session ID.');
|
||||
|
||||
return $this->createErrorResponse($error, 400);
|
||||
}
|
||||
|
||||
if (\is_callable($this->sessionEndListener)) {
|
||||
\call_user_func($this->sessionEndListener, $this->sessionId);
|
||||
}
|
||||
|
||||
return $this->withCorsHeaders($this->responseFactory->createResponse(204));
|
||||
}
|
||||
|
||||
protected function handleUnsupportedRequest(): ResponseInterface
|
||||
{
|
||||
$this->logger->warning('Unsupported HTTP method received.', [
|
||||
'method' => $this->request->getMethod(),
|
||||
]);
|
||||
|
||||
$response = $this->createErrorResponse(Error::forInvalidRequest('Method Not Allowed'), 405);
|
||||
|
||||
return $this->withCorsHeaders($response);
|
||||
}
|
||||
|
||||
protected function withCorsHeaders(ResponseInterface $response): ResponseInterface
|
||||
{
|
||||
foreach ($this->corsHeaders as $name => $value) {
|
||||
$response = $response->withHeader($name, $value);
|
||||
}
|
||||
|
||||
return $response;
|
||||
}
|
||||
|
||||
protected function createErrorResponse(Error $jsonRpcError, int $statusCode): ResponseInterface
|
||||
{
|
||||
$errorPayload = json_encode($jsonRpcError, \JSON_THROW_ON_ERROR);
|
||||
|
||||
return $this->responseFactory->createResponse($statusCode)
|
||||
->withHeader('Content-Type', 'application/json')
|
||||
->withBody($this->streamFactory->createStream($errorPayload));
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue