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 !