Stateless Mode memory leak

[algirdasci]

Describe the bug
Stateless mode with Streamable-HTTP leaks memory. The StreamableHTTPSessionManager stateless server task group never exits and the _task list grows with each request leading to always increasing memory usage.

Expected behavior
Stateless server exits after handling the request freeing up the resources

Steps to reproduce

1. Create minimal MCP server:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("StatelessServer", stateless_http=True)


@mcp.tool()
def task_size() -> int:
    return len(mcp.session_manager._task_group._tasks)


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

2. Create minimal MCP client:

import asyncio

from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
from mcp.types import CallToolResult

async def execute_mcp_tool_call():
    transport = StreamableHttpTransport("http://localhost:8000/mcp/")

    async with Client(transport) as client:
        tools = await client.list_tools()
        print(tools)

        result = await client.call_tool("task_size")
        print(result)


asyncio.run(execute_mcp_tool_call())

3. Do multiple requests
4. Tool will return the size of pending tasks in session manager task group and it is always increasing:

[TextContent(type='text', text='23', annotations=None)]
[TextContent(type='text', text='27', annotations=None)]
[TextContent(type='text', text='31', annotations=None)]

[algirdasci]

I've managed to workaround this by adding a timeout for stateless server, which works in my case, but maybe there would be better solution for that? Unfortunately, I'm not that good in python, so I'm not sure what the better approach could be.

26b9502

[czx2zch]

Describe the bug Stateless mode with Streamable-HTTP leaks memory. The StreamableHTTPSessionManager stateless server task group never exits and the _task list grows with each request leading to always increasing memory usage.

Expected behavior Stateless server exits after handling the request freeing up the resources

Steps to reproduce

1. Create minimal MCP server:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("StatelessServer", stateless_http=True)

@mcp.tool()
def task_size() -> int:
return len(mcp.session_manager._task_group._tasks)

if name == "main":
mcp.run(transport="streamable-http")
2. Create minimal MCP client:

import asyncio

from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
from mcp.types import CallToolResult

async def execute_mcp_tool_call():
transport = StreamableHttpTransport("http://localhost:8000/mcp/")

async with Client(transport) as client:
    tools = await client.list_tools()
    print(tools)

    result = await client.call_tool("task_size")
    print(result)

asyncio.run(execute_mcp_tool_call())
3. Do multiple requests
4. Tool will return the size of pending tasks in session manager task group and it is always increasing:

[TextContent(type='text', text='23', annotations=None)]
[TextContent(type='text', text='27', annotations=None)]
[TextContent(type='text', text='31', annotations=None)]

i tried to handle it with this method:

async def _ultimate_fixed_handle_stateless_request(
self,
scope: Scope,
receive: Receive,
send: Send,
) -> None:
logger.debug("Stateless mode: Creating new transport for this request")

http_transport = StreamableHTTPServerTransport(
    mcp_session_id=None,
    is_json_response_enabled=self.json_response,
    event_store=None,
)

# 使用超时确保任务不会无限运行
async with anyio.create_task_group() as request_tg:
    server_ready = anyio.Event()
    request_completed = anyio.Event()

    async def run_stateless_server():
        try:
            async with http_transport.connect() as streams:
                read_stream, write_stream = streams
                server_ready.set()
                
                # 关键修复：使用超时运行服务器
                with anyio.move_on_after(30):  # 30秒超时
                    await self.app.run(
                        read_stream,
                        write_stream,
                        self.app.create_initialization_options(),
                        stateless=True,
                    )
        except Exception as e:
            logger.debug(f"Stateless server ended: {e}")
        finally:
            logger.debug("Stateless server task ending")

    async def handle_request_and_signal():
        """处理请求并发信号结束服务器"""
        try:
            await http_transport.handle_request(scope, receive, send)
        finally:
            request_completed.set()
            # 给服务器一点时间处理完最后的消息
            await anyio.sleep(0.1)
            # 取消整个任务组
            request_tg.cancel_scope.cancel()

    # 启动两个任务
    request_tg.start_soon(run_stateless_server)
    await server_ready.wait()
    
    request_tg.start_soon(handle_request_and_signal)
    
    # 等待请求完成
    await request_completed.wait()

StreamableHTTPSessionManager._handle_stateless_request = _ultimate_fixed_handle_stateless_request

and then memory leak not happened any more

[czx2zch]

When will this issue be resolved？ @ihrpr

[MadanMaram]

Description:

1. We tested MCP SDK server, Creating dummy MCP tool using both FastMCP and Python SDK (both scripts attached below)
2. We have tested by sending the request sequentially and here is the code async def main():
   while True:
   await call_tool("Bob")
3. We observed a memory leak(continuous increase in %MEM and RSS) in the MCP server, as client continuously calls the simple tool (e.g., a greet function), eventually leading to an Out of Memory (OOM)

Issue:
Here is the Logs:Memory will keep increasing
memory_log_pythonmcp.txt

This issue is reproducible both in the fastmcp library and the MCP SDK.

Example Code
Server.py

from mcp.server.fastmcp import FastMCP

#Create an MCP server
mcp = FastMCP("Demo Server")

#Add a simple tool
@mcp.tool()
def greet(name: str) -> str:
    """Return a greeting message."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    #Run the server with streamable HTTP transport
    mcp.run(transport="streamable-http")

---

client.py(（with a while loop）

import asyncio
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession

async def call_tool(name: str):
    #Connect to the MCP server using streamable HTTP
    async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            #initialize the connection
            await session.initialize()
            #Call the 'greet' tool
            result = await session.call_tool("greet", {"name": name})
            print(result)

async def main():
    while True:
        await call_tool("Bob")

if __name__ == "__main__":
    asyncio.run(main())

---

Python & MCP Python SDK
Version:

Python: 3.12.1

MCP Python SDK: 1.10.1

[nathan-thillairajah]

I spent some time debugging this problem (before finding this Github issue) and just wanted to clarify a few details from the other comments - @felixweinberger hopefully this is helpful and lmk if you have any questions!

Every HTTP request to a stateless MCP server goes through StreamableHTTPSessionManager._handle_stateless_request() which creates 3 background async tasks:

• run_stateless_server
• _receive_loop
• message_router

These tasks wait on the http_transport._read_stream_writer and http_transport._write_stream transport streams to close as their exit condition, but they are never closed. So the tasks persist indefinitely.

Setup

• MCP SDK 1.10.1
• FastMCP 2.10.2
• Using FastMCP(..., stateless_http=True) in a FastAPI app

How I confirmed the task growth

I vibe coded simple task counter that I added to my locally running MCP server to confirm the unbounded task growth:

# Start with 4 normal framework tasks
curl -s http://localhost:8080/task-debug/count | jq '.total_tasks'  # Returns: 4

# Make one MCP request
curl -X POST http://localhost:8080/websearch/mcp/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'

# Now we have 7 tasks (+3 leaked)
curl -s http://localhost:8080/task-debug/count | jq '.total_tasks'  # Returns: 7

# Keeps growing: 10, 13, 16, 19... tasks are never cleaned up which causes unbounded memory growth

(happy to share that task counter code if you want it)

My fix

In my local Python code, I applied a patch for _handle_stateless_request() with a finally block that cleans up the transport streams, and using that impl allowed the background tasks to close at the end of processing each request (which fixed the memory leak). Not sure if this is the ideal solution but sharing just in case:

async def _handle_stateless_request(self, scope, receive, send):
    """Process request in stateless mode - creating a new transport for each request."""
    logger.debug("Stateless mode: Creating new transport for this request")
    
    http_transport = StreamableHTTPServerTransport(
        mcp_session_id=None,
        is_json_response_enabled=self.json_response,
        event_store=None,
    )

    async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED):
        async with http_transport.connect() as streams:
            read_stream, write_stream = streams
            task_status.started()
            await self.app.run(
                read_stream,
                write_stream,
                self.app.create_initialization_options(),
                stateless=True,
            )

    assert self._task_group is not None
    await self._task_group.start(run_stateless_server)
    
    try:
        await http_transport.handle_request(scope, receive, send)
    finally:
        # close the streams so tasks can exit
        try:
            if hasattr(http_transport, "_read_stream_writer") and http_transport._read_stream_writer:
                await http_transport._read_stream_writer.aclose()
            if hasattr(http_transport, "_write_stream") and http_transport._write_stream:
                await http_transport._write_stream.aclose()
            if hasattr(http_transport, "close"):
                await http_transport.close()
        except Exception as e:
            logger.debug(f"Error during transport cleanup: {e}")

[MadanMaram]

Thank you for the suggestion @nathan-thillairajah. Since I am at exploration stage of FastMCP can you share how you integrated the above code with FastMCP library.

for ex:
I have initiated the FastMCP, stateless server in this way

server.py

from fastmcp import FastMCP

mcp = FastMCP("My MCP Server",stateless_http=True)

@mcp.tool
def greet(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
    

client.py

import asyncio
from fastmcp import FastMCP, Client

async def call_tool(name: str):
    async with Client("http://127.0.0.1:8000/mcp") as client:
        result = await client.call_tool("greet", {"name": name})
        print(result)

while True:
    asyncio.run(call_tool("Bob"))

[nathan-thillairajah]

@MadanMaram I just patched the MCP SDK library itself in my own code. To be clear this is just a temporary workaround until the bug is fixed in the library.

But I did this by creating a new file called mcp_memory_leak_fix.py:

import logging
from typing import Any
from anyio.abc import TaskStatus
import anyio

logger = logging.getLogger(__name__)

def apply_mcp_memory_leak_fix():
    """Monkey-patch the MCP SDK to fix the stateless memory leak."""
    try:
        from mcp.server import streamable_http_manager
        from mcp.server.streamable_http import StreamableHTTPServerTransport
        
        # Get the original class
        original_class = streamable_http_manager.StreamableHTTPSessionManager
        original_init = original_class.__init__
        
        def patched_init(self, *args, **kwargs):
            # Call original init
            original_init(self, *args, **kwargs)
            
            # Replace the problematic method with our fixed version
            async def fixed_handle_stateless_request(scope, receive, send):
                """Fixed version with proper cleanup."""
                logger.debug("Stateless mode: Creating new transport for this request")
                
                http_transport = StreamableHTTPServerTransport(
                    mcp_session_id=None,
                    is_json_response_enabled=self.json_response,
                    event_store=None,
                )

                async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED):
                    try:
                        async with http_transport.connect() as streams:
                            read_stream, write_stream = streams
                            task_status.started()
                            await self.app.run(
                                read_stream,
                                write_stream,
                                self.app.create_initialization_options(),
                                stateless=True,
                            )
                    except Exception:
                        pass

                assert self._task_group is not None
                await self._task_group.start(run_stateless_server)
                
                try:
                    await http_transport.handle_request(scope, receive, send)
                finally:
                    # FIX: Close transport streams so tasks can exit
                    try:
                        if hasattr(http_transport, "_read_stream_writer") and http_transport._read_stream_writer:
                            await http_transport._read_stream_writer.aclose()
                        if hasattr(http_transport, "_write_stream") and http_transport._write_stream:
                            await http_transport._write_stream.aclose()
                        if hasattr(http_transport, "close"):
                            await http_transport.close()
                    except Exception as e:
                        logger.debug(f"Error during transport cleanup: {e}")
            
            self._handle_stateless_request = fixed_handle_stateless_request
        
        # Apply the patch
        original_class.__init__ = patched_init
        logger.info("Applied MCP memory leak fix")
        return True
        
    except Exception as e:
        logger.error(f"Failed to apply MCP memory leak fix: {e}")
        return False

And then you'd update your server.py to apply the patch

from mcp_memory_leak_fix import apply_mcp_memory_leak_fix
apply_mcp_memory_leak_fix()
from fastmcp import FastMCP

mcp = FastMCP("My MCP Server", stateless_http=True)

@mcp.tool
def greet(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

(no changes to client)

[MadanMaram]

@nathan-thillairajah Thank you for the temp solution.

[ketkarah]

Hi @nathan-thillairajah I too was facing a similar issue of memory leak, and your solution really works for stateless http. However, I tried following a similar approach for stateful connections, and getting task group initialization error. Here is my code

def apply_mcp_memory_leak_fix_stateful():
    """Monkey-patch the MCP SDK to fix the stateless memory leak."""
    try:
        from mcp.server import streamable_http_manager
        from mcp.server.streamable_http import StreamableHTTPServerTransport, MCP_SESSION_ID_HEADER, EventStore
        from starlette.requests import Request
        from starlette.responses import Response
        from uuid import uuid4
        from http import HTTPStatus

        
        # Get the original class
        original_class = streamable_http_manager.StreamableHTTPSessionManager
        original_init = original_class.__init__
        
        def patched_init(self, *args, **kwargs):
            # Call original init
            original_init(self, *args, **kwargs)  
            async def fixed_handle_stateful_request(scope,receive,send) -> None:
                """
                Process request in stateful mode - maintaining session state between requests.

                Args:
                    scope: ASGI scope
                    receive: ASGI receive function
                    send: ASGI send function
                """
                request = Request(scope, receive)
                request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)

                # Existing session case
                if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:
                    transport = self._server_instances[request_mcp_session_id]
                    #logger.debug("Session already exists, handling request directly")
                    await transport.handle_request(scope, receive, send)
                    return

                if request_mcp_session_id is None:
                    # New session case
                    #logger.debug("Creating new transport")
                    async with self._session_creation_lock:
                        new_session_id = uuid4().hex
                        http_transport = StreamableHTTPServerTransport(
                            mcp_session_id=new_session_id,
                            is_json_response_enabled=self.json_response,
                            event_store=self.event_store,  # May be None (no resumability)
                            security_settings=self.security_settings,
                        )

                        assert http_transport.mcp_session_id is not None
                        self._server_instances[http_transport.mcp_session_id] = http_transport
                        #logger.info(f"Created new transport with session ID: {new_session_id}")

                        # Define the server runner
                        async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
                            async with http_transport.connect() as streams:
                                read_stream, write_stream = streams
                                task_status.started()
                                try:
                                    await self.app.run(
                                        read_stream,
                                        write_stream,
                                        self.app.create_initialization_options(),
                                        stateless=False,  # Stateful mode
                                    )
                                except Exception as e:
                                    """logger.error(
                                        f"Session {http_transport.mcp_session_id} crashed: {e}",
                                        exc_info=True,
                                    )"""
                                    pass
                                finally:
                                    # Only remove from instances if not terminated
                                    if (
                                        http_transport.mcp_session_id
                                        and http_transport.mcp_session_id in self._server_instances
                                        and not http_transport.is_terminated
                                    ):
                                        """logger.info(
                                            "Cleaning up crashed session "
                                            f"{http_transport.mcp_session_id} from "
                                            "active instances."
                                        )"""
                                        del self._server_instances[http_transport.mcp_session_id]

                        # Assert task group is not None for type checking
                        assert self._task_group is not None
                        # Start the server task
                        await self._task_group.start(run_server)

                        # Handle the HTTP request and return the response
                        try:
                            await http_transport.handle_request(scope, receive, send)
                            # FIX: Close transport streams so tasks can exit
                        finally:
                            try:
                                if hasattr(http_transport, "_read_stream_writer") and http_transport._read_stream_writer:
                                    await http_transport._read_stream_writer.aclose()
                                if hasattr(http_transport, "_write_stream") and http_transport._write_stream:
                                    await http_transport._write_stream.aclose()
                                if hasattr(http_transport, "close"):
                                    await http_transport.close()
                            except Exception as e:
                            #logger.debug(f"Error during transport cleanup: {e}")
                                pass
                else:
                    # Invalid session ID
                    response = Response(
                        "Bad Request: No valid session ID provided",
                        status_code=HTTPStatus.BAD_REQUEST,
                    )
                    await response(scope, receive, send)
            self._handle_stateful_request = fixed_handle_stateful_request
        
        # Apply the patch
        original_class.__init__ = patched_init
        #logger.info("Applied MCP memory leak fix")
        return True
        
    except Exception as e:
        #logger.error(f"Failed to apply MCP memory leak fix: {e}")
        return False

This is the main server file. Credits: @MadanMaram

from patched_stateless_http import apply_mcp_memory_leak_fix_stateful
apply_mcp_memory_leak_fix_stateful()
from fastmcp import FastMCP
    

mcp = FastMCP("Dummy Server", host="0.0.0.0", port=8000, stateless_http=False)

#Add a simple tool
@mcp.tool
def greet(name: str) -> str:
    #print(mcp.http_app().routes)
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

However, I am getting 'task group not initialized error' when I tried with the same client code as @MadanMaram
Here is the error trace:

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/fastmcp/server/http.py", line 285, in handle_streamable_http
    await session_manager.handle_request(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/mcp/server/streamable_http_manager.py", line 140, in handle_request
    raise RuntimeError("Task group is not initialized. Make sure to use run().")
RuntimeError: Task group is not initialized. Make sure to use run().

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/uvicorn/protocols/http/h11_impl.py", line 403, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/starlette/applications.py", line 112, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 187, in __call__
    raise exc
  File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 165, in __call__
    await self.app(scope, receive, _send)
  File "/usr/local/lib/python3.12/site-packages/fastmcp/server/http.py", line 67, in __call__
    await self.app(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/usr/local/lib/python3.12/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 715, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 735, in app
    await route.handle(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 460, in handle
    await self.app(scope, receive, send)
  File "/usr/local/lib/python3.12/site-packages/fastmcp/server/http.py", line 303, in handle_streamable_http
    raise RuntimeError(f"{new_error_message}\\nOriginal error: {e}") from e
RuntimeError: FastMCP's StreamableHTTPSessionManager task group was not initialized. This commonly occurs when the FastMCP application's lifespan is not passed to the parent ASGI application (e.g., FastAPI or Starlette). Please ensure you are setting `lifespan=mcp_app.lifespan` in your parent app's constructor, where `mcp_app` is the application instance returned by `fastmcp_instance.http_app()`. \nFor more details, see the FastMCP ASGI integration documentation: https://gofastmcp.com/deployment/asgi\nOriginal error: Task group is not initialized. Make sure to use run().

[ketkarah]

Update:
Resolved stateful for now.

class StatefulMemoryLeakFixer:
    """Class to handle patching the MCP SDK to fix the stateful memory leak."""
    
    def __init__(self):
        self.patched = False
    
    def apply_fix(self):
        """Monkey-patch the MCP SDK to fix the stateful memory leak."""
        try:
            from mcp.server import streamable_http_manager
            from mcp.server.streamable_http import StreamableHTTPServerTransport, MCP_SESSION_ID_HEADER, EventStore
            from starlette.requests import Request
            from starlette.responses import Response
            from uuid import uuid4
            from http import HTTPStatus

            # Get the original class
            original_class = streamable_http_manager.StreamableHTTPSessionManager
            original_init = original_class.__init__
            
            # Add a session tracking mechanism
            def patched_init(self, *args, **kwargs):
                # Call original init
                original_init(self, *args, **kwargs)
                
                
                # Helper function to close transport streams
                async def close_transport_streams(transport):
                    """Close all streams associated with a transport"""
                    try:
                        # Close read stream
                        if hasattr(transport, "_read_stream_writer") and transport._read_stream_writer:
                            print(f"Closing read stream for {transport.mcp_session_id}")
                            await transport._read_stream_writer.aclose()
                            transport._read_stream_writer = None
                        
                        # Close write stream
                        if hasattr(transport, "_write_stream") and transport._write_stream:
                            print(f"Closing write stream for {transport.mcp_session_id}")
                            await transport._write_stream.aclose()
                            transport._write_stream = None
                        
                        # Call transport's close method if it exists
                        if hasattr(transport, "close"):
                            print(f"Closing transport for {transport.mcp_session_id}")
                            await transport.close()
                    except Exception as e:
                        print(f"Error closing transport streams: {e}")
                
                self._close_transport_streams = close_transport_streams
                
                # Patch the handle_stateful_request method
                async def fixed_handle_stateful_request(scope, receive, send) -> None:
                    """
                    Process request in stateful mode with improved cleanup.
                    """
                    print("Starting stateful request handler")
                    request = Request(scope, receive)
                    request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
                    
                    # Run session cleanup
                    #await self._cleanup_inactive_sessions()
                    
                    # Existing session case
                    if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:
                        print(f"Using existing session: {request_mcp_session_id}")
                        transport = self._server_instances[request_mcp_session_id]
                        
                        try:
                            await transport.handle_request(scope, receive, send)
                        except Exception as e:
                            print(f"Error handling request for existing session {request_mcp_session_id}: {e}")
                            
                            # Clean up the session if there's an error
                            await self._close_transport_streams(transport)
                            transport._terminated = True
                            del self._server_instances[request_mcp_session_id]
                            
                            # Send error response
                            response = Response(
                                "Internal Server Error", 
                                status_code=HTTPStatus.INTERNAL_SERVER_ERROR
                            )
                            await response(scope, receive, send)
                        return
                    
                    # New session case
                    if request_mcp_session_id is None:
                        print("Creating new transport for stateful session")
                        async with self._session_creation_lock:
                            new_session_id = uuid4().hex
                            http_transport = StreamableHTTPServerTransport(
                                mcp_session_id=new_session_id,
                                is_json_response_enabled=self.json_response,
                                event_store=self.event_store,
                                security_settings=self.security_settings,
                            )
                            
                            assert http_transport.mcp_session_id is not None
                            self._server_instances[http_transport.mcp_session_id] = http_transport
                            
                            print(f"Created new transport with session ID: {new_session_id}")
                            
                            # Define the server runner with better cleanup
                            async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
                                session_id = http_transport.mcp_session_id
                                try:
                                    async with http_transport.connect() as streams:
                                        read_stream, write_stream = streams
                                        task_status.started()
                                        try:
                                            print(f"Running server for session {session_id}")
                                            await self.app.run(
                                                read_stream,
                                                write_stream,
                                                self.app.create_initialization_options(),
                                                stateless=False,
                                            )
                                        except Exception as e:
                                            print(f"Server exception for session {session_id}: {e}")
                                except Exception as e:
                                    print(f"Connection error for session {session_id}: {e}")
                                finally:
                                    print(f"Server task finished for session {session_id}")
                                    # Always clean up when the server task finishes
                                    if session_id in self._server_instances:
                                        try:
                                            await self._close_transport_streams(http_transport)
                                            del self._server_instances[session_id]
                                            print(f"Cleaned up session {session_id} after server task finished")
                                        except Exception as e:
                                            print(f"Error during final cleanup for session {session_id}: {e}")
                            
                            # Start the server task
                            assert self._task_group is not None
                            await self._task_group.start(run_server)
                            
                            try:
                                # Handle the HTTP request
                                print(f"Handling HTTP request for new session {new_session_id}")
                                await http_transport.handle_request(scope, receive, send)
                            except Exception as e:
                                print(f"Error handling request for new session {new_session_id}: {e}")
                                
                                # Clean up on error
                                await self._close_transport_streams(http_transport)
                                if new_session_id in self._server_instances:
                                    del self._server_instances[new_session_id]
                                
                                # Send error response
                                response = Response(
                                    "Internal Server Error", 
                                    status_code=HTTPStatus.INTERNAL_SERVER_ERROR
                                )
                                await response(scope, receive, send)
                    else:
                        # Invalid session ID
                        print(f"Invalid session ID: {request_mcp_session_id}")
                        response = Response(
                            "Bad Request: No valid session ID provided",
                            status_code=HTTPStatus.BAD_REQUEST,
                        )
                        await response(scope, receive, send)
                
                # Replace the original method with our fixed version
                self._handle_stateful_request = fixed_handle_stateful_request
            
            # Apply the patch
            original_class.__init__ = patched_init
            print("Applied MCP memory leak fix for stateful connections")
            self.patched = True
            return True
            
        except Exception as e:
            print(f"Failed to apply MCP memory leak fix: {e}")
            return False

In the run_server function there is a finally to clean up sessions once tasks finish. The other finally after

await http_transport.handle_request(scope, receive, send)

was triggering session cleanup prematurely. Added transport cleanup and session cleanup, both in run_server function and it fixed the issue

Thanks @nathan-thillairajah for the code to patch stateless, as similar approach worked for stateful, albeit with some modifications.
Thanks @MadanMaram for the load-test code for memory leak !