diff --git a/docs/content/docs/framework/tools/builtin.en.mdx b/docs/content/docs/framework/tools/builtin.en.mdx index 881971503..2ad6791ba 100644 --- a/docs/content/docs/framework/tools/builtin.en.mdx +++ b/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" diff --git a/docs/content/docs/framework/tools/builtin.mdx b/docs/content/docs/framework/tools/builtin.mdx index a9086b09d..15a2900c7 100644 --- a/docs/content/docs/framework/tools/builtin.mdx +++ b/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" diff --git a/tests/integrations/agentkit/test_app.py b/tests/integrations/agentkit/test_app.py index 6dede85dd..c8c4e22f8 100644 --- a/tests/integrations/agentkit/test_app.py +++ b/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( diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 59391a023..64513602a 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 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,6 +205,51 @@ 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("", 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 ") + class TestExecuteSkillsSkillApi(unittest.TestCase): def _tool_context(self): @@ -206,9 +257,26 @@ def _tool_context(self): 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"], ) diff --git a/tests/utils/test_auth_tip_token_key.py b/tests/utils/test_auth_tip_token_key.py new file mode 100644 index 000000000..9bbda708c --- /dev/null +++ b/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 + ) diff --git a/veadk/integrations/agentkit/app.py b/veadk/integrations/agentkit/app.py index acf5d1fe9..0a81f153a 100644 --- a/veadk/integrations/agentkit/app.py +++ b/veadk/integrations/agentkit/app.py @@ -54,6 +54,11 @@ mount_session_capability_routes, ) from veadk.memory.short_term_memory import ShortTermMemory +from veadk.utils.auth import ( + TIP_TOKEN_KEY_METADATA_KEY, + TIP_TOKEN_KEY_STATE_KEY, + extract_tip_token_key_from_request_parts, +) if TYPE_CHECKING: from veadk.runner import Runner @@ -612,6 +617,34 @@ def _run_request_custom_metadata(req: RunAgentRequest) -> dict[str, Any] | None: return metadata if isinstance(metadata, dict) and metadata else None +def _custom_metadata_with_tip_token_key( + custom_metadata: Mapping[str, Any] | None, + tip_token_key: str | None, +) -> dict[str, Any] | None: + metadata = dict(custom_metadata or {}) + metadata.pop(TIP_TOKEN_KEY_METADATA_KEY, None) + metadata.pop(TIP_TOKEN_KEY_STATE_KEY, None) + if tip_token_key: + metadata[TIP_TOKEN_KEY_METADATA_KEY] = tip_token_key + return metadata or None + + +async def _request_json_payload(request: Request) -> Any: + try: + return await request.json() + except Exception: + return None + + +async def _request_tip_token_key(request: Request, payload: Any = None) -> str | None: + if payload is None: + payload = await _request_json_payload(request) + return extract_tip_token_key_from_request_parts( + headers=request.headers, + payload=payload, + ) + + def _resolve_invoke_app_name( services: _RuntimeServices, root_agent: BaseAgent, @@ -669,7 +702,10 @@ async def run_agent_dynamic( root_agent=root_agent, prompt=_content_text(req.new_message), ) - custom_metadata = _run_request_custom_metadata(req) + custom_metadata = _custom_metadata_with_tip_token_key( + _run_request_custom_metadata(req), + await _request_tip_token_key(request), + ) run_config = ( RunConfig(custom_metadata=custom_metadata) if custom_metadata else None ) @@ -710,7 +746,10 @@ async def monitor() -> None: monitor_task.cancel() @app.post("/run_sse") - async def run_agent_sse_dynamic(req: RunAgentRequest) -> StreamingResponse: + async def run_agent_sse_dynamic( + req: RunAgentRequest, + request: Request, + ) -> StreamingResponse: app_name = _resolve_run_app_name(services, root_agent, req) runner = _dynamic_runner( services, @@ -719,7 +758,10 @@ async def run_agent_sse_dynamic(req: RunAgentRequest) -> StreamingResponse: prompt=_content_text(req.new_message), ) stream_mode = StreamingMode.SSE if req.streaming else StreamingMode.NONE - custom_metadata = _run_request_custom_metadata(req) + custom_metadata = _custom_metadata_with_tip_token_key( + _run_request_custom_metadata(req), + await _request_tip_token_key(request), + ) if not runner.auto_create_session: session = await session_service.get_session( @@ -782,6 +824,8 @@ async def invoke_agent_dynamic(request: Request) -> StreamingResponse: app_name = _resolve_invoke_app_name(services, root_agent) user_id = request.headers.get("user_id") or "agentkit_user" session_id = request.headers.get("session_id") or "" + payload = await _request_json_payload(request) + tip_token_key = await _request_tip_token_key(request, payload) prompt = await _invoke_text(request) content = types.UserContent(parts=[types.Part(text=prompt or "")]) @@ -811,7 +855,13 @@ async def event_generator(): user_id=user_id, session_id=session_id, new_message=content, - run_config=RunConfig(streaming_mode=StreamingMode.SSE), + run_config=RunConfig( + streaming_mode=StreamingMode.SSE, + custom_metadata=_custom_metadata_with_tip_token_key( + None, + tip_token_key, + ), + ), ) ) as agen: async for event in agen: diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py index 75059ab54..9a92db4ee 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 from collections.abc import Iterable from typing import Optional @@ -30,6 +29,10 @@ resolve_agentkit_tool_id, ) from veadk.tools.builtin_tools.run_sandbox_agent import run_sandbox_agent +from veadk.utils.auth import ( + TIP_TOKEN_KEY_ENV, + TIP_TOKEN_KEY_METADATA_KEY, +) _SKILL_API_UPGRADE_STATUS_CODES = frozenset({404, 405}) @@ -61,13 +64,17 @@ def _tool_user_session_id(tool_context: ToolContext) -> str: 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 - ) + run_config = getattr(tool_context, "run_config", None) + if run_config is None: + invocation_context = getattr(tool_context, "_invocation_context", None) + run_config = getattr(invocation_context, "run_config", None) + custom_metadata = getattr(run_config, "custom_metadata", None) + if isinstance(custom_metadata, dict): + value = custom_metadata.get(TIP_TOKEN_KEY_METADATA_KEY) + if isinstance(value, str) and value.strip(): + return value.strip() + + return None def _skill_api_url(endpoint: str, path: str) -> str: @@ -287,6 +294,9 @@ def execute_skills( if env_vars: account_id = get_agentkit_account_id(tool_context.state) extra_env_vars = dict(env_vars) + tip_token_key = _tip_token_key(tool_context) + if tip_token_key: + extra_env_vars[TIP_TOKEN_KEY_ENV] = tip_token_key if account_id: extra_env_vars.setdefault( "TOS_SKILLS_DIR", diff --git a/veadk/tools/builtin_tools/run_sandbox_agent.py b/veadk/tools/builtin_tools/run_sandbox_agent.py index bde11cadf..f6ba2e12a 100644 --- a/veadk/tools/builtin_tools/run_sandbox_agent.py +++ b/veadk/tools/builtin_tools/run_sandbox_agent.py @@ -15,7 +15,7 @@ import json import os import re -from typing import Optional +from typing import Any, Optional from google.adk.tools import ToolContext @@ -26,6 +26,7 @@ _ENV_VAR_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _PROTECTED_ENV_VARS = frozenset({"TOOL_USER_SESSION_ID", "USER_SESSION_ID"}) +_SENSITIVE_ENV_VARS = frozenset({"TIP_TOKEN_KEY"}) def _merge_execution_env_vars( @@ -55,6 +56,32 @@ def _merge_execution_env_vars( return merged_env_vars +def _redact_sensitive_env_values(value: Any, env_vars: dict[str, str]) -> Any: + sensitive_values = { + env_value + for env_name, env_value in env_vars.items() + if env_name in _SENSITIVE_ENV_VARS and env_value + } + if not sensitive_values: + return value + + if isinstance(value, str): + redacted = value + for sensitive_value in sensitive_values: + redacted = redacted.replace(sensitive_value, "") + return redacted + if isinstance(value, dict): + return { + key: _redact_sensitive_env_values(item, env_vars) + for key, item in value.items() + } + if isinstance(value, list): + return [_redact_sensitive_env_values(item, env_vars) for item in value] + if isinstance(value, tuple): + return tuple(_redact_sensitive_env_values(item, env_vars) for item in value) + return value + + def _clean_ansi_codes(text: str) -> str: """Remove ANSI escape sequences (color codes, etc.).""" ansi_escape = re.compile(r"\x1b\[[0-9;]*m") @@ -227,9 +254,11 @@ def run_sandbox_agent( logger.debug("Invoke run sandbox agent completed") try: - return _format_execution_result(res["Result"]["Result"]) + result = _format_execution_result(res["Result"]["Result"]) + return _redact_sensitive_env_values(result, env_vars) except KeyError as e: + redacted_res = _redact_sensitive_env_values(res, env_vars) logger.error( - f"Error occurred while running sandbox agent: {e}, response is {res}" + f"Error occurred while running sandbox agent: {e}, response is {redacted_res}" ) - return res + return redacted_res diff --git a/veadk/utils/auth.py b/veadk/utils/auth.py index 565a26b7c..169174ce6 100644 --- a/veadk/utils/auth.py +++ b/veadk/utils/auth.py @@ -23,7 +23,8 @@ import base64 import json -from typing import Literal, Optional +from collections.abc import Mapping +from typing import Any, Literal, Optional from google.adk.auth.auth_credential import ( AuthCredential, @@ -38,6 +39,51 @@ VE_TIP_TOKEN_HEADER = "X-Ve-TIP-Token" VE_TIP_TOKEN_CREDENTIAL_KEY = "ve_tip_token" +# TIP token key is an authorization handle injected per request by AgentKit +# Gateway/upstream identity services. It is distinct from a raw TIP token. +TIP_TOKEN_KEY_HEADER = "X-Tip-Token-Key" +TIP_TOKEN_KEY_ENV = "TIP_TOKEN_KEY" +TIP_TOKEN_KEY_METADATA_KEY = "tip_token_key" +TIP_TOKEN_KEY_STATE_KEY = "TIP_TOKEN_KEY" + + +def _clean_tip_token_key(value: Any) -> str | None: + if not isinstance(value, str): + return None + cleaned = value.strip() + return cleaned or None + + +def extract_tip_token_key_from_headers( + headers: Mapping[str, Any] | None, +) -> str | None: + """Extract a TIP token key from request headers without accepting raw tokens.""" + if not headers: + return None + normalized = {str(key).lower(): value for key, value in headers.items()} + return _clean_tip_token_key(normalized.get(TIP_TOKEN_KEY_HEADER.lower())) + + +def extract_tip_token_key_from_payload(payload: Any) -> str | None: + """Extract a TIP token key from a JSON body ``auth.tip_token_key`` field.""" + if not isinstance(payload, Mapping): + return None + auth = payload.get("auth") + if not isinstance(auth, Mapping): + return None + return _clean_tip_token_key(auth.get(TIP_TOKEN_KEY_METADATA_KEY)) + + +def extract_tip_token_key_from_request_parts( + *, + headers: Mapping[str, Any] | None, + payload: Any = None, +) -> str | None: + """Resolve request-scoped TIP token key with header taking precedence.""" + return extract_tip_token_key_from_headers( + headers + ) or extract_tip_token_key_from_payload(payload) + def strip_bearer_prefix(token: str) -> str: """Remove 'Bearer ' prefix from token if present.