Draft: Streamable Http

src/mcp-bundle/composer.json

@@ -21,7 +21,8 @@
     },
     "require-dev": {
         "phpstan/phpstan": "^2.1",
-        "phpunit/phpunit": "^11.5"
+        "phpunit/phpunit": "^11.5",
+        "symfony/security-bundle": "^7.3"
     },
     "config": {
         "sort-packages": true

src/mcp-bundle/config/options.php

@@ -59,6 +59,7 @@
                 ->children()
                     ->booleanNode('stdio')->defaultFalse()->end()
                     ->booleanNode('sse')->defaultFalse()->end()
+                    ->booleanNode('http_stream')->defaultTrue()->end() // @todo change to default false
                 ->end()
             ->end()
         ->end()

src/mcp-bundle/config/routes.php

@@ -9,16 +9,29 @@
  * file that was distributed with this source code.
  */
 
-use Symfony\AI\McpBundle\Controller\McpController;
+use Symfony\AI\McpBundle\Controller\McpSseController;
+use Symfony\AI\McpBundle\Controller\McpHttpStreamController;
 use Symfony\Component\Routing\Loader\Configurator\RoutingConfigurator;
 
 return function (RoutingConfigurator $routes): void {
     $routes->add('_mcp_sse', '/sse')
-        ->controller([McpController::class, 'sse'])
+        ->controller([McpSseController::class, 'sse'])
         ->methods(['GET'])
     ;
     $routes->add('_mcp_messages', '/messages/{id}')
-        ->controller([McpController::class, 'messages'])
+        ->controller([McpSseController::class, 'messages'])
         ->methods(['POST'])
     ;
+    $routes->add('_mcp_http', '/http/')
+        ->controller([McpHttpStreamController::class, 'endpoint'])
+        ->methods(['POST'])
+    ;
+    $routes->add('_mcp_http_initiate_sse', '/http/')
+        ->controller([McpHttpStreamController::class, 'initiateSseFromStream'])
+        ->methods(['GET'])
+    ;
+    $routes->add('_mcp_http_delete_session', '/http/')
+        ->controller([McpHttpStreamController::class, 'deleteSession'])
+        ->methods(['DELETE'])
+    ;
 };

src/mcp-bundle/config/services.php

@@ -11,6 +11,9 @@
 
 namespace Symfony\Component\DependencyInjection\Loader\Configurator;
 
+use Symfony\AI\McpBundle\Session\SessionIdentifierResolver;
+use Symfony\AI\McpBundle\Session\SessionResolver;
+use Symfony\AI\McpBundle\Session\SessionSubscriber;
 use Symfony\AI\McpSdk\Capability\ToolChain;
 use Symfony\AI\McpSdk\Message\Factory;
 use Symfony\AI\McpSdk\Server;
@@ -21,6 +24,8 @@
 use Symfony\AI\McpSdk\Server\RequestHandler\ToolCallHandler;
 use Symfony\AI\McpSdk\Server\RequestHandler\ToolListHandler;
 use Symfony\AI\McpSdk\Server\Transport\Sse\Store\CachePoolStore;
+use Symfony\AI\McpSdk\Server\Transport\StreamableHttp\Session\SessionIdentifierFactory;
+use Symfony\AI\McpSdk\Server\Transport\StreamableHttp\Session\SessionPoolStorage;
 
 return static function (ContainerConfigurator $container): void {
     $container->services()
@@ -50,13 +55,15 @@
 
         ->set('mcp.message_factory', Factory::class)
             ->args([])
+            ->alias(Factory::class, 'mcp.message_factory')
         ->set('mcp.server.json_rpc', JsonRpcHandler::class)
             ->args([
                 service('mcp.message_factory'),
                 tagged_iterator('mcp.server.request_handler'),
                 tagged_iterator('mcp.server.notification_handler'),
                 service('logger')->ignoreOnInvalid(),
             ])
+            ->alias(JsonRpcHandler::class, 'mcp.server.json_rpc')
         ->set('mcp.server', Server::class)
             ->args([
                 service('mcp.server.json_rpc'),
@@ -67,6 +74,34 @@
             ->args([
                 service('cache.app'),
             ])
+        ->set('mcp.server.http_stream.session.identifier_factory', SessionIdentifierFactory::class)
+            ->args([
+                service('security')->nullOnInvalid(),
+            ])
+            ->alias(SessionIdentifierFactory::class, 'mcp.server.http_stream.session.identifier_factory')
+        ->set('mcp.server.http_stream.session.identifier_resolver', SessionIdentifierResolver::class)
+            ->tag('controller.argument_value_resolver')
+        ->set('mcp.server.http_stream.session.resolver', SessionResolver::class)
+            ->tag('controller.argument_value_resolver')
+        ->set('mcp.server.http_stream.session.pool', SessionPoolStorage::class)
+            ->args([
+                service('cache.app'),
+                param('mcp.http_stream.session.ttl'),
+            ])
+            ->alias(Server\Transport\StreamableHttp\SessionStorageInterface::class, 'mcp.server.http_stream.session.pool')
+        ->set('mcp.server.http_stream.session.subscriber', SessionSubscriber::class)
+            ->args([
+                service('mcp.server.http_stream.session.identifier_factory'),
+                service('mcp.server.http_stream.session.factory'),
+            ])
+            ->tag('kernel.event_subscriber')
+            ->alias(SessionSubscriber::class, 'mcp.server.http_stream.session.subscriber')
+        ->set('mcp.server.http_stream.session.factory', Server\Transport\StreamableHttp\Session\SessionFactory::class)
+            ->args([
+                service('mcp.server.http_stream.session.identifier_factory'),
+                service('mcp.server.http_stream.session.pool'),
+            ])
+            ->alias(Server\Transport\StreamableHttp\Session\SessionFactory::class, 'mcp.server.http_stream.session.factory')
         ->set('mcp.tool_chain', ToolChain::class)
             ->args([
                 tagged_iterator('mcp.tool'),

src/mcp-bundle/src/Controller/McpHttpStreamController.php

@@ -0,0 +1,142 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Symfony\AI\McpBundle\Controller;
+
+use Symfony\AI\McpSdk\Message\Factory;
+use Symfony\AI\McpSdk\Message\Notification;
+use Symfony\AI\McpSdk\Message\StreamableResponse;
+use Symfony\AI\McpSdk\Server;
+use Symfony\AI\McpSdk\Server\Transport\StreamableHttp\Session\Session;
+use Symfony\AI\McpSdk\Server\Transport\StreamableHttp\Session\SessionFactory;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Request;
+use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\HttpFoundation\StreamedResponse;
+use Symfony\Component\HttpKernel\Exception\BadRequestHttpException;
+use Symfony\Component\Uid\Uuid;
+
+final readonly class McpHttpStreamController
+{
+    public function __construct(
+        private Server\JsonRpcHandler    $handler,
+        private Factory $messageFactory,
+        private SessionFactory $sessionFactory,
+    ) {
+    }
+    public function endpoint(Request $request, ?Session $session = null): Response
+    {
+        $message = $this->messageFactory->create($request->getContent());
+        if ($session === null) {
+            // Must be an "initialize" request. If not ==> 404.
+            if ($message->method !== 'initialize') { // @todo do better
+                return new Response(null, Response::HTTP_NOT_FOUND);
+            }
+            $session = $this->sessionFactory->get();
+            $session->save();
+        }
+
+        // Handle the input
+        // If response is streamable ==> open an SSE Stream and store all responses in session for later replay
+        // If response is not ==> JSON
+
+        $response = $this->handler->handleMessage($message);
+
+        if ($message instanceof Notification) {
+            return new Response(null, Response::HTTP_ACCEPTED);
+        }
+        if ($response instanceof StreamableResponse) {
+            //$transport = new Server\Transport\StreamableHttp\StreamTransport($session->addNewStream(), $session, $response->responses);
+            return new StreamedResponse(function () use ($session, $response) {
+                $streamId = $session->addNewStream();
+                foreach (($response->responses)() as $response) {
+                    $eventId = Uuid::v4()->toString();
+                    if (is_array($response)) {
+                        $rawResponse = json_encode($response, \JSON_THROW_ON_ERROR);
+                    } else {
+                        $rawResponse = $this->handler->encodeResponse($response);
+                    }
+                    $session->addEventOnStream($streamId, $eventId, $rawResponse);
+                    echo "id: $eventId\n";
+                    echo "type: notification\n";
+                    echo "data: " . $rawResponse . "\n\n";
+                    if (false !== ob_get_length()) {
+                        ob_flush();
+                    }
+                    flush();
+                }
+            }, headers: [
+                'Content-Type' => 'text/event-stream',
+                'Cache-Control' => 'no-cache',
+                'X-Accel-Buffering' => 'no',
+                'Mcp-Session-Id' => $session->sessionIdentifier->sessionId->toString(),
+            ]);
+        }
+        return new JsonResponse($this->handler->encodeResponse($response), Response::HTTP_OK, [
+            'Content-Type' => 'application/json',
+            'Cache-Control' => 'no-cache',
+            'Mcp-Session-Id' => $session->sessionIdentifier->sessionId->toString(),
+        ], true);
+    }
+
+    /**
+     * Clients that no longer need a particular session (e.g., because the user is leaving the client application) SHOULD send an HTTP DELETE to the MCP endpoint with the Mcp-Session-Id header, to explicitly terminate the session.
+     * @see{https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#session-management}
+     *
+     * @param Session $session
+     * @return Response
+     */
+    public function deleteSession(Session $session): Response
+    {
+        $session->delete();
+        return new Response(null, Response::HTTP_NO_CONTENT);
+    }
+
+    /**
+     * @param Request $request
+     * @param Session $session
+     * @return Response
+     */
+    public function initiateSseFromStream(Request $request, Session $session): Response
+    {
+        if ($request->headers->has('Last-Event-ID')) {
+            try {
+                $session->getStreamIdForEvent($request->headers->get('Last-Event-ID'));
+            } catch (\InvalidArgumentException $e) {
+                throw new BadRequestHttpException($e->getMessage());
+            }
+            $lastEventId = $request->headers->get('Last-Event-ID');
+            return new StreamedResponse(function () use ($session, $lastEventId) {
+                $i = 0;
+                do {
+                    $events = $session->getEventsAfterId($lastEventId);
+                    $lastEvent = null;
+                    foreach ($events as $event) {
+                        $lastEventId = $event['id'];
+                        $lastEvent = $event['event'];
+                        echo 'id: ' . $lastEventId . \PHP_EOL;
+                        echo 'data: ' . $lastEvent . \PHP_EOL . \PHP_EOL;
+                        if (false !== ob_get_length()) {
+                            ob_flush();
+                        }
+                        flush();
+                    }
+                    if ($events === []) {
+                        usleep(1000);
+                    }
+                    // @todo we should detect here that the "real" response has been sent and close the stream
+                } while (! ($lastEvent instanceof \Symfony\AI\McpSdk\Message\Response) && $i++ < 50);
+            }, headers: [
+                'Content-Type' => 'text/event-stream',
+                'Cache-Control' => 'no-cache',
+                'X-Accel-Buffering' => 'no',
+            ]);
+
+
+        } else {
+            // At this point server cannot attach to this stream to send request / notifications, so act like we don't support
+            return new Response(null, Response::HTTP_METHOD_NOT_ALLOWED);
+        }
+    }
+}

src/mcp-bundle/src/Controller/McpController.php → src/mcp-bundle/src/Controller/McpSseController.php

@@ -20,7 +20,7 @@
 use Symfony\Component\Routing\Generator\UrlGeneratorInterface;
 use Symfony\Component\Uid\Uuid;
 
-final readonly class McpController
+final readonly class McpSseController
 {
     public function __construct(
         private Server $server,

src/mcp-bundle/src/McpBundle.php

@@ -12,7 +12,8 @@
 namespace Symfony\AI\McpBundle;
 
 use Symfony\AI\McpBundle\Command\McpCommand;
-use Symfony\AI\McpBundle\Controller\McpController;
+use Symfony\AI\McpBundle\Controller\McpHttpStreamController;
+use Symfony\AI\McpBundle\Controller\McpSseController;
 use Symfony\AI\McpBundle\Routing\RouteLoader;
 use Symfony\AI\McpSdk\Capability\Tool\IdentifierInterface;
 use Symfony\AI\McpSdk\Server\NotificationHandlerInterface;
@@ -40,6 +41,7 @@ public function loadExtension(array $config, ContainerConfigurator $container, C
         $builder->setParameter('mcp.app', $config['app']);
         $builder->setParameter('mcp.version', $config['version']);
         $builder->setParameter('mcp.page_size', $config['page_size']);
+        $builder->setParameter('mcp.http_stream.session.ttl', 3600);
 
         if (isset($config['client_transports'])) {
             $this->configureClient($config['client_transports'], $builder);
@@ -52,11 +54,11 @@ public function loadExtension(array $config, ContainerConfigurator $container, C
     }
 
     /**
-     * @param array{stdio: bool, sse: bool} $transports
+     * @param array{stdio: bool, sse: bool, http_stream: bool} $transports
      */
     private function configureClient(array $transports, ContainerBuilder $container): void
     {
-        if (!$transports['stdio'] && !$transports['sse']) {
+        if (!$transports['stdio'] && !$transports['sse'] && !$transports['http_stream']) {
             return;
         }
 
@@ -74,7 +76,7 @@ private function configureClient(array $transports, ContainerBuilder $container)
         }
 
         if ($transports['sse']) {
-            $container->register('mcp.server.controller', McpController::class)
+            $container->register('mcp.server.sse.controller', McpSseController::class)
                 ->setArguments([
                     new Reference('mcp.server'),
                     new Reference('mcp.server.sse.store.cache_pool'),
@@ -84,6 +86,19 @@ private function configureClient(array $transports, ContainerBuilder $container)
                 ->addTag('controller.service_arguments');
         }
 
+        if ($transports['http_stream']) {
+            $container->register('mcp.server.http_stream.controller', McpHttpStreamController::class)
+                ->setArguments([
+                    new Reference('mcp.server.json_rpc'),
+                    new Reference('mcp.message_factory'),
+                    new Reference('mcp.server.http_stream.session.factory'),
+                ])
+                ->setPublic(true)
+                ->addTag('controller.service_arguments')
+            ;
+            $container->setAlias(McpHttpStreamController::class, 'mcp.server.http_stream.controller');
+        }
+
         $container->register('mcp.server.route_loader', RouteLoader::class)
             ->setArgument(0, $transports['sse'])
             ->addTag('routing.route_loader');

src/mcp-bundle/src/Session/SessionIdentifierResolver.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace Symfony\AI\McpBundle\Session;
+
+use Symfony\AI\McpSdk\Exception\InvalidSessionIdException;
+use Symfony\AI\McpSdk\Server\Transport\StreamableHttp\SessionIdentifier;
+use Symfony\Component\HttpFoundation\Request;
+use Symfony\Component\HttpKernel\Controller\ValueResolverInterface;
+use Symfony\Component\HttpKernel\ControllerMetadata\ArgumentMetadata;
+
+readonly class SessionIdentifierResolver implements ValueResolverInterface
+{
+    public function resolve(Request $request, ArgumentMetadata $argument): iterable
+    {
+        if ($argument->getType() !== SessionIdentifier::class) {
+            return [];
+        }
+
+        if (!$request->attributes->has('_mcp_session_id')) {
+            return match($argument->isNullable()) {
+                true => [null],
+                false => []
+            };
+        }
+
+        $sessionIdentifier = $request->attributes->get('_mcp_session_id');
+        if (!$sessionIdentifier instanceof SessionIdentifier) {
+            throw new InvalidSessionIdException(sprintf('Session "%s" not found.', $sessionIdentifier));
+        }
+
+        return [$sessionIdentifier];
+    }
+}