Runs the tool by converting tool call to env action and stepping env.
(self, *, args: Dict[str, Any], tool_context: Any)
| 86 | return self._function_declaration |
| 87 | |
| 88 | async def run_async(self, *, args: Dict[str, Any], tool_context: Any) -> str: |
| 89 | """Runs the tool by converting tool call to env action and stepping env.""" |
| 90 | env_response = self._env.step( |
| 91 | types.Part(function_call=types.FunctionCall(name=self.name, args=args)) |
| 92 | ) |
| 93 | # We modify the ADK session state with the updates from the environment, |
| 94 | # in particular `done` and `reward`. These can be consumed downstream for |
| 95 | # instance to extract the trajectory reward or interrupt the loop. |
| 96 | tool_context.actions.state_delta['done'] = env_response.done |
| 97 | tool_context.actions.state_delta['reward'] = env_response.reward |
| 98 | tool_context.actions.skip_summarization = True |
| 99 | if env_response.done: |
| 100 | tool_context.actions.escalate = True |
| 101 | return env_response.observation |
| 102 | |
| 103 | |
| 104 | def _default_retry_options() -> types.HttpRetryOptions: |
no test coverage detected