Merge branch 'main' into main

Author: zhangp365

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
   "click>=8.1.8, <9.0.0",                                    # For CLI tools
   "fastapi>=0.115.0, <0.124.0",                              # FastAPI framework
   "google-api-python-client>=2.157.0, <3.0.0",               # Google API client discovery
+  "google-auth>=2.47.0",                                     # Google Auth library
   "google-cloud-aiplatform[agent_engines]>=1.132.0, <2.0.0", # For VertexAI integrations, e.g. example store.
   "google-cloud-bigquery-storage>=2.0.0",
   "google-cloud-bigquery>=2.2.0",

src/google/adk/agents/live_request_queue.py

@@ -45,15 +45,6 @@ class LiveRequestQueue:
   """Queue used to send LiveRequest in a live(bidirectional streaming) way."""
 
   def __init__(self):
-    # Ensure there's an event loop available in this thread
-    try:
-      asyncio.get_running_loop()
-    except RuntimeError:
-      # No running loop, create one
-      loop = asyncio.new_event_loop()
-      asyncio.set_event_loop(loop)
-
-    # Now create the queue (it will use the event loop we just ensured exists)
     self._queue = asyncio.Queue()
 
   def close(self):

src/google/adk/auth/credential_manager.py

@@ -176,17 +176,13 @@ async def get_auth_credential(
   async def _load_existing_credential(
       self, callback_context: CallbackContext
   ) -> Optional[AuthCredential]:
-    """Load existing credential from credential service or cached exchanged credential."""
+    """Load existing credential from credential service."""
 
     # Try loading from credential service first
     credential = await self._load_from_credential_service(callback_context)
     if credential:
       return credential
 
-    # Check if we have a cached exchanged credential
-    if self._auth_config.exchanged_auth_credential:
-      return self._auth_config.exchanged_auth_credential
-
     return None
 
   async def _load_from_credential_service(

src/google/adk/cli/adk_web_server.py

@@ -1547,14 +1547,31 @@ async def event_generator():
               )
           ) as agen:
             async for event in agen:
-              # Format as SSE data
-              sse_event = event.model_dump_json(
-                  exclude_none=True, by_alias=True
-              )
-              logger.debug(
-                  "Generated event in agent run streaming: %s", sse_event
-              )
-              yield f"data: {sse_event}\n\n"
+              # ADK Web renders artifacts from `actions.artifactDelta`
+              # during part processing *and* during action processing
+              # 1) the original event with `artifactDelta` cleared (content)
+              # 2) a content-less "action-only" event carrying `artifactDelta`
+              events_to_stream = [event]
+              if (
+                  event.actions.artifact_delta
+                  and event.content
+                  and event.content.parts
+              ):
+                content_event = event.model_copy(deep=True)
+                content_event.actions.artifact_delta = {}
+                artifact_event = event.model_copy(deep=True)
+                artifact_event.content = None
+                events_to_stream = [content_event, artifact_event]
+
+              for event_to_stream in events_to_stream:
+                sse_event = event_to_stream.model_dump_json(
+                    exclude_none=True,
+                    by_alias=True,
+                )
+                logger.debug(
+                    "Generated event in agent run streaming: %s", sse_event
+                )
+                yield f"data: {sse_event}\n\n"
         except Exception as e:
           logger.exception("Error in event_generator: %s", e)
           # You might want to yield an error event here

src/google/adk/cli/cli_tools_click.py

@@ -36,6 +36,7 @@
 from . import cli_deploy
 from .. import version
 from ..evaluation.constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
+from ..sessions.migration import migration_runner
 from .cli import run_cli
 from .fast_api import get_fast_api_app
 from .utils import envs
@@ -1507,6 +1508,47 @@ def cli_deploy_cloud_run(
     click.secho(f"Deploy failed: {e}", fg="red", err=True)
 
 
+@main.group()
+def migrate():
+  """ADK migration commands."""
+  pass
+
+
+@migrate.command("session", cls=HelpfulCommand)
+@click.option(
+    "--source_db_url",
+    required=True,
+    help=(
+        "SQLAlchemy URL of source database in database session service, e.g."
+        " sqlite:///source.db."
+    ),
+)
+@click.option(
+    "--dest_db_url",
+    required=True,
+    help=(
+        "SQLAlchemy URL of destination database in database session service,"
+        " e.g. sqlite:///dest.db."
+    ),
+)
+@click.option(
+    "--log_level",
+    type=LOG_LEVELS,
+    default="INFO",
+    help="Optional. Set the logging level",
+)
+def cli_migrate_session(
+    *, source_db_url: str, dest_db_url: str, log_level: str
+):
+  """Migrates a session database to the latest schema version."""
+  logs.setup_adk_logger(getattr(logging, log_level.upper()))
+  try:
+    migration_runner.upgrade(source_db_url, dest_db_url)
+    click.secho("Migration check and upgrade process finished.", fg="green")
+  except Exception as e:
+    click.secho(f"Migration failed: {e}", fg="red", err=True)
+
+
 @deploy.command("agent_engine")
 @click.option(
     "--api_key",

src/google/adk/cli/utils/envs.py

@@ -12,12 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
+import functools
 import logging
 import os
 
 from dotenv import load_dotenv
 
-logger = logging.getLogger(__file__)
+from ...utils.env_utils import is_env_enabled
+
+logger = logging.getLogger('google_adk.' + __name__)
+
+_ADK_DISABLE_LOAD_DOTENV_ENV_VAR = 'ADK_DISABLE_LOAD_DOTENV'
+
+
+@functools.lru_cache(maxsize=1)
+def _get_explicit_env_keys() -> frozenset[str]:
+  """Returns env var keys set before ADK loads any `.env` files.
+
+  This snapshot is used to preserve user-provided environment variables while
+  still allowing later `.env` files to override earlier ones via
+  `override=True`.
+  """
+  return frozenset(os.environ)
 
 
 def _walk_to_root_until_found(folder, filename) -> str:
@@ -35,15 +53,33 @@ def _walk_to_root_until_found(folder, filename) -> str:
 def load_dotenv_for_agent(
     agent_name: str, agent_parent_folder: str, filename: str = '.env'
 ):
-  """Loads the .env file for the agent module."""
+  """Loads the `.env` file for the agent module.
+
+  Explicit environment variables (present before the first `.env` load) are
+  preserved, while values loaded from `.env` may be overridden by later `.env`
+  loads.
+  """
+  if is_env_enabled(_ADK_DISABLE_LOAD_DOTENV_ENV_VAR):
+    logger.info(
+        'Skipping %s loading because %s is enabled.',
+        filename,
+        _ADK_DISABLE_LOAD_DOTENV_ENV_VAR,
+    )
+    return
 
   # Gets the folder of agent_module as starting_folder
   starting_folder = os.path.abspath(
       os.path.join(agent_parent_folder, agent_name)
   )
   dotenv_file_path = _walk_to_root_until_found(starting_folder, filename)
   if dotenv_file_path:
+    explicit_env_keys = _get_explicit_env_keys()
+    explicit_env = {
+        key: os.environ[key] for key in explicit_env_keys if key in os.environ
+    }
+
     load_dotenv(dotenv_file_path, override=True, verbose=True)
+    os.environ.update(explicit_env)
     logger.info(
         'Loaded %s file for %s at %s',
         filename,

src/google/adk/models/lite_llm.py

@@ -584,9 +584,12 @@ async def _get_content(
     parts: Iterable[types.Part],
     *,
     provider: str = "",
-) -> Union[OpenAIMessageContent, str]:
+) -> OpenAIMessageContent:
   """Converts a list of parts to litellm content.
 
+  Thought parts represent internal model reasoning and are always dropped so
+  they are not replayed back to the model in subsequent turns.
+
   Args:
     parts: The parts to convert.
     provider: The LLM provider name (e.g., "openai", "azure").
@@ -595,11 +598,25 @@ async def _get_content(
     The litellm content.
   """
 
+  parts_without_thought = [part for part in parts if not part.thought]
+  if len(parts_without_thought) == 1:
+    part = parts_without_thought[0]
+    if part.text:
+      return part.text
+    if (
+        part.inline_data
+        and part.inline_data.data
+        and part.inline_data.mime_type
+        and part.inline_data.mime_type.startswith("text/")
+    ):
+      return _decode_inline_text_data(part.inline_data.data)
+
   content_objects = []
-  for part in parts:
+  for part in parts_without_thought:
+    # Skip thought parts to prevent reasoning from being replayed in subsequent
+    # turns. Thought parts are internal model reasoning and should not be sent
+    # back to the model.
     if part.text:
-      if len(parts) == 1:
-        return part.text
       content_objects.append({
           "type": "text",
           "text": part.text,
@@ -611,8 +628,6 @@ async def _get_content(
     ):
       if part.inline_data.mime_type.startswith("text/"):
         decoded_text = _decode_inline_text_data(part.inline_data.data)
-        if len(parts) == 1:
-          return decoded_text
         content_objects.append({
             "type": "text",
             "text": decoded_text,

src/google/adk/sessions/database_session_service.py

@@ -178,12 +178,9 @@ async def _prepare_tables(self):
           self._db_schema_version = await conn.run_sync(
               _schema_check_utils.get_db_schema_version_from_connection
           )
-      except Exception:
-        # If inspection fails, assume the latest schema
-        logger.warning(
-            "Failed to inspect database tables, assuming the latest schema."
-        )
-        self._db_schema_version = _schema_check_utils.LATEST_SCHEMA_VERSION
+      except Exception as e:
+        logger.error("Failed to inspect database tables: %s", e)
+        raise
 
     # Check if tables are created and create them if not
     if self._tables_created:

src/google/adk/sessions/migration/_schema_check_utils.py

@@ -16,6 +16,7 @@
 
 import logging
 
+from sqlalchemy import create_engine as create_sync_engine
 from sqlalchemy import inspect
 from sqlalchemy import text
 
@@ -38,14 +39,16 @@ def _get_schema_version_impl(inspector, connection) -> str:
       if result:
         return result[0]
       else:
-        return LATEST_SCHEMA_VERSION
+        raise ValueError(
+            "Schema version not found in adk_internal_metadata. The database"
+            " might be malformed."
+        )
     except Exception as e:
-      logger.warning(
-          "Failed to query schema version from adk_internal_metadata,"
-          " assuming the latest schema: %s.",
+      logger.error(
+          "Failed to query schema version from adk_internal_metadata: %s.",
           e,
       )
-      return LATEST_SCHEMA_VERSION
+      raise
   # Metadata table doesn't exist, check for v0 schema.
   # V0 schema has an 'events' table with an 'actions' column.
   if inspector.has_table("events"):
@@ -57,17 +60,57 @@ def _get_schema_version_impl(inspector, connection) -> str:
             " serialize event actions. The v0 schema will not be supported"
             " going forward and will be deprecated in a few rollouts. Please"
             " migrate to the v1 schema which uses JSON serialization for event"
-            " data. The migration command and script will be provided soon."
+            " data. You can use `adk migrate session` command to migrate your"
+            " database."
         )
         return SCHEMA_VERSION_0_PICKLE
     except Exception as e:
-      logger.warning("Failed to inspect 'events' table columns: %s", e)
-      return LATEST_SCHEMA_VERSION
-  # New database, assume the latest schema.
+      logger.error("Failed to inspect 'events' table columns: %s", e)
+      raise
+  # New database, use the latest schema.
   return LATEST_SCHEMA_VERSION
 
 
 def get_db_schema_version_from_connection(connection) -> str:
   """Gets DB schema version from a DB connection."""
   inspector = inspect(connection)
   return _get_schema_version_impl(inspector, connection)
+
+
+def _to_sync_url(db_url: str) -> str:
+  """Removes '+driver' from SQLAlchemy URL."""
+  if "://" in db_url:
+    scheme, _, rest = db_url.partition("://")
+    if "+" in scheme:
+      dialect = scheme.split("+", 1)[0]
+      return f"{dialect}://{rest}"
+  return db_url
+
+
+def get_db_schema_version(db_url: str) -> str:
+  """Reads schema version from DB.
+
+  Checks metadata table first and then falls back to table structure.
+
+  Args:
+    db_url: The database URL.
+
+  Returns:
+    The detected schema version as a string. Returns `LATEST_SCHEMA_VERSION`
+    if it's a new database.
+  """
+  engine = None
+  try:
+    engine = create_sync_engine(_to_sync_url(db_url))
+    with engine.connect() as connection:
+      inspector = inspect(connection)
+      return _get_schema_version_impl(inspector, connection)
+  except Exception:
+    logger.warning(
+        "Failed to get schema version from database %s.",
+        db_url,
+    )
+    raise
+  finally:
+    if engine:
+      engine.dispose()