async def create_session(self) -> ClientSession:
    """Creates and initializes an MCP client session.

    Returns:
        ClientSession: The initialized MCP client session.
    """
    if self._session is not None:
      return self._session

    # Create a new exit stack for this session
    self._exit_stack = AsyncExitStack()

    try:
      if isinstance(self._connection_params, StdioConnectionParams):
        client = stdio_client(
            server=self._connection_params.server_params,
            errlog=self._errlog,
        )
      elif isinstance(self._connection_params, SseConnectionParams):
        client = sse_client(
            url=self._connection_params.url,
            headers=self._connection_params.headers,
            timeout=self._connection_params.timeout,
            sse_read_timeout=self._connection_params.sse_read_timeout,
        )
      elif isinstance(self._connection_params, StreamableHTTPConnectionParams):
        client = streamablehttp_client(
            url=self._connection_params.url,
            headers=self._connection_params.headers,
            timeout=timedelta(seconds=self._connection_params.timeout),
            sse_read_timeout=timedelta(
                seconds=self._connection_params.sse_read_timeout
            ),
            terminate_on_close=self._connection_params.terminate_on_close,
        )
      else:
        raise ValueError(
            'Unable to initialize connection. Connection should be'
            ' StdioServerParameters or SseServerParams, but got'
            f' {self._connection_params}'
        )

      transports = await self._exit_stack.enter_async_context(client)
      # The streamable http client returns a GetSessionCallback in addition to the read/write MemoryObjectStreams
      # needed to build the ClientSession, we limit then to the two first values to be compatible with all clients.
      if isinstance(self._connection_params, StdioConnectionParams):
        session = await self._exit_stack.enter_async_context(
            ClientSession(
                *transports[:2],
                read_timeout_seconds=timedelta(
                    seconds=self._connection_params.timeout
                ),
            )
        )
      else:
        session = await self._exit_stack.enter_async_context(
            ClientSession(*transports[:2])
        )
      await session.initialize()

      self._session = session
      return session

    except Exception:
      # If session creation fails, clean up the exit stack
      if self._exit_stack:
        await self._exit_stack.aclose()
        self._exit_stack = None
      raise
Creates and initializes an MCP client session.

    Returns:
        ClientSession: The initialized MCP client session.
    
src/google/adk/tools/mcp_tool/mcp_session_manager.py