google · harshal-96 · Sep 17, 2026 · Sep 18, 2026 · Sep 23, 2026 · Sep 24, 2026
diff --git a/src/google/adk/tools/mcp_tool/_agent_to_mcp.py b/src/google/adk/tools/mcp_tool/_agent_to_mcp.py
@@ -16,8 +16,12 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
+import contextvars
+import logging
 from typing import Any
+from typing import Iterator
 from typing import MutableMapping
 from typing import Optional
 import weakref
@@ -37,8 +41,15 @@
 from ...runners import Runner
 from ...sessions.in_memory_session_service import InMemorySessionService
 
+logger = logging.getLogger("google_adk." + __name__)
+
 _MCP_USER_ID = "mcp_user"
 _INLINE_RESOURCE_URI = "resource://adk-agent/inline-data"
+# Caps the deletes one reap has in flight, so a large backlog does not hit a
+# database-backed session service all at once. Arbitrary, not measured.
+_MAX_CONCURRENT_DELETES = 8
+# Same bound runners.py gives toolset cleanup. Arbitrary, not measured.
+_DELETE_TIMEOUT_SECONDS = 10.0
 
 
 def _build_runner(agent: BaseAgent) -> Runner:
@@ -109,11 +120,183 @@ class to tool functions yet.
   return getattr(session, "_connection", session)
 
 
+async def _delete_session_with_timeout(runner: Runner, session_id: str) -> None:
+  """Deletes one ADK session, giving up after ``_DELETE_TIMEOUT_SECONDS``.
+
+  Only one reap runs at a time, so a delete that never returns would stop all
+  reaping; timing it out lets a later call retry it instead. This uses
+  ``asyncio.wait`` rather than ``asyncio.wait_for``: before Python 3.12,
+  ``wait_for`` can swallow a cancellation that arrives as the delete
+  finishes, which would let a cancelled reap keep draining its backlog.
+
+  Args:
+    runner: The Runner whose session service owns the session.
+    session_id: The id of the session to delete.
+
+  Raises:
+    TimeoutError: If the delete does not finish within the timeout.
+  """
+  delete = asyncio.ensure_future(
+      runner.session_service.delete_session(
+          app_name=runner.app_name,
+          user_id=_MCP_USER_ID,
+          session_id=session_id,
+      )
+  )
+  try:
+    await asyncio.wait({delete}, timeout=_DELETE_TIMEOUT_SECONDS)
+  except asyncio.CancelledError:
+    delete.cancel()
+    if delete.done() and not delete.cancelled():
+      # Retrieve the outcome so asyncio does not report it as unretrieved.
+      delete.exception()
+    raise
+  if not delete.done():
+    delete.cancel()
+    raise TimeoutError(
+        f"delete_session did not finish within {_DELETE_TIMEOUT_SECONDS}s"
+    )
+  delete.result()
+
+
+async def _reap_orphaned_sessions(
+    runner: Runner,
+    sessions: MutableMapping[object, str],
+    created: set[str],
+) -> None:
+  """Deletes ADK sessions whose MCP connection is gone.
+
+  ``sessions`` holds its connections weakly, so an entry vanishes when its
+  connection is garbage-collected; the ADK session it pointed to would stay
+  in the session service forever. Under a stateless streamable HTTP transport
+  the connection lives for a single request, which turns that into one leaked
+  session per tool call. Reaping is started from tool calls rather than from
+  a GC callback, because a GC callback may fire without a running event loop.
+
+  The reap repeats until no orphan is left, including connections that close
+  while it runs, so a backlog drains without waiting for further tool calls.
+  Each batch is deleted by ``_MAX_CONCURRENT_DELETES`` workers sharing one
+  iterator: concurrently, because deleting one at a time caps throughput at
+  one session per delete round trip, which a slow session service cannot
+  sustain under frequent short connections; and through a fixed pool rather
+  than a task per session, because after an outage the backlog can hold
+  many thousands of ids. Once as many deletes have failed as there are
+  workers, the workers stop deleting and the pass ends, so an unavailable
+  service sees fewer than twice the pool size of attempts per pass, while a
+  single session that keeps failing does not hold up the others. Every id
+  the pass did not delete is put back, and a later tool call retries it.
+
+  Args:
+    runner: The Runner whose session service owns the sessions.
+    sessions: Per-connection map from MCP connection to ADK session id.
+    created: Ids of every session ever entered into ``sessions``. Ids no
+      longer reachable through ``sessions`` are deleted and removed from it.
+  """
+
+  async def delete_pending(
+      pending: Iterator[str], errors: list[Exception], max_errors: int
+  ) -> None:
+    for session_id in pending:
+      if len(errors) >= max_errors:
+        created.add(session_id)
+        continue
+      try:
+        await _delete_session_with_timeout(runner, session_id)
+      except asyncio.CancelledError:
+        # The cancelled delete may or may not have landed; deleting a missing
+        # session is a no-op for the built-in services, so a retry is safe.
+        created.add(session_id)
+        raise
+      except Exception as e:  # pylint: disable=broad-exception-caught
+        # Reaping is housekeeping; it must not fail the tool call that
+        # triggered it. Put the id back so a later call retries it.
+        created.add(session_id)
+        errors.append(e)
+
+  while True:
+    batch = list(created - set(sessions.values()))
+    if not batch:
+      return
+    # Claimed in the same synchronous stretch that computed it, so a
+    # concurrent reap never deletes the same id twice.
+    created.difference_update(batch)
+    pending = iter(batch)
+    errors: list[Exception] = []
+    workers = min(_MAX_CONCURRENT_DELETES, len(batch))
+    # Stopping only once there have been as many failures as workers tells an
+    # outage apart from a single session that keeps failing, which must not
+    # hold up the rest of the batch.
+    try:
+      await asyncio.gather(
+          *(delete_pending(pending, errors, workers) for _ in range(workers))
+      )
+    finally:
+      # If the reap is cancelled (e.g. at loop shutdown), ids no worker had
+      # started yet go back too, so they are not dropped from tracking.
+      created.update(pending)
+    if errors:
+      logger.warning(
+          "Failed to delete %d orphaned MCP agent session(s); will retry on a"
+          " later tool call.",
+          len(errors),
+          exc_info=errors[0],
+      )
+      return
+
+
+def _log_reap_failure(task: asyncio.Task[None]) -> None:
+  """Logs an unexpected error from a background reap.
+
+  Retrieving the exception here also keeps asyncio from reporting it later as
+  "Task exception was never retrieved".
+
+  Args:
+    task: The finished reap task.
+  """
+  if task.cancelled():
+    return
+  exc = task.exception()
+  if exc is not None:
+    logger.warning(
+        "Background reap of orphaned MCP agent sessions failed.",
+        exc_info=exc,
+    )
+
+
+def _start_background_reap(
+    runner: Runner,
+    sessions: MutableMapping[object, str],
+    created: set[str],
+) -> asyncio.Task[None]:
+  """Starts `_reap_orphaned_sessions` as a task with failure logging.
+
+  Callers run this inside an empty ``contextvars.Context``. Both the task and
+  its done callback capture the current context, and a copy of the calling
+  tool request's context would keep that request, and through it the MCP
+  connection, alive for as long as the task exists (MCP SDK 1.x stores the
+  request in a ContextVar). A stuck reap would then pin a connection whose
+  session it should delete. Running in an empty context also works on Python
+  3.10, which lacks ``create_task(context=...)``.
+
+  Args:
+    runner: The Runner whose session service owns the sessions.
+    sessions: Per-connection map from MCP connection to ADK session id.
+    created: Ids of every session entered into ``sessions``.
+
+  Returns:
+    The started reap task.
+  """
+  task = asyncio.create_task(_reap_orphaned_sessions(runner, sessions, created))
+  task.add_done_callback(_log_reap_failure)
+  return task
+
+
 async def _run_agent(
     runner: Runner,
     request: str,
     ctx: Optional[Context[ServerSession, Any]] = None,
     sessions: Optional[MutableMapping[object, str]] = None,
+    created: Optional[set[str]] = None,
 ) -> list[mcp_types.ContentBlock]:
   """Runs the agent for one request and returns its final response content.
 
@@ -128,6 +311,8 @@ async def _run_agent(
     request: The user request text for this call.
     ctx: The MCP tool call context, used for progress and session reuse.
     sessions: Per-connection map from MCP connection to ADK session id.
+    created: Set recording the id of every session entered into ``sessions``,
+      so `_reap_orphaned_sessions` can delete the ones whose connection dies.
 
   Returns:
     The agent's final response as a list of MCP content blocks (text plus any
@@ -144,7 +329,11 @@ async def _run_agent(
     )
     session_id = session.id
     if sessions is not None and connection is not None:
+      # No await between the two writes: an id is either absent from both or
+      # present in both, so the reaper never sees a session it cannot delete.
       sessions[connection] = session_id
+      if created is not None:
+        created.add(session_id)
   new_message = types.Content(role="user", parts=[types.Part(text=request)])
   final_content: list[mcp_types.ContentBlock] = []
   async for event in runner.run_async(
@@ -173,6 +362,7 @@ def to_mcp_server(
     name: Optional[str] = None,
     instructions: Optional[str] = None,
     runner: Optional[Runner] = None,
+    delete_orphaned_sessions: bool = True,
 ) -> FastMCP:
   """Exposes an ADK agent as an MCP server.
 
@@ -183,18 +373,33 @@ def to_mcp_server(
   lets harnesses that speak MCP drive an ADK agent.
 
   One ADK session is kept per MCP connection, so successive tool calls on the
-  same connection form a single multi-turn conversation.
+  same connection form a single multi-turn conversation. When a connection
+  goes away its ADK session is deleted from the session service in the
+  background, by a reap that tool calls start, so a long-running server does
+  not accumulate dead conversations.
 
   The caller chooses the transport, e.g. ``server.run(transport="stdio")`` for
   a local host or ``server.run(transport="streamable-http")`` for a networked
-  one.
+  one. A stateless streamable HTTP deployment (``stateless_http=True``, e.g.
+  behind an autoscaler) gets a fresh connection per request, so every tool
+  call is its own single-turn conversation whose session is likewise
+  reclaimed.
 
   Args:
     agent: The ADK agent to serve.
     name: The MCP server and tool name. Defaults to the agent's name.
     instructions: Optional instructions the MCP host may show to its model.
     runner: A pre-built Runner. If omitted, one is created with in-memory
       services.
+    delete_orphaned_sessions: Whether to delete a connection's ADK session
+      from the session service once the connection is gone. Defaults to True,
+      which keeps a long-running server's memory bounded. Set to False to
+      retain finished conversations in the session service, e.g. when a
+      caller-supplied ``runner`` uses a persistent session service whose
+      records are read after the fact; the caller then owns their cleanup.
+      Only sessions created by this server process are tracked, so with a
+      persistent session service, sessions orphaned before a restart are not
+      deleted.
 
   Returns:
     A ``FastMCP`` server exposing the agent as a single tool.
@@ -213,11 +418,40 @@ def to_mcp_server(
   # WeakKeyDictionary() instantiation below as abstract-class-instantiated.
   # pylint: disable-next=abstract-class-instantiated
   sessions: MutableMapping[object, str] = weakref.WeakKeyDictionary()
+  # Ids of every session in `sessions`, kept strongly so the sessions of
+  # collected connections can still be found and deleted. None disables the
+  # tracking and with it the reaping.
+  created_session_ids: Optional[set[str]] = (
+      set() if delete_orphaned_sessions else None
+  )
+
+  # The in-flight reap, if any. Held here so the task is not garbage-collected
+  # mid-run, and so at most one reap runs at a time.
+  reap_task: Optional[asyncio.Task[None]] = None
 
   async def call_agent(
       request: str, ctx: Context[ServerSession, Any]
   ) -> list[mcp_types.ContentBlock]:
-    return await _run_agent(agent_runner, request, ctx, sessions)
+    nonlocal reap_task
+    # A task from another event loop may never finish from this loop's point
+    # of view (e.g. that loop stopped without cancelling it), so it must not
+    # block reaping here.
+    if created_session_ids is not None and (
+        reap_task is None
+        or reap_task.done()
+        or reap_task.get_loop() is not asyncio.get_running_loop()
+    ):
+      # Reap in the background so a slow session service (e.g. a database)
+      # never adds delete latency to the live call.
+      reap_task = contextvars.Context().run(
+          _start_background_reap,
+          agent_runner,
+          sessions,
+          created_session_ids,
+      )
+    return await _run_agent(
+        agent_runner, request, ctx, sessions, created_session_ids
+    )
 
   server.add_tool(
       call_agent,