From b41c8ae4c574e4304613a2aac5d428ecb984b9a9 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 19:57:52 +0800 Subject: [PATCH 1/9] feat(tools): allow model-selected execution timeouts --- tests/tools/builtin_tools/test_run_code.py | 117 ++++++++++++++++++ .../builtin_tools/test_run_sandbox_agent.py | 70 +++++++++++ veadk/tools/builtin_tools/execute_skills.py | 16 ++- veadk/tools/builtin_tools/run_code.py | 22 +++- 4 files changed, 219 insertions(+), 6 deletions(-) create mode 100644 tests/tools/builtin_tools/test_run_code.py diff --git a/tests/tools/builtin_tools/test_run_code.py b/tests/tools/builtin_tools/test_run_code.py new file mode 100644 index 000000000..25bdbe36e --- /dev/null +++ b/tests/tools/builtin_tools/test_run_code.py @@ -0,0 +1,117 @@ +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from google.adk.tools import FunctionTool + +from veadk.tools.builtin_tools.execute_skills import execute_skills +from veadk.tools.builtin_tools import run_code as run_code_module + + +def _tool_context(): + return SimpleNamespace( + state={}, + _invocation_context=SimpleNamespace( + session=SimpleNamespace(id="session-id"), + agent=SimpleNamespace(name="agent-name"), + user_id="user-id", + ), + ) + + +@pytest.mark.parametrize( + ("function", "maximum", "range_text"), + [ + (execute_skills, 900, "between 1 and 900 seconds"), + (run_code_module.run_code, 300, "between 1 and 300 seconds"), + ], +) +def test_timeout_is_exposed_in_function_call_declaration(function, maximum, range_text): + declaration = FunctionTool(function)._get_declaration() + + assert declaration is not None + assert declaration.parameters_json_schema is not None + timeout_schema = declaration.parameters_json_schema["properties"]["timeout"] + assert timeout_schema["default"] == maximum + assert timeout_schema["type"] == "integer" + description = " ".join(declaration.description.split()) + assert range_text in description + + +@pytest.mark.parametrize("timeout", [0, -1, 301, 1.5, True]) +def test_run_code_rejects_invalid_timeout(timeout): + with pytest.raises( + ValueError, + match=r"timeout must be an integer between 1 and 300 seconds", + ): + run_code_module.run_code( + "print('hello')", + "python3", + _tool_context(), + timeout=timeout, + ) + + +@pytest.mark.parametrize("hard_timeout", [0, -1, 301, 1.5, True]) +def test_run_code_rejects_invalid_hard_timeout(hard_timeout): + with pytest.raises( + ValueError, + match=r"hard_timeout must be an integer between 1 and 300 seconds", + ): + run_code_module.run_code( + "echo hello", + "bash", + _tool_context(), + hard_timeout=hard_timeout, + ) + + +def test_run_code_uses_default_timeout_for_python(): + with ( + patch.object(run_code_module, "resolve_agentkit_tool_id", return_value="tool"), + patch.object( + run_code_module, + "get_agentkit_endpoint_config", + return_value=("service", "region", "host", "https"), + ), + patch.object( + run_code_module, + "invoke_agentkit_run_code", + return_value={"Result": {"Result": "ok"}}, + ) as invoke, + ): + result = run_code_module.run_code( + "print('hello')", + "python3", + _tool_context(), + ) + + assert result == "ok" + assert invoke.call_args.kwargs["timeout"] == 300 + + +def test_run_code_forwards_custom_timeouts_for_bash(): + with ( + patch.object(run_code_module, "resolve_agentkit_tool_id", return_value="tool"), + patch.object( + run_code_module, + "get_agentkit_endpoint_config", + return_value=("service", "region", "host", "https"), + ), + patch.object( + run_code_module, + "invoke_agentkit_exec_bash", + return_value={"Result": {"Result": "ok"}}, + ) as invoke, + ): + result = run_code_module.run_code( + "echo hello", + "bash", + _tool_context(), + timeout=120, + hard_timeout=180, + ) + + assert result == "ok" + assert invoke.call_args.kwargs["timeout"] == 120 + assert invoke.call_args.kwargs["hard_timeout"] == 180 diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 59391a023..307842949 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -254,6 +254,76 @@ def fake_urlopen(request, timeout=None): self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) self.assertIn(b'"prompt": "do work"', request_obj.data) + def test_execute_skills_forwards_custom_timeout(self): + session_kwargs = [] + request_timeouts = [] + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return b'{"content": "api result"}' + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **kwargs: ( + session_kwargs.append(kwargs) or "https://sandbox.test" + ), + ) + + with patch.object( + module.request, + "urlopen", + lambda _request, timeout=None: ( + request_timeouts.append(timeout) or FakeResponse() + ), + ): + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + timeout=120, + ) + + self.assertEqual("api result", result) + self.assertEqual(1800, session_kwargs[0]["ttl"]) + self.assertEqual([120], request_timeouts) + + def test_execute_skills_forwards_custom_timeout_to_legacy_path(self): + captured_kwargs = {} + module = _load_execute_skills_module( + run_sandbox_agent=lambda **kwargs: ( + captured_kwargs.update(kwargs) or "legacy result" + ), + ) + + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + env_vars={"CUSTOM_VALUE": "custom"}, + timeout=120, + ) + + self.assertEqual("legacy result", result) + self.assertEqual(120, captured_kwargs["timeout"]) + + def test_execute_skills_rejects_invalid_timeout(self): + module = _load_execute_skills_module() + + for timeout in (0, -1, 901, 1.5, True): + with self.subTest(timeout=timeout): + with self.assertRaisesRegex( + ValueError, + r"timeout must be an integer between 1 and 900 seconds", + ): + module.execute_skills( + "do work", + tool_context=self._tool_context(), + timeout=timeout, + ) + def test_health_check_retries_502_until_upstream_is_ready(self): attempts = [] diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 75059ab54..bf2628c33 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -40,6 +40,13 @@ _SKILL_API_HEALTH_REQUEST_TIMEOUT = 5.0 +def _validate_timeout(timeout: int) -> None: + if type(timeout) is not int or not 1 <= timeout <= _SKILL_API_TIMEOUT: + raise ValueError( + f"timeout must be an integer between 1 and {_SKILL_API_TIMEOUT} seconds" + ) + + def _skill_api_upgrade_hint(path: str) -> str: api_path = ( "/v1/skills/stream" @@ -266,6 +273,7 @@ def execute_skills( tool_context: ToolContext = None, env_vars: Optional[dict[str, str]] = None, prefer_stream: bool = False, + timeout: int = _SKILL_API_TIMEOUT, ) -> str: """Execute skills in a sandbox and return the output. @@ -276,12 +284,16 @@ def execute_skills( env_vars (Optional[dict[str, str]]): Environment variables passed to the skill agent process for this execution only. Requests with custom environment variables use the legacy RunCode execution path. + timeout (int, optional): Maximum execution time in seconds. Defaults to + 900. The value can be adjusted for each call but must be between 1 + and 900 seconds. Returns: str: The output of the code execution. """ if tool_context is None: raise ValueError("tool_context is required for execute_skills") + _validate_timeout(timeout) tool_id = resolve_agentkit_tool_id("AGENTKIT_TOOL_ID_SKILLS") if env_vars: @@ -296,7 +308,7 @@ def execute_skills( workflow_prompt=workflow_prompt, tool_id=tool_id, tool_context=tool_context, - timeout=_SKILL_API_TIMEOUT, + timeout=timeout, extra_env_vars=extra_env_vars, ) @@ -305,5 +317,5 @@ def execute_skills( tool_id=tool_id, tool_context=tool_context, prefer_stream=prefer_stream, - timeout=_SKILL_API_TIMEOUT, + timeout=timeout, ) diff --git a/veadk/tools/builtin_tools/run_code.py b/veadk/tools/builtin_tools/run_code.py index 7497a208c..9a3249cf6 100644 --- a/veadk/tools/builtin_tools/run_code.py +++ b/veadk/tools/builtin_tools/run_code.py @@ -26,15 +26,24 @@ logger = get_logger(__name__) +_MAX_TIMEOUT = 300 + + +def _validate_timeout(name: str, value: int) -> None: + if type(value) is not int or not 1 <= value <= _MAX_TIMEOUT: + raise ValueError( + f"{name} must be an integer between 1 and {_MAX_TIMEOUT} seconds" + ) + def run_code( code: str, language: str, tool_context: ToolContext, - timeout: int = 30, + timeout: int = _MAX_TIMEOUT, exec_dir: str = "/tmp", env: dict[str, str] | None = None, - hard_timeout: int = 300, + hard_timeout: int = _MAX_TIMEOUT, max_output_length: int = 30000, ) -> str: """Run code in a code sandbox and return the output. @@ -43,16 +52,21 @@ def run_code( Args: code (str): The code to run. language (str): The execution language. Use ``python3`` for code or ``bash`` for shell scripts. - timeout (int, optional): The timeout in seconds for the code execution. Defaults to 30. + timeout (int, optional): The timeout in seconds for the code execution. + Defaults to 300 and must be between 1 and 300 seconds. exec_dir (str, optional): Working directory for Bash execution. Defaults to ``/tmp``. env (dict[str, str], optional): Environment variables for Bash execution. - hard_timeout (int, optional): Hard timeout for Bash execution. Defaults to 300 seconds. + hard_timeout (int, optional): Hard timeout for Bash execution. Defaults + to 300 and must be between 1 and 300 seconds. max_output_length (int, optional): Maximum Bash output length. Defaults to 30000. Returns: str: The output of the code execution. """ + _validate_timeout("timeout", timeout) + _validate_timeout("hard_timeout", hard_timeout) + tool_id = resolve_agentkit_tool_id("AGENTKIT_TOOL_ID_SCRIPT") service, region, host, _ = get_agentkit_endpoint_config() logger.debug(f"tools endpoint: {host}") From 50857094856e5ae6158a615fe69558c96a3a1ee7 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 21:05:21 +0800 Subject: [PATCH 2/9] feat(tools): extend execute skills timeout to 30 minutes --- tests/tools/builtin_tools/test_run_code.py | 2 +- tests/tools/builtin_tools/test_run_sandbox_agent.py | 10 +++++----- veadk/tools/builtin_tools/execute_skills.py | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_code.py b/tests/tools/builtin_tools/test_run_code.py index 25bdbe36e..ce2711381 100644 --- a/tests/tools/builtin_tools/test_run_code.py +++ b/tests/tools/builtin_tools/test_run_code.py @@ -22,7 +22,7 @@ def _tool_context(): @pytest.mark.parametrize( ("function", "maximum", "range_text"), [ - (execute_skills, 900, "between 1 and 900 seconds"), + (execute_skills, 1800, "between 1 and 1800 seconds"), (run_code_module.run_code, 300, "between 1 and 300 seconds"), ], ) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index fe349aef0..08b72337f 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -250,7 +250,7 @@ def fake_urlopen(request, timeout=None): self.assertEqual(1, len(captured_requests)) request_obj, timeout = captured_requests[0] self.assertEqual("https://sandbox.test/v1/skills/execute", request_obj.full_url) - self.assertEqual(900, timeout) + self.assertEqual(1800, timeout) self.assertEqual("POST", request_obj.get_method()) self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) self.assertIn(b'"prompt": "do work"', request_obj.data) @@ -313,11 +313,11 @@ def test_execute_skills_forwards_custom_timeout_to_legacy_path(self): def test_execute_skills_rejects_invalid_timeout(self): module = _load_execute_skills_module() - for timeout in (0, -1, 901, 1.5, True): + for timeout in (0, -1, 1801, 1.5, True): with self.subTest(timeout=timeout): with self.assertRaisesRegex( ValueError, - r"timeout must be an integer between 1 and 900 seconds", + r"timeout must be an integer between 1 and 1800 seconds", ): module.execute_skills( "do work", @@ -501,7 +501,7 @@ def fake_urlopen(request, timeout=None): self.assertEqual(result, "hello world") request_obj, timeout = captured_requests[0] self.assertEqual("https://sandbox.test/run_sse", request_obj.full_url) - self.assertEqual(900, timeout) + self.assertEqual(1800, timeout) self.assertIn(b'"app_name": "agent"', request_obj.data) self.assertIn(b'"session_id": "session-1"', request_obj.data) self.assertIn(b'"text": "do work"', request_obj.data) @@ -548,7 +548,7 @@ def fake_urlopen(request, timeout=None): request_obj, timeout = captured_requests[0] payload = json.loads(request_obj.data.decode()) self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) - self.assertEqual(900, timeout) + self.assertEqual(1800, timeout) self.assertEqual("message/send", payload["method"]) self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) self.assertTrue(payload["params"]["configuration"]["blocking"]) diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 33773d32e..0fd70c6e2 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -33,7 +33,7 @@ _SKILL_API_UPGRADE_STATUS_CODES = frozenset({404, 405}) _SKILL_API_TRANSIENT_STATUS_CODES = frozenset({502, 503, 504}) -_SKILL_API_TIMEOUT = 900 +_SKILL_API_TIMEOUT = 1800 _SKILL_API_HEALTH_TIMEOUT = 30.0 _SKILL_API_HEALTH_POLL_INTERVAL = 1.0 _SKILL_API_HEALTH_REQUEST_TIMEOUT = 5.0 @@ -459,8 +459,8 @@ def execute_skills( Supported values are "execute" (default), "run_sse", "a2a", and "python_agent". It can also be set with AGENTKIT_SKILL_INVOCATION_MODE. timeout (int, optional): Maximum execution time in seconds. Defaults to - 900. The value can be adjusted for each call but must be between 1 - and 900 seconds. + 1800. The value can be adjusted for each call but must be between 1 + and 1800 seconds. Returns: str: The output of the code execution. From caeb12d731d66c2fc4b746c706a24d2862084cba Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 21:32:58 +0800 Subject: [PATCH 3/9] refactor(agentkit-session): reuse existing session before creating new one --- tests/tools/builtin_tools/test_run_code.py | 14 +++++ veadk/tools/builtin_tools/_agentkit.py | 66 ++++++++++++++++++++-- 2 files changed, 74 insertions(+), 6 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_code.py b/tests/tools/builtin_tools/test_run_code.py index ce2711381..11970127d 100644 --- a/tests/tools/builtin_tools/test_run_code.py +++ b/tests/tools/builtin_tools/test_run_code.py @@ -1,3 +1,17 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + from types import SimpleNamespace from unittest.mock import patch diff --git a/veadk/tools/builtin_tools/_agentkit.py b/veadk/tools/builtin_tools/_agentkit.py index 80b9e5830..4fa1bf7cb 100644 --- a/veadk/tools/builtin_tools/_agentkit.py +++ b/veadk/tools/builtin_tools/_agentkit.py @@ -232,6 +232,61 @@ def invoke_agentkit_exec_bash( ) +def _get_or_create_agentkit_session( + *, + client, + tool_id: str, + tool_user_session_id: str, + ttl: int, +): + """Return the newest reusable session for ``tool_user_session_id`` or create one.""" + from agentkit.sdk.tools import types as tools_types + + try: + listing = client.list_sessions( + tools_types.ListSessionsRequest( + ToolId=tool_id, + Filters=[ + tools_types.FiltersItemForListSessions( + Name="UserSessionId", + Values=[tool_user_session_id], + ) + ], + PageSize=20, + ) + ) + except Exception as exc: # noqa: BLE001 + logger.debug(f"AgentKit ListSessions failed, falling back to create: {exc}") + listing = None + + candidates = getattr(listing, "session_infos", None) or [] + reusable = [ + info + for info in candidates + if getattr(info, "user_session_id", None) == tool_user_session_id + and (getattr(info, "status", None) or "").strip().lower() + not in _SESSION_TERMINAL_STATUSES + ] + if reusable: + reusable.sort( + key=lambda info: getattr(info, "created_at", "") or "", reverse=True + ) + chosen = reusable[0] + logger.debug( + f"Reusing AgentKit session {getattr(chosen, 'session_id', None)} " + f"for UserSessionId={tool_user_session_id}" + ) + return chosen + + return client.create_session( + tools_types.CreateSessionRequest( + ToolId=tool_id, + UserSessionId=tool_user_session_id, + Ttl=ttl, + ) + ) + + def ensure_agentkit_session_endpoint( *, tool_id: str, @@ -262,12 +317,11 @@ def ensure_agentkit_session_endpoint( region=region, session_token=session_token, ) - session = client.create_session( - tools_types.CreateSessionRequest( - ToolId=tool_id, - UserSessionId=tool_user_session_id, - Ttl=ttl, - ) + session = _get_or_create_agentkit_session( + client=client, + tool_id=tool_id, + tool_user_session_id=tool_user_session_id, + ttl=ttl, ) if not wait_until_ready: public_endpoint = getattr(session, "endpoint", None) From 9dfb1bf2432b5ee84e27bea8f6b70d2160f93bf5 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 22:02:30 +0800 Subject: [PATCH 4/9] feat(tools): use nonblocking A2A polling for execute skills --- .../builtin_tools/test_run_sandbox_agent.py | 170 +++++++++++++- veadk/tools/builtin_tools/execute_skills.py | 207 +++++++++++++++++- 2 files changed, 358 insertions(+), 19 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 08b72337f..30e9c71ea 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -213,7 +213,7 @@ def _tool_context(self): _invocation_context=invocation_context, ) - def test_prefers_new_skill_execute_api_when_endpoint_is_available(self): + def test_execute_mode_posts_skill_execute_api_when_endpoint_is_available(self): captured_requests = [] health_endpoints = [] session_kwargs = [] @@ -242,7 +242,11 @@ def fake_urlopen(request, timeout=None): ) with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills("do work", tool_context=self._tool_context()) + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="execute", + ) self.assertEqual(result, "api result") self.assertTrue(session_kwargs[0]["wait_until_ready"]) @@ -255,7 +259,7 @@ def fake_urlopen(request, timeout=None): self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) self.assertIn(b'"prompt": "do work"', request_obj.data) - def test_execute_skills_forwards_custom_timeout(self): + def test_execute_mode_forwards_custom_timeout(self): session_kwargs = [] request_timeouts = [] @@ -285,6 +289,7 @@ def read(self): result = module.execute_skills( "do work", tool_context=self._tool_context(), + invocation_mode="execute", timeout=120, ) @@ -506,7 +511,96 @@ def fake_urlopen(request, timeout=None): self.assertIn(b'"session_id": "session-1"', request_obj.data) self.assertIn(b'"text": "do work"', request_obj.data) - def test_a2a_mode_posts_message_send_and_returns_text(self): + def test_default_a2a_mode_sends_nonblocking_message_and_polls_task(self): + captured_requests = [] + responses = [ + { + "jsonrpc": "2.0", + "id": "send", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "working"}, + }, + }, + { + "jsonrpc": "2.0", + "id": "get", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "completed"}, + "artifacts": [ + { + "parts": [ + {"kind": "text", "text": "a2a "}, + { + "kind": "text", + "text": "thought", + "metadata": {"adk_thought": True}, + }, + ] + }, + {"parts": [{"kind": "text", "text": "result"}]}, + ], + }, + }, + ] + + class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return json.dumps(self.payload).encode() + + def fake_urlopen(request, timeout=None): + captured_requests.append((request, timeout)) + return FakeResponse(responses.pop(0)) + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) + + with ( + patch.object(module.request, "urlopen", fake_urlopen), + patch.object(module.time, "sleep") as sleep, + ): + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + ) + + self.assertEqual(result, "a2a \nresult") + self.assertEqual(2, len(captured_requests)) + send_request, send_timeout = captured_requests[0] + send_payload = json.loads(send_request.data.decode()) + self.assertEqual("https://sandbox.test/a2a", send_request.full_url) + self.assertEqual(60, send_timeout) + self.assertEqual("message/send", send_payload["method"]) + self.assertEqual( + "do work", send_payload["params"]["message"]["parts"][0]["text"] + ) + self.assertFalse(send_payload["params"]["configuration"]["blocking"]) + self.assertEqual(20, send_payload["params"]["configuration"]["historyLength"]) + self.assertEqual("user", send_payload["params"]["metadata"]["user_id"]) + self.assertEqual("session-1", send_payload["params"]["metadata"]["session_id"]) + + get_request, get_timeout = captured_requests[1] + get_payload = json.loads(get_request.data.decode()) + self.assertEqual("https://sandbox.test/a2a", get_request.full_url) + self.assertEqual(60, get_timeout) + self.assertEqual("tasks/get", get_payload["method"]) + self.assertEqual("task-1", get_payload["params"]["id"]) + sleep.assert_called_once() + + def test_a2a_blocking_mode_posts_blocking_message_send(self): captured_requests = [] class FakeResponse: @@ -541,7 +635,7 @@ def fake_urlopen(request, timeout=None): result = module.execute_skills( "do work", tool_context=self._tool_context(), - invocation_mode="a2a", + invocation_mode="a2a_blocking", ) self.assertEqual(result, "a2a result") @@ -553,6 +647,54 @@ def fake_urlopen(request, timeout=None): self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) self.assertTrue(payload["params"]["configuration"]["blocking"]) + def test_a2a_mode_raises_when_task_fails(self): + responses = [ + { + "jsonrpc": "2.0", + "id": "send", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "working"}, + }, + }, + { + "jsonrpc": "2.0", + "id": "get", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "failed"}, + }, + }, + ] + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return json.dumps(responses.pop(0)).encode() + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) + + with ( + patch.object( + module.request, "urlopen", lambda *_args, **_kwargs: FakeResponse() + ), + patch.object(module.time, "sleep"), + ): + with self.assertRaisesRegex( + RuntimeError, + r"A2A task task-1 ended with state failed", + ): + module.execute_skills("do work", tool_context=self._tool_context()) + def test_unsupported_invocation_mode_raises_value_error(self): module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( @@ -606,7 +748,11 @@ def fake_urlopen(_request, **_kwargs): RuntimeError, r"HTTP 404.*(?:升级|upgrade).*Skill", ): - module.execute_skills("do work", tool_context=self._tool_context()) + module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="execute", + ) def test_requires_sandbox_upgrade_when_skill_api_returns_405(self): class MethodNotAllowedResponse: @@ -634,7 +780,11 @@ def fake_urlopen(_request, **_kwargs): RuntimeError, r"HTTP 405.*(?:升级|upgrade).*Skill", ): - module.execute_skills("do work", tool_context=self._tool_context()) + module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="execute", + ) def test_raises_runtime_error_when_session_endpoint_is_unavailable(self): def raise_endpoint_error(**_kwargs): @@ -682,7 +832,11 @@ def fake_urlopen(_request, **_kwargs): with patch.object(module.request, "urlopen", fake_urlopen): with self.assertRaisesRegex(RuntimeError, "HTTP 500: internal error"): - module.execute_skills("do work", tool_context=self._tool_context()) + module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="execute", + ) if __name__ == "__main__": diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 0fd70c6e2..6c8aadfde 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -39,7 +39,20 @@ _SKILL_API_HEALTH_REQUEST_TIMEOUT = 5.0 _SKILL_INVOCATION_MODE_ENV = "AGENTKIT_SKILL_INVOCATION_MODE" _SKILL_INVOCATION_MODES = frozenset( - {"execute", "skill_api", "run_sse", "a2a", "python_agent"} + {"execute", "skill_api", "run_sse", "a2a", "a2a_blocking", "python_agent"} +) +_A2A_POLL_INTERVAL = 2.0 +_A2A_REQUEST_TIMEOUT = 60 +_A2A_HISTORY_LENGTH = 20 +_A2A_TERMINAL_STATES = frozenset( + { + "completed", + "failed", + "canceled", + "rejected", + "input-required", + "auth-required", + } ) @@ -89,16 +102,15 @@ def _skill_api_url(endpoint: str, path: str) -> str: def _resolve_skill_invocation_mode(mode: str | None = None) -> str: - # 默认保持 /v1/skills/execute；新沙箱可通过参数或环境变量显式切换后端。 - resolved = (mode or os.getenv(_SKILL_INVOCATION_MODE_ENV) or "execute").strip() + resolved = (mode or os.getenv(_SKILL_INVOCATION_MODE_ENV) or "a2a").strip() if not resolved: - return "execute" + return "a2a" normalized = resolved.lower().replace("-", "_") if normalized not in _SKILL_INVOCATION_MODES: raise ValueError( "Unsupported AgentKit Skill invocation mode " f"{resolved!r}. Expected one of: " - "execute, run_sse, a2a, python_agent." + "execute, run_sse, a2a, a2a_blocking, python_agent." ) return "execute" if normalized == "skill_api" else normalized @@ -249,6 +261,8 @@ def _extract_text_from_parts(parts: object) -> str: for part in parts: if not isinstance(part, dict): continue + if _is_adk_thought_part(part): + continue text = part.get("text") if isinstance(text, str): chunks.append(text) @@ -259,6 +273,11 @@ def _extract_text_from_parts(parts: object) -> str: return "".join(chunks) +def _is_adk_thought_part(part: dict) -> bool: + metadata = part.get("metadata") + return isinstance(metadata, dict) and metadata.get("adk_thought") is True + + def _extract_text_from_a2a_result(result: object) -> str: if not isinstance(result, dict): return "" @@ -287,6 +306,90 @@ def _extract_text_from_a2a_result(result: object) -> str: return "" +def _a2a_result_task(operation: str, response: object) -> dict: + if not isinstance(response, dict): + raise RuntimeError(f"{operation} response JSON is not an object") + if response.get("error") is not None: + raise RuntimeError(json.dumps(response["error"], ensure_ascii=False)) + result = response.get("result") + if not isinstance(result, dict): + raise RuntimeError(f"{operation} response does not contain result task") + if result.get("kind") != "task" and "status" not in result: + raise RuntimeError(f"{operation} response result is not an A2A task") + return result + + +def _a2a_task_id(task: dict) -> str: + value = task.get("id") + if not isinstance(value, str) or not value: + raise RuntimeError("A2A message/send response task does not contain id") + return value + + +def _a2a_task_state(task: dict) -> str | None: + status = task.get("status") + if not isinstance(status, dict): + return None + state = status.get("state") + return state if isinstance(state, str) else None + + +def _a2a_task_result_text(task: dict) -> str: + artifacts = task.get("artifacts") + if isinstance(artifacts, list): + chunks = [ + _extract_text_from_parts(artifact.get("parts")) + for artifact in artifacts + if isinstance(artifact, dict) + ] + text = "\n".join(chunk for chunk in chunks if chunk) + if text: + return text + + status = task.get("status") + if isinstance(status, dict): + message = status.get("message") + if isinstance(message, dict): + text = _extract_text_from_parts(message.get("parts")) + if text: + return text + + history = task.get("history") + if isinstance(history, list): + for message in reversed(history): + if isinstance(message, dict) and message.get("role") in { + "agent", + "assistant", + }: + text = _extract_text_from_parts(message.get("parts")) + if text: + return text + return "" + + +def _post_a2a_jsonrpc( + *, + endpoint: str, + payload: dict[str, object], + timeout: int, +) -> dict: + raw = _post_json(endpoint=endpoint, path="/a2a", payload=payload, timeout=timeout) + try: + response = json.loads(raw.decode("utf-8")) + except json.JSONDecodeError as exc: + raise RuntimeError("A2A JSON-RPC response is not valid JSON") from exc + if not isinstance(response, dict): + raise RuntimeError("A2A JSON-RPC response JSON is not an object") + return response + + +def _a2a_request_timeout(deadline: float) -> int: + remaining = deadline - time.monotonic() + if remaining <= 0: + return 0 + return max(1, int(min(_A2A_REQUEST_TIMEOUT, remaining))) + + def _parse_run_sse_response(raw: bytes) -> str: chunks: list[str] = [] for raw_line in raw.splitlines(): @@ -338,7 +441,82 @@ def _execute_skills_via_a2a( timeout: int, ) -> str: invocation_context = tool_context._invocation_context - # A2A 沙箱使用 JSON-RPC message/send，同步等待最终结果。 + deadline = time.monotonic() + timeout + message = { + "kind": "message", + "messageId": uuid.uuid4().hex, + "role": "user", + "parts": [{"kind": "text", "text": workflow_prompt}], + } + metadata = { + "user_id": invocation_context.user_id, + "session_id": invocation_context.session.id, + } + task = _a2a_result_task( + "A2ASendMessage", + _post_a2a_jsonrpc( + endpoint=endpoint, + payload={ + "jsonrpc": "2.0", + "id": uuid.uuid4().hex, + "method": "message/send", + "params": { + "message": message, + "metadata": metadata, + "configuration": { + "blocking": False, + "historyLength": _A2A_HISTORY_LENGTH, + }, + }, + }, + timeout=_a2a_request_timeout(deadline), + ), + ) + task_id = _a2a_task_id(task) + + while _a2a_task_state(task) not in _A2A_TERMINAL_STATES: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"Timed out while waiting for A2A task {task_id}") + time.sleep(min(_A2A_POLL_INTERVAL, remaining)) + task = _a2a_result_task( + "A2AGetTask", + _post_a2a_jsonrpc( + endpoint=endpoint, + payload={ + "jsonrpc": "2.0", + "id": uuid.uuid4().hex, + "method": "tasks/get", + "params": { + "id": task_id, + "historyLength": _A2A_HISTORY_LENGTH, + }, + }, + timeout=_a2a_request_timeout(deadline), + ), + ) + + state = _a2a_task_state(task) + if state != "completed": + raise RuntimeError( + f"A2A task {task_id} ended with state {state}: " + f"{json.dumps(task, ensure_ascii=False)}" + ) + + text = _a2a_task_result_text(task) + if text: + return text + return json.dumps(task, ensure_ascii=False) + + +def _execute_skills_via_a2a_blocking( + *, + workflow_prompt: str, + endpoint: str, + tool_context: ToolContext, + timeout: int, +) -> str: + invocation_context = tool_context._invocation_context payload = { "jsonrpc": "2.0", "id": uuid.uuid4().hex, @@ -357,9 +535,8 @@ def _execute_skills_via_a2a( "configuration": {"blocking": True}, }, } - raw = _post_json(endpoint=endpoint, path="/a2a", payload=payload, timeout=timeout) - response = json.loads(raw.decode("utf-8")) - if isinstance(response, dict) and response.get("error"): + response = _post_a2a_jsonrpc(endpoint=endpoint, payload=payload, timeout=timeout) + if response.get("error"): raise RuntimeError(json.dumps(response["error"], ensure_ascii=False)) result = response.get("result") if isinstance(response, dict) else None text = _extract_text_from_a2a_result(result) @@ -428,6 +605,13 @@ def _execute_skills_via_skill_api( tool_context=tool_context, timeout=timeout, ) + if invocation_mode == "a2a_blocking": + return _execute_skills_via_a2a_blocking( + workflow_prompt=workflow_prompt, + endpoint=endpoint, + tool_context=tool_context, + timeout=timeout, + ) _wait_for_skill_api_health(endpoint=endpoint) return _post_skill_api_json( @@ -456,8 +640,9 @@ def execute_skills( skill agent process for this execution only. Requests with custom environment variables use the legacy RunCode execution path. invocation_mode (Optional[str]): AgentKit Skill sandbox invocation backend. - Supported values are "execute" (default), "run_sse", "a2a", and - "python_agent". It can also be set with AGENTKIT_SKILL_INVOCATION_MODE. + Supported values are "a2a" (default), "execute", "run_sse", + "a2a_blocking", and "python_agent". It can also be set with + AGENTKIT_SKILL_INVOCATION_MODE. timeout (int, optional): Maximum execution time in seconds. Defaults to 1800. The value can be adjusted for each call but must be between 1 and 1800 seconds. From 209624ab63fb08e3b02e10d5ecbeb17f98f7fb80 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 22:21:08 +0800 Subject: [PATCH 5/9] refactor: simplify execute_skills to A2A polling --- .../builtin_tools/test_run_sandbox_agent.py | 449 ++---------------- veadk/tools/builtin_tools/execute_skills.py | 350 +------------- 2 files changed, 53 insertions(+), 746 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 30e9c71ea..645e851d8 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -81,7 +81,6 @@ def _load_execute_skills_module( *, ensure_agentkit_session_endpoint=lambda **_kwargs: "", run_sandbox_agent=lambda **_kwargs: "", - wait_for_skill_api_health=lambda **_kwargs: None, ): module_path = ( Path(__file__).resolve().parents[3] @@ -140,8 +139,6 @@ def _load_execute_skills_module( assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - if wait_for_skill_api_health is not None: - module._wait_for_skill_api_health = wait_for_skill_api_health return module @@ -213,108 +210,6 @@ def _tool_context(self): _invocation_context=invocation_context, ) - def test_execute_mode_posts_skill_execute_api_when_endpoint_is_available(self): - captured_requests = [] - health_endpoints = [] - session_kwargs = [] - - class FakeResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - def read(self): - return b'{"content": "api result"}' - - def fake_urlopen(request, timeout=None): - captured_requests.append((request, timeout)) - return FakeResponse() - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **kwargs: ( - session_kwargs.append(kwargs) or "https://sandbox.test" - ), - wait_for_skill_api_health=lambda **kwargs: health_endpoints.append( - kwargs["endpoint"] - ), - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - ) - - self.assertEqual(result, "api result") - self.assertTrue(session_kwargs[0]["wait_until_ready"]) - self.assertEqual(["https://sandbox.test"], health_endpoints) - self.assertEqual(1, len(captured_requests)) - request_obj, timeout = captured_requests[0] - self.assertEqual("https://sandbox.test/v1/skills/execute", request_obj.full_url) - self.assertEqual(1800, timeout) - self.assertEqual("POST", request_obj.get_method()) - self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) - self.assertIn(b'"prompt": "do work"', request_obj.data) - - def test_execute_mode_forwards_custom_timeout(self): - session_kwargs = [] - request_timeouts = [] - - class FakeResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - def read(self): - return b'{"content": "api result"}' - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **kwargs: ( - session_kwargs.append(kwargs) or "https://sandbox.test" - ), - ) - - with patch.object( - module.request, - "urlopen", - lambda _request, timeout=None: ( - request_timeouts.append(timeout) or FakeResponse() - ), - ): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - timeout=120, - ) - - self.assertEqual("api result", result) - self.assertEqual(1800, session_kwargs[0]["ttl"]) - self.assertEqual([120], request_timeouts) - - def test_execute_skills_forwards_custom_timeout_to_legacy_path(self): - captured_kwargs = {} - module = _load_execute_skills_module( - run_sandbox_agent=lambda **kwargs: ( - captured_kwargs.update(kwargs) or "legacy result" - ), - ) - - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - env_vars={"CUSTOM_VALUE": "custom"}, - timeout=120, - ) - - self.assertEqual("legacy result", result) - self.assertEqual(120, captured_kwargs["timeout"]) - def test_execute_skills_rejects_invalid_timeout(self): module = _load_execute_skills_module() @@ -330,187 +225,37 @@ def test_execute_skills_rejects_invalid_timeout(self): timeout=timeout, ) - def test_health_check_retries_502_until_upstream_is_ready(self): - attempts = [] - - class HealthyResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - class ErrorResponse: - def read(self): - return b"bad gateway" - - def close(self): - return None - - module = _load_execute_skills_module(wait_for_skill_api_health=None) - - def fake_urlopen(req, **_kwargs): - attempts.append((req.full_url, req.get_method())) - if len(attempts) == 1: - raise module.error.HTTPError( - url=req.full_url, - code=502, - msg="Bad Gateway", - hdrs={}, - fp=ErrorResponse(), - ) - return HealthyResponse() - - with ( - patch.object(module.request, "urlopen", fake_urlopen), - patch.object(module.time, "sleep") as sleep, - ): - module._wait_for_skill_api_health(endpoint="https://sandbox.test") - - self.assertEqual( - [ - ("https://sandbox.test/v1/skills/healthz", "GET"), - ("https://sandbox.test/v1/skills/healthz", "GET"), - ], - attempts, - ) - sleep.assert_called_once_with(1.0) - - def test_health_check_allows_images_without_health_endpoint(self): - class NotFoundResponse: - def read(self): - return b"not found" - - def close(self): - return None - - module = _load_execute_skills_module(wait_for_skill_api_health=None) - - def fake_urlopen(req, **_kwargs): - raise module.error.HTTPError( - url=req.full_url, - code=404, - msg="Not Found", - hdrs={}, - fp=NotFoundResponse(), - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - module._wait_for_skill_api_health(endpoint="https://sandbox.test") - - def test_env_vars_use_legacy_runcode_execution(self): - captured_kwargs = {} - - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Skill API must not be used when env_vars are provided" - ), - run_sandbox_agent=fake_run_sandbox_agent, - ) - - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - env_vars={"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, - ) - - self.assertEqual(result, "legacy result") - self.assertEqual( - {"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, - captured_kwargs["extra_env_vars"], - ) - - def test_python_agent_mode_uses_legacy_runcode_execution(self): - captured_kwargs = {} - - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Session endpoint must not be used in python_agent mode" - ), - run_sandbox_agent=fake_run_sandbox_agent, - ) - - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="python_agent", - ) - - self.assertEqual(result, "legacy result") - self.assertEqual("do work", captured_kwargs["workflow_prompt"]) - self.assertEqual("test-tool", captured_kwargs["tool_id"]) - - def test_invocation_mode_can_be_read_from_environment(self): - captured_kwargs = {} - - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" - + def test_env_vars_are_not_supported(self): module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Session endpoint must not be used in python_agent mode" - ), - run_sandbox_agent=fake_run_sandbox_agent, + "env_vars must be rejected before endpoint resolution" + ) ) - with patch.dict( - module.os.environ, - {"AGENTKIT_SKILL_INVOCATION_MODE": "python_agent"}, - ): - result = module.execute_skills("do work", tool_context=self._tool_context()) - - self.assertEqual(result, "legacy result") - self.assertEqual("do work", captured_kwargs["workflow_prompt"]) - - def test_run_sse_mode_posts_run_sse_and_aggregates_event_text(self): - captured_requests = [] - - class FakeResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - def read(self): - return ( - b'data: {"content":{"parts":[{"text":"hello "}]}}\n\n' - b'data: {"content":{"parts":[{"text":"world"}]}}\n\n' - b"data: [DONE]\n\n" - ) + with self.assertRaisesRegex(ValueError, "env_vars is not supported"): + module.execute_skills( + "do work", + tool_context=self._tool_context(), + env_vars={"CUSTOM_VALUE": "custom"}, + ) - def fake_urlopen(request, timeout=None): - captured_requests.append((request, timeout)) - return FakeResponse() + with self.assertRaisesRegex(ValueError, "env_vars is not supported"): + module.execute_skills( + "do work", + tool_context=self._tool_context(), + env_vars={}, + ) - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) + def test_execute_skills_does_not_accept_invocation_mode(self): + module = _load_execute_skills_module() - with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills( + with self.assertRaises(TypeError): + module.execute_skills( "do work", tool_context=self._tool_context(), - invocation_mode="run_sse", + invocation_mode="execute", ) - self.assertEqual(result, "hello world") - request_obj, timeout = captured_requests[0] - self.assertEqual("https://sandbox.test/run_sse", request_obj.full_url) - self.assertEqual(1800, timeout) - self.assertIn(b'"app_name": "agent"', request_obj.data) - self.assertIn(b'"session_id": "session-1"', request_obj.data) - self.assertIn(b'"text": "do work"', request_obj.data) - def test_default_a2a_mode_sends_nonblocking_message_and_polls_task(self): captured_requests = [] responses = [ @@ -565,7 +310,9 @@ def fake_urlopen(request, timeout=None): return FakeResponse(responses.pop(0)) module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ensure_agentkit_session_endpoint=lambda **kwargs: ( + self.assertEqual(1800, kwargs["ttl"]) or "https://sandbox.test" + ), ) with ( @@ -600,8 +347,8 @@ def fake_urlopen(request, timeout=None): self.assertEqual("task-1", get_payload["params"]["id"]) sleep.assert_called_once() - def test_a2a_blocking_mode_posts_blocking_message_send(self): - captured_requests = [] + def test_a2a_forwards_custom_timeout_to_requests(self): + captured_timeouts = [] class FakeResponse: def __enter__(self): @@ -614,38 +361,41 @@ def read(self): return json.dumps( { "jsonrpc": "2.0", - "id": "req", + "id": "send", "result": { - "kind": "message", - "role": "agent", - "parts": [{"kind": "text", "text": "a2a result"}], + "kind": "task", + "id": "task-1", + "status": { + "state": "completed", + "message": { + "parts": [ + {"kind": "text", "text": "custom timeout"} + ] + }, + }, }, } ).encode() - def fake_urlopen(request, timeout=None): - captured_requests.append((request, timeout)) + def fake_urlopen(_request, timeout=None): + captured_timeouts.append(timeout) return FakeResponse() module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ensure_agentkit_session_endpoint=lambda **kwargs: ( + self.assertEqual(1800, kwargs["ttl"]) or "https://sandbox.test" + ), ) with patch.object(module.request, "urlopen", fake_urlopen): result = module.execute_skills( "do work", tool_context=self._tool_context(), - invocation_mode="a2a_blocking", + timeout=120, ) - self.assertEqual(result, "a2a result") - request_obj, timeout = captured_requests[0] - payload = json.loads(request_obj.data.decode()) - self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) - self.assertEqual(1800, timeout) - self.assertEqual("message/send", payload["method"]) - self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) - self.assertTrue(payload["params"]["configuration"]["blocking"]) + self.assertEqual("custom timeout", result) + self.assertEqual([60], captured_timeouts) def test_a2a_mode_raises_when_task_fails(self): responses = [ @@ -695,97 +445,19 @@ def read(self): ): module.execute_skills("do work", tool_context=self._tool_context()) - def test_unsupported_invocation_mode_raises_value_error(self): - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Invalid mode must be rejected before endpoint resolution" - ), - ) - - with self.assertRaisesRegex(ValueError, "Unsupported AgentKit Skill"): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="stream", - ) - def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: "", ) self.assertEqual( - "https://sandbox.test/v1/skills/execute?faasInstanceName=inst&Authorization=key", + "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", module._skill_api_url( "https://sandbox.test/?faasInstanceName=inst&Authorization=key", - "/v1/skills/execute", + "/a2a", ), ) - def test_requires_sandbox_upgrade_when_skill_api_returns_404(self): - class NotFoundResponse: - def read(self): - return b"not found" - - def close(self): - return None - - def fake_urlopen(_request, **_kwargs): - raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", - code=404, - msg="Not Found", - hdrs={}, - fp=NotFoundResponse(), - ) - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - with self.assertRaisesRegex( - RuntimeError, - r"HTTP 404.*(?:升级|upgrade).*Skill", - ): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - ) - - def test_requires_sandbox_upgrade_when_skill_api_returns_405(self): - class MethodNotAllowedResponse: - def read(self): - return b"method not allowed" - - def close(self): - return None - - def fake_urlopen(_request, **_kwargs): - raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", - code=405, - msg="Method Not Allowed", - hdrs={}, - fp=MethodNotAllowedResponse(), - ) - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - with self.assertRaisesRegex( - RuntimeError, - r"HTTP 405.*(?:升级|upgrade).*Skill", - ): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - ) - def test_raises_runtime_error_when_session_endpoint_is_unavailable(self): def raise_endpoint_error(**_kwargs): raise RuntimeError("session unsupported") @@ -809,35 +481,6 @@ def test_missing_tool_context_raises_value_error(self): with self.assertRaisesRegex(ValueError, r"tool_context is required"): module.execute_skills("do work", tool_context=None) - def test_non_compatibility_skill_api_http_error_is_not_swallowed(self): - class ServerErrorResponse: - def read(self): - return b"internal error" - - def close(self): - return None - - def fake_urlopen(_request, **_kwargs): - raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", - code=500, - msg="Internal Server Error", - hdrs={}, - fp=ServerErrorResponse(), - ) - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - with self.assertRaisesRegex(RuntimeError, "HTTP 500: internal error"): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - ) - if __name__ == "__main__": unittest.main() diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 6c8aadfde..11a009745 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -15,7 +15,6 @@ from __future__ import annotations import json -import os import time import uuid from typing import Optional @@ -26,21 +25,10 @@ from veadk.tools.builtin_tools._agentkit import ( ensure_agentkit_session_endpoint, - get_agentkit_account_id, resolve_agentkit_tool_id, ) -from veadk.tools.builtin_tools.run_sandbox_agent import run_sandbox_agent -_SKILL_API_UPGRADE_STATUS_CODES = frozenset({404, 405}) -_SKILL_API_TRANSIENT_STATUS_CODES = frozenset({502, 503, 504}) _SKILL_API_TIMEOUT = 1800 -_SKILL_API_HEALTH_TIMEOUT = 30.0 -_SKILL_API_HEALTH_POLL_INTERVAL = 1.0 -_SKILL_API_HEALTH_REQUEST_TIMEOUT = 5.0 -_SKILL_INVOCATION_MODE_ENV = "AGENTKIT_SKILL_INVOCATION_MODE" -_SKILL_INVOCATION_MODES = frozenset( - {"execute", "skill_api", "run_sse", "a2a", "a2a_blocking", "python_agent"} -) _A2A_POLL_INTERVAL = 2.0 _A2A_REQUEST_TIMEOUT = 60 _A2A_HISTORY_LENGTH = 20 @@ -63,14 +51,6 @@ def _validate_timeout(timeout: int) -> None: ) -def _skill_api_upgrade_hint() -> str: - return ( - "提示：当前 Skill 沙箱镜像未实现 /v1/skills/execute 接口，" - "可能是旧版沙箱镜像。" - "请升级 Skill 沙箱镜像或切换到支持 Skill HTTP API 的新版沙箱。" - ) - - def _tool_user_session_id(tool_context: ToolContext) -> str: invocation_context = tool_context._invocation_context session_id = invocation_context.session.id @@ -79,16 +59,6 @@ def _tool_user_session_id(tool_context: ToolContext) -> str: return agent_name + "_" + user_id + "_" + session_id -def _tip_token_key(tool_context: ToolContext) -> str | None: - state = tool_context.state or {} - return ( - state.get("TIP_TOKEN_KEY") - or state.get("tip_token_key") - or os.getenv("TIP_TOKEN_KEY") - or None - ) - - def _skill_api_url(endpoint: str, path: str) -> str: if not endpoint: raise RuntimeError("AgentKit session endpoint is empty") @@ -101,59 +71,6 @@ def _skill_api_url(endpoint: str, path: str) -> str: ) -def _resolve_skill_invocation_mode(mode: str | None = None) -> str: - resolved = (mode or os.getenv(_SKILL_INVOCATION_MODE_ENV) or "a2a").strip() - if not resolved: - return "a2a" - normalized = resolved.lower().replace("-", "_") - if normalized not in _SKILL_INVOCATION_MODES: - raise ValueError( - "Unsupported AgentKit Skill invocation mode " - f"{resolved!r}. Expected one of: " - "execute, run_sse, a2a, a2a_blocking, python_agent." - ) - return "execute" if normalized == "skill_api" else normalized - - -def _post_skill_api_json( - *, - endpoint: str, - path: str, - payload: dict[str, object], - tip_token_key: str | None, - timeout: int, -) -> str: - headers = { - "Content-Type": "application/json", - "Accept": "application/json", - } - if tip_token_key: - headers["X-Tip-Token-Key"] = tip_token_key - - req = request.Request( - _skill_api_url(endpoint, path), - data=json.dumps(payload).encode("utf-8"), - headers=headers, - method="POST", - ) - try: - with request.urlopen(req, timeout=timeout) as response: - return _parse_skill_execute_response(response.read()) - except error.HTTPError as exc: - if exc.code in _SKILL_API_UPGRADE_STATUS_CODES: - raise RuntimeError( - f"Skill HTTP API returned HTTP {exc.code}. {_skill_api_upgrade_hint()}" - ) from exc - detail = exc.read().decode("utf-8", errors="replace") - raise RuntimeError( - f"Skill HTTP API request failed with HTTP {exc.code}: {detail}" - ) from exc - except error.URLError as exc: - raise RuntimeError( - f"Skill HTTP API endpoint is not reachable: {exc.reason}" - ) from exc - - def _post_json( *, endpoint: str, @@ -182,78 +99,6 @@ def _post_json( ) from exc -def _wait_for_skill_api_health( - *, - endpoint: str, - timeout: float = _SKILL_API_HEALTH_TIMEOUT, - poll_interval: float = _SKILL_API_HEALTH_POLL_INTERVAL, -) -> None: - """Wait until the Skill API upstream is reachable through the session endpoint.""" - deadline = time.monotonic() + timeout - last_error = "unknown error" - while True: - req = request.Request( - _skill_api_url(endpoint, "/v1/skills/healthz"), - headers={"Accept": "application/json"}, - method="GET", - ) - try: - remaining = max(0.001, deadline - time.monotonic()) - with request.urlopen( - req, - timeout=min(_SKILL_API_HEALTH_REQUEST_TIMEOUT, remaining), - ): - return - except error.HTTPError as exc: - if exc.code in _SKILL_API_UPGRADE_STATUS_CODES: - # Some compatible images predate the dedicated health endpoint. - return - if exc.code not in _SKILL_API_TRANSIENT_STATUS_CODES: - detail = exc.read().decode("utf-8", errors="replace") - raise RuntimeError( - f"Skill HTTP API health check failed with HTTP {exc.code}: {detail}" - ) from exc - last_error = f"HTTP {exc.code}" - except error.URLError as exc: - last_error = str(exc.reason) - - remaining = deadline - time.monotonic() - if remaining <= 0: - raise RuntimeError( - f"Timed out waiting for Skill HTTP API health check: {last_error}" - ) - time.sleep(min(poll_interval, remaining)) - - -def _parse_skill_execute_response(raw: bytes) -> str: - try: - payload = json.loads(raw.decode("utf-8")) - except json.JSONDecodeError: - return raw.decode("utf-8", errors="replace") - - if isinstance(payload, dict): - if isinstance(payload.get("content"), str): - return payload["content"] - data = payload.get("data") - if isinstance(data, dict) and isinstance(data.get("content"), str): - return data["content"] - return json.dumps(payload, ensure_ascii=False) - - -def _run_request_payload(workflow_prompt: str, tool_context: ToolContext) -> dict: - invocation_context = tool_context._invocation_context - return { - "app_name": invocation_context.agent.name, - "user_id": invocation_context.user_id, - "session_id": invocation_context.session.id, - "new_message": { - "role": "user", - "parts": [{"text": workflow_prompt}], - }, - "streaming": True, - } - - def _extract_text_from_parts(parts: object) -> str: if not isinstance(parts, list): return "" @@ -278,34 +123,6 @@ def _is_adk_thought_part(part: dict) -> bool: return isinstance(metadata, dict) and metadata.get("adk_thought") is True -def _extract_text_from_a2a_result(result: object) -> str: - if not isinstance(result, dict): - return "" - if result.get("kind") == "message": - return _extract_text_from_parts(result.get("parts")) - artifacts = result.get("artifacts") - if isinstance(artifacts, list): - chunks = [ - _extract_text_from_parts(artifact.get("parts")) - for artifact in artifacts - if isinstance(artifact, dict) - ] - text = "".join(chunks) - if text: - return text - history = result.get("history") - if isinstance(history, list): - for message in reversed(history): - if isinstance(message, dict) and message.get("role") in { - "agent", - "assistant", - }: - text = _extract_text_from_parts(message.get("parts")) - if text: - return text - return "" - - def _a2a_result_task(operation: str, response: object) -> dict: if not isinstance(response, dict): raise RuntimeError(f"{operation} response JSON is not an object") @@ -390,49 +207,6 @@ def _a2a_request_timeout(deadline: float) -> int: return max(1, int(min(_A2A_REQUEST_TIMEOUT, remaining))) -def _parse_run_sse_response(raw: bytes) -> str: - chunks: list[str] = [] - for raw_line in raw.splitlines(): - line = raw_line.decode("utf-8", errors="replace") - if not line.startswith("data:"): - continue - data = line[len("data:") :].strip() - if not data or data == "[DONE]": - continue - try: - event = json.loads(data) - except json.JSONDecodeError: - continue - if isinstance(event, dict) and isinstance(event.get("error"), str): - raise RuntimeError(event["error"]) - if not isinstance(event, dict): - continue - content = event.get("content") - if isinstance(content, dict): - text = _extract_text_from_parts(content.get("parts")) - if text: - chunks.append(text) - return "".join(chunks) - - -def _execute_skills_via_run_sse( - *, - workflow_prompt: str, - endpoint: str, - tool_context: ToolContext, - timeout: int, -) -> str: - # run_sse 复用 ADK 运行入口，适配只暴露 ADK Runtime 接口的 Skill 沙箱。 - raw = _post_json( - endpoint=endpoint, - path="/run_sse", - payload=_run_request_payload(workflow_prompt, tool_context), - timeout=timeout, - accept="text/event-stream", - ) - return _parse_run_sse_response(raw) - - def _execute_skills_via_a2a( *, workflow_prompt: str, @@ -509,74 +283,12 @@ def _execute_skills_via_a2a( return json.dumps(task, ensure_ascii=False) -def _execute_skills_via_a2a_blocking( - *, - workflow_prompt: str, - endpoint: str, - tool_context: ToolContext, - timeout: int, -) -> str: - invocation_context = tool_context._invocation_context - payload = { - "jsonrpc": "2.0", - "id": uuid.uuid4().hex, - "method": "message/send", - "params": { - "message": { - "kind": "message", - "messageId": uuid.uuid4().hex, - "role": "user", - "parts": [{"kind": "text", "text": workflow_prompt}], - }, - "metadata": { - "user_id": invocation_context.user_id, - "session_id": invocation_context.session.id, - }, - "configuration": {"blocking": True}, - }, - } - response = _post_a2a_jsonrpc(endpoint=endpoint, payload=payload, timeout=timeout) - if response.get("error"): - raise RuntimeError(json.dumps(response["error"], ensure_ascii=False)) - result = response.get("result") if isinstance(response, dict) else None - text = _extract_text_from_a2a_result(result) - if text: - return text - return json.dumps(result, ensure_ascii=False) - - -def _execute_skills_via_python_agent( - *, - workflow_prompt: str, - tool_id: str, - tool_context: ToolContext, - timeout: int, - env_vars: Optional[dict[str, str]] = None, -) -> str: - # python_agent 是旧版 RunCode 路径，本质是在沙箱内执行 python agent.py。 - account_id = get_agentkit_account_id(tool_context.state) - extra_env_vars = dict(env_vars or {}) - if account_id: - extra_env_vars.setdefault( - "TOS_SKILLS_DIR", - f"tos://agentkit-platform-{account_id}/skills/", - ) - return run_sandbox_agent( - workflow_prompt=workflow_prompt, - tool_id=tool_id, - tool_context=tool_context, - timeout=timeout, - extra_env_vars=extra_env_vars, - ) - - def _execute_skills_via_skill_api( *, workflow_prompt: str, tool_id: str, tool_context: ToolContext, timeout: int, - invocation_mode: str, ) -> str: try: endpoint = ensure_agentkit_session_endpoint( @@ -591,34 +303,10 @@ def _execute_skills_via_skill_api( f"AgentKit session endpoint is not available: {exc}" ) from exc - if invocation_mode == "run_sse": - return _execute_skills_via_run_sse( - workflow_prompt=workflow_prompt, - endpoint=endpoint, - tool_context=tool_context, - timeout=timeout, - ) - if invocation_mode == "a2a": - return _execute_skills_via_a2a( - workflow_prompt=workflow_prompt, - endpoint=endpoint, - tool_context=tool_context, - timeout=timeout, - ) - if invocation_mode == "a2a_blocking": - return _execute_skills_via_a2a_blocking( - workflow_prompt=workflow_prompt, - endpoint=endpoint, - tool_context=tool_context, - timeout=timeout, - ) - - _wait_for_skill_api_health(endpoint=endpoint) - return _post_skill_api_json( + return _execute_skills_via_a2a( + workflow_prompt=workflow_prompt, endpoint=endpoint, - path="/v1/skills/execute", - payload={"prompt": workflow_prompt}, - tip_token_key=_tip_token_key(tool_context), + tool_context=tool_context, timeout=timeout, ) @@ -627,7 +315,6 @@ def execute_skills( workflow_prompt: str, tool_context: ToolContext = None, env_vars: Optional[dict[str, str]] = None, - invocation_mode: Optional[str] = None, timeout: int = _SKILL_API_TIMEOUT, ) -> str: """Execute skills in a sandbox and return the output. @@ -636,13 +323,8 @@ def execute_skills( Args: workflow_prompt (str): instruction of workflow - env_vars (Optional[dict[str, str]]): Environment variables passed to the - skill agent process for this execution only. Requests with custom - environment variables use the legacy RunCode execution path. - invocation_mode (Optional[str]): AgentKit Skill sandbox invocation backend. - Supported values are "a2a" (default), "execute", "run_sse", - "a2a_blocking", and "python_agent". It can also be set with - AGENTKIT_SKILL_INVOCATION_MODE. + env_vars (Optional[dict[str, str]]): Unsupported. AgentKit Skill execution + uses A2A and does not support per-call process environment injection. timeout (int, optional): Maximum execution time in seconds. Defaults to 1800. The value can be adjusted for each call but must be between 1 and 1800 seconds. @@ -653,31 +335,13 @@ def execute_skills( if tool_context is None: raise ValueError("tool_context is required for execute_skills") _validate_timeout(timeout) + if env_vars is not None: + raise ValueError("env_vars is not supported for execute_skills A2A execution") tool_id = resolve_agentkit_tool_id("AGENTKIT_TOOL_ID_SKILLS") - if env_vars: - # env_vars 依赖进程级环境变量注入，只能走 legacy python agent.py 路径。 - return _execute_skills_via_python_agent( - workflow_prompt=workflow_prompt, - tool_id=tool_id, - tool_context=tool_context, - timeout=timeout, - env_vars=env_vars, - ) - - mode = _resolve_skill_invocation_mode(invocation_mode) - if mode == "python_agent": - return _execute_skills_via_python_agent( - workflow_prompt=workflow_prompt, - tool_id=tool_id, - tool_context=tool_context, - timeout=timeout, - ) - return _execute_skills_via_skill_api( workflow_prompt=workflow_prompt, tool_id=tool_id, tool_context=tool_context, timeout=timeout, - invocation_mode=mode, ) From 0083777db5037f06eeb4927ce363c78abd6da4a8 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 23:09:11 +0800 Subject: [PATCH 6/9] fix(execute-skills): retry transient a2a gateway failures --- .../builtin_tools/test_run_sandbox_agent.py | 164 ++++++++++++++++++ veadk/tools/builtin_tools/execute_skills.py | 54 +++++- 2 files changed, 217 insertions(+), 1 deletion(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 645e851d8..5fc11699d 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -13,6 +13,7 @@ # limitations under the License. import importlib.util +import io import json import sys import types @@ -397,6 +398,125 @@ def fake_urlopen(_request, timeout=None): self.assertEqual("custom timeout", result) self.assertEqual([60], captured_timeouts) + def test_a2a_posts_to_vefaas_a2a_endpoint_with_query_auth(self): + captured_requests = [] + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return json.dumps( + { + "jsonrpc": "2.0", + "id": "test-1", + "result": { + "kind": "task", + "id": "task-1", + "status": { + "state": "completed", + "message": { + "parts": [{"kind": "text", "text": "hello result"}] + }, + }, + }, + } + ).encode() + + def fake_urlopen(request, timeout=None): + captured_requests.append((request, timeout)) + return FakeResponse() + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: ( + "https://sandbox.test/?faasInstanceName=inst&Authorization=key" + ), + ) + + with patch.object(module.request, "urlopen", fake_urlopen): + result = module.execute_skills("hello", tool_context=self._tool_context()) + + self.assertEqual("hello result", result) + self.assertEqual(1, len(captured_requests)) + send_request, send_timeout = captured_requests[0] + send_payload = json.loads(send_request.data.decode()) + self.assertEqual( + "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", + send_request.full_url, + ) + self.assertEqual("application/json", send_request.get_header("Content-type")) + self.assertIsNone(send_request.get_header("Accept")) + self.assertEqual(60, send_timeout) + self.assertEqual("2.0", send_payload["jsonrpc"]) + self.assertEqual("message/send", send_payload["method"]) + self.assertEqual("message", send_payload["params"]["message"]["kind"]) + self.assertEqual("user", send_payload["params"]["message"]["role"]) + self.assertEqual( + [{"kind": "text", "text": "hello"}], + send_payload["params"]["message"]["parts"], + ) + self.assertFalse(send_payload["params"]["configuration"]["blocking"]) + self.assertEqual(20, send_payload["params"]["configuration"]["historyLength"]) + self.assertEqual("session-1", send_payload["params"]["metadata"]["session_id"]) + self.assertEqual("user", send_payload["params"]["metadata"]["user_id"]) + + def test_retry_502(self): + captured_timeouts = [] + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return json.dumps( + { + "jsonrpc": "2.0", + "id": "send", + "result": { + "kind": "task", + "id": "task-1", + "status": { + "state": "completed", + "message": { + "parts": [{"kind": "text", "text": "retry ok"}] + }, + }, + }, + } + ).encode() + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) + + def fake_urlopen(request, timeout=None): + captured_timeouts.append(timeout) + if len(captured_timeouts) == 1: + raise module.error.HTTPError( + request.full_url, + 502, + "Bad Gateway", + hdrs=None, + fp=io.BytesIO(b"temporary gateway error"), + ) + return FakeResponse() + + with ( + patch.object(module.request, "urlopen", fake_urlopen), + patch.object(module.time, "sleep") as sleep, + ): + result = module.execute_skills("hello", tool_context=self._tool_context()) + + self.assertEqual("retry ok", result) + self.assertEqual([60, 60], captured_timeouts) + sleep.assert_called_once_with(2.0) + def test_a2a_mode_raises_when_task_fails(self): responses = [ { @@ -458,6 +578,50 @@ def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): ), ) + def test_a2a_jsonrpc_url_appends_a2a_before_vefaas_query(self): + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "", + ) + + self.assertEqual( + "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", + module._a2a_jsonrpc_url( + "https://sandbox.test/?faasInstanceName=inst&Authorization=key" + ), + ) + self.assertEqual( + "https://sc56tro0thc3nstnfkabv.apigateway-cn-beijing.volceapi.com/a2a" + "?faasInstanceName=vefaas-example-sandbox&Authorization=test-token", + module._a2a_jsonrpc_url( + "https://sc56tro0thc3nstnfkabv.apigateway-cn-beijing.volceapi.com" + "?faasInstanceName=vefaas-example-sandbox&Authorization=test-token" + ), + ) + + def test_a2a_jsonrpc_url_keeps_explicit_a2a_endpoint(self): + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "", + ) + + endpoint = "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key" + self.assertEqual(endpoint, module._a2a_jsonrpc_url(endpoint)) + self.assertEqual( + endpoint, + module._a2a_jsonrpc_url( + "https://sandbox.test/a2a/?faasInstanceName=inst&Authorization=key" + ), + ) + + def test_a2a_jsonrpc_url_appends_a2a_for_plain_endpoint(self): + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "", + ) + + self.assertEqual( + "https://sandbox.test/a2a", + module._a2a_jsonrpc_url("https://sandbox.test"), + ) + def test_raises_runtime_error_when_session_endpoint_is_unavailable(self): def raise_endpoint_error(**_kwargs): raise RuntimeError("session unsupported") diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 11a009745..c0dd595b5 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -32,6 +32,7 @@ _A2A_POLL_INTERVAL = 2.0 _A2A_REQUEST_TIMEOUT = 60 _A2A_HISTORY_LENGTH = 20 +_A2A_RETRY_STATUS_CODES = frozenset({502, 503, 504}) _A2A_TERMINAL_STATES = frozenset( { "completed", @@ -71,6 +72,24 @@ def _skill_api_url(endpoint: str, path: str) -> str: ) +def _a2a_jsonrpc_url(endpoint: str) -> str: + if not endpoint: + raise RuntimeError("AgentKit session endpoint is empty") + parts = urlsplit(endpoint) + normalized_path = parts.path.rstrip("/") + if normalized_path.endswith("/a2a"): + return urlunsplit( + ( + parts.scheme, + parts.netloc, + normalized_path, + parts.query, + parts.fragment, + ) + ) + return _skill_api_url(endpoint, "/a2a") + + def _post_json( *, endpoint: str, @@ -189,8 +208,39 @@ def _post_a2a_jsonrpc( endpoint: str, payload: dict[str, object], timeout: int, + retry_until: float | None = None, ) -> dict: - raw = _post_json(endpoint=endpoint, path="/a2a", payload=payload, timeout=timeout) + url = _a2a_jsonrpc_url(endpoint) + while True: + request_timeout = ( + _a2a_request_timeout(retry_until) if retry_until is not None else timeout + ) + if request_timeout <= 0: + raise TimeoutError("Timed out while waiting for A2A endpoint") + req = request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with request.urlopen(req, timeout=request_timeout) as response: + raw = response.read() + break + except error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + if exc.code in _A2A_RETRY_STATUS_CODES and retry_until is not None: + remaining = retry_until - time.monotonic() + if remaining > 0: + time.sleep(min(_A2A_POLL_INTERVAL, remaining)) + continue + raise RuntimeError( + f"AgentKit Skill /a2a request failed with HTTP {exc.code}: {detail}" + ) from exc + except error.URLError as exc: + raise RuntimeError( + f"AgentKit Skill /a2a endpoint is not reachable: {exc.reason}" + ) from exc try: response = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as exc: @@ -244,6 +294,7 @@ def _execute_skills_via_a2a( }, }, timeout=_a2a_request_timeout(deadline), + retry_until=deadline, ), ) task_id = _a2a_task_id(task) @@ -267,6 +318,7 @@ def _execute_skills_via_a2a( }, }, timeout=_a2a_request_timeout(deadline), + retry_until=deadline, ), ) From 507673b57d378d0f08cfea8c770442a495cdbf14 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Tue, 11 Aug 2026 23:48:49 +0800 Subject: [PATCH 7/9] perf(skills): add capped exponential backoff for A2A polling --- .../builtin_tools/test_run_sandbox_agent.py | 39 ++++++++++++++++++- veadk/tools/builtin_tools/execute_skills.py | 5 ++- 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 5fc11699d..ef314c2f6 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -19,7 +19,7 @@ import types import unittest from pathlib import Path -from unittest.mock import patch +from unittest.mock import call, patch def _load_run_sandbox_agent_module(): @@ -346,7 +346,42 @@ def fake_urlopen(request, timeout=None): self.assertEqual(60, get_timeout) self.assertEqual("tasks/get", get_payload["method"]) self.assertEqual("task-1", get_payload["params"]["id"]) - sleep.assert_called_once() + sleep.assert_called_once_with(2.0) + + def test_a2a_task_polling_uses_capped_exponential_backoff(self): + states = ["working", "working", "working", "working", "working", "completed"] + + def fake_post_a2a_jsonrpc(**_kwargs): + state = states.pop(0) + return { + "jsonrpc": "2.0", + "result": { + "kind": "task", + "id": "task-1", + "status": { + "state": state, + "message": {"parts": [{"kind": "text", "text": "done"}]}, + }, + }, + } + + module = _load_execute_skills_module() + with ( + patch.object(module, "_post_a2a_jsonrpc", fake_post_a2a_jsonrpc), + patch.object(module.time, "sleep") as sleep, + ): + result = module._execute_skills_via_a2a( + workflow_prompt="do work", + endpoint="https://sandbox.test", + tool_context=self._tool_context(), + timeout=1800, + ) + + self.assertEqual("done", result) + self.assertEqual( + [call(2.0), call(4.0), call(8.0), call(16.0), call(16.0)], + sleep.call_args_list, + ) def test_a2a_forwards_custom_timeout_to_requests(self): captured_timeouts = [] diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index c0dd595b5..7c56e544a 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -30,6 +30,7 @@ _SKILL_API_TIMEOUT = 1800 _A2A_POLL_INTERVAL = 2.0 +_A2A_MAX_POLL_INTERVAL = 16.0 _A2A_REQUEST_TIMEOUT = 60 _A2A_HISTORY_LENGTH = 20 _A2A_RETRY_STATUS_CODES = frozenset({502, 503, 504}) @@ -298,12 +299,13 @@ def _execute_skills_via_a2a( ), ) task_id = _a2a_task_id(task) + poll_interval = _A2A_POLL_INTERVAL while _a2a_task_state(task) not in _A2A_TERMINAL_STATES: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError(f"Timed out while waiting for A2A task {task_id}") - time.sleep(min(_A2A_POLL_INTERVAL, remaining)) + time.sleep(min(poll_interval, remaining)) task = _a2a_result_task( "A2AGetTask", _post_a2a_jsonrpc( @@ -321,6 +323,7 @@ def _execute_skills_via_a2a( retry_until=deadline, ), ) + poll_interval = min(poll_interval * 2, _A2A_MAX_POLL_INTERVAL) state = _a2a_task_state(task) if state != "completed": From 8c9442a28daf74c9e7db7cecb2a7567cf17279a3 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 12 Aug 2026 00:02:17 +0800 Subject: [PATCH 8/9] fix --- .../builtin_tools/test_run_sandbox_agent.py | 608 ++++++++---------- 1 file changed, 271 insertions(+), 337 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index ef314c2f6..b042402e4 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -13,13 +13,12 @@ # limitations under the License. import importlib.util -import io import json import sys import types import unittest from pathlib import Path -from unittest.mock import call, patch +from unittest.mock import patch def _load_run_sandbox_agent_module(): @@ -82,6 +81,7 @@ def _load_execute_skills_module( *, ensure_agentkit_session_endpoint=lambda **_kwargs: "", run_sandbox_agent=lambda **_kwargs: "", + wait_for_skill_api_health=lambda **_kwargs: None, ): module_path = ( Path(__file__).resolve().parents[3] @@ -140,6 +140,8 @@ def _load_execute_skills_module( assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) + if wait_for_skill_api_health is not None: + module._wait_for_skill_api_health = wait_for_skill_api_health return module @@ -211,92 +213,12 @@ def _tool_context(self): _invocation_context=invocation_context, ) - def test_execute_skills_rejects_invalid_timeout(self): - module = _load_execute_skills_module() - - for timeout in (0, -1, 1801, 1.5, True): - with self.subTest(timeout=timeout): - with self.assertRaisesRegex( - ValueError, - r"timeout must be an integer between 1 and 1800 seconds", - ): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - timeout=timeout, - ) - - def test_env_vars_are_not_supported(self): - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "env_vars must be rejected before endpoint resolution" - ) - ) - - with self.assertRaisesRegex(ValueError, "env_vars is not supported"): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - env_vars={"CUSTOM_VALUE": "custom"}, - ) - - with self.assertRaisesRegex(ValueError, "env_vars is not supported"): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - env_vars={}, - ) - - def test_execute_skills_does_not_accept_invocation_mode(self): - module = _load_execute_skills_module() - - with self.assertRaises(TypeError): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="execute", - ) - - def test_default_a2a_mode_sends_nonblocking_message_and_polls_task(self): + def test_prefers_new_skill_execute_api_when_endpoint_is_available(self): captured_requests = [] - responses = [ - { - "jsonrpc": "2.0", - "id": "send", - "result": { - "kind": "task", - "id": "task-1", - "status": {"state": "working"}, - }, - }, - { - "jsonrpc": "2.0", - "id": "get", - "result": { - "kind": "task", - "id": "task-1", - "status": {"state": "completed"}, - "artifacts": [ - { - "parts": [ - {"kind": "text", "text": "a2a "}, - { - "kind": "text", - "text": "thought", - "metadata": {"adk_thought": True}, - }, - ] - }, - {"parts": [{"kind": "text", "text": "result"}]}, - ], - }, - }, - ] + health_endpoints = [] + session_kwargs = [] class FakeResponse: - def __init__(self, payload): - self.payload = payload - def __enter__(self): return self @@ -304,136 +226,177 @@ def __exit__(self, *_args): return None def read(self): - return json.dumps(self.payload).encode() + return b'{"content": "api result"}' def fake_urlopen(request, timeout=None): captured_requests.append((request, timeout)) - return FakeResponse(responses.pop(0)) + return FakeResponse() module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **kwargs: ( - self.assertEqual(1800, kwargs["ttl"]) or "https://sandbox.test" + session_kwargs.append(kwargs) or "https://sandbox.test" + ), + wait_for_skill_api_health=lambda **kwargs: health_endpoints.append( + kwargs["endpoint"] ), ) + with patch.object(module.request, "urlopen", fake_urlopen): + result = module.execute_skills("do work", tool_context=self._tool_context()) + + self.assertEqual(result, "api result") + self.assertTrue(session_kwargs[0]["wait_until_ready"]) + self.assertEqual(["https://sandbox.test"], health_endpoints) + self.assertEqual(1, len(captured_requests)) + request_obj, timeout = captured_requests[0] + self.assertEqual("https://sandbox.test/v1/skills/execute", request_obj.full_url) + self.assertEqual(900, timeout) + self.assertEqual("POST", request_obj.get_method()) + self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) + self.assertIn(b'"prompt": "do work"', request_obj.data) + + def test_health_check_retries_502_until_upstream_is_ready(self): + attempts = [] + + class HealthyResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + class ErrorResponse: + def read(self): + return b"bad gateway" + + def close(self): + return None + + module = _load_execute_skills_module(wait_for_skill_api_health=None) + + def fake_urlopen(req, **_kwargs): + attempts.append((req.full_url, req.get_method())) + if len(attempts) == 1: + raise module.error.HTTPError( + url=req.full_url, + code=502, + msg="Bad Gateway", + hdrs={}, + fp=ErrorResponse(), + ) + return HealthyResponse() + with ( patch.object(module.request, "urlopen", fake_urlopen), patch.object(module.time, "sleep") as sleep, ): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - ) + module._wait_for_skill_api_health(endpoint="https://sandbox.test") - self.assertEqual(result, "a2a \nresult") - self.assertEqual(2, len(captured_requests)) - send_request, send_timeout = captured_requests[0] - send_payload = json.loads(send_request.data.decode()) - self.assertEqual("https://sandbox.test/a2a", send_request.full_url) - self.assertEqual(60, send_timeout) - self.assertEqual("message/send", send_payload["method"]) self.assertEqual( - "do work", send_payload["params"]["message"]["parts"][0]["text"] + [ + ("https://sandbox.test/v1/skills/healthz", "GET"), + ("https://sandbox.test/v1/skills/healthz", "GET"), + ], + attempts, ) - self.assertFalse(send_payload["params"]["configuration"]["blocking"]) - self.assertEqual(20, send_payload["params"]["configuration"]["historyLength"]) - self.assertEqual("user", send_payload["params"]["metadata"]["user_id"]) - self.assertEqual("session-1", send_payload["params"]["metadata"]["session_id"]) - - get_request, get_timeout = captured_requests[1] - get_payload = json.loads(get_request.data.decode()) - self.assertEqual("https://sandbox.test/a2a", get_request.full_url) - self.assertEqual(60, get_timeout) - self.assertEqual("tasks/get", get_payload["method"]) - self.assertEqual("task-1", get_payload["params"]["id"]) - sleep.assert_called_once_with(2.0) - - def test_a2a_task_polling_uses_capped_exponential_backoff(self): - states = ["working", "working", "working", "working", "working", "completed"] - - def fake_post_a2a_jsonrpc(**_kwargs): - state = states.pop(0) - return { - "jsonrpc": "2.0", - "result": { - "kind": "task", - "id": "task-1", - "status": { - "state": state, - "message": {"parts": [{"kind": "text", "text": "done"}]}, - }, - }, - } - - module = _load_execute_skills_module() - with ( - patch.object(module, "_post_a2a_jsonrpc", fake_post_a2a_jsonrpc), - patch.object(module.time, "sleep") as sleep, - ): - result = module._execute_skills_via_a2a( - workflow_prompt="do work", - endpoint="https://sandbox.test", - tool_context=self._tool_context(), - timeout=1800, + sleep.assert_called_once_with(1.0) + + def test_health_check_allows_images_without_health_endpoint(self): + class NotFoundResponse: + def read(self): + return b"not found" + + def close(self): + return None + + module = _load_execute_skills_module(wait_for_skill_api_health=None) + + def fake_urlopen(req, **_kwargs): + raise module.error.HTTPError( + url=req.full_url, + code=404, + msg="Not Found", + hdrs={}, + fp=NotFoundResponse(), ) - self.assertEqual("done", result) + with patch.object(module.request, "urlopen", fake_urlopen): + module._wait_for_skill_api_health(endpoint="https://sandbox.test") + + def test_env_vars_use_legacy_runcode_execution(self): + captured_kwargs = {} + + def fake_run_sandbox_agent(**kwargs): + captured_kwargs.update(kwargs) + return "legacy result" + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( + "Skill API must not be used when env_vars are provided" + ), + run_sandbox_agent=fake_run_sandbox_agent, + ) + + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + env_vars={"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, + ) + + self.assertEqual(result, "legacy result") self.assertEqual( - [call(2.0), call(4.0), call(8.0), call(16.0), call(16.0)], - sleep.call_args_list, + {"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, + captured_kwargs["extra_env_vars"], ) - def test_a2a_forwards_custom_timeout_to_requests(self): - captured_timeouts = [] + def test_python_agent_mode_uses_legacy_runcode_execution(self): + captured_kwargs = {} - class FakeResponse: - def __enter__(self): - return self + def fake_run_sandbox_agent(**kwargs): + captured_kwargs.update(kwargs) + return "legacy result" - def __exit__(self, *_args): - return None + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( + "Session endpoint must not be used in python_agent mode" + ), + run_sandbox_agent=fake_run_sandbox_agent, + ) - def read(self): - return json.dumps( - { - "jsonrpc": "2.0", - "id": "send", - "result": { - "kind": "task", - "id": "task-1", - "status": { - "state": "completed", - "message": { - "parts": [ - {"kind": "text", "text": "custom timeout"} - ] - }, - }, - }, - } - ).encode() + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="python_agent", + ) - def fake_urlopen(_request, timeout=None): - captured_timeouts.append(timeout) - return FakeResponse() + self.assertEqual(result, "legacy result") + self.assertEqual("do work", captured_kwargs["workflow_prompt"]) + self.assertEqual("test-tool", captured_kwargs["tool_id"]) + + def test_invocation_mode_can_be_read_from_environment(self): + captured_kwargs = {} + + def fake_run_sandbox_agent(**kwargs): + captured_kwargs.update(kwargs) + return "legacy result" module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **kwargs: ( - self.assertEqual(1800, kwargs["ttl"]) or "https://sandbox.test" + ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( + "Session endpoint must not be used in python_agent mode" ), + run_sandbox_agent=fake_run_sandbox_agent, ) - with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - timeout=120, - ) + with patch.dict( + module.os.environ, + {"AGENTKIT_SKILL_INVOCATION_MODE": "python_agent"}, + ): + result = module.execute_skills("do work", tool_context=self._tool_context()) - self.assertEqual("custom timeout", result) - self.assertEqual([60], captured_timeouts) + self.assertEqual(result, "legacy result") + self.assertEqual("do work", captured_kwargs["workflow_prompt"]) - def test_a2a_posts_to_vefaas_a2a_endpoint_with_query_auth(self): + def test_run_sse_mode_posts_run_sse_and_aggregates_event_text(self): captured_requests = [] class FakeResponse: @@ -444,62 +407,37 @@ def __exit__(self, *_args): return None def read(self): - return json.dumps( - { - "jsonrpc": "2.0", - "id": "test-1", - "result": { - "kind": "task", - "id": "task-1", - "status": { - "state": "completed", - "message": { - "parts": [{"kind": "text", "text": "hello result"}] - }, - }, - }, - } - ).encode() + return ( + b'data: {"content":{"parts":[{"text":"hello "}]}}\n\n' + b'data: {"content":{"parts":[{"text":"world"}]}}\n\n' + b"data: [DONE]\n\n" + ) def fake_urlopen(request, timeout=None): captured_requests.append((request, timeout)) return FakeResponse() module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: ( - "https://sandbox.test/?faasInstanceName=inst&Authorization=key" - ), + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", ) with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills("hello", tool_context=self._tool_context()) + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="run_sse", + ) - self.assertEqual("hello result", result) - self.assertEqual(1, len(captured_requests)) - send_request, send_timeout = captured_requests[0] - send_payload = json.loads(send_request.data.decode()) - self.assertEqual( - "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", - send_request.full_url, - ) - self.assertEqual("application/json", send_request.get_header("Content-type")) - self.assertIsNone(send_request.get_header("Accept")) - self.assertEqual(60, send_timeout) - self.assertEqual("2.0", send_payload["jsonrpc"]) - self.assertEqual("message/send", send_payload["method"]) - self.assertEqual("message", send_payload["params"]["message"]["kind"]) - self.assertEqual("user", send_payload["params"]["message"]["role"]) - self.assertEqual( - [{"kind": "text", "text": "hello"}], - send_payload["params"]["message"]["parts"], - ) - self.assertFalse(send_payload["params"]["configuration"]["blocking"]) - self.assertEqual(20, send_payload["params"]["configuration"]["historyLength"]) - self.assertEqual("session-1", send_payload["params"]["metadata"]["session_id"]) - self.assertEqual("user", send_payload["params"]["metadata"]["user_id"]) + self.assertEqual(result, "hello world") + request_obj, timeout = captured_requests[0] + self.assertEqual("https://sandbox.test/run_sse", request_obj.full_url) + self.assertEqual(900, timeout) + self.assertIn(b'"app_name": "agent"', request_obj.data) + self.assertIn(b'"session_id": "session-1"', request_obj.data) + self.assertIn(b'"text": "do work"', request_obj.data) - def test_retry_502(self): - captured_timeouts = [] + def test_a2a_mode_posts_message_send_and_returns_text(self): + captured_requests = [] class FakeResponse: def __enter__(self): @@ -512,93 +450,52 @@ def read(self): return json.dumps( { "jsonrpc": "2.0", - "id": "send", + "id": "req", "result": { - "kind": "task", - "id": "task-1", - "status": { - "state": "completed", - "message": { - "parts": [{"kind": "text", "text": "retry ok"}] - }, - }, + "kind": "message", + "role": "agent", + "parts": [{"kind": "text", "text": "a2a result"}], }, } ).encode() - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) - def fake_urlopen(request, timeout=None): - captured_timeouts.append(timeout) - if len(captured_timeouts) == 1: - raise module.error.HTTPError( - request.full_url, - 502, - "Bad Gateway", - hdrs=None, - fp=io.BytesIO(b"temporary gateway error"), - ) + captured_requests.append((request, timeout)) return FakeResponse() - with ( - patch.object(module.request, "urlopen", fake_urlopen), - patch.object(module.time, "sleep") as sleep, - ): - result = module.execute_skills("hello", tool_context=self._tool_context()) - - self.assertEqual("retry ok", result) - self.assertEqual([60, 60], captured_timeouts) - sleep.assert_called_once_with(2.0) - - def test_a2a_mode_raises_when_task_fails(self): - responses = [ - { - "jsonrpc": "2.0", - "id": "send", - "result": { - "kind": "task", - "id": "task-1", - "status": {"state": "working"}, - }, - }, - { - "jsonrpc": "2.0", - "id": "get", - "result": { - "kind": "task", - "id": "task-1", - "status": {"state": "failed"}, - }, - }, - ] - - class FakeResponse: - def __enter__(self): - return self + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) - def __exit__(self, *_args): - return None + with patch.object(module.request, "urlopen", fake_urlopen): + result = module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="a2a", + ) - def read(self): - return json.dumps(responses.pop(0)).encode() + self.assertEqual(result, "a2a result") + request_obj, timeout = captured_requests[0] + payload = json.loads(request_obj.data.decode()) + self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) + self.assertEqual(900, timeout) + self.assertEqual("message/send", payload["method"]) + self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) + self.assertTrue(payload["params"]["configuration"]["blocking"]) + def test_unsupported_invocation_mode_raises_value_error(self): module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( + "Invalid mode must be rejected before endpoint resolution" + ), ) - with ( - patch.object( - module.request, "urlopen", lambda *_args, **_kwargs: FakeResponse() - ), - patch.object(module.time, "sleep"), - ): - with self.assertRaisesRegex( - RuntimeError, - r"A2A task task-1 ended with state failed", - ): - module.execute_skills("do work", tool_context=self._tool_context()) + with self.assertRaisesRegex(ValueError, "Unsupported AgentKit Skill"): + module.execute_skills( + "do work", + tool_context=self._tool_context(), + invocation_mode="stream", + ) def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): module = _load_execute_skills_module( @@ -606,56 +503,68 @@ def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): ) self.assertEqual( - "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", + "https://sandbox.test/v1/skills/execute?faasInstanceName=inst&Authorization=key", module._skill_api_url( "https://sandbox.test/?faasInstanceName=inst&Authorization=key", - "/a2a", + "/v1/skills/execute", ), ) - def test_a2a_jsonrpc_url_appends_a2a_before_vefaas_query(self): - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "", - ) + def test_requires_sandbox_upgrade_when_skill_api_returns_404(self): + class NotFoundResponse: + def read(self): + return b"not found" - self.assertEqual( - "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key", - module._a2a_jsonrpc_url( - "https://sandbox.test/?faasInstanceName=inst&Authorization=key" - ), - ) - self.assertEqual( - "https://sc56tro0thc3nstnfkabv.apigateway-cn-beijing.volceapi.com/a2a" - "?faasInstanceName=vefaas-example-sandbox&Authorization=test-token", - module._a2a_jsonrpc_url( - "https://sc56tro0thc3nstnfkabv.apigateway-cn-beijing.volceapi.com" - "?faasInstanceName=vefaas-example-sandbox&Authorization=test-token" - ), - ) + def close(self): + return None + + def fake_urlopen(_request, **_kwargs): + raise module.error.HTTPError( + url="https://sandbox.test/v1/skills/execute", + code=404, + msg="Not Found", + hdrs={}, + fp=NotFoundResponse(), + ) - def test_a2a_jsonrpc_url_keeps_explicit_a2a_endpoint(self): module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "", + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", ) - endpoint = "https://sandbox.test/a2a?faasInstanceName=inst&Authorization=key" - self.assertEqual(endpoint, module._a2a_jsonrpc_url(endpoint)) - self.assertEqual( - endpoint, - module._a2a_jsonrpc_url( - "https://sandbox.test/a2a/?faasInstanceName=inst&Authorization=key" - ), - ) + with patch.object(module.request, "urlopen", fake_urlopen): + with self.assertRaisesRegex( + RuntimeError, + r"HTTP 404.*(?:升级|upgrade).*Skill", + ): + module.execute_skills("do work", tool_context=self._tool_context()) + + def test_requires_sandbox_upgrade_when_skill_api_returns_405(self): + class MethodNotAllowedResponse: + def read(self): + return b"method not allowed" + + def close(self): + return None + + def fake_urlopen(_request, **_kwargs): + raise module.error.HTTPError( + url="https://sandbox.test/v1/skills/execute", + code=405, + msg="Method Not Allowed", + hdrs={}, + fp=MethodNotAllowedResponse(), + ) - def test_a2a_jsonrpc_url_appends_a2a_for_plain_endpoint(self): module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "", + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", ) - self.assertEqual( - "https://sandbox.test/a2a", - module._a2a_jsonrpc_url("https://sandbox.test"), - ) + with patch.object(module.request, "urlopen", fake_urlopen): + with self.assertRaisesRegex( + RuntimeError, + r"HTTP 405.*(?:升级|upgrade).*Skill", + ): + module.execute_skills("do work", tool_context=self._tool_context()) def test_raises_runtime_error_when_session_endpoint_is_unavailable(self): def raise_endpoint_error(**_kwargs): @@ -680,6 +589,31 @@ def test_missing_tool_context_raises_value_error(self): with self.assertRaisesRegex(ValueError, r"tool_context is required"): module.execute_skills("do work", tool_context=None) + def test_non_compatibility_skill_api_http_error_is_not_swallowed(self): + class ServerErrorResponse: + def read(self): + return b"internal error" + + def close(self): + return None + + def fake_urlopen(_request, **_kwargs): + raise module.error.HTTPError( + url="https://sandbox.test/v1/skills/execute", + code=500, + msg="Internal Server Error", + hdrs={}, + fp=ServerErrorResponse(), + ) + + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) + + with patch.object(module.request, "urlopen", fake_urlopen): + with self.assertRaisesRegex(RuntimeError, "HTTP 500: internal error"): + module.execute_skills("do work", tool_context=self._tool_context()) + if __name__ == "__main__": unittest.main() From 8a0451f27921c25dbc2ad2587228dd87597faec1 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 12 Aug 2026 00:24:01 +0800 Subject: [PATCH 9/9] test(execute-skills): align sandbox agent tests with a2a execution --- .../builtin_tools/test_run_sandbox_agent.py | 312 +++++++++--------- 1 file changed, 163 insertions(+), 149 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index b042402e4..09d440684 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -213,9 +213,8 @@ def _tool_context(self): _invocation_context=invocation_context, ) - def test_prefers_new_skill_execute_api_when_endpoint_is_available(self): + def test_execute_skills_posts_a2a_message_send_and_returns_artifact_text(self): captured_requests = [] - health_endpoints = [] session_kwargs = [] class FakeResponse: @@ -226,7 +225,20 @@ def __exit__(self, *_args): return None def read(self): - return b'{"content": "api result"}' + return json.dumps( + { + "jsonrpc": "2.0", + "id": "req", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "completed"}, + "artifacts": [ + {"parts": [{"kind": "text", "text": "a2a result"}]} + ], + }, + } + ).encode() def fake_urlopen(request, timeout=None): captured_requests.append((request, timeout)) @@ -236,26 +248,30 @@ def fake_urlopen(request, timeout=None): ensure_agentkit_session_endpoint=lambda **kwargs: ( session_kwargs.append(kwargs) or "https://sandbox.test" ), - wait_for_skill_api_health=lambda **kwargs: health_endpoints.append( - kwargs["endpoint"] - ), ) with patch.object(module.request, "urlopen", fake_urlopen): result = module.execute_skills("do work", tool_context=self._tool_context()) - self.assertEqual(result, "api result") + self.assertEqual(result, "a2a result") self.assertTrue(session_kwargs[0]["wait_until_ready"]) - self.assertEqual(["https://sandbox.test"], health_endpoints) + self.assertEqual("test-tool", session_kwargs[0]["tool_id"]) + self.assertEqual( + "agent_user_session-1", session_kwargs[0]["tool_user_session_id"] + ) self.assertEqual(1, len(captured_requests)) request_obj, timeout = captured_requests[0] - self.assertEqual("https://sandbox.test/v1/skills/execute", request_obj.full_url) - self.assertEqual(900, timeout) + payload = json.loads(request_obj.data.decode()) + self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) + self.assertEqual(60, timeout) self.assertEqual("POST", request_obj.get_method()) - self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"]) - self.assertIn(b'"prompt": "do work"', request_obj.data) + self.assertEqual("message/send", payload["method"]) + self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) + self.assertFalse(payload["params"]["configuration"]["blocking"]) + self.assertEqual("user", payload["params"]["metadata"]["user_id"]) + self.assertEqual("session-1", payload["params"]["metadata"]["session_id"]) - def test_health_check_retries_502_until_upstream_is_ready(self): + def test_a2a_retries_502_until_upstream_is_ready(self): attempts = [] class HealthyResponse: @@ -265,6 +281,22 @@ def __enter__(self): def __exit__(self, *_args): return None + def read(self): + return json.dumps( + { + "jsonrpc": "2.0", + "id": "req", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "completed"}, + "artifacts": [ + {"parts": [{"kind": "text", "text": "ready"}]} + ], + }, + } + ).encode() + class ErrorResponse: def read(self): return b"bad gateway" @@ -272,7 +304,9 @@ def read(self): def close(self): return None - module = _load_execute_skills_module(wait_for_skill_api_health=None) + module = _load_execute_skills_module( + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", + ) def fake_urlopen(req, **_kwargs): attempts.append((req.full_url, req.get_method())) @@ -290,113 +324,94 @@ def fake_urlopen(req, **_kwargs): patch.object(module.request, "urlopen", fake_urlopen), patch.object(module.time, "sleep") as sleep, ): - module._wait_for_skill_api_health(endpoint="https://sandbox.test") + result = module.execute_skills("do work", tool_context=self._tool_context()) + self.assertEqual(result, "ready") self.assertEqual( [ - ("https://sandbox.test/v1/skills/healthz", "GET"), - ("https://sandbox.test/v1/skills/healthz", "GET"), + ("https://sandbox.test/a2a", "POST"), + ("https://sandbox.test/a2a", "POST"), ], attempts, ) - sleep.assert_called_once_with(1.0) - - def test_health_check_allows_images_without_health_endpoint(self): - class NotFoundResponse: - def read(self): - return b"not found" - - def close(self): - return None - - module = _load_execute_skills_module(wait_for_skill_api_health=None) - - def fake_urlopen(req, **_kwargs): - raise module.error.HTTPError( - url=req.full_url, - code=404, - msg="Not Found", - hdrs={}, - fp=NotFoundResponse(), - ) - - with patch.object(module.request, "urlopen", fake_urlopen): - module._wait_for_skill_api_health(endpoint="https://sandbox.test") - - def test_env_vars_use_legacy_runcode_execution(self): - captured_kwargs = {} - - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" + sleep.assert_called_once_with(2.0) + def test_env_vars_are_rejected_for_a2a_execution(self): module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Skill API must not be used when env_vars are provided" + "env_vars must be rejected before endpoint resolution" ), - run_sandbox_agent=fake_run_sandbox_agent, - ) - - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - env_vars={"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, ) - self.assertEqual(result, "legacy result") - self.assertEqual( - {"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""}, - captured_kwargs["extra_env_vars"], - ) - - def test_python_agent_mode_uses_legacy_runcode_execution(self): - captured_kwargs = {} - - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" + with self.assertRaisesRegex(ValueError, "env_vars is not supported"): + module.execute_skills( + "do work", + tool_context=self._tool_context(), + env_vars={"CUSTOM_VALUE": "custom"}, + ) - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Session endpoint must not be used in python_agent mode" - ), - run_sandbox_agent=fake_run_sandbox_agent, - ) + def test_a2a_polls_task_until_completed_and_returns_status_message_text(self): + captured_requests = [] + responses = [ + { + "jsonrpc": "2.0", + "id": "send", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "working"}, + }, + }, + { + "jsonrpc": "2.0", + "id": "get", + "result": { + "kind": "task", + "id": "task-1", + "status": { + "state": "completed", + "message": {"parts": [{"kind": "text", "text": "poll result"}]}, + }, + }, + }, + ] - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="python_agent", - ) + class FakeResponse: + def __enter__(self): + return self - self.assertEqual(result, "legacy result") - self.assertEqual("do work", captured_kwargs["workflow_prompt"]) - self.assertEqual("test-tool", captured_kwargs["tool_id"]) + def __exit__(self, *_args): + return None - def test_invocation_mode_can_be_read_from_environment(self): - captured_kwargs = {} + def read(self): + return json.dumps(responses.pop(0)).encode() - def fake_run_sandbox_agent(**kwargs): - captured_kwargs.update(kwargs) - return "legacy result" + def fake_urlopen(request, timeout=None): + captured_requests.append((request, timeout)) + return FakeResponse() module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Session endpoint must not be used in python_agent mode" - ), - run_sandbox_agent=fake_run_sandbox_agent, + ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", ) - with patch.dict( - module.os.environ, - {"AGENTKIT_SKILL_INVOCATION_MODE": "python_agent"}, + with ( + patch.object(module.request, "urlopen", fake_urlopen), + patch.object(module.time, "sleep") as sleep, ): result = module.execute_skills("do work", tool_context=self._tool_context()) - self.assertEqual(result, "legacy result") - self.assertEqual("do work", captured_kwargs["workflow_prompt"]) - - def test_run_sse_mode_posts_run_sse_and_aggregates_event_text(self): + self.assertEqual(result, "poll result") + self.assertEqual(2, len(captured_requests)) + send_request, _send_timeout = captured_requests[0] + get_request, _get_timeout = captured_requests[1] + send_payload = json.loads(send_request.data.decode()) + get_payload = json.loads(get_request.data.decode()) + self.assertEqual("message/send", send_payload["method"]) + self.assertEqual("tasks/get", get_payload["method"]) + self.assertEqual("task-1", get_payload["params"]["id"]) + sleep.assert_called_once_with(2.0) + + def test_a2a_returns_history_text_when_artifacts_and_status_are_empty(self): captured_requests = [] class FakeResponse: @@ -407,11 +422,29 @@ def __exit__(self, *_args): return None def read(self): - return ( - b'data: {"content":{"parts":[{"text":"hello "}]}}\n\n' - b'data: {"content":{"parts":[{"text":"world"}]}}\n\n' - b"data: [DONE]\n\n" - ) + return json.dumps( + { + "jsonrpc": "2.0", + "id": "req", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "completed"}, + "history": [ + { + "role": "user", + "parts": [{"kind": "text", "text": "do work"}], + }, + { + "role": "agent", + "parts": [ + {"kind": "text", "text": "history result"} + ], + }, + ], + }, + } + ).encode() def fake_urlopen(request, timeout=None): captured_requests.append((request, timeout)) @@ -422,23 +455,18 @@ def fake_urlopen(request, timeout=None): ) with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="run_sse", - ) + result = module.execute_skills("do work", tool_context=self._tool_context()) - self.assertEqual(result, "hello world") + self.assertEqual(result, "history result") request_obj, timeout = captured_requests[0] - self.assertEqual("https://sandbox.test/run_sse", request_obj.full_url) - self.assertEqual(900, timeout) - self.assertIn(b'"app_name": "agent"', request_obj.data) - self.assertIn(b'"session_id": "session-1"', request_obj.data) - self.assertIn(b'"text": "do work"', request_obj.data) - - def test_a2a_mode_posts_message_send_and_returns_text(self): - captured_requests = [] + payload = json.loads(request_obj.data.decode()) + self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) + self.assertEqual(60, timeout) + self.assertEqual("message/send", payload["method"]) + self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) + self.assertFalse(payload["params"]["configuration"]["blocking"]) + def test_a2a_rejects_non_task_message_send_response(self): class FakeResponse: def __enter__(self): return self @@ -459,42 +487,28 @@ def read(self): } ).encode() - def fake_urlopen(request, timeout=None): - captured_requests.append((request, timeout)) - return FakeResponse() - module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", ) - with patch.object(module.request, "urlopen", fake_urlopen): - result = module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="a2a", - ) - - self.assertEqual(result, "a2a result") - request_obj, timeout = captured_requests[0] - payload = json.loads(request_obj.data.decode()) - self.assertEqual("https://sandbox.test/a2a", request_obj.full_url) - self.assertEqual(900, timeout) - self.assertEqual("message/send", payload["method"]) - self.assertEqual("do work", payload["params"]["message"]["parts"][0]["text"]) - self.assertTrue(payload["params"]["configuration"]["blocking"]) + with patch.object( + module.request, "urlopen", lambda *_args, **_kwargs: FakeResponse() + ): + with self.assertRaisesRegex( + RuntimeError, "A2ASendMessage response result is not an A2A task" + ): + module.execute_skills("do work", tool_context=self._tool_context()) - def test_unsupported_invocation_mode_raises_value_error(self): + def test_timeout_must_be_within_a2a_execution_limit(self): module = _load_execute_skills_module( ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail( - "Invalid mode must be rejected before endpoint resolution" + "Invalid timeout must be rejected before endpoint resolution" ), ) - with self.assertRaisesRegex(ValueError, "Unsupported AgentKit Skill"): + with self.assertRaisesRegex(ValueError, "between 1 and 1800 seconds"): module.execute_skills( - "do work", - tool_context=self._tool_context(), - invocation_mode="stream", + "do work", tool_context=self._tool_context(), timeout=0 ) def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): @@ -510,7 +524,7 @@ def test_skill_api_url_preserves_agentkit_endpoint_query_auth(self): ), ) - def test_requires_sandbox_upgrade_when_skill_api_returns_404(self): + def test_a2a_http_404_is_not_swallowed(self): class NotFoundResponse: def read(self): return b"not found" @@ -520,7 +534,7 @@ def close(self): def fake_urlopen(_request, **_kwargs): raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", + url="https://sandbox.test/a2a", code=404, msg="Not Found", hdrs={}, @@ -534,11 +548,11 @@ def fake_urlopen(_request, **_kwargs): with patch.object(module.request, "urlopen", fake_urlopen): with self.assertRaisesRegex( RuntimeError, - r"HTTP 404.*(?:升级|upgrade).*Skill", + r"AgentKit Skill /a2a request failed with HTTP 404: not found", ): module.execute_skills("do work", tool_context=self._tool_context()) - def test_requires_sandbox_upgrade_when_skill_api_returns_405(self): + def test_a2a_http_405_is_not_swallowed(self): class MethodNotAllowedResponse: def read(self): return b"method not allowed" @@ -548,7 +562,7 @@ def close(self): def fake_urlopen(_request, **_kwargs): raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", + url="https://sandbox.test/a2a", code=405, msg="Method Not Allowed", hdrs={}, @@ -562,7 +576,7 @@ def fake_urlopen(_request, **_kwargs): with patch.object(module.request, "urlopen", fake_urlopen): with self.assertRaisesRegex( RuntimeError, - r"HTTP 405.*(?:升级|upgrade).*Skill", + r"AgentKit Skill /a2a request failed with HTTP 405: method not allowed", ): module.execute_skills("do work", tool_context=self._tool_context()) @@ -589,7 +603,7 @@ def test_missing_tool_context_raises_value_error(self): with self.assertRaisesRegex(ValueError, r"tool_context is required"): module.execute_skills("do work", tool_context=None) - def test_non_compatibility_skill_api_http_error_is_not_swallowed(self): + def test_a2a_http_500_is_not_swallowed(self): class ServerErrorResponse: def read(self): return b"internal error" @@ -599,7 +613,7 @@ def close(self): def fake_urlopen(_request, **_kwargs): raise module.error.HTTPError( - url="https://sandbox.test/v1/skills/execute", + url="https://sandbox.test/a2a", code=500, msg="Internal Server Error", hdrs={},