feat(agents): add SkillSandboxAgent

tests/agents/test_skill_sandbox_agent.py

@@ -0,0 +1,186 @@
+# 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.
+
+
+"""Public SkillSandboxAgent API tests."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestServer
+from google.adk.models.base_llm import BaseLlm
+from google.adk.models.llm_response import LlmResponse
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+from google.genai import types
+from pydantic import PrivateAttr
+
+from veadk import Agent, SkillSandboxAgent
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+class ParentModel(BaseLlm):
+    model: str = "offline-parent"
+    _calls: int = PrivateAttr(default=0)
+
+    async def generate_content_async(self, llm_request, stream=False):
+        self._calls += 1
+        yield LlmResponse(
+            content=types.Content(
+                role="model",
+                parts=[
+                    types.Part(
+                        function_call=types.FunctionCall(
+                            name="transfer_to_agent", args={"agent_name": "sandbox"}
+                        )
+                    )
+                ],
+            )
+        )
+
+
+async def make_runner(child):
+    root = Agent(
+        name="coordinator",
+        model=ParentModel(),
+        model_api_key="fixture",
+        sub_agents=[child],
+    )
+    sessions = InMemorySessionService()
+    await sessions.create_session(
+        app_name="probe", user_id="user", session_id="session"
+    )
+    return Runner(agent=root, app_name="probe", session_service=sessions)
+
+
+async def collect(runner):
+    result = []
+    async for event in runner.run_async(
+        user_id="user",
+        session_id="session",
+        new_message=types.Content(
+            role="user",
+            parts=[types.Part(text="List the available skills")],
+        ),
+    ):
+        result.append(event)
+    return result
+
+
+def test_skill_sandbox_agent_is_exported_and_defaults_to_skill():
+    agent = SkillSandboxAgent(name="sandbox")
+
+    assert agent.tool_type == "Skill"
+    assert agent.tool_id_env == "AGENTKIT_TOOL_ID_SKILLS"
+
+
+@pytest.mark.anyio
+async def test_skill_sandbox_agent_resolves_skills_tool_env_by_default():
+    agent = SkillSandboxAgent(name="sandbox")
+    ctx = SimpleNamespace(
+        app_name="app", user_id="user", session=SimpleNamespace(state={})
+    )
+    lease = SimpleNamespace(
+        session_id="physical", select_endpoint=lambda **kw: "https://example.test"
+    )
+
+    with (
+        patch(
+            "veadk.agents.agentkit_remote_sandbox_agent.resolve_agentkit_tool_id",
+            return_value="skills-tool",
+        ) as resolve_tool_id,
+        patch(
+            "veadk.agents.agentkit_remote_sandbox_agent.ensure_agentkit_session_lease",
+            return_value=lease,
+        ),
+    ):
+        kind, _, _, _ = await agent._resolve(ctx, "logical")
+
+    assert kind == "Skill"
+    resolve_tool_id.assert_called_once_with("AGENTKIT_TOOL_ID_SKILLS")
+
+
+@pytest.mark.anyio
+async def test_skill_sandbox_agent_sends_guarded_user_task_to_a2a():
+    received_messages = []
+
+    def task(state="working"):
+        result = {
+            "kind": "task",
+            "id": "a2a-task",
+            "contextId": "remote-context",
+            "status": {"state": state},
+        }
+        if state == "completed":
+            result["artifacts"] = [
+                {
+                    "artifactId": "answer",
+                    "parts": [{"kind": "text", "text": "Guarded task accepted"}],
+                }
+            ]
+        return result
+
+    async def handle(request):
+        if request.method == "GET":
+            return web.json_response(
+                {
+                    "name": "skill",
+                    "description": "fixture",
+                    "version": "1",
+                    "url": "http://127.0.0.1:1/a2a",
+                    "capabilities": {"streaming": False},
+                    "defaultInputModes": ["text/plain"],
+                    "defaultOutputModes": ["text/plain"],
+                    "skills": [],
+                }
+            )
+        body = await request.json()
+        if body["method"].startswith("message/"):
+            received_messages.append(body["params"]["message"])
+
+        def reply(result):
+            return {"jsonrpc": "2.0", "id": body["id"], "result": result}
+
+        if body["method"] == "tasks/get":
+            return web.json_response(reply(task("completed")))
+        return web.json_response(reply(task()))
+
+    app = web.Application()
+    app.router.add_route("*", "/{path:.*}", handle)
+    async with TestServer(app) as server:
+        runner = await make_runner(
+            SkillSandboxAgent(
+                name="sandbox",
+                guardrails="Only use bash_tool.",
+                endpoint=str(server.make_url("/?route=fixture")),
+            )
+        )
+        events = await collect(runner)
+
+    assert not [e.error_message for e in events if e.error_message]
+    assert received_messages
+    submitted_text = "\n".join(
+        part["text"]
+        for part in received_messages[0]["parts"]
+        if part.get("kind") == "text"
+    )
+    assert "Only use bash_tool." in submitted_text
+    assert "用户请求 / User request:" in submitted_text
+    assert "List the available skills" in submitted_text

veadk/__init__.py

@@ -18,6 +18,7 @@
 
 if TYPE_CHECKING:
     from veadk.agent import Agent, ModelFallbackEndpoint
+    from veadk.agents.skill_sandbox_agent import SkillSandboxAgent
     from veadk.runner import Runner
 
 
@@ -41,6 +42,10 @@ def __getattr__(name):
         )
 
         return AgentkitRemoteSandboxAgent
+    if name == "SkillSandboxAgent":
+        from veadk.agents.skill_sandbox_agent import SkillSandboxAgent
+
+        return SkillSandboxAgent
     raise AttributeError(f"module 'veadk' has no attribute '{name}'")
 
 
@@ -49,5 +54,6 @@ def __getattr__(name):
     "ModelFallbackEndpoint",
     "Runner",
     "AgentkitRemoteSandboxAgent",
+    "SkillSandboxAgent",
     "VERSION",
 ]

veadk/agents/agentkit_remote_sandbox_agent.py

@@ -55,6 +55,7 @@ class AgentkitRemoteSandboxAgent(BaseAgent):
     """
 
     tool_id: str | None = None
+    tool_id_env: str | None = None
     tool_type: Literal["Skill", "CodeEnv"] | None = None
     request_timeout: float = Field(default=900, gt=0, lt=86000)
     expiry_buffer: float = Field(default=90, ge=0, lt=86400)
@@ -90,7 +91,11 @@ def _discover_type(self, tool_id, state):
     async def _resolve(self, ctx, logical):
         if self.endpoint:
             return self.tool_type, self.endpoint, binding_key(self.endpoint), None
-        tool_id = self.tool_id or resolve_agentkit_tool_id()
+        tool_id = self.tool_id or (
+            resolve_agentkit_tool_id(self.tool_id_env)
+            if self.tool_id_env
+            else resolve_agentkit_tool_id()
+        )
         state = dict(ctx.session.state)
         kind = self.tool_type
         if kind is None:
@@ -154,6 +159,12 @@ async def _headers(self, ctx):
                     headers["inbound_auth"] = token
         return headers
 
+    def _task_from_content(self, content):
+        return "\n".join(p.text for p in content.parts if p.text)
+
+    def _prepare_task(self, task):
+        return task
+
     async def _run_async_impl(self, ctx):
         from veadk.agents._remote_sandbox.code import code_events
         from veadk.agents._remote_sandbox.skill import skill_events
@@ -193,7 +204,7 @@ async def _run_async_impl(self, ctx):
                 content = user_event.content if user_event else ctx.user_content
                 if not content or not content.parts:
                     raise SandboxAgentError("A user task is required")
-                task = "\n".join(p.text for p in content.parts if p.text)
+                task = self._prepare_task(self._task_from_content(content))
                 if not task.strip():
                     raise SandboxAgentError(
                         "This sandbox entry currently requires a text task"

veadk/agents/skill_sandbox_agent.py

@@ -0,0 +1,45 @@
+# 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.
+
+
+"""Public Skill Sandbox agent backed by AgentKit Skill A2A sessions."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import Field
+
+from veadk.agents.agentkit_remote_sandbox_agent import AgentkitRemoteSandboxAgent
+
+
+class SkillSandboxAgent(AgentkitRemoteSandboxAgent):
+    """Delegate an ADK turn directly to an AgentKit Skills Sandbox session."""
+
+    tool_type: Literal["Skill"] = "Skill"
+    tool_id_env: str | None = "AGENTKIT_TOOL_ID_SKILLS"
+    guardrails: str = ""
+    timeout: float | None = Field(default=None, gt=0, lt=86000, exclude=True)
+
+    def model_post_init(self, context):
+        if self.timeout is not None:
+            self.request_timeout = self.timeout
+            self.ttl = max(int(self.timeout), self.ttl)
+        super().model_post_init(context)
+
+    def _prepare_task(self, task):
+        guardrails = self.guardrails.strip()
+        if not guardrails:
+            return task
+        return f"{guardrails}\n\n用户请求 / User request:\n{task}"