WIP: feat(skill sandbox): runtime tip transfer support

docs/content/docs/framework/tools/builtin.en.mdx

@@ -530,6 +530,15 @@ Environment variables:
 - `AGENTKIT_TOOL_SERVICE_CODE`: ServiceCode for calling AgentKit Tools
 - `AGENTKIT_TOOL_REGION`: region for calling AgentKit Tools; defaults to `cn-beijing`
 
+TIP token key propagation:
+
+- Runtime requests support the `X-Tip-Token-Key` header, falling back to
+  `auth.tip_token_key`; VeADK forwards it only for the current tool call and
+  does not reuse historical session state.
+- The legacy RunCode path injects `TIP_TOKEN_KEY` for the current execution; do
+  not store token keys in `agentkit.yaml`, `runtime_envs`, or other persistent
+  config. Skill Sandbox does not need to install `agentkit-cli`.
+
 Additional `config.yaml` keys:
 
 ```yaml title="config.yaml"

docs/content/docs/framework/tools/builtin.mdx

@@ -526,6 +526,13 @@ if __name__ == "__main__":
 - `AGENTKIT_TOOL_SERVICE_CODE`：调用 AgentKit Tools 的 ServiceCode
 - `AGENTKIT_TOOL_REGION`：调用 AgentKit Tools 的地域，默认 `cn-beijing`
 
+TIP token key 透传：
+
+- Runtime 请求支持 `X-Tip-Token-Key` 请求头，兜底读取 `auth.tip_token_key`；
+  VeADK 只在本轮工具调用中透传，不复用历史 session。
+- legacy RunCode 路径会为本次执行注入 `TIP_TOKEN_KEY`；不要写入 `agentkit.yaml`、
+  `runtime_envs` 等持久化配置，也不需要在 Skill Sandbox 内安装 `agentkit-cli`。
+
 新增的 `config.yaml` 配置项：
 
 ```yaml title="config.yaml"

tests/integrations/agentkit/test_app.py

@@ -182,6 +182,87 @@ def fake_runner(**kwargs: Any) -> object:
     assert isinstance(plugins[0], FrontendInvocationPlugin)
 
 
+def test_tip_token_key_metadata_is_request_scoped_and_preserves_other_metadata() -> (
+    None
+):
+    metadata = agentkit_app._custom_metadata_with_tip_token_key(
+        {"keep": "value", "tip_token_key": "stale"},
+        "fresh-key",
+    )
+
+    assert metadata == {"keep": "value", "tip_token_key": "fresh-key"}
+
+
+def test_missing_tip_token_key_removes_stale_metadata() -> None:
+    metadata = agentkit_app._custom_metadata_with_tip_token_key(
+        {"keep": "value", "tip_token_key": "stale"},
+        None,
+    )
+
+    assert metadata == {"keep": "value"}
+
+
+def test_run_route_passes_tip_token_key_as_request_scoped_metadata(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured_runs: list[dict[str, Any]] = []
+
+    class FakeRunner:
+        auto_create_session = True
+
+        async def run_async(self, **kwargs: Any):
+            captured_runs.append(kwargs)
+            yield {"ok": True}
+
+    app = FastAPI()
+    app._tmpl_attrs = {  # type: ignore[attr-defined]
+        "session_service": object(),
+        "auto_create_session": True,
+    }
+    monkeypatch.setattr(
+        agentkit_app, "_dynamic_runner", lambda *_args, **_kwargs: FakeRunner()
+    )
+    agentkit_app._configure_dynamic_a2a_routes(app, _root_agent())
+
+    payload = {
+        "app_name": "agent",
+        "user_id": "user",
+        "session_id": "session",
+        "new_message": {"role": "user", "parts": [{"text": "hello"}]},
+        "custom_metadata": {
+            "keep": "value",
+            "tip_token_key": "stale-metadata",
+            "TIP_TOKEN_KEY": "stale-state-key",
+        },
+        "auth": {"tip_token_key": "body-key"},
+    }
+    client = TestClient(app)
+
+    with_header = client.post(
+        "/run",
+        json=payload,
+        headers={"X-Tip-Token-Key": "header-key"},
+    )
+    with_body_only = client.post(
+        "/run",
+        json=payload,
+    )
+    without_token = client.post(
+        "/run",
+        json={**payload, "auth": {}},
+    )
+
+    assert with_header.status_code == 200
+    assert with_body_only.status_code == 200
+    assert without_token.status_code == 200
+    first_metadata = captured_runs[0]["run_config"].custom_metadata
+    second_metadata = captured_runs[1]["run_config"].custom_metadata
+    third_metadata = captured_runs[2]["run_config"].custom_metadata
+    assert first_metadata == {"keep": "value", "tip_token_key": "header-key"}
+    assert second_metadata == {"keep": "value", "tip_token_key": "body-key"}
+    assert third_metadata == {"keep": "value"}
+
+
 def test_agent_info_exposes_mounted_skills_and_components() -> None:
     root_agent = _root_agent()
     setattr(

tests/tools/builtin_tools/test_run_sandbox_agent.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import importlib.util
+import os
 import sys
 import types
 import unittest
@@ -111,6 +112,10 @@ def _load_execute_skills_module(
     fake_runner.run_sandbox_agent = run_sandbox_agent
     fake_utils = types.ModuleType("veadk.utils")
     fake_utils.__path__ = []  # type: ignore[attr-defined]
+    fake_auth = types.ModuleType("veadk.utils.auth")
+    fake_auth.TIP_TOKEN_KEY_ENV = "TIP_TOKEN_KEY"
+    fake_auth.TIP_TOKEN_KEY_METADATA_KEY = "tip_token_key"
+    fake_auth.TIP_TOKEN_KEY_STATE_KEY = "TIP_TOKEN_KEY"
     fake_logger = types.ModuleType("veadk.utils.logger")
     fake_logger.get_logger = lambda _name: types.SimpleNamespace(
         debug=lambda *_args, **_kwargs: None,
@@ -128,6 +133,7 @@ def _load_execute_skills_module(
         "veadk.tools.builtin_tools._agentkit": fake_agentkit,
         "veadk.tools.builtin_tools.run_sandbox_agent": fake_runner,
         "veadk.utils": fake_utils,
+        "veadk.utils.auth": fake_auth,
         "veadk.utils.logger": fake_logger,
     }
 
@@ -199,16 +205,78 @@ def test_runner_code_overrides_the_sandbox_process_environment(self):
         self.assertNotIn("if key not in env", code)
         self.assertIn('srv_pythonpath = env.get("SRV_PYTHONPATH")', code)
 
+    def test_redacts_tip_token_key_from_echoed_sandbox_response(self):
+        response = {
+            "Result": {"Result": "runner code contains {'TIP_TOKEN_KEY': 'secret-key'}"}
+        }
+
+        redacted = self.module._redact_sensitive_env_values(
+            response,
+            {"TIP_TOKEN_KEY": "secret-key", "CUSTOM_VALUE": "visible"},
+        )
+
+        self.assertNotIn("secret-key", str(redacted))
+        self.assertIn("<redacted>", str(redacted))
+
+    def test_redacts_tip_token_key_from_success_output(self):
+        tool_context = types.SimpleNamespace(
+            _invocation_context=types.SimpleNamespace(
+                session=types.SimpleNamespace(id="session"),
+                agent=types.SimpleNamespace(name="agent"),
+                user_id="user",
+            ),
+            state={},
+        )
+        response = {
+            "Result": {
+                "Result": (
+                    '{"success": true, "data": {"outputs": '
+                    '[{"text": "printed secret-key"}]}}'
+                )
+            }
+        }
+
+        with patch.object(
+            self.module,
+            "invoke_agentkit_run_code",
+            return_value=response,
+        ):
+            result = self.module.run_sandbox_agent(
+                workflow_prompt="do work",
+                tool_id="tool",
+                tool_context=tool_context,
+                extra_env_vars={"TIP_TOKEN_KEY": "secret-key"},
+            )
+
+        self.assertEqual(result, "printed <redacted>")
+
 
 class TestExecuteSkillsSkillApi(unittest.TestCase):
     def _tool_context(self):
         invocation_context = types.SimpleNamespace(
             session=types.SimpleNamespace(id="session-1"),
             agent=types.SimpleNamespace(name="agent"),
             user_id="user",
+            run_config=types.SimpleNamespace(
+                custom_metadata={"tip_token_key": "tip-from-metadata"}
+            ),
         )
         return types.SimpleNamespace(
-            state={"TIP_TOKEN_KEY": "tip-from-state"},
+            state={},
+            run_config=invocation_context.run_config,
+            _invocation_context=invocation_context,
+        )
+
+    def _tool_context_without_request_token(self):
+        invocation_context = types.SimpleNamespace(
+            session=types.SimpleNamespace(id="session-1"),
+            agent=types.SimpleNamespace(name="agent"),
+            user_id="user",
+            run_config=types.SimpleNamespace(custom_metadata={}),
+        )
+        return types.SimpleNamespace(
+            state={"TIP_TOKEN_KEY": "stale-session-key"},
+            run_config=invocation_context.run_config,
             _invocation_context=invocation_context,
         )
 
@@ -251,9 +319,80 @@ def fake_urlopen(request, timeout=None):
         self.assertEqual("https://sandbox.test/v1/skills/execute", request_obj.full_url)
         self.assertEqual(900, timeout)
         self.assertEqual("POST", request_obj.get_method())
-        self.assertEqual("tip-from-state", request_obj.headers["X-tip-token-key"])
+        self.assertEqual("tip-from-metadata", request_obj.headers["X-tip-token-key"])
         self.assertIn(b'"prompt": "do work"', request_obj.data)
 
+    def test_skill_api_does_not_reuse_stale_session_state_tip_token_key(self):
+        captured_requests = []
+
+        class FakeResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return None
+
+            def read(self):
+                return b'{"content": "api result"}'
+
+        module = _load_execute_skills_module(
+            ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test",
+        )
+
+        with patch.object(
+            module.request,
+            "urlopen",
+            lambda request, **_kwargs: captured_requests.append(request)
+            or FakeResponse(),
+        ):
+            result = module.execute_skills(
+                "do work",
+                tool_context=self._tool_context_without_request_token(),
+            )
+
+        self.assertEqual(result, "api result")
+        self.assertNotIn(
+            "X-tip-token-key",
+            captured_requests[0].headers,
+        )
+
+    def test_skill_api_ignores_process_env_tip_token_key_by_default(self):
+        captured_requests = []
+
+        class FakeResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return None
+
+            def read(self):
+                return b'{"content": "api result"}'
+
+        module = _load_execute_skills_module(
+            ensure_agentkit_session_endpoint=lambda **_kwargs: "https://sandbox.test",
+        )
+
+        with (
+            patch.dict(os.environ, {"TIP_TOKEN_KEY": "process-env-key"}),
+            patch.object(
+                module.request,
+                "urlopen",
+                lambda request, **_kwargs: captured_requests.append(request)
+                or FakeResponse(),
+            ),
+        ):
+            result = module.execute_skills(
+                "do work",
+                tool_context=self._tool_context_without_request_token(),
+            )
+
+        self.assertEqual(result, "api result")
+        self.assertNotIn(
+            "X-tip-token-key",
+            captured_requests[0].headers,
+        )
+
     def test_health_check_retries_502_until_upstream_is_ready(self):
         attempts = []
 
@@ -344,7 +483,41 @@ def fake_run_sandbox_agent(**kwargs):
 
         self.assertEqual(result, "legacy result")
         self.assertEqual(
-            {"CUSTOM_VALUE": "custom", "TOS_SKILLS_DIR": ""},
+            {
+                "CUSTOM_VALUE": "custom",
+                "TOS_SKILLS_DIR": "",
+                "TIP_TOKEN_KEY": "tip-from-metadata",
+            },
+            captured_kwargs["extra_env_vars"],
+        )
+
+    def test_legacy_runcode_execution_injects_tip_token_key_from_metadata(self):
+        captured_kwargs = {}
+
+        def fake_run_sandbox_agent(**kwargs):
+            captured_kwargs.update(kwargs)
+            return "legacy result"
+
+        module = _load_execute_skills_module(
+            ensure_agentkit_session_endpoint=lambda **_kwargs: self.fail(
+                "Skill API must not be used when env_vars are provided"
+            ),
+            run_sandbox_agent=fake_run_sandbox_agent,
+        )
+
+        result = module.execute_skills(
+            "do work",
+            tool_context=self._tool_context(),
+            env_vars={"CUSTOM_VALUE": "custom", "TIP_TOKEN_KEY": "model-provided"},
+        )
+
+        self.assertEqual(result, "legacy result")
+        self.assertEqual(
+            {
+                "CUSTOM_VALUE": "custom",
+                "TIP_TOKEN_KEY": "tip-from-metadata",
+                "TOS_SKILLS_DIR": "tos://agentkit-platform-test-account/skills/",
+            },
             captured_kwargs["extra_env_vars"],
         )
 

tests/utils/test_auth_tip_token_key.py

@@ -0,0 +1,42 @@
+from veadk.utils.auth import (
+    TIP_TOKEN_KEY_HEADER,
+    extract_tip_token_key_from_payload,
+    extract_tip_token_key_from_request_parts,
+)
+
+
+def test_extract_tip_token_key_prefers_header_over_body():
+    token_key = extract_tip_token_key_from_request_parts(
+        headers={TIP_TOKEN_KEY_HEADER: " header-key "},
+        payload={"auth": {"tip_token_key": "body-key"}},
+    )
+
+    assert token_key == "header-key"
+
+
+def test_extract_tip_token_key_reads_body_auth_field_when_header_missing():
+    token_key = extract_tip_token_key_from_payload(
+        {"auth": {"tip_token_key": " body-key "}}
+    )
+
+    assert token_key == "body-key"
+
+
+def test_extract_tip_token_key_ignores_ve_tip_token_header():
+    token_key = extract_tip_token_key_from_request_parts(
+        headers={"X-Ve-TIP-Token": "raw-or-unknown"},
+        payload={},
+    )
+
+    assert token_key is None
+
+
+def test_extract_tip_token_key_returns_none_when_missing_or_empty():
+    assert extract_tip_token_key_from_request_parts(headers={}, payload={}) is None
+    assert (
+        extract_tip_token_key_from_request_parts(
+            headers={TIP_TOKEN_KEY_HEADER: " "},
+            payload={"auth": {"tip_token_key": ""}},
+        )
+        is None
+    )