fix(agents): keep Literal and Annotated output schemas

src/google/adk/agents/config_agent_utils.py

@@ -91,6 +91,9 @@ def _is_schema_type(annotation: Any) -> bool:
   origin = typing.get_origin(annotation)
   args = typing.get_args(annotation)
 
+  if origin is typing.Annotated:
+    return _is_schema_type(args[0])
+
   if _is_union(origin):
     return any(_is_schema_type(arg) for arg in args)
 

src/google/adk/models/anthropic_llm.py

@@ -1360,14 +1360,13 @@ async def generate_content_async(
     # A requested schema rides along as a tool, since forcing a call on a
     # request that already declares tools would cut that turn short.
     output_schema_tool: _OutputSchemaTool | None = None
-    if (
-        llm_request.config
-        and llm_request.config.response_schema
-        and not function_declarations
-    ):
-      output_schema_tool = _OutputSchemaTool.for_schema(
-          llm_request.config.response_schema
-      )
+    response_schema = None
+    if llm_request.config:
+      response_schema = llm_request.config.response_schema
+      if response_schema is None:
+        response_schema = llm_request.config.response_json_schema
+    if response_schema and not function_declarations:
+      output_schema_tool = _OutputSchemaTool.for_schema(response_schema)
       tools = [output_schema_tool.param]
       tool_choice = output_schema_tool.choice(
           thinking_enabled=_thinking_is_enabled(thinking)

src/google/adk/models/llm_request.py

@@ -15,6 +15,9 @@
 from __future__ import annotations
 
 import logging
+from typing import Annotated
+from typing import get_origin
+from typing import Literal
 
 from google.genai import types
 from pydantic import BaseModel
@@ -24,6 +27,7 @@
 
 from ..agents.context_cache_config import ContextCacheConfig
 from ..tools.base_tool import BaseTool
+from ..utils._schema_utils import schema_to_json_schema
 from ..utils._schema_utils import SchemaType
 from ._service_tier import ServiceTier
 from .cache_metadata import CacheMetadata
@@ -403,5 +407,9 @@ def set_output_schema(
           " Pass output_schema=<your_schema> (base_model is deprecated)."
       )
 
-    self.config.response_schema = schema
+    if get_origin(schema) in (Literal, Annotated):
+      self.config.response_schema = None
+      self.config.response_json_schema = schema_to_json_schema(schema)
+    else:
+      self.config.response_schema = schema
     self.config.response_mime_type = "application/json"

src/google/adk/utils/_schema_utils.py

@@ -30,21 +30,34 @@
 from typing import cast
 from typing import get_args
 from typing import get_origin
+from typing import Literal
 from typing import Union
 
 from google.genai import types
 from pydantic import BaseModel
 from pydantic import TypeAdapter
+from pydantic import ValidatorFunctionWrapHandler
+from pydantic import WrapValidator
 from typing_extensions import Annotated
 
 from . import _json_utils
 from .content_utils import extract_text_from_content
 
 logger = logging.getLogger("google_adk." + __name__)
 
+
+def _preserve_schema_alias(
+    value: Any, handler: ValidatorFunctionWrapHandler
+) -> Any:
+  """Keeps typing aliases that Schema would otherwise coerce from attributes."""
+  if get_origin(value) in (Literal, Annotated):
+    return value
+  return handler(value)
+
+
 # Use SchemaUnion from google.genai.types to support all schema types
 # that the underlying API supports.
-SchemaType = types.SchemaUnion
+SchemaType = Annotated[types.SchemaUnion, WrapValidator(_preserve_schema_alias)]
 """Type for schema fields (e.g., output_schema, input_schema).
 
 Supports all schema types that the underlying Google GenAI API supports:

tests/unittests/agents/test_agent_config.py

@@ -17,6 +17,7 @@
 import os
 from pathlib import Path
 from textwrap import dedent
+from typing import Annotated
 from typing import Any
 from typing import Literal
 from unittest import mock
@@ -39,6 +40,7 @@
 from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
 import jsonschema
 from pydantic import BaseModel
+from pydantic import Field
 from pydantic import ValidationError
 import pytest
 import yaml
@@ -1938,3 +1940,31 @@ def test_shipped_agent_config_json_rejects_invalid_configs(
   assert isinstance(schema, dict)
   with pytest.raises(jsonschema.ValidationError):
     jsonschema.validate(invalid_config_dict, schema)
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        Literal["yes", "no"],
+        Annotated[BaseModel, Field(title="Answer")],
+        Annotated[int, Field(gt=0)],
+    ],
+)
+def test_from_config_resolves_output_schema_alias(tmp_path, schema):
+  """YAML schema references still resolve through the wrapped schema type."""
+  config_file = tmp_path / "agent.yaml"
+  config_file.write_text(
+      yaml.safe_dump({
+          "name": "test_agent",
+          "instruction": "",
+          "output_schema": {"name": "schemas.Answer"},
+      })
+  )
+
+  with mock.patch.object(
+      config_agent_utils, "resolve_code_reference", return_value=schema
+  ) as resolve:
+    agent = config_agent_utils.from_config(str(config_file))
+
+  resolve.assert_called_once_with(CodeConfig(name="schemas.Answer"))
+  assert agent.output_schema is schema

tests/unittests/agents/test_llm_agent_fields.py

@@ -16,7 +16,9 @@
 
 import asyncio
 import logging
+from typing import Annotated
 from typing import Any
+from typing import Literal
 from unittest import mock
 import warnings
 
@@ -42,6 +44,7 @@
 from google.adk.workflow._function_node import FunctionNode
 from google.genai import types
 from pydantic import BaseModel
+from pydantic import Field
 import pytest
 
 
@@ -1441,3 +1444,19 @@ def my_tool():
   children = list(parent._iter_static_children())
   assert [c.name for c in children] == ['sub', 'tool_wf', 'appended_node']
   assert agent_tool_agent not in children
+
+
+@pytest.mark.parametrize(
+    'schema',
+    [
+        Literal['yes', 'no'],
+        Annotated[BaseModel, Field(title='Answer')],
+        Annotated[int, Field(gt=0)],
+    ],
+)
+def test_output_schema_preserves_typing_alias(schema):
+  """Typing aliases retain their constraints instead of becoming empty schemas."""
+  agent = LlmAgent(name='test_agent', output_schema=schema)
+
+  assert agent.output_schema is schema
+  assert agent.model_dump()['output_schema'] is schema

tests/unittests/events/test_request_input.py

@@ -16,10 +16,15 @@
 
 from __future__ import annotations
 
+from typing import Annotated
+from typing import Literal
 import uuid
 
 from google.adk.events.request_input import RequestInput
 from google.adk.platform import uuid as platform_uuid
+from pydantic import BaseModel
+from pydantic import Field
+import pytest
 
 
 class TestRequestInputInterruptId:
@@ -54,3 +59,19 @@ def test_explicit_interrupt_id_is_preserved(self):
     request = RequestInput(interrupt_id="explicit-id")
 
     assert request.interrupt_id == "explicit-id"
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        Literal["yes", "no"],
+        Annotated[BaseModel, Field(title="Answer")],
+        Annotated[int, Field(gt=0)],
+    ],
+)
+def test_response_schema_preserves_typing_alias(schema):
+  """Input requests retain the declared response schema."""
+  request = RequestInput(response_schema=schema)
+
+  assert request.response_schema is schema
+  assert request.model_dump()["response_schema"] is schema

tests/unittests/models/test_anthropic_llm.py

@@ -18,7 +18,9 @@
 import os
 import re
 import threading
+from typing import Annotated
 from typing import Any
+from typing import Literal
 from unittest import mock
 from unittest.mock import AsyncMock
 from unittest.mock import MagicMock
@@ -46,6 +48,7 @@
 from google.genai.types import Part
 import httpx
 from pydantic import BaseModel
+from pydantic import Field
 import pytest
 
 
@@ -5263,3 +5266,52 @@ async def test_tool_config_none_mode_is_kept_with_thinking():
   )
 
   assert kwargs["tool_choice"] == {"type": "none"}
+
+
+@pytest.mark.parametrize("schema_kind", ["literal", "annotated"])
+@pytest.mark.parametrize("with_response_schema", [False, True])
+async def test_output_schema_json_schema_fallback(
+    schema_kind, with_response_schema
+):
+  """Typing aliases reach Claude as tool schemas, with response_schema preferred."""
+  schema = (
+      Literal["a", "b"]
+      if schema_kind == "literal"
+      else Annotated[_Person, Field(description="A person")]
+  )
+  request = LlmRequest(
+      contents=[Content(role="user", parts=[Part.from_text(text="Who?")])],
+  )
+  request.set_output_schema(schema)
+  if with_response_schema:
+    request.config.response_schema = _Person
+  llm = AnthropicLlm(model="claude-sonnet-4-20250514")
+  mock_client = MagicMock()
+  mock_client.messages.create = AsyncMock(
+      return_value=_tool_use_message({"name": "Ada", "age": 36})
+  )
+
+  with mock.patch.object(llm, "_anthropic_client", mock_client):
+    async for _ in llm.generate_content_async(request):
+      pass
+
+  input_schema = mock_client.messages.create.call_args.kwargs["tools"][0][
+      "input_schema"
+  ]
+  if with_response_schema:
+    assert input_schema == _Person.model_json_schema()
+  elif schema_kind == "literal":
+    assert input_schema == {
+        "type": "object",
+        "properties": {"response": {"type": "string", "enum": ["a", "b"]}},
+        "required": ["response"],
+    }
+  else:
+    assert input_schema == {
+        "type": "object",
+        "properties": {
+            "response": {"$ref": "#/$defs/_Person", "description": "A person"}
+        },
+        "required": ["response"],
+        "$defs": {"_Person": _Person.model_json_schema()},
+    }

tests/unittests/models/test_llm_request.py

@@ -16,6 +16,8 @@
 
 import asyncio
 import logging
+from typing import Annotated
+from typing import Literal
 
 from google.adk.agents.invocation_context import InvocationContext
 from google.adk.agents.sequential_agent import SequentialAgent
@@ -24,7 +26,11 @@
 from google.adk.tools.base_tool import BaseTool
 from google.adk.tools.function_tool import FunctionTool
 from google.adk.tools.tool_context import ToolContext
+from google.adk.utils._schema_utils import schema_to_json_schema
 from google.genai import types
+from google.genai.models import _GenerateContentConfig_to_mldev
+from pydantic import BaseModel
+from pydantic import Field
 import pytest
 
 
@@ -932,3 +938,54 @@ def test_insert_transient_user_content_goes_before_trailing_user_turns():
       'follow-up part 1',
       'follow-up part 2',
   ]
+
+
+@pytest.mark.parametrize(
+    'schema',
+    [
+        Literal['yes', 'no'],
+        Annotated[BaseModel, Field(title='Answer')],
+        Annotated[int, Field(gt=0)],
+    ],
+)
+@pytest.mark.parametrize('argument', ['output_schema', 'base_model'])
+def test_set_output_schema_sends_typing_alias_as_json_schema(schema, argument):
+  """Typing aliases use the JSON schema path supported by GenAI."""
+  request = LlmRequest()
+  request.config.response_schema = str
+
+  request.set_output_schema(**{argument: schema})
+
+  converted = _GenerateContentConfig_to_mldev(None, request.config)
+  assert converted['responseJsonSchema'] == schema_to_json_schema(schema)
+  assert 'responseSchema' not in converted
+  assert request.config.response_schema is None
+  assert request.config.response_json_schema == schema_to_json_schema(schema)
+  assert request.config.response_mime_type == 'application/json'
+  assert request.model_dump()['config']['response_json_schema'] == (
+      schema_to_json_schema(schema)
+  )
+
+
+@pytest.mark.parametrize(
+    'schema',
+    [
+        BaseModel,
+        list[BaseModel],
+        list[str],
+        dict[str, int],
+        int,
+        int | str,
+        {'type': 'string'},
+        types.Schema(type=types.Type.STRING),
+    ],
+)
+def test_set_output_schema_keeps_other_schema_types(schema):
+  """Existing schema inputs continue to use response_schema unchanged."""
+  request = LlmRequest()
+
+  request.set_output_schema(schema)
+
+  assert request.config.response_schema is schema
+  assert request.config.response_json_schema is None
+  assert request.config.response_mime_type == 'application/json'