Add Middleware handlers to StreamableHttpTransport (#218)

composer.json

@@ -28,6 +28,8 @@
     "psr/event-dispatcher": "^1.0",
     "psr/http-factory": "^1.1",
     "psr/http-message": "^1.1 || ^2.0",
+    "psr/http-server-handler": "^1.0",
+    "psr/http-server-middleware": "^1.0",
     "psr/log": "^1.0 || ^2.0 || ^3.0",
     "symfony/finder": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/uid": "^5.4 || ^6.4 || ^7.3 || ^8.0"

docs/transports.md

@@ -179,6 +179,46 @@ Default CORS headers:
 - `Access-Control-Allow-Methods: GET, POST, DELETE, OPTIONS`
 - `Access-Control-Allow-Headers: Content-Type, Mcp-Session-Id, Mcp-Protocol-Version, Last-Event-ID, Authorization, Accept`
 
+### PSR-15 Middleware
+
+`StreamableHttpTransport` can run a PSR-15 middleware chain before it processes the request. Middleware can log,
+enforce auth, or short-circuit with a response for any HTTP method.
+
+```php
+use Mcp\Server\Transport\StreamableHttpTransport;
+use Psr\Http\Message\ResponseFactoryInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Server\MiddlewareInterface;
+use Psr\Http\Server\RequestHandlerInterface;
+
+final class AuthMiddleware implements MiddlewareInterface
+{
+    public function __construct(private ResponseFactoryInterface $responses)
+    {
+    }
+
+    public function process(ServerRequestInterface $request, RequestHandlerInterface $handler)
+    {
+        if (!$request->hasHeader('Authorization')) {
+            return $this->responses->createResponse(401);
+        }
+
+        return $handler->handle($request);
+    }
+}
+
+$transport = new StreamableHttpTransport(
+    $request,
+    $responseFactory,
+    $streamFactory,
+    [],
+    $logger,
+    [new AuthMiddleware($responseFactory)],
+);
+```
+
+If middleware returns a response, the transport will still ensure CORS headers are present unless you set them yourself.
+
 ### Architecture
 
 The HTTP transport doesn't run its own web server. Instead, it processes PSR-7 requests and returns PSR-7 responses that

src/Server/Transport/Http/MiddlewareRequestHandler.php

@@ -0,0 +1,47 @@
+<?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\Http;
+
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Server\MiddlewareInterface;
+use Psr\Http\Server\RequestHandlerInterface;
+
+/**
+ * A request handler that processes a middleware pipeline before dispatching
+ * the request to the core transport handler.
+ *
+ * @author Volodymyr Panivko <sveneld300@gmail.com>
+ *
+ * @internal
+ */
+class MiddlewareRequestHandler implements RequestHandlerInterface
+{
+    /**
+     * @param list<MiddlewareInterface> $middleware
+     */
+    public function __construct(
+        private array $middleware,
+        private \Closure $application,
+    ) {
+    }
+
+    public function handle(ServerRequestInterface $request): ResponseInterface
+    {
+        $middleware = array_shift($this->middleware);
+        if (null === $middleware) {
+            return ($this->application)($request);
+        }
+
+        return $middleware->process($request, $this);
+    }
+}

src/Server/Transport/StreamableHttpTransport.php

@@ -12,11 +12,14 @@
 namespace Mcp\Server\Transport;
 
 use Http\Discovery\Psr17FactoryDiscovery;
+use Mcp\Exception\InvalidArgumentException;
 use Mcp\Schema\JsonRpc\Error;
+use Mcp\Server\Transport\Http\MiddlewareRequestHandler;
 use Psr\Http\Message\ResponseFactoryInterface;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\ServerRequestInterface;
 use Psr\Http\Message\StreamFactoryInterface;
+use Psr\Http\Server\MiddlewareInterface;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Uid\Uuid;
 
@@ -36,19 +39,22 @@ class StreamableHttpTransport extends BaseTransport
     /** @var array<string, string> */
     private array $corsHeaders;
 
+    /** @var list<MiddlewareInterface> */
+    private array $middleware = [];
+
     /**
-     * @param array<string, string> $corsHeaders
+     * @param array<string, string>         $corsHeaders
+     * @param iterable<MiddlewareInterface> $middleware
      */
     public function __construct(
-        private readonly ServerRequestInterface $request,
+        private ServerRequestInterface $request,
         ?ResponseFactoryInterface $responseFactory = null,
         ?StreamFactoryInterface $streamFactory = null,
         array $corsHeaders = [],
         ?LoggerInterface $logger = null,
+        iterable $middleware = [],
     ) {
         parent::__construct($logger);
-        $sessionIdString = $this->request->getHeaderLine('Mcp-Session-Id');
-        $this->sessionId = $sessionIdString ? Uuid::fromString($sessionIdString) : null;
 
         $this->responseFactory = $responseFactory ?? Psr17FactoryDiscovery::findResponseFactory();
         $this->streamFactory = $streamFactory ?? Psr17FactoryDiscovery::findStreamFactory();
@@ -59,6 +65,13 @@ class StreamableHttpTransport extends BaseTransport
             'Access-Control-Allow-Headers' => 'Content-Type, Mcp-Session-Id, Mcp-Protocol-Version, Last-Event-ID, Authorization, Accept',
             'Access-Control-Expose-Headers' => 'Mcp-Session-Id',
         ], $corsHeaders);
+
+        foreach ($middleware as $m) {
+            if (!$m instanceof MiddlewareInterface) {
+                throw new InvalidArgumentException('Streamable HTTP middleware must implement Psr\\Http\\Server\\MiddlewareInterface.');
+            }
+            $this->middleware[] = $m;
+        }
     }
 
     public function send(string $data, array $context): void
@@ -69,17 +82,17 @@ class StreamableHttpTransport extends BaseTransport
 
     public function listen(): ResponseInterface
     {
-        return match ($this->request->getMethod()) {
-            'OPTIONS' => $this->handleOptionsRequest(),
-            'POST' => $this->handlePostRequest(),
-            'DELETE' => $this->handleDeleteRequest(),
-            default => $this->createErrorResponse(Error::forInvalidRequest('Method Not Allowed'), 405),
-        };
+        $handler = new MiddlewareRequestHandler(
+            $this->middleware,
+            \Closure::fromCallable([$this, 'handleRequest']),
+        );
+
+        return $this->withCorsHeaders($handler->handle($this->request));
     }
 
     protected function handleOptionsRequest(): ResponseInterface
     {
-        return $this->withCorsHeaders($this->responseFactory->createResponse(204));
+        return $this->responseFactory->createResponse(204);
     }
 
     protected function handlePostRequest(): ResponseInterface
@@ -92,7 +105,7 @@ class StreamableHttpTransport extends BaseTransport
                 ->withHeader('Content-Type', 'application/json')
                 ->withBody($this->streamFactory->createStream($this->immediateResponse));
 
-            return $this->withCorsHeaders($response);
+            return $response;
         }
 
         if (null !== $this->sessionFiber) {
@@ -112,7 +125,7 @@ class StreamableHttpTransport extends BaseTransport
 
         $this->handleSessionEnd($this->sessionId);
 
-        return $this->withCorsHeaders($this->responseFactory->createResponse(200));
+        return $this->responseFactory->createResponse(200);
     }
 
     protected function createJsonResponse(): ResponseInterface
@@ -120,7 +133,7 @@ class StreamableHttpTransport extends BaseTransport
         $outgoingMessages = $this->getOutgoingMessages($this->sessionId);
 
         if (empty($outgoingMessages)) {
-            return $this->withCorsHeaders($this->responseFactory->createResponse(202));
+            return $this->responseFactory->createResponse(202);
         }
 
         $messages = array_column($outgoingMessages, 'message');
@@ -134,7 +147,7 @@ class StreamableHttpTransport extends BaseTransport
             $response = $response->withHeader('Mcp-Session-Id', $this->sessionId->toRfc4122());
         }
 
-        return $this->withCorsHeaders($response);
+        return $response;
     }
 
     protected function createStreamedResponse(): ResponseInterface
@@ -201,7 +214,7 @@ class StreamableHttpTransport extends BaseTransport
             $response = $response->withHeader('Mcp-Session-Id', $this->sessionId->toRfc4122());
         }
 
-        return $this->withCorsHeaders($response);
+        return $response;
     }
 
     protected function handleFiberTermination(): void
@@ -246,15 +259,31 @@ class StreamableHttpTransport extends BaseTransport
             $response = $response->withHeader('Allow', 'POST, DELETE, OPTIONS');
         }
 
-        return $this->withCorsHeaders($response);
+        return $response;
     }
 
     protected function withCorsHeaders(ResponseInterface $response): ResponseInterface
     {
         foreach ($this->corsHeaders as $name => $value) {
-            $response = $response->withHeader($name, $value);
+            if (!$response->hasHeader($name)) {
+                $response = $response->withHeader($name, $value);
+            }
         }
 
         return $response;
     }
+
+    private function handleRequest(ServerRequestInterface $request): ResponseInterface
+    {
+        $this->request = $request;
+        $sessionIdString = $request->getHeaderLine('Mcp-Session-Id');
+        $this->sessionId = $sessionIdString ? Uuid::fromString($sessionIdString) : null;
+
+        return match ($request->getMethod()) {
+            'OPTIONS' => $this->handleOptionsRequest(),
+            'POST' => $this->handlePostRequest(),
+            'DELETE' => $this->handleDeleteRequest(),
+            default => $this->createErrorResponse(Error::forInvalidRequest('Method Not Allowed'), 405),
+        };
+    }
 }

tests/Unit/Server/Transport/StreamableHttpTransportTest.php

@@ -0,0 +1,164 @@
+<?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\Tests\Unit\Server\Transport;
+
+use Mcp\Server\Transport\StreamableHttpTransport;
+use Nyholm\Psr7\Factory\Psr17Factory;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\Attributes\TestDox;
+use PHPUnit\Framework\TestCase;
+use Psr\Http\Message\ResponseFactoryInterface;
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Server\MiddlewareInterface;
+use Psr\Http\Server\RequestHandlerInterface;
+
+final class StreamableHttpTransportTest extends TestCase
+{
+    public static function corsHeaderProvider(): iterable
+    {
+        yield 'GET (middleware returns 401)' => ['GET', false, 401];
+        yield 'POST (middleware returns 401)' => ['POST', false, 401];
+        yield 'DELETE (middleware returns 401)' => ['DELETE', false, 401];
+        yield 'OPTIONS (middleware delegates -> transport handles preflight)' => ['OPTIONS', true, 204];
+        yield 'GET (middleware delegates -> transport handles preflight)' => ['GET', true, 405];
+        yield 'POST (middleware delegates -> transport handles preflight)' => ['POST', true, 202];
+        yield 'DELETE (middleware delegates -> transport handles preflight)' => ['DELETE', true, 400];
+    }
+
+    #[DataProvider('corsHeaderProvider')]
+    #[TestDox('CORS headers are always applied')]
+    public function testCorsHeader(string $method, bool $middlewareDelegatesToTransport, int $expectedStatusCode): void
+    {
+        $factory = new Psr17Factory();
+        $request = $factory->createServerRequest($method, 'https://example.com');
+
+        $middleware = new class($factory, $expectedStatusCode, $middlewareDelegatesToTransport) implements MiddlewareInterface {
+            public function __construct(
+                private ResponseFactoryInterface $responseFactory,
+                private int $expectedStatusCode,
+                private bool $middlewareDelegatesToTransport,
+            ) {
+            }
+
+            public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
+            {
+                if ($this->middlewareDelegatesToTransport) {
+                    return $handler->handle($request);
+                }
+
+                return $this->responseFactory->createResponse($this->expectedStatusCode);
+            }
+        };
+
+        $transport = new StreamableHttpTransport(
+            $request,
+            $factory,
+            $factory,
+            [],
+            null,
+            [$middleware],
+        );
+
+        $response = $transport->listen();
+
+        $this->assertSame($expectedStatusCode, $response->getStatusCode(), $response->getBody()->getContents());
+        $this->assertTrue($response->hasHeader('Access-Control-Allow-Origin'));
+        $this->assertTrue($response->hasHeader('Access-Control-Allow-Methods'));
+        $this->assertTrue($response->hasHeader('Access-Control-Allow-Headers'));
+        $this->assertTrue($response->hasHeader('Access-Control-Expose-Headers'));
+
+        $this->assertSame('*', $response->getHeaderLine('Access-Control-Allow-Origin'));
+        $this->assertSame('GET, POST, DELETE, OPTIONS', $response->getHeaderLine('Access-Control-Allow-Methods'));
+        $this->assertSame(
+            'Content-Type, Mcp-Session-Id, Mcp-Protocol-Version, Last-Event-ID, Authorization, Accept',
+            $response->getHeaderLine('Access-Control-Allow-Headers')
+        );
+        $this->assertSame('Mcp-Session-Id', $response->getHeaderLine('Access-Control-Expose-Headers'));
+    }
+
+    #[TestDox('transport replaces existing CORS headers on the response')]
+    public function testCorsHeadersAreReplacedWhenAlreadyPresent(): void
+    {
+        $factory = new Psr17Factory();
+        $request = $factory->createServerRequest('GET', 'https://example.com');
+
+        $middleware = new class($factory) implements MiddlewareInterface {
+            public function __construct(private ResponseFactoryInterface $responses)
+            {
+            }
+
+            public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
+            {
+                return $this->responses->createResponse(200)
+                    ->withHeader('Access-Control-Allow-Origin', 'https://another.com');
+            }
+        };
+
+        $transport = new StreamableHttpTransport(
+            $request,
+            $factory,
+            $factory,
+            [],
+            null,
+            [$middleware],
+        );
+
+        $response = $transport->listen();
+
+        $this->assertSame(200, $response->getStatusCode());
+
+        $this->assertSame('https://another.com', $response->getHeaderLine('Access-Control-Allow-Origin'));
+        $this->assertSame('GET, POST, DELETE, OPTIONS', $response->getHeaderLine('Access-Control-Allow-Methods'));
+        $this->assertSame(
+            'Content-Type, Mcp-Session-Id, Mcp-Protocol-Version, Last-Event-ID, Authorization, Accept',
+            $response->getHeaderLine('Access-Control-Allow-Headers')
+        );
+        $this->assertSame('Mcp-Session-Id', $response->getHeaderLine('Access-Control-Expose-Headers'));
+    }
+
+    #[TestDox('middleware runs before transport handles the request')]
+    public function testMiddlewareRunsBeforeTransportHandlesRequest(): void
+    {
+        $factory = new Psr17Factory();
+        $request = $factory->createServerRequest('OPTIONS', 'https://example.com');
+
+        $state = new \stdClass();
+        $state->called = false;
+        $middleware = new class($state) implements MiddlewareInterface {
+            public function __construct(private \stdClass $state)
+            {
+            }
+
+            public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
+            {
+                $this->state->called = true;
+
+                return $handler->handle($request);
+            }
+        };
+
+        $transport = new StreamableHttpTransport(
+            $request,
+            $factory,
+            $factory,
+            [],
+            null,
+            [$middleware],
+        );
+
+        $response = $transport->listen();
+
+        $this->assertTrue($state->called);
+        $this->assertSame(204, $response->getStatusCode());
+    }
+}