From 600fcda94207b6a8e629b776cd213ca788450541 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 16:06:40 +0800 Subject: [PATCH 1/6] fix(a2a): adapt skill sandbox invocation contract --- tests/a2a/test_remote_ve_agent.py | 106 ++++++++++++++ .../builtin_tools/test_run_sandbox_agent.py | 80 ---------- veadk/a2a/remote_ve_agent.py | 138 ++++++++++++++---- veadk/tools/builtin_tools/execute_skills.py | 73 +-------- 4 files changed, 219 insertions(+), 178 deletions(-) create mode 100644 tests/a2a/test_remote_ve_agent.py diff --git a/tests/a2a/test_remote_ve_agent.py b/tests/a2a/test_remote_ve_agent.py new file mode 100644 index 000000000..9ed76d119 --- /dev/null +++ b/tests/a2a/test_remote_ve_agent.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import asyncio +from unittest.mock import Mock, patch + +from veadk.a2a.remote_ve_agent import RemoteVeAgent + + +def _agent_card(url: str) -> dict: + return { + "name": "skill-agent", + "description": "Skill sandbox agent", + "url": url, + "version": "1.0.0", + "capabilities": {"streaming": True}, + "defaultInputModes": ["text"], + "defaultOutputModes": ["text"], + "skills": [ + { + "id": "chat", + "name": "chat", + "description": "Chat with the skill agent", + "tags": ["chat"], + } + ], + } + + +def _build_agent(endpoint: str, card_url: str) -> tuple[RemoteVeAgent, Mock]: + response = Mock() + response.json.return_value = _agent_card(card_url) + with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response) as get: + agent = RemoteVeAgent(name="remote", url=endpoint) + return agent, get + + +def _close(agent: RemoteVeAgent) -> None: + asyncio.run(agent._httpx_client.aclose()) + + +def test_remote_agent_preserves_a2a_path_from_agent_card() -> None: + agent, get = _build_agent("https://sandbox.test", "https://sandbox.test/a2a") + try: + assert agent._agent_card.url == "https://sandbox.test/a2a" + assert get.call_args.args[0] == ( + "https://sandbox.test/.well-known/agent-card.json" + ) + finally: + _close(agent) + + +def test_remote_agent_preserves_same_origin_session_authorization_query() -> None: + agent, get = _build_agent( + "https://sandbox.test/?faasInstanceName=inst&Authorization=key", + "https://sandbox.test/a2a", + ) + try: + assert get.call_args.args[0] == ( + "https://sandbox.test/.well-known/agent-card.json" + ) + assert get.call_args.kwargs["params"] == { + "faasInstanceName": "inst", + "Authorization": "key", + } + assert dict(agent._httpx_client.params) == { + "faasInstanceName": "inst", + "Authorization": "key", + } + finally: + _close(agent) + + +def test_remote_agent_does_not_forward_session_query_to_different_origin() -> None: + agent, _ = _build_agent( + "https://sandbox.test/?Authorization=key", + "https://different.test/a2a", + ) + try: + assert agent._agent_card.url == "https://different.test/a2a" + assert "Authorization" not in dict(agent._httpx_client.params) + finally: + _close(agent) + + +def test_remote_agent_keeps_explicit_query_auth_for_different_origin() -> None: + response = Mock() + response.json.return_value = _agent_card("https://different.test/a2a") + with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response): + agent = RemoteVeAgent( + name="remote", + url="https://sandbox.test", + auth_token="explicit-token", + auth_method="querystring", + ) + try: + assert dict(agent._httpx_client.params) == {"token": "explicit-token"} + finally: + _close(agent) + + +def test_remote_agent_replaces_loopback_host_but_keeps_card_path() -> None: + agent, _ = _build_agent("https://sandbox.test", "http://localhost:8000/a2a") + try: + assert agent._agent_card.url == "https://sandbox.test/a2a" + finally: + _close(agent) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 59391a023..d692bdb82 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -465,86 +465,6 @@ def fake_urlopen(_request, **_kwargs): with self.assertRaisesRegex(RuntimeError, "HTTP 500: internal error"): module.execute_skills("do work", tool_context=self._tool_context()) - def test_stream_mode_aggregates_text_chunks_from_skill_api_sse(self): - sse_body = ( - "event: chunk\n" - 'data: {"request_id":"req_1","type":"progress","content":"started","metadata":{}}\n\n' - "event: chunk\n" - 'data: {"request_id":"req_1","type":"text","content":"hello ","metadata":{}}\n\n' - "event: chunk\n" - 'data: {"request_id":"req_1","type":"text","content":"world","metadata":{}}\n\n' - "event: done\n" - 'data: {"request_id":"req_1","type":"progress","content":"done","metadata":{}}\n\n' - ).encode() - - class FakeResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - def read(self): - raise AssertionError("stream response must not be buffered with read()") - - def __iter__(self): - return iter(sse_body.splitlines(keepends=True)) - - captured_urls = [] - - def fake_urlopen(request, **_kwargs): - captured_urls.append(request.full_url) - 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(), - prefer_stream=True, - ) - - self.assertEqual(result, "hello world") - self.assertEqual(["https://sandbox.test/v1/skills/stream"], captured_urls) - - def test_stream_mode_raises_skill_api_error_event(self): - sse_body = ( - "event: error\n" - 'data: {"request_id":"req_1","type":"text","content":"skill failed","metadata":{}}\n\n' - ).encode() - - class FakeResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return None - - def read(self): - raise AssertionError("stream response must not be buffered with read()") - - def __iter__(self): - return iter(sse_body.splitlines(keepends=True)) - - module = _load_execute_skills_module( - ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test", - ) - - with patch.object( - module.request, - "urlopen", - lambda *_args, **_kwargs: FakeResponse(), - ): - with self.assertRaisesRegex(RuntimeError, "skill failed"): - module.execute_skills( - "do work", - tool_context=self._tool_context(), - prefer_stream=True, - ) - if __name__ == "__main__": unittest.main() diff --git a/veadk/a2a/remote_ve_agent.py b/veadk/a2a/remote_ve_agent.py index 1eddedaa9..7801ba541 100644 --- a/veadk/a2a/remote_ve_agent.py +++ b/veadk/a2a/remote_ve_agent.py @@ -12,27 +12,28 @@ # See the License for the specific language governing permissions and # limitations under the License. -import json import functools +import json from typing import AsyncGenerator, Literal, Optional +from urllib.parse import parse_qsl, urljoin, urlsplit, urlunsplit -from a2a.client.base_client import BaseClient import httpx import requests +from a2a.client.base_client import BaseClient from a2a.types import AgentCard +from google.adk.agents.invocation_context import InvocationContext from google.adk.agents.remote_a2a_agent import RemoteA2aAgent +from google.adk.events.event import Event +from google.adk.utils.context_utils import Aclosing from veadk.integrations.ve_identity.utils import generate_headers from veadk.utils.auth import VE_TIP_TOKEN_CREDENTIAL_KEY, VE_TIP_TOKEN_HEADER from veadk.utils.logger import get_logger -from google.adk.utils.context_utils import Aclosing -from google.adk.events.event import Event -from google.adk.agents.invocation_context import InvocationContext - logger = get_logger(__name__) AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent-card.json" +_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: @@ -41,6 +42,73 @@ def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: return agent_card_object +def _endpoint_parts(url: str): + parts = urlsplit(url) + if parts.scheme not in {"http", "https"} or not parts.hostname: + raise ValueError(f"Invalid A2A endpoint URL: {url}") + if parts.username or parts.password or parts.fragment: + raise ValueError("A2A endpoint URL must not contain userinfo or a fragment") + return parts + + +def _agent_card_discovery_url(endpoint: str) -> str: + parts = _endpoint_parts(endpoint) + base_path = parts.path.rstrip("/") + return urlunsplit( + ( + parts.scheme, + parts.netloc, + f"{base_path}{AGENT_CARD_WELL_KNOWN_PATH}", + "", + "", + ) + ) + + +def _endpoint_query_params(endpoint: str) -> dict[str, str]: + return dict(parse_qsl(_endpoint_parts(endpoint).query, keep_blank_values=True)) + + +def _origin(parts) -> tuple[str, str, int | None]: + return parts.scheme.lower(), (parts.hostname or "").lower(), parts.port + + +def _resolve_agent_card_rpc_url(card_url: str | None, endpoint: str) -> str: + endpoint_parts = _endpoint_parts(endpoint) + clean_endpoint = urlunsplit( + ( + endpoint_parts.scheme, + endpoint_parts.netloc, + endpoint_parts.path or "/", + "", + "", + ) + ) + if not card_url: + return clean_endpoint + + resolved_url = urljoin(clean_endpoint, card_url) + card_parts = _endpoint_parts(resolved_url) + if ( + card_parts.hostname in _LOOPBACK_HOSTS + and endpoint_parts.hostname not in _LOOPBACK_HOSTS + ): + resolved_url = urlunsplit( + ( + endpoint_parts.scheme, + endpoint_parts.netloc, + card_parts.path, + card_parts.query, + "", + ) + ) + return resolved_url + + +def _is_same_origin(first_url: str, second_url: str) -> bool: + return _origin(_endpoint_parts(first_url)) == _origin(_endpoint_parts(second_url)) + + class RemoteVeAgent(RemoteA2aAgent): """Connect to a remote agent on the VeFaaS platform. @@ -168,13 +236,16 @@ def __init__( f"Unsupported auth method {auth_method}, use `header` or `querystring` instead." ) + endpoint_query_params = _endpoint_query_params(effective_url) + discovery_params = {**endpoint_query_params, **req_params} agent_card_dict = requests.get( - effective_url + AGENT_CARD_WELL_KNOWN_PATH, + _agent_card_discovery_url(effective_url), headers=req_headers, - params=req_params, + params=discovery_params, ).json() - # replace agent_card_url with actual host - agent_card_dict["url"] = effective_url + agent_card_dict["url"] = _resolve_agent_card_rpc_url( + agent_card_dict.get("url"), effective_url + ) agent_card_object = _convert_agent_card_dict_to_obj(agent_card_dict) @@ -184,27 +255,31 @@ def __init__( client_to_use = httpx_client if client_was_provided: - # If a client was provided, update it with auth info - if auth_token: - if auth_method == "header": - client_to_use.headers.update(req_headers) - elif auth_method == "querystring": - new_params = dict(client_to_use.params) - new_params.update(req_params) - client_to_use.params = new_params + if auth_token and auth_method == "header": + client_to_use.headers.update(req_headers) else: - # If no client was provided, create a new one with auth info - if auth_token: - if auth_method == "header": - client_to_use = httpx.AsyncClient( - base_url=effective_url, headers=req_headers, timeout=600 - ) - elif auth_method == "querystring": - client_to_use = httpx.AsyncClient( - base_url=effective_url, params=req_params, timeout=600 - ) - else: # No auth, no client provided - client_to_use = httpx.AsyncClient(base_url=effective_url, timeout=600) + endpoint_parts = _endpoint_parts(effective_url) + clean_base_url = urlunsplit( + ( + endpoint_parts.scheme, + endpoint_parts.netloc, + endpoint_parts.path or "/", + "", + "", + ) + ) + client_to_use = httpx.AsyncClient( + base_url=clean_base_url, + headers=req_headers, + timeout=600, + ) + + new_params = dict(client_to_use.params) + new_params.update(req_params) + if _is_same_origin(agent_card_dict["url"], effective_url): + new_params.update(discovery_params) + if new_params: + client_to_use.params = new_params super().__init__( name=name, agent_card=agent_card_object, httpx_client=client_to_use @@ -327,9 +402,10 @@ async def _inject_auth_token(self, ctx: InvocationContext) -> None: return try: - from veadk.utils.auth import build_auth_config from google.adk.agents.callback_context import CallbackContext + from veadk.utils.auth import build_auth_config + # Inject TIP token via header workload_auth_config = build_auth_config( auth_method="apikey", diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 75059ab54..158790c8d 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -17,7 +17,6 @@ import json import os import time -from collections.abc import Iterable from typing import Optional from urllib import error, request from urllib.parse import urlsplit, urlunsplit @@ -31,7 +30,6 @@ ) 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 @@ -40,14 +38,10 @@ _SKILL_API_HEALTH_REQUEST_TIMEOUT = 5.0 -def _skill_api_upgrade_hint(path: str) -> str: - api_path = ( - "/v1/skills/stream" - if path.rstrip("/").endswith("/stream") - else "/v1/skills/execute" - ) +def _skill_api_upgrade_hint() -> str: return ( - f"提示：当前 Skill 沙箱镜像未实现 {api_path} 接口，可能是旧版沙箱镜像。" + "提示：当前 Skill 沙箱镜像未实现 /v1/skills/execute 接口，" + "可能是旧版沙箱镜像。" "请升级 Skill 沙箱镜像或切换到支持 Skill HTTP API 的新版沙箱。" ) @@ -89,11 +83,10 @@ def _post_skill_api_json( payload: dict[str, object], tip_token_key: str | None, timeout: int, - stream: bool, ) -> str: headers = { "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", + "Accept": "application/json", } if tip_token_key: headers["X-Tip-Token-Key"] = tip_token_key @@ -106,14 +99,11 @@ def _post_skill_api_json( ) try: with request.urlopen(req, timeout=timeout) as response: - if stream: - return _parse_skill_stream_response(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}. " - f"{_skill_api_upgrade_hint(path)}" + f"Skill HTTP API returned HTTP {exc.code}. {_skill_api_upgrade_hint()}" ) from exc detail = exc.read().decode("utf-8", errors="replace") raise RuntimeError( @@ -183,58 +173,11 @@ def _parse_skill_execute_response(raw: bytes) -> str: return json.dumps(payload, ensure_ascii=False) -def _parse_skill_stream_response(raw: bytes | Iterable[bytes]) -> str: - chunks: list[str] = [] - event_name = "message" - data_lines: list[str] = [] - - def flush_event() -> None: - nonlocal event_name, data_lines - if not data_lines: - event_name = "message" - return - data = "\n".join(data_lines) - try: - payload = json.loads(data) - except json.JSONDecodeError: - payload = {} - - if event_name == "error": - content = payload.get("content") if isinstance(payload, dict) else None - if isinstance(content, str): - raise RuntimeError(content) - raise RuntimeError(data) - if isinstance(payload, dict) and payload.get("type") == "text": - content = payload.get("content") - if isinstance(content, str): - chunks.append(content) - - event_name = "message" - data_lines = [] - - raw_lines = raw.splitlines() if isinstance(raw, bytes) else raw - for raw_line in raw_lines: - line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n") - if not line: - flush_event() - continue - if line.startswith(":"): - continue - if line.startswith("event:"): - event_name = line[len("event:") :].strip() - elif line.startswith("data:"): - data_lines.append(line[len("data:") :].strip()) - - flush_event() - return "".join(chunks) - - def _execute_skills_via_skill_api( *, workflow_prompt: str, tool_id: str, tool_context: ToolContext, - prefer_stream: bool, timeout: int, ) -> str: try: @@ -250,14 +193,12 @@ def _execute_skills_via_skill_api( f"AgentKit session endpoint is not available: {exc}" ) from exc _wait_for_skill_api_health(endpoint=endpoint) - path = "/v1/skills/stream" if prefer_stream else "/v1/skills/execute" return _post_skill_api_json( endpoint=endpoint, - path=path, + path="/v1/skills/execute", payload={"prompt": workflow_prompt}, tip_token_key=_tip_token_key(tool_context), timeout=timeout, - stream=prefer_stream, ) @@ -265,7 +206,6 @@ def execute_skills( workflow_prompt: str, tool_context: ToolContext = None, env_vars: Optional[dict[str, str]] = None, - prefer_stream: bool = False, ) -> str: """Execute skills in a sandbox and return the output. @@ -304,6 +244,5 @@ def execute_skills( workflow_prompt=workflow_prompt, tool_id=tool_id, tool_context=tool_context, - prefer_stream=prefer_stream, timeout=_SKILL_API_TIMEOUT, ) From 7956657be4b42d02208cee636bbfa0535b46f117 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 16:30:19 +0800 Subject: [PATCH 2/6] fix(a2a): harden remote agent card discovery --- tests/a2a/test_remote_ve_agent.py | 33 ++++++++++++++++++++++++++ veadk/a2a/remote_ve_agent.py | 39 ++++++++++++++++++++++++++++--- 2 files changed, 69 insertions(+), 3 deletions(-) diff --git a/tests/a2a/test_remote_ve_agent.py b/tests/a2a/test_remote_ve_agent.py index 9ed76d119..b9eb50049 100644 --- a/tests/a2a/test_remote_ve_agent.py +++ b/tests/a2a/test_remote_ve_agent.py @@ -3,6 +3,9 @@ import asyncio from unittest.mock import Mock, patch +import pytest +import requests + from veadk.a2a.remote_ve_agent import RemoteVeAgent @@ -49,6 +52,17 @@ def test_remote_agent_preserves_a2a_path_from_agent_card() -> None: _close(agent) +def test_remote_agent_discovers_root_card_when_url_is_a2a_rpc_path() -> None: + agent, get = _build_agent("https://sandbox.test/a2a", "https://sandbox.test/a2a") + try: + assert get.call_args.args[0] == ( + "https://sandbox.test/.well-known/agent-card.json" + ) + assert agent._agent_card.url == "https://sandbox.test/a2a" + finally: + _close(agent) + + def test_remote_agent_preserves_same_origin_session_authorization_query() -> None: agent, get = _build_agent( "https://sandbox.test/?faasInstanceName=inst&Authorization=key", @@ -104,3 +118,22 @@ def test_remote_agent_replaces_loopback_host_but_keeps_card_path() -> None: assert agent._agent_card.url == "https://sandbox.test/a2a" finally: _close(agent) + + +def test_remote_agent_raises_clear_error_for_agent_card_http_failure() -> None: + response = Mock() + response.status_code = 503 + response.reason = "Service Unavailable" + response.raise_for_status.side_effect = requests.HTTPError(response=response) + + with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response): + with pytest.raises( + RuntimeError, + match="Failed to fetch A2A Agent Card: HTTP 503 Service Unavailable", + ): + RemoteVeAgent(name="remote", url="https://sandbox.test") + + +def test_remote_agent_rejects_invalid_endpoint_scheme() -> None: + with pytest.raises(ValueError, match="Invalid A2A endpoint URL"): + RemoteVeAgent(name="remote", url="ftp://sandbox.test") diff --git a/veadk/a2a/remote_ve_agent.py b/veadk/a2a/remote_ve_agent.py index 7801ba541..c47853ec8 100644 --- a/veadk/a2a/remote_ve_agent.py +++ b/veadk/a2a/remote_ve_agent.py @@ -34,6 +34,7 @@ AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent-card.json" _LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) +_A2A_RPC_PATHS = frozenset({"/a2a", "/a2a/"}) def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: @@ -54,6 +55,8 @@ def _endpoint_parts(url: str): def _agent_card_discovery_url(endpoint: str) -> str: parts = _endpoint_parts(endpoint) base_path = parts.path.rstrip("/") + if parts.path in _A2A_RPC_PATHS: + base_path = "" return urlunsplit( ( parts.scheme, @@ -109,6 +112,36 @@ def _is_same_origin(first_url: str, second_url: str) -> bool: return _origin(_endpoint_parts(first_url)) == _origin(_endpoint_parts(second_url)) +def _fetch_agent_card_dict( + *, + discovery_url: str, + headers: dict[str, str], + params: dict[str, str], +) -> dict: + try: + response = requests.get(discovery_url, headers=headers, params=params) + response.raise_for_status() + except requests.HTTPError as exc: + status_code = ( + exc.response.status_code if exc.response is not None else "unknown" + ) + reason = exc.response.reason if exc.response is not None else "" + detail = f" {reason}" if reason else "" + raise RuntimeError( + f"Failed to fetch A2A Agent Card: HTTP {status_code}{detail}" + ) from exc + except requests.RequestException as exc: + raise RuntimeError(f"Failed to fetch A2A Agent Card: {exc}") from exc + + try: + agent_card_dict = response.json() + except ValueError as exc: + raise RuntimeError("Failed to parse A2A Agent Card response as JSON") from exc + if not isinstance(agent_card_dict, dict): + raise RuntimeError("A2A Agent Card response must be a JSON object") + return agent_card_dict + + class RemoteVeAgent(RemoteA2aAgent): """Connect to a remote agent on the VeFaaS platform. @@ -238,11 +271,11 @@ def __init__( endpoint_query_params = _endpoint_query_params(effective_url) discovery_params = {**endpoint_query_params, **req_params} - agent_card_dict = requests.get( - _agent_card_discovery_url(effective_url), + agent_card_dict = _fetch_agent_card_dict( + discovery_url=_agent_card_discovery_url(effective_url), headers=req_headers, params=discovery_params, - ).json() + ) agent_card_dict["url"] = _resolve_agent_card_rpc_url( agent_card_dict.get("url"), effective_url ) From b83c5869c3a30048cf2855121d5ab9591580fb6c Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 16:46:50 +0800 Subject: [PATCH 3/6] fix(a2a): minimize remote agent card discovery changes --- tests/a2a/test_remote_ve_agent.py | 5 -- veadk/a2a/remote_ve_agent.py | 133 ++++++++++++------------------ 2 files changed, 52 insertions(+), 86 deletions(-) diff --git a/tests/a2a/test_remote_ve_agent.py b/tests/a2a/test_remote_ve_agent.py index b9eb50049..d75bccece 100644 --- a/tests/a2a/test_remote_ve_agent.py +++ b/tests/a2a/test_remote_ve_agent.py @@ -132,8 +132,3 @@ def test_remote_agent_raises_clear_error_for_agent_card_http_failure() -> None: match="Failed to fetch A2A Agent Card: HTTP 503 Service Unavailable", ): RemoteVeAgent(name="remote", url="https://sandbox.test") - - -def test_remote_agent_rejects_invalid_endpoint_scheme() -> None: - with pytest.raises(ValueError, match="Invalid A2A endpoint URL"): - RemoteVeAgent(name="remote", url="ftp://sandbox.test") diff --git a/veadk/a2a/remote_ve_agent.py b/veadk/a2a/remote_ve_agent.py index c47853ec8..0179d2a36 100644 --- a/veadk/a2a/remote_ve_agent.py +++ b/veadk/a2a/remote_ve_agent.py @@ -34,7 +34,6 @@ AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent-card.json" _LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) -_A2A_RPC_PATHS = frozenset({"/a2a", "/a2a/"}) def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: @@ -43,19 +42,10 @@ def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: return agent_card_object -def _endpoint_parts(url: str): - parts = urlsplit(url) - if parts.scheme not in {"http", "https"} or not parts.hostname: - raise ValueError(f"Invalid A2A endpoint URL: {url}") - if parts.username or parts.password or parts.fragment: - raise ValueError("A2A endpoint URL must not contain userinfo or a fragment") - return parts - - def _agent_card_discovery_url(endpoint: str) -> str: - parts = _endpoint_parts(endpoint) + parts = urlsplit(endpoint) base_path = parts.path.rstrip("/") - if parts.path in _A2A_RPC_PATHS: + if base_path == "/a2a": base_path = "" return urlunsplit( ( @@ -69,29 +59,17 @@ def _agent_card_discovery_url(endpoint: str) -> str: def _endpoint_query_params(endpoint: str) -> dict[str, str]: - return dict(parse_qsl(_endpoint_parts(endpoint).query, keep_blank_values=True)) - - -def _origin(parts) -> tuple[str, str, int | None]: - return parts.scheme.lower(), (parts.hostname or "").lower(), parts.port + return dict(parse_qsl(urlsplit(endpoint).query, keep_blank_values=True)) def _resolve_agent_card_rpc_url(card_url: str | None, endpoint: str) -> str: - endpoint_parts = _endpoint_parts(endpoint) - clean_endpoint = urlunsplit( - ( - endpoint_parts.scheme, - endpoint_parts.netloc, - endpoint_parts.path or "/", - "", - "", - ) - ) + endpoint_parts = urlsplit(endpoint) + clean_endpoint = urlunsplit(endpoint_parts._replace(query="", fragment="")) if not card_url: return clean_endpoint resolved_url = urljoin(clean_endpoint, card_url) - card_parts = _endpoint_parts(resolved_url) + card_parts = urlsplit(resolved_url) if ( card_parts.hostname in _LOOPBACK_HOSTS and endpoint_parts.hostname not in _LOOPBACK_HOSTS @@ -109,37 +87,17 @@ def _resolve_agent_card_rpc_url(card_url: str | None, endpoint: str) -> str: def _is_same_origin(first_url: str, second_url: str) -> bool: - return _origin(_endpoint_parts(first_url)) == _origin(_endpoint_parts(second_url)) - - -def _fetch_agent_card_dict( - *, - discovery_url: str, - headers: dict[str, str], - params: dict[str, str], -) -> dict: - try: - response = requests.get(discovery_url, headers=headers, params=params) - response.raise_for_status() - except requests.HTTPError as exc: - status_code = ( - exc.response.status_code if exc.response is not None else "unknown" - ) - reason = exc.response.reason if exc.response is not None else "" - detail = f" {reason}" if reason else "" - raise RuntimeError( - f"Failed to fetch A2A Agent Card: HTTP {status_code}{detail}" - ) from exc - except requests.RequestException as exc: - raise RuntimeError(f"Failed to fetch A2A Agent Card: {exc}") from exc - - try: - agent_card_dict = response.json() - except ValueError as exc: - raise RuntimeError("Failed to parse A2A Agent Card response as JSON") from exc - if not isinstance(agent_card_dict, dict): - raise RuntimeError("A2A Agent Card response must be a JSON object") - return agent_card_dict + first = urlsplit(first_url) + second = urlsplit(second_url) + return ( + first.scheme.lower(), + (first.hostname or "").lower(), + first.port, + ) == ( + second.scheme.lower(), + (second.hostname or "").lower(), + second.port, + ) class RemoteVeAgent(RemoteA2aAgent): @@ -271,11 +229,23 @@ def __init__( endpoint_query_params = _endpoint_query_params(effective_url) discovery_params = {**endpoint_query_params, **req_params} - agent_card_dict = _fetch_agent_card_dict( - discovery_url=_agent_card_discovery_url(effective_url), + response = requests.get( + _agent_card_discovery_url(effective_url), headers=req_headers, params=discovery_params, ) + try: + response.raise_for_status() + except requests.HTTPError as exc: + status_code = ( + exc.response.status_code if exc.response is not None else "unknown" + ) + reason = exc.response.reason if exc.response is not None else "" + detail = f" {reason}" if reason else "" + raise RuntimeError( + f"Failed to fetch A2A Agent Card: HTTP {status_code}{detail}" + ) from exc + agent_card_dict = response.json() agent_card_dict["url"] = _resolve_agent_card_rpc_url( agent_card_dict.get("url"), effective_url ) @@ -288,30 +258,31 @@ def __init__( client_to_use = httpx_client if client_was_provided: - if auth_token and auth_method == "header": - client_to_use.headers.update(req_headers) + # If a client was provided, update it with auth info + if auth_token: + if auth_method == "header": + client_to_use.headers.update(req_headers) + elif auth_method == "querystring": + new_params = dict(client_to_use.params) + new_params.update(req_params) + client_to_use.params = new_params else: - endpoint_parts = _endpoint_parts(effective_url) - clean_base_url = urlunsplit( - ( - endpoint_parts.scheme, - endpoint_parts.netloc, - endpoint_parts.path or "/", - "", - "", - ) - ) - client_to_use = httpx.AsyncClient( - base_url=clean_base_url, - headers=req_headers, - timeout=600, - ) + # If no client was provided, create a new one with auth info + if auth_token: + if auth_method == "header": + client_to_use = httpx.AsyncClient( + base_url=effective_url, headers=req_headers, timeout=600 + ) + elif auth_method == "querystring": + client_to_use = httpx.AsyncClient( + base_url=effective_url, params=req_params, timeout=600 + ) + else: # No auth, no client provided + client_to_use = httpx.AsyncClient(base_url=effective_url, timeout=600) - new_params = dict(client_to_use.params) - new_params.update(req_params) if _is_same_origin(agent_card_dict["url"], effective_url): + new_params = dict(client_to_use.params) new_params.update(discovery_params) - if new_params: client_to_use.params = new_params super().__init__( From 481bd84754b13286a30caf35c434b3a748bf4d98 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 16:53:39 +0800 Subject: [PATCH 4/6] feat(agentkit): support skill sandbox invocation modes --- .../builtin_tools/test_run_sandbox_agent.py | 149 +++++++++++ veadk/tools/builtin_tools/execute_skills.py | 251 +++++++++++++++++- 2 files changed, 391 insertions(+), 9 deletions(-) diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index d692bdb82..b042402e4 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 json import sys import types import unittest @@ -348,6 +349,154 @@ def fake_run_sandbox_agent(**kwargs): 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" + + 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, + ) + + 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" + ) + + 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="run_sse", + ) + + 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_a2a_mode_posts_message_send_and_returns_text(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": "req", + "result": { + "kind": "message", + "role": "agent", + "parts": [{"kind": "text", "text": "a2a 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", + ) + + 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"]) + + 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: "", diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 158790c8d..b877ed617 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -17,6 +17,7 @@ import json import os import time +import uuid from typing import Optional from urllib import error, request from urllib.parse import urlsplit, urlunsplit @@ -36,6 +37,10 @@ _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"} +) def _skill_api_upgrade_hint() -> str: @@ -76,6 +81,20 @@ 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 "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." + ) + return "execute" if normalized == "skill_api" else normalized + + def _post_skill_api_json( *, endpoint: str, @@ -115,6 +134,34 @@ def _post_skill_api_json( ) from exc +def _post_json( + *, + endpoint: str, + path: str, + payload: dict[str, object], + timeout: int, + accept: str = "application/json", +) -> bytes: + req = request.Request( + _skill_api_url(endpoint, path), + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json", "Accept": accept}, + method="POST", + ) + try: + with request.urlopen(req, timeout=timeout) as response: + return response.read() + except error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError( + f"AgentKit Skill {path} request failed with HTTP {exc.code}: {detail}" + ) from exc + except error.URLError as exc: + raise RuntimeError( + f"AgentKit Skill {path} endpoint is not reachable: {exc.reason}" + ) from exc + + def _wait_for_skill_api_health( *, endpoint: str, @@ -173,12 +220,175 @@ def _parse_skill_execute_response(raw: bytes) -> str: 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 "" + chunks: list[str] = [] + for part in parts: + if not isinstance(part, dict): + continue + text = part.get("text") + if isinstance(text, str): + chunks.append(text) + continue + text_part = part.get("textPart") + if isinstance(text_part, dict) and isinstance(text_part.get("text"), str): + chunks.append(text_part["text"]) + return "".join(chunks) + + +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 _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: + 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, + 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}, + }, + } + 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) + + +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: + 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( @@ -192,6 +402,22 @@ def _execute_skills_via_skill_api( raise RuntimeError( 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( endpoint=endpoint, @@ -206,6 +432,7 @@ def execute_skills( workflow_prompt: str, tool_context: ToolContext = None, env_vars: Optional[dict[str, str]] = None, + invocation_mode: Optional[str] = None, ) -> str: """Execute skills in a sandbox and return the output. @@ -216,6 +443,9 @@ 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. + 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. Returns: str: The output of the code execution. @@ -225,19 +455,21 @@ def execute_skills( tool_id = resolve_agentkit_tool_id("AGENTKIT_TOOL_ID_SKILLS") if env_vars: - account_id = get_agentkit_account_id(tool_context.state) - extra_env_vars = dict(env_vars) - if account_id: - extra_env_vars.setdefault( - "TOS_SKILLS_DIR", - f"tos://agentkit-platform-{account_id}/skills/", - ) - return run_sandbox_agent( + 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, - extra_env_vars=extra_env_vars, ) return _execute_skills_via_skill_api( @@ -245,4 +477,5 @@ def execute_skills( tool_id=tool_id, tool_context=tool_context, timeout=_SKILL_API_TIMEOUT, + invocation_mode=mode, ) From 743abc62f29f7b57e8a2bc0a2862eafcd3a20b46 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 16:57:05 +0800 Subject: [PATCH 5/6] docs(a2a): add comments for skill sandbox invocation --- veadk/a2a/remote_ve_agent.py | 3 +++ veadk/tools/builtin_tools/execute_skills.py | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/veadk/a2a/remote_ve_agent.py b/veadk/a2a/remote_ve_agent.py index 0179d2a36..a20edef08 100644 --- a/veadk/a2a/remote_ve_agent.py +++ b/veadk/a2a/remote_ve_agent.py @@ -45,6 +45,7 @@ def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: def _agent_card_discovery_url(endpoint: str) -> str: parts = urlsplit(endpoint) base_path = parts.path.rstrip("/") + # AgentKit Skill 沙箱的 RPC 地址通常是 /a2a，但 Agent Card 暴露在根路径。 if base_path == "/a2a": base_path = "" return urlunsplit( @@ -68,6 +69,7 @@ def _resolve_agent_card_rpc_url(card_url: str | None, endpoint: str) -> str: if not card_url: return clean_endpoint + # 保留 Agent Card 声明的 RPC path，避免把 /a2a 覆盖成沙箱根地址。 resolved_url = urljoin(clean_endpoint, card_url) card_parts = urlsplit(resolved_url) if ( @@ -281,6 +283,7 @@ def __init__( client_to_use = httpx.AsyncClient(base_url=effective_url, timeout=600) if _is_same_origin(agent_card_dict["url"], effective_url): + # 同源 RPC 继续携带 session query 鉴权；跨域 Card URL 不透传用户查询参数。 new_params = dict(client_to_use.params) new_params.update(discovery_params) client_to_use.params = new_params diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index b877ed617..cd19749f7 100644 --- a/veadk/tools/builtin_tools/execute_skills.py +++ b/veadk/tools/builtin_tools/execute_skills.py @@ -82,6 +82,7 @@ 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" @@ -311,6 +312,7 @@ def _execute_skills_via_run_sse( tool_context: ToolContext, timeout: int, ) -> str: + # run_sse 复用 ADK 运行入口，适配只暴露 ADK Runtime 接口的 Skill 沙箱。 raw = _post_json( endpoint=endpoint, path="/run_sse", @@ -329,6 +331,7 @@ 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, @@ -366,6 +369,7 @@ def _execute_skills_via_python_agent( 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: @@ -455,6 +459,7 @@ def execute_skills( 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, From 0c28f17515344e6092f38059dd7cb1a7a456a870 Mon Sep 17 00:00:00 2001 From: "lixuefei.nice" Date: Wed, 5 Aug 2026 17:45:53 +0800 Subject: [PATCH 6/6] feat: del useless code --- tests/a2a/test_remote_ve_agent.py | 134 ------------------------------ veadk/a2a/remote_ve_agent.py | 109 +++--------------------- 2 files changed, 13 insertions(+), 230 deletions(-) delete mode 100644 tests/a2a/test_remote_ve_agent.py diff --git a/tests/a2a/test_remote_ve_agent.py b/tests/a2a/test_remote_ve_agent.py deleted file mode 100644 index d75bccece..000000000 --- a/tests/a2a/test_remote_ve_agent.py +++ /dev/null @@ -1,134 +0,0 @@ -from __future__ import annotations - -import asyncio -from unittest.mock import Mock, patch - -import pytest -import requests - -from veadk.a2a.remote_ve_agent import RemoteVeAgent - - -def _agent_card(url: str) -> dict: - return { - "name": "skill-agent", - "description": "Skill sandbox agent", - "url": url, - "version": "1.0.0", - "capabilities": {"streaming": True}, - "defaultInputModes": ["text"], - "defaultOutputModes": ["text"], - "skills": [ - { - "id": "chat", - "name": "chat", - "description": "Chat with the skill agent", - "tags": ["chat"], - } - ], - } - - -def _build_agent(endpoint: str, card_url: str) -> tuple[RemoteVeAgent, Mock]: - response = Mock() - response.json.return_value = _agent_card(card_url) - with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response) as get: - agent = RemoteVeAgent(name="remote", url=endpoint) - return agent, get - - -def _close(agent: RemoteVeAgent) -> None: - asyncio.run(agent._httpx_client.aclose()) - - -def test_remote_agent_preserves_a2a_path_from_agent_card() -> None: - agent, get = _build_agent("https://sandbox.test", "https://sandbox.test/a2a") - try: - assert agent._agent_card.url == "https://sandbox.test/a2a" - assert get.call_args.args[0] == ( - "https://sandbox.test/.well-known/agent-card.json" - ) - finally: - _close(agent) - - -def test_remote_agent_discovers_root_card_when_url_is_a2a_rpc_path() -> None: - agent, get = _build_agent("https://sandbox.test/a2a", "https://sandbox.test/a2a") - try: - assert get.call_args.args[0] == ( - "https://sandbox.test/.well-known/agent-card.json" - ) - assert agent._agent_card.url == "https://sandbox.test/a2a" - finally: - _close(agent) - - -def test_remote_agent_preserves_same_origin_session_authorization_query() -> None: - agent, get = _build_agent( - "https://sandbox.test/?faasInstanceName=inst&Authorization=key", - "https://sandbox.test/a2a", - ) - try: - assert get.call_args.args[0] == ( - "https://sandbox.test/.well-known/agent-card.json" - ) - assert get.call_args.kwargs["params"] == { - "faasInstanceName": "inst", - "Authorization": "key", - } - assert dict(agent._httpx_client.params) == { - "faasInstanceName": "inst", - "Authorization": "key", - } - finally: - _close(agent) - - -def test_remote_agent_does_not_forward_session_query_to_different_origin() -> None: - agent, _ = _build_agent( - "https://sandbox.test/?Authorization=key", - "https://different.test/a2a", - ) - try: - assert agent._agent_card.url == "https://different.test/a2a" - assert "Authorization" not in dict(agent._httpx_client.params) - finally: - _close(agent) - - -def test_remote_agent_keeps_explicit_query_auth_for_different_origin() -> None: - response = Mock() - response.json.return_value = _agent_card("https://different.test/a2a") - with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response): - agent = RemoteVeAgent( - name="remote", - url="https://sandbox.test", - auth_token="explicit-token", - auth_method="querystring", - ) - try: - assert dict(agent._httpx_client.params) == {"token": "explicit-token"} - finally: - _close(agent) - - -def test_remote_agent_replaces_loopback_host_but_keeps_card_path() -> None: - agent, _ = _build_agent("https://sandbox.test", "http://localhost:8000/a2a") - try: - assert agent._agent_card.url == "https://sandbox.test/a2a" - finally: - _close(agent) - - -def test_remote_agent_raises_clear_error_for_agent_card_http_failure() -> None: - response = Mock() - response.status_code = 503 - response.reason = "Service Unavailable" - response.raise_for_status.side_effect = requests.HTTPError(response=response) - - with patch("veadk.a2a.remote_ve_agent.requests.get", return_value=response): - with pytest.raises( - RuntimeError, - match="Failed to fetch A2A Agent Card: HTTP 503 Service Unavailable", - ): - RemoteVeAgent(name="remote", url="https://sandbox.test") diff --git a/veadk/a2a/remote_ve_agent.py b/veadk/a2a/remote_ve_agent.py index a20edef08..1eddedaa9 100644 --- a/veadk/a2a/remote_ve_agent.py +++ b/veadk/a2a/remote_ve_agent.py @@ -12,28 +12,27 @@ # See the License for the specific language governing permissions and # limitations under the License. -import functools import json +import functools from typing import AsyncGenerator, Literal, Optional -from urllib.parse import parse_qsl, urljoin, urlsplit, urlunsplit +from a2a.client.base_client import BaseClient import httpx import requests -from a2a.client.base_client import BaseClient from a2a.types import AgentCard -from google.adk.agents.invocation_context import InvocationContext from google.adk.agents.remote_a2a_agent import RemoteA2aAgent -from google.adk.events.event import Event -from google.adk.utils.context_utils import Aclosing from veadk.integrations.ve_identity.utils import generate_headers from veadk.utils.auth import VE_TIP_TOKEN_CREDENTIAL_KEY, VE_TIP_TOKEN_HEADER from veadk.utils.logger import get_logger +from google.adk.utils.context_utils import Aclosing +from google.adk.events.event import Event +from google.adk.agents.invocation_context import InvocationContext + logger = get_logger(__name__) AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent-card.json" -_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: @@ -42,66 +41,6 @@ def _convert_agent_card_dict_to_obj(agent_card_dict: dict) -> AgentCard: return agent_card_object -def _agent_card_discovery_url(endpoint: str) -> str: - parts = urlsplit(endpoint) - base_path = parts.path.rstrip("/") - # AgentKit Skill 沙箱的 RPC 地址通常是 /a2a，但 Agent Card 暴露在根路径。 - if base_path == "/a2a": - base_path = "" - return urlunsplit( - ( - parts.scheme, - parts.netloc, - f"{base_path}{AGENT_CARD_WELL_KNOWN_PATH}", - "", - "", - ) - ) - - -def _endpoint_query_params(endpoint: str) -> dict[str, str]: - return dict(parse_qsl(urlsplit(endpoint).query, keep_blank_values=True)) - - -def _resolve_agent_card_rpc_url(card_url: str | None, endpoint: str) -> str: - endpoint_parts = urlsplit(endpoint) - clean_endpoint = urlunsplit(endpoint_parts._replace(query="", fragment="")) - if not card_url: - return clean_endpoint - - # 保留 Agent Card 声明的 RPC path，避免把 /a2a 覆盖成沙箱根地址。 - resolved_url = urljoin(clean_endpoint, card_url) - card_parts = urlsplit(resolved_url) - if ( - card_parts.hostname in _LOOPBACK_HOSTS - and endpoint_parts.hostname not in _LOOPBACK_HOSTS - ): - resolved_url = urlunsplit( - ( - endpoint_parts.scheme, - endpoint_parts.netloc, - card_parts.path, - card_parts.query, - "", - ) - ) - return resolved_url - - -def _is_same_origin(first_url: str, second_url: str) -> bool: - first = urlsplit(first_url) - second = urlsplit(second_url) - return ( - first.scheme.lower(), - (first.hostname or "").lower(), - first.port, - ) == ( - second.scheme.lower(), - (second.hostname or "").lower(), - second.port, - ) - - class RemoteVeAgent(RemoteA2aAgent): """Connect to a remote agent on the VeFaaS platform. @@ -229,28 +168,13 @@ def __init__( f"Unsupported auth method {auth_method}, use `header` or `querystring` instead." ) - endpoint_query_params = _endpoint_query_params(effective_url) - discovery_params = {**endpoint_query_params, **req_params} - response = requests.get( - _agent_card_discovery_url(effective_url), + agent_card_dict = requests.get( + effective_url + AGENT_CARD_WELL_KNOWN_PATH, headers=req_headers, - params=discovery_params, - ) - try: - response.raise_for_status() - except requests.HTTPError as exc: - status_code = ( - exc.response.status_code if exc.response is not None else "unknown" - ) - reason = exc.response.reason if exc.response is not None else "" - detail = f" {reason}" if reason else "" - raise RuntimeError( - f"Failed to fetch A2A Agent Card: HTTP {status_code}{detail}" - ) from exc - agent_card_dict = response.json() - agent_card_dict["url"] = _resolve_agent_card_rpc_url( - agent_card_dict.get("url"), effective_url - ) + params=req_params, + ).json() + # replace agent_card_url with actual host + agent_card_dict["url"] = effective_url agent_card_object = _convert_agent_card_dict_to_obj(agent_card_dict) @@ -282,12 +206,6 @@ def __init__( else: # No auth, no client provided client_to_use = httpx.AsyncClient(base_url=effective_url, timeout=600) - if _is_same_origin(agent_card_dict["url"], effective_url): - # 同源 RPC 继续携带 session query 鉴权；跨域 Card URL 不透传用户查询参数。 - new_params = dict(client_to_use.params) - new_params.update(discovery_params) - client_to_use.params = new_params - super().__init__( name=name, agent_card=agent_card_object, httpx_client=client_to_use ) @@ -409,9 +327,8 @@ async def _inject_auth_token(self, ctx: InvocationContext) -> None: return try: - from google.adk.agents.callback_context import CallbackContext - from veadk.utils.auth import build_auth_config + from google.adk.agents.callback_context import CallbackContext # Inject TIP token via header workload_auth_config = build_auth_config(