Open
Description
Describe the bug
When using a langchaintool with run_live, it currently throws the following error: NameError: name 'Union' is not defined. This gets triggered by the get_type_hints in the run_live function, likely because it is getting the hint on the LC wrapper and not the tool itself.
To Reproduce
from google.adk.agents import Agent, LiveRequestQueue
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from pydantic import BaseModel, Field
from langchain.tools.base import StructuredTool
from langchain.tools import tool
from typing import get_type_hints
from google.adk.tools.langchain_tool import LangchainTool
@tool
def my_custom_tool(input: str) -> str:
"""
A custom tool that processes the input string and returns a modified version.
Args:
input (str): The input string to be processed.
Returns:
str: The processed output string.
"""
# Example processing: simply return the input string in uppercase
return 'Result from my_custom_tool'
class CustomToolInput(BaseModel):
input: str = Field(description="The input string to be processed.")
custom_tool_structured = StructuredTool.from_function(
func=my_custom_tool,
name='my custom tool',
description='my custom tool',
args_schema=CustomToolInput
)
adk_custom_tool = LangchainTool(
custom_tool_structured
)
agent = Agent(
name="live_agent",
model="gemini-live-2.5-flash",
instruction="test!",
tools=[adk_custom_tool],
)
session_service = InMemorySessionService()
runner = Runner(
app_name="voice_assistant", agent=agent, session_service=session_service
)
live_request_queue = LiveRequestQueue()
VOICE_NAME = "Puck"
session = await session_service.create_session(
app_name="voice_assistant",
user_id="default_user"
)
run_config = RunConfig(
streaming_mode=StreamingMode.BIDI,
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=VOICE_NAME)
)
),
response_modalities=["AUDIO"],
output_audio_transcription=types.AudioTranscriptionConfig(),
input_audio_transcription=types.AudioTranscriptionConfig(),
)
async for event in runner.run_live(
session=session,
live_request_queue=live_request_queue,
run_config=run_config
):
print("hi")
/var/tmp/ipykernel_106190/965267043.py:62: DeprecationWarning: The `session` parameter is deprecated. Please use `user_id` and `session_id` instead.
async for event in runner.run_live(
---------------------------------------------------------------------------
NameError Traceback (most recent call last)
Cell In[5], line 62
46 session = await session_service.create_session(
47 app_name="voice_assistant",
48 user_id="default_user"
49 )
50 run_config = RunConfig(
51 streaming_mode=StreamingMode.BIDI,
52 speech_config=types.SpeechConfig(
(...)
59 input_audio_transcription=types.AudioTranscriptionConfig(),
60 )
---> 62 async for event in runner.run_live(
63 session=session,
64 live_request_queue=live_request_queue,
65 run_config=run_config
66 ):
67 print("hi")
File /opt/conda/lib/python3.10/site-packages/google/adk/runners.py:319, in Runner.run_live(self, user_id, session_id, live_request_queue, run_config, session)
314 for tool in invocation_context.agent.tools:
315 # replicate a LiveRequestQueue for streaming tools that relis on
316 # LiveRequestQueue
317 from typing import get_type_hints
--> 319 type_hints = get_type_hints(tool)
320 for arg_type in type_hints.values():
321 if arg_type is LiveRequestQueue:
File /opt/conda/lib/python3.10/typing.py:1871, in get_type_hints(obj, globalns, localns, include_extras)
1863 if isinstance(value, str):
1864 # class-level forward refs were handled above, this must be either
1865 # a module-level annotation or a function argument annotation
1866 value = ForwardRef(
1867 value,
1868 is_argument=not isinstance(obj, types.ModuleType),
1869 is_class=False,
1870 )
-> 1871 value = _eval_type(value, globalns, localns)
1872 if name in defaults and defaults[name] is None:
1873 value = Optional[value]
File /opt/conda/lib/python3.10/typing.py:327, in _eval_type(t, globalns, localns, recursive_guard)
321 """Evaluate all forward references in the given type t.
322 For use of globalns and localns see the docstring for get_type_hints().
323 recursive_guard is used to prevent infinite recursion with a recursive
324 ForwardRef.
325 """
326 if isinstance(t, ForwardRef):
--> 327 return t._evaluate(globalns, localns, recursive_guard)
328 if isinstance(t, (_GenericAlias, GenericAlias, types.UnionType)):
329 ev_args = tuple(_eval_type(a, globalns, localns, recursive_guard) for a in t.__args__)
File /opt/conda/lib/python3.10/typing.py:694, in ForwardRef._evaluate(self, globalns, localns, recursive_guard)
689 if self.__forward_module__ is not None:
690 globalns = getattr(
691 sys.modules.get(self.__forward_module__, None), '__dict__', globalns
692 )
693 type_ = _type_check(
--> 694 eval(self.__forward_code__, globalns, localns),
695 "Forward references must evaluate to types.",
696 is_argument=self.__forward_is_argument__,
697 allow_special_forms=self.__forward_is_class__,
698 )
699 self.__forward_value__ = _eval_type(
700 type_, globalns, localns, recursive_guard | {self.__forward_arg__}
701 )
702 self.__forward_evaluated__ = True
File <string>:1
NameError: name 'Union' is not defined
Expected behavior
A clear and concise description of what you expected to happen.
I expect to call the agent successfully, not get a typing error back
Desktop (please complete the following information):
- OS: [e.g. iOS]
- Python version(python -V): 3.10
- ADK version(pip show google-adk): 1.5.0
Model Information:
gemini-2.5-live
Additional context
Add any other context about the problem here.