to_mcp_server never deletes the ADK session of a dead MCP connection; a stateless streamable HTTP deployment leaks one session per tool call

Describe the bug

to_mcp_server keeps one ADK session per MCP connection in a weakref.WeakKeyDictionary (src/google/adk/tools/mcp_tool/_agent_to_mcp.py). When a connection is garbage collected, the weak map drops the entry, but nothing ever calls session_service.delete_session, so the ADK session, with its full event history, stays in the session service forever. The default runner built by to_mcp_server uses InMemorySessionService, which is a plain dict with no eviction.

Two consequences:

1. Stateful serving: a long running server accumulates one dead conversation per closed connection.
2. Stateless streamable HTTP, which is the mode you want behind an autoscaler such as Cloud Run: the MCP SDK builds a completely fresh transport for every request (StreamableHTTPSessionManager with stateless=True), so _connection_key returns a per request object and every tool call creates a new ADK session that is never deleted. That is one leaked session per tool call, growing linearly with traffic, and the sessions hold full conversation content.

To Reproduce

The script below makes 10 short lived in-memory MCP connections with one tool call each, which is exactly the connection pattern a stateless streamable HTTP deployment produces, then counts the sessions left in the service.

import asyncio
from contextlib import asynccontextmanager
import gc
from typing import AsyncGenerator

import anyio
from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.auth.credential_service.in_memory_credential_service import (
    InMemoryCredentialService,
)
from google.adk.events.event import Event
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.tools.mcp_tool import to_mcp_server
from google.genai import types
from mcp.client.session import ClientSession
from mcp.shared.memory import create_client_server_memory_streams


@asynccontextmanager
async def connected_client(lowlevel_server):
  """One in-memory MCP connection, the same shape a network client makes."""
  async with create_client_server_memory_streams() as (
      client_streams,
      server_streams,
  ):
    client_read, client_write = client_streams
    server_read, server_write = server_streams
    async with anyio.create_task_group() as tg:
      tg.start_soon(
          lambda: lowlevel_server.run(
              server_read,
              server_write,
              lowlevel_server.create_initialization_options(),
          )
      )
      try:
        async with ClientSession(
            read_stream=client_read, write_stream=client_write
        ) as session:
          await session.initialize()
          yield session
      finally:
        tg.cancel_scope.cancel()


class EchoAgent(BaseAgent):

  async def _run_async_impl(
      self, ctx: InvocationContext
  ) -> AsyncGenerator[Event, None]:
    yield Event(
        author=self.name,
        content=types.Content(role="model", parts=[types.Part(text="ok")]),
    )


async def main():
  agent = EchoAgent(name="echo")
  session_service = InMemorySessionService()
  runner = Runner(
      app_name="echo",
      agent=agent,
      session_service=session_service,
      artifact_service=InMemoryArtifactService(),
      memory_service=InMemoryMemoryService(),
      credential_service=InMemoryCredentialService(),
  )
  server = to_mcp_server(agent, runner=runner)
  lowlevel = getattr(server, "_mcp_server", None) or getattr(
      server, "_lowlevel_server", server
  )

  for i in range(10):
    async with connected_client(lowlevel) as client:
      await client.call_tool("echo", {"request": f"call {i}"})
    gc.collect()

  remaining = await session_service.list_sessions(
      app_name="echo", user_id="mcp_user"
  )
  print(f"sessions left in the service: {len(remaining.sessions)}")


asyncio.run(main())

Output on main: sessions left in the service: 10

Also reproduced over a real transport: running the server with run_streamable_http_async(host="127.0.0.1", port=8765, stateless_http=True) and making each call over a fresh streamable_http_client connection, 9 calls leave 9 sessions in the service.

Expected behavior

Sessions whose connection is gone are deleted from the session service. With the fix below, the same runs leave 1 session (the most recent call's, reclaimed on the next call).

Environment

google-adk main (7ae1c9b), mcp 2.2.0 (also reproduces on 1.24.0 and 1.26.0), Python 3.12.

Proposed fix (PR to follow)

Record the id of every session entered into the connection map, and at the start of each tool call delete the sessions that are no longer reachable through the weak map. Reaping runs lazily from the tool call rather than from a GC finalizer, because finalizers can fire without a running event loop. This adds no TTL or retention policy of its own: connection lifetime is already governed by the MCP transport layer (client disconnects, the SDK's idle session timeout, stateless per request teardown), and the fix makes ADK sessions follow that lifetime instead of outliving it.

One design note for review: with a caller supplied Runner backed by a persistent session service, deleting orphaned sessions is a behavior change, since conversations would no longer remain readable after their connection dies. I think cleanup is the right default here, because the module creates these sessions under its internal mcp_user id and unbounded growth is the worse failure, but I am happy to add an opt-out flag if retention matters for some deployments.

Possible follow-ups, feedback welcome

1. A stateless=True option on to_mcp_server that skips the connection map and deletes each session eagerly after its call, as explicit support for deployments that run the server with stateless_http=True.
2. An optional conversation id argument on the generated tool, so a client of a stateless deployment can keep multi-turn conversations by passing the id back with each call. Today stateless mode silently degrades to one single turn conversation per call, because there is no connection identity to thread a conversation on.

I have the first ready and would build the second if there is interest.