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 00000000..11970127 --- /dev/null +++ b/tests/tools/builtin_tools/test_run_code.py @@ -0,0 +1,131 @@ +# 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 + +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, 1800, "between 1 and 1800 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 b042402e..09d44068 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={}, diff --git a/veadk/tools/builtin_tools/_agentkit.py b/veadk/tools/builtin_tools/_agentkit.py index 80b9e583..4fa1bf7c 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) diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index cd19749f..7c56e544 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,29 +25,32 @@ 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 = 900 -_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", "python_agent"} + +_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}) +_A2A_TERMINAL_STATES = frozenset( + { + "completed", + "failed", + "canceled", + "rejected", + "input-required", + "auth-required", + } ) -def _skill_api_upgrade_hint() -> str: - return ( - "提示：当前 Skill 沙箱镜像未实现 /v1/skills/execute 接口，" - "可能是旧版沙箱镜像。" - "请升级 Skill 沙箱镜像或切换到支持 Skill HTTP API 的新版沙箱。" - ) +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 _tool_user_session_id(tool_context: ToolContext) -> str: @@ -59,16 +61,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") @@ -81,58 +73,22 @@ 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() - if not resolved: - return "execute" - 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." +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 "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 + return _skill_api_url(endpoint, "/a2a") def _post_json( @@ -163,78 +119,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 "" @@ -242,6 +126,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) @@ -252,22 +138,60 @@ def _extract_text_from_parts(parts: object) -> str: return "".join(chunks) -def _extract_text_from_a2a_result(result: object) -> str: +def _is_adk_thought_part(part: dict) -> bool: + metadata = part.get("metadata") + return isinstance(metadata, dict) and metadata.get("adk_thought") is True + + +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): - return "" - if result.get("kind") == "message": - return _extract_text_from_parts(result.get("parts")) - artifacts = result.get("artifacts") + 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 = "".join(chunks) + text = "\n".join(chunk for chunk in chunks if chunk) if text: return text - history = result.get("history") + + 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 { @@ -280,47 +204,58 @@ def _extract_text_from_a2a_result(result: object) -> str: return "" -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( +def _post_a2a_jsonrpc( *, - workflow_prompt: str, endpoint: str, - tool_context: ToolContext, + payload: dict[str, object], 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) + retry_until: float | None = None, +) -> dict: + 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: + 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 _execute_skills_via_a2a( @@ -331,59 +266,76 @@ def _execute_skills_via_a2a( timeout: int, ) -> str: invocation_context = tool_context._invocation_context - # A2A 沙箱使用 JSON-RPC message/send，同步等待最终结果。 - 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}, - }, + deadline = time.monotonic() + timeout + message = { + "kind": "message", + "messageId": uuid.uuid4().hex, + "role": "user", + "parts": [{"kind": "text", "text": workflow_prompt}], } - 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"): - 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) + 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), + retry_until=deadline, + ), + ) + 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(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), + retry_until=deadline, + ), + ) + poll_interval = min(poll_interval * 2, _A2A_MAX_POLL_INTERVAL) -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/", + 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)}" ) - return run_sandbox_agent( - workflow_prompt=workflow_prompt, - tool_id=tool_id, - tool_context=tool_context, - timeout=timeout, - extra_env_vars=extra_env_vars, - ) + + text = _a2a_task_result_text(task) + if text: + return text + return json.dumps(task, ensure_ascii=False) def _execute_skills_via_skill_api( @@ -392,7 +344,6 @@ def _execute_skills_via_skill_api( tool_id: str, tool_context: ToolContext, timeout: int, - invocation_mode: str, ) -> str: try: endpoint = ensure_agentkit_session_endpoint( @@ -407,27 +358,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, - ) - - _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, ) @@ -436,7 +370,7 @@ 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. @@ -444,43 +378,25 @@ 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 "execute" (default), "run_sse", "a2a", 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. 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) + 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=_SKILL_API_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=_SKILL_API_TIMEOUT, - ) - return _execute_skills_via_skill_api( workflow_prompt=workflow_prompt, tool_id=tool_id, tool_context=tool_context, - timeout=_SKILL_API_TIMEOUT, - invocation_mode=mode, + timeout=timeout, ) diff --git a/veadk/tools/builtin_tools/run_code.py b/veadk/tools/builtin_tools/run_code.py index 7497a208..9a3249cf 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}")