KsADK

工具与 Skill Runtime

KsADK 可以通过框架原生工具、MCP/A2A 集成和可选的 Skill Runtime 向 Agent 暴露工具。公开规则是:工具必须显式声明,在运行时可验证,并且和 Agent 不 需要的密钥或本地文件隔离。

工具层次

Tool Gateway 审批流

框架适配器在 Agent 执行前归一化工具定义。Agent 拿到的是稳定工具描述、收窄的 输入 schema 和可在本地 Web UI 渲染或写回会话历史的结构化结果;而 ToolGateway.invoke() 在调用真实实现前先校验 ToolPolicy,strict 模式下 medium/high/critical 风险工具会返回 approval_required,等待人工确认后再走 execute 分支。

优先使用能满足需求的最简单路径。不要把一个确定性的本地 helper 包成远端 runtime,只为了让它看起来像工具。

什么时候使用哪条路径

需求推荐路径
Agent 项目内的简单 Python helper框架原生 function tool
自带生命周期的外部工具服务MCP toolset
Agent-to-Agent 协议集成A2A client 或 adapter
可复用、可沙箱执行的技能Skill Runtime
仅本地开发local backend,使用显式路径和测试数据
不可信或成本较高的执行受审查的 sandbox backend 和限制
AgentEngine 常用内置能力ksadk.toolsets 的 focused profile
低频或高风险内置能力agentengine_tool_dispatcher 按需列出、描述和调用

AgentEngine 内置工具

KsADK Toolsets 层次结构

ksadk.toolsets 提供 SDK 内置工具入口。get_agentengine_tools() 无参时仍返回 全量内置工具,保持历史兼容;新项目推荐显式选择工具集合:

agent.py
from ksadk.toolsets import describe_agentengine_tools, get_agentengine_tools

tools = get_agentengine_tools(include=["focused", "agentengine_tool_dispatcher"])
tool_descriptions = describe_agentengine_tools(include=["focused", "agentengine_tool_dispatcher"])

include 可以混用工具组、profile 和具体工具名:

include含义
skill / workspace / platform / sandbox绑定对应内置工具组
focused / core绑定常用低风险工具集合
run_code 等具体工具名显式扩展单个工具

focused/core 默认直接暴露:

  • list_skillssearch_skillsload_skill
  • workspace_statussearch_workspace_files
  • edit_workspace_filelint_workspace_file
  • component_statussandbox_status

execute_skillsrun_commandrun_codedelete_workspace_file 和整文件写入类 工具不会默认进入 focused profile。

需要这些能力时,可以显式绑定工具名,或通过 dispatcher 渐进式披露。

工具分发与渐进式披露

tool_dispatcher(action, tool_name=None, arguments=None, include=None, profile="default")ksadk/toolsets/__init__.py 里的低风险索引工具,让模型只在需要时按名称列出、描述并调用 KsADK 本地内置工具,从而压缩上下文里的工具描述数量。它调用真实工具对象,因此仍受 Tool Gateway 审批策略约束。

tool_dispatcher 三分支渐进式披露

三种 action 共享同一份内置工具注册表,且都排除了 tool_dispatcher / agentengine_tool_dispatcher 自身,避免递归调用。call 分支最终走的是 ToolGateway.invoke,所以 strict 模式下 run_commandrun_codeexecute_skills 和 workspace 写入/删除工具依旧会返回 approval_required envelope,由调用方 (UI 或外层 runtime)将 approval 结果回传。

action行为
list列出可调度工具,不包含 dispatcher 自身,返回 tools + tool_count
describe返回单个工具的描述、risk_levelrequires_approvalboundary
call按名称调用 KsADK 本地内置工具,结果回包 ok + tool_name + result

dispatcher 只调度 KsADK 本地内置工具,不连接控制台 Tool Space 数据库,也不做远端动态工具绑定。它调用真实工具对象,因此不会绕过 Tool Gateway 审批策略;例如 strict 模式下调用 run_commandrun_codeexecute_skills 或 workspace 写入/删除工具时,仍会返回 approval_required envelope。

Platform 工具参考

component_status 是 platform 组唯一的低风险只读工具,用于让 Agent 在运行时 快速了解 AgentEngine 绑定的模型、知识库、长期记忆、Skill Space、沙箱和 workspace 是否就绪。它读取环境变量与已绑定状态,不产生任何副作用。

component_status

签名
from ksadk.toolsets.platform import component_status

component_status() -> dict
风险等级low
是否需要审批
副作用无(只读)
所属组platform
入口get_agentengine_tools(include=["platform"])

返回结构

返回示例
{
  "ok": True,
  "summary": {
    "model": "glm-5.1",
    "knowledge_base_bound": True,
    "long_term_memory_bound": False,
    "skill_space_bound": True,
    "isolated_execution": "enabled",
    "sandbox_direct_tools": "enabled",
  },
  "skill_space": {"space_ids": [...], "tools": [...]},
  "skill_runtime": {"backend": "e2b", "enabled": True, "template_bound": True},
  "sandbox": {"backend": "e2b", "enabled": True, "tools": ["sandbox_status", "run_command", "run_code"]},
  "workspace": {"root": "/.../workspace", "tools": [...]},
}

相关环境变量

变量作用
OPENAI_MODEL_NAME / MODEL_NAME体现在 summary.model
KSADK_KB_DATASET_ID知识库是否绑定
KSADK_LTM_NAMESPACE长期记忆是否绑定
KSADK_SANDBOX_TEMPLATE_ID / KSADK_SKILL_RUNTIME_TEMPLATE_ID沙箱模板是否绑定
KSADK_SKILL_RUNTIME_BACKENDSkill Runtime 后端类型

Workspace 工具参考

Workspace 工具被限制在 AgentEngine UI workspace 目录内(resolve_local_session_dir() / workspace)。所有路径都被归一化并拒绝越界访问。写操作按 medium 风险处理, delete_workspace_filehigh 风险处理;strict 模式下需要 approval。

0.6.7 会话级 read_state 隔离

read_workspace_file 会记录 WorkspaceReadState(mtime、size、行范围)。 edit_workspace_file / multi_edit_workspace_file 在写入前必须校验该状态: 未读返回 file_not_read,文件被外部修改返回 file_modified_since_read。 状态按 session_id 隔离——workspace_state.py(session_id, path) 作为键, 默认从 ToolExecutionContext 取当前会话 ID,因此不同会话/账号的 read_state 互不串扰。

workspace_status

签名
workspace_status() -> dict
风险等级low
是否需要审批
副作用无(只读,会 mkdir 根目录)
所属组workspace / focused

返回 workspace root、最多 50 个文件的采样列表与 file_count_sampled

list_workspace_files

签名
list_workspace_files(
    path: str = ".",
    glob: str | None = None,
    recursive: bool = False,
    include_dirs: bool = True,
    max_results: int = 500,
    sort_by: str = "name",  # name | mtime | size
) -> dict
风险等级low
是否需要审批
副作用无(只读)
所属组workspace / focused

返回 entriesname / path / type / size / mtime_ns)和 truncated 标志; max_results 上限 5000,path 不存在返回 ok=False + error_message

read_workspace_file

签名
read_workspace_file(
    path: str,
    start_line: int | None = None,
    end_line: int | None = None,
    max_chars: int | None = None,
    include_line_numbers: bool = True,
) -> dict
风险等级low
是否需要审批
副作用无(只读,但会写入 read_state)
所属组workspace / focused

读取 UTF-8 文本(>2MB 拒绝),支持行范围、行号、字符预算(1000–100000)。 返回 content / read_range / total_lines / partial,超出预算会持久化为 preview。同时记录 WorkspaceReadState 供后续 edit 校验。

write_workspace_file

签名
write_workspace_file(
    path: str,
    content: str,
    overwrite: bool = True,
    approval: dict | None = None,  
) -> dict
风险等级medium
是否需要审批strict 模式下需要
副作用workspace_write
所属组workspace(默认不进 focused)

整文件写入。overwrite=False 时文件已存在则返回 ok=False。strict 模式下未带 approval 返回 approval_required envelope。

write_workspace_files

签名
write_workspace_files(
    files: list[dict],  # [{path, content}, ...]
    overwrite: bool = True,
    approval: dict | None = None,  
) -> dict
风险等级medium
是否需要审批strict 模式下需要
副作用workspace_write
所属组workspace(默认不进 focused)

批量整文件写入,单次最多 100 个文件,返回 written 列表与 truncated

edit_workspace_file

签名
edit_workspace_file(
    path: str,
    old_text: str,
    new_text: str,
    expected_replacements: int = 1,
    replace_all: bool = False,
    approval: dict | None = None,  
) -> dict
风险等级medium
是否需要审批strict 模式下需要
副作用workspace_edit
所属组workspace / focused

精确片段替换。需先 read_workspace_file,否则返回 file_not_readold_text 未命中返回 snippet_not_found(附带 nearby_candidates),匹配次数与 expected_replacements 不一致返回 ambiguous_edit。返回 unified diff(按预算 持久化)。支持引号归一化匹配(直引号/弯引号互通)。

multi_edit_workspace_file

签名
multi_edit_workspace_file(
    path: str,
    edits: list[dict],  # [{old_text, new_text, expected_replacements?, replace_all?}, ...]
    replace_all: bool = False,
    approval: dict | None = None,  
) -> dict
风险等级medium
是否需要审批strict 模式下需要
副作用workspace_edit
所属组workspace

原子化多片段编辑。任一 edit 失败整体不写入,返回 failed_edit_index。 成功返回 edit_count / replacements / 每条 edit 的 diagnostics

lint_workspace_file

签名
lint_workspace_file(
    path: str,
    language: str = "auto",  # auto | python | json | markdown | text
) -> dict
风险等级low
是否需要审批
副作用无(只读)
所属组workspace / focused

轻量内置检查:Python 用 ast.parse、JSON 用 json.loads、其余扫描 NUL 字节 和尾部空白。ok 仅在无 error 级 issue 时为真。lint_model 固定为 built_in_lightweight——这不是项目级 formatter 或完整 lint 工具链。

search_workspace_files

签名
search_workspace_files(
    query: str,
    path: str = ".",
    is_regex: bool = False,
    glob: str | None = None,
    case_sensitive: bool = False,
    context_lines: int = 0,
    max_results: int = 100,
) -> dict
风险等级low
是否需要审批
副作用无(只读)
所属组workspace / focused

若环境装了 rg(ripgrep)优先走 rg(更快),否则回退纯 Python。返回 resultspath / line / text / context_before / context_after)、 match_counttruncatedsearch_backendrgpython)。 单文件 >1MB 跳过;context_lines 上限 20;max_results 上限 5000。

delete_workspace_file

签名
delete_workspace_file(
    path: str,
    approval: dict | None = None,  
) -> dict
风险等级high
是否需要审批strict 模式下需要
副作用workspace_delete
所属组workspace(默认不进 focused)

删除文件或空目录。拒绝删除 workspace root。非空目录需先清空子项。

多框架接入示例

agent.py (LangGraph)
from langgraph.prebuilt import create_react_agent
from ksadk.toolsets import get_agentengine_tools

# focused 默认含 workspace 读/search/edit/lint + component_status
tools = get_agentengine_tools(include=["focused"])

# 显式加上高风险删除能力(strict 模式下需 approval)
tools += get_agentengine_tools(include=["delete_workspace_file"])

agent = create_react_agent(model="openai/glm-5.1", tools=tools)
agent.py (LangChain)
from langchain.agents import create_tool_calling_agent
from ksadk.toolsets import get_agentengine_tools

# as_tool() 已把 workspace 函数包成 LangChain Tool
tools = get_agentengine_tools(include=["workspace"])

# 可单独取某个工具
from ksadk.toolsets.workspace import edit_workspace_file
tools.append(edit_workspace_file)
agent.py (Google ADK)
from google.adk.agents import LlmAgent
from ksadk.toolsets import get_agentengine_tools

# ADK Agent 直接接收工具列表
agent = LlmAgent(
    name="workspace_agent",
    model="openai/glm-5.1",
    instruction="Use workspace tools to read and edit files.",
    tools=get_agentengine_tools(include=["focused"]),
)

写操作需要 approval

strict 模式下,write_workspace_file / write_workspace_files / edit_workspace_file / multi_edit_workspace_file(medium)和 delete_workspace_file(high)会先返回 approval_required envelope。 调用方(本地 Web UI 或外层 runtime)需把用户确认结果通过 approval 参数回传, 才会进入真实执行分支。

Tool Gateway 与人工确认

Tool Gateway 负责工具风险、审批和执行边界,不负责上下文压缩。公开 SDK 支持 用环境变量启用严格审批:

strict 模式审批

strict 模式下,medium / high / critical 风险工具会先返回结构化 approval_required,而不是直接执行。UI 或外层 runtime 可以展示审批请求,并在用户确认后把 approval 结果传回工具调用。

工具类型默认风险行为
workspace read/search/statuslow直接执行
edit_workspace_file / 写文件mediumstrict 模式需要审批
delete_workspace_filehighstrict 模式需要审批
execute_skillshighstrict 模式需要审批
run_command / run_codehighstrict 模式需要审批,且只进 sandbox backend

edit_workspace_file 是 exact snippet replacement:old_text 未命中时返回 snippet_not_found,匹配次数与 expected_replacements 不一致时返回 ambiguous_editlint_workspace_file 是 SDK 内置轻量检查,支持 Python AST、JSON parse 和通用文本检查;它不是项目级 formatter 或完整 lint 工具链。

Skill 加载时序

ksadk/skills/runtime/loader.pyload_skills() 在 Agent 装载期把本地技能与 远程技能 manifest 合并。远程技能先经 Skill Service 列举、获取下载地址,再做 sha256 校验与安全解压,最后交给 load_local_skill() 装入 LocalSkill

KsADK Skill 加载时序

下面的时序图给出同一流程的 mermaid 文本版本:

这条链路解释了 Skill Runtime 的安全边界:远程 zip 必须通过 sha256 校验和 路径穿越检查才会被解压;只有 SKILL.md 被定位到时,技能才会进入 LocalSkillKSADK_SKILL_ALLOW_HASH_MISMATCH 仅在显式开启时跳过哈希失败,但仍会把警告 写回 SkillLoadResult.warnings

这个顺序也说明了边界:Skill Runtime 与 MCP 是 ADK Agent 装载期扩展;A2A 是 runner 外层协议适配。通过 A2A 暴露的 runner 可以间接使用已注入的工具, 但 A2A Server 自身不负责解析 MCP,也不负责执行沙箱。

Skill 工具参考

ksadk.toolsets.skills 暴露四个 Skill 工具。前三个 (list_skills / search_skills / load_skill) 风险等级 low,直接读 Skill Space;execute_skills 风险等级 high,走隔离 runtime。所有工具的签名、参数、返回结构和风险都遵循下表。

list_skills

列出当前账号可发现的远程技能 manifest(用户 Skill Space + 公共预置空间,去重后)。

签名
list_skills() -> dict[str, Any]
风险等级low
是否需要审批
副作用
所属组skill / focused
env varKSADK_SKILL_SPACE_IDS / KSADK_PUBLIC_SKILL_SPACE_IDS / KSADK_SKILL_SERVICE_URL

返回 okskills(manifest 列表,含 name/description/version/space_id)和 skill_space_ids;Skill Service 未配置时返回 ok=False + 错误结构。

agent.py
from ksadk.toolsets.skills import list_skills

result = list_skills()
# {"ok": True, "skills": [...], "skill_space_ids": ["space_xxx"]}

search_skills

按名称、别名、标签、描述和 examples 模糊匹配技能。

签名
search_skills(query: str, max_results: int = 10) -> dict[str, Any]
风险等级low
是否需要审批
副作用
所属组skill / focused
env varlist_skills

返回 okqueryresults(每条含 name/score/description);query 为空 返回 ok=False

agent.py
from ksadk.toolsets.skills import search_skills

hits = search_skills("diagram", max_results=5)
# {"ok": True, "query": "diagram", "results": [...]}

load_skill

下载并加载某个技能的 SKILL.md 指令到外层 Agent 上下文(不走隔离 runtime)。

签名
load_skill(skill_name: str) -> dict[str, Any]
风险等级low
是否需要审批
副作用skill_cache_write(写入本地缓存目录)
所属组skill / focused
env varKSADK_SKILL_SERVICE_URL / KSADK_SKILL_CACHE_DIR

返回 okinstructionsnameversionroot_dirhas_scripts_dirscript_filescache_hitusage。技能不存在时返回 ok=False + available_skills

usage 字段提示模型:在外层 Agent 中直接完成任务,除非技能显式要求隔离执行。需要隔离执行时应改用 execute_skills

agent.py
from ksadk.toolsets.skills import load_skill

skill = load_skill("baoyu-diagram")
# skill["instructions"] 是 SKILL.md 正文,可直接读入 Agent 上下文

execute_skills

把一段 workflow prompt 交给 Skill Runtime 隔离执行,可选预绑定技能名。

签名
execute_skills(
    workflow_prompt: str,
    skill_names: list[str] | str | None = None,
    approval: dict[str, Any] | None = None,  
) -> dict[str, Any]
风险等级high
是否需要审批strict 模式下需要
副作用isolated_runtime_execution
所属组skill(默认不进 focused)
env varKSADK_SKILL_RUNTIME_BACKEND / KSADK_SANDBOX_TEMPLATE_ID / KSADK_SKILL_RUNTIME_TIMEOUT

返回 runtime 结果(okexit_codestdoutstderrduration_mstimed_outoutput_files 等),并附 execution_context=f"skill-runtime/{backend}"。 runtime 未配置时返回 skill_runtime_disabled + hint。

execute_skills 需要审批与隔离 runtime

execute_skills 风险等级为 high,side effect 是隔离 runtime 执行。在 KSADK_TOOL_APPROVAL_MODE=strict 下,调用会先返回 approval_required 信封,由 UI 或外层 runtime 把 approval={"approved": True} 回传后再执行。runtime 未配置时返回 skill_runtime_disabled 错误,提示设置 KSADK_SKILL_RUNTIME_BACKEND=local_process|e2b 或 sandbox template id。

execute_skills 编排流

execute_skillsToolGateway.invoke 审批后进入 _execute_skills_impl, 最终落到 build_execute_skills_tool 构造的 runtime tool。链路如下:

execute_skills 编排流

runtime 内部的「发现 → 下载 → sha256 → 解压 → loader → agent」与上面 Skill 加载时序同构:SkillServiceClient 列举技能、download_skill_archive 拉取 zip、PackageStore.store_archive 做 sha256 校验和路径穿越保护,再交给 load_local_skill 装入 runtime 子进程内的 Agent。

Skill 工具多框架接入

agent.py
from ksadk.toolsets.skills import get_skill_tools

# as_tool 自动把 Python 函数转成框架工具
tools = get_skill_tools()  # [list_skills, search_skills, load_skill, execute_skills]
agent.py
from google.adk.agents import LlmAgent
from ksadk.toolsets import get_agentengine_tools

# focused profile 已含 list_skills / search_skills / load_skill
tools = get_agentengine_tools(include=["focused"])
root_agent = LlmAgent(name="skill_agent", model=model, tools=tools)
agent.py
from ksadk.toolsets import get_agentengine_tools

# focused 默认不含 execute_skills,需显式扩展
tools = get_agentengine_tools(include=["focused", "execute_skills"])

Dispatcher 工具参考

tool_dispatcher 和兼容别名 agentengine_tool_dispatcher 是低风险索引工具, 让模型按名称 list / describe / call KsADK 本地内置工具,避免一次性把全量工具 描述塞进上下文。tool_search 是只读搜索工具,返回匹配工具的描述、风险等级和边界。

tool_dispatcher

签名
tool_dispatcher(
    action: str,  # list | describe | call
    tool_name: str | None = None,
    arguments: dict[str, Any] | str | None = None,
    include: str | Iterable[str] | None = None,
    profile: str = "default",  # default | coding
) -> dict[str, Any]
风险等级low(自身);call 分支继承目标工具策略
是否需要审批dispatcher 自身不需要;callToolGateway.invoke
边界local_ksadk_builtin_tools
所属组dispatcher
参数类型说明
actionstrlist / describe / call
tool_namestrdescribe / call 时必填;不能是 dispatcher 自身
argumentsdict / JSON 字符串call 时传给目标工具;JSON 字符串会自动 json.loads
includestr / list[str]限定 list 的工具集
profilestrdefault(全组)或 coding(focused)

渐进式披露(list → describe → call)

三种 action 都共享同一份注册表,并都排除 tool_dispatcher / agentengine_tool_dispatcher 自身以避免递归。call 分支最终走 ToolGateway.invoke,所以 strict 模式下 run_commandrun_codeexecute_skills 和 workspace 写入/删除工具依旧会返回 approval_required 信封。

action行为
list列出可调度工具,排除 dispatcher 自身,返回 tools + tool_count
describe返回单个工具的 risk_level / requires_approval / boundary
call按名称调用工具,结果回包 ok + tool_name + resultapproval_required 透传并附 dispatched_tool_name
agent.py
from ksadk.toolsets import tool_dispatcher

# 1. list 看有哪些工具可调度
tools = tool_dispatcher(action="list", profile="coding")
# 2. describe 查 run_command 的风险与边界
spec = tool_dispatcher(action="describe", tool_name="run_command")
# 3. call 真正执行(strict 下可能先返回 approval_required)
out = tool_dispatcher(action="call", tool_name="run_command",
                     arguments={"command": "ls -la"})
签名
tool_search(
    query: str,
    profile: str = "coding",
    max_results: int = 8,
    include_disabled: bool = False,
) -> dict[str, Any]
风险等级low
是否需要审批
副作用无(只读)
边界local_ksadk_builtin_tool_registry
所属组tools

返回 okqueryresults(每条含 name/score/group/risk_level/boundary/ execution)和 deferred_tool_namestool_search 同时搜索内置工具描述和 register_external_tools 注册的框架管理工具描述,但不会执行任何工具——它只返回 deferred_tool_names,让调用方按需再走 tool_dispatcher(action="call")

agent.py
from ksadk.toolsets import tool_search

hits = tool_search("read file", profile="coding", max_results=5)
# {"ok": True, "results": [{"name": "read_workspace_file", "score": 4.0, ...}], ...}

内置工具模式与 Profile

通过两个环境变量控制 KsADK 向 runtime 暴露多少内置工具:

变量取值行为
KSADK_BUILTIN_TOOLS_MODEoff不注入任何内置工具
direct(默认)按 profile 全量注入工具对象
focused注入 focused 工具集(含 tool_search
dispatcher / deferred只注入 tool_dispatcher / tool_search,按需展开
变量取值行为
KSADK_BUILTIN_TOOLS_PROFILEdefault全部内置工具组(skill/workspace/platform/sandbox/web)
codingfocused 子集(workspace 读写、search、sandbox status、web 等)

内置工具注入模式决策

agent.py
from ksadk.toolsets import builtin_tools_for_runtime

# 读 KSADK_BUILTIN_TOOLS_MODE + KSADK_BUILTIN_TOOLS_PROFILE
tools = builtin_tools_for_runtime()  # mode=off 时返回 []

mode=deferred 适合上下文预算紧张的 hosted 场景:runtime 只看到 tool_searchtool_dispatcher,模型先搜索再按需 call,把全量工具描述挡在 context 之外。

mode暴露工具适用场景
off自带工具集的纯框架 Agent
direct + default全量内置工具本地开发 / 调试
direct + codingfocused 工具集通用 coding agent
focusedfocused + tool_search预算有限的 coding agent
dispatcher / deferredtool_dispatcher / tool_searchhosted / 多租户,按需披露

Skill Service 集成

Skill 工具通过 SkillServiceClient 访问 Skill Service。公开配置只需变量名, 不需要在文档里写真实 token 或控制面地址。

变量作用默认 / 回退
KSADK_SKILL_SERVICE_URLSkill Service 基址由 AICP 连接推断
KSADK_SKILL_SERVICE_ENDPOINT / _SCHEME显式 endpoint / schemeAICP 模式推断
KSADK_SKILL_SERVICE_TOKENBearer token
KSADK_SKILL_SERVICE_ACCESS_KEY访问密钥KSYUN_ACCESS_KEY / KS3_ACCESS_KEY
KSADK_SKILL_SERVICE_SECRET_KEY私有密钥KSYUN_SECRET_KEY / KS3_SECRET_KEY
KSADK_SKILL_SERVICE_ACCOUNT_ID账号 IDKSYUN_ACCOUNT_ID
KSADK_SKILL_SERVICE_REGION逻辑 regionKSYUN_REGION / cn-beijing-6
KSADK_SKILL_SERVICE_API_VERSIONAPI 版本2024-06-12
KSADK_SKILL_SERVICE_SIGN_SERVICE签名服务名aicp
KSADK_SKILL_SPACE_IDS用户 Skill Space(逗号分隔)SKILL_SPACE_ID 回退
KSADK_PUBLIC_SKILL_SPACE_IDS公共预置空间
KSADK_PUBLIC_SKILL_ALLOWLIST公共技能白名单空(全放行)
KSADK_SKILL_CACHE_DIR技能包缓存目录系统临时目录
KSADK_SKILL_MANIFEST_LIMITmanifest 最大条数30
KSADK_SKILL_MANIFEST_TIMEOUTmanifest 拉取超时(秒)10

配置 KSADK_SKILL_SPACE_IDS 后,runtime_agent_env_from_process 会把上述变量注入子进程 / sandbox 环境,使 runtime 内的 Skill 工具能访问同一个 Skill Service。should_resolve_child_skill_service_url() 为真时还会把推断出的 service URL 写入子进程,避免子进程再次走 AICP 推断。

.env
# 只写变量名,不要提交真实值
export KSADK_SKILL_SPACE_IDS=space_xxx
export KSADK_SKILL_SERVICE_URL=https://skill-service.example.com
export KSADK_SKILL_SERVICE_TOKEN=$SKILL_SERVICE_TOKEN

Tool Gateway 审批

ToolGateway.invoke 在调用真实实现前读取 ToolPolicy,按 KSADK_TOOL_APPROVAL_MODE 决定是否拦截:

mode行为
off(默认)不拦截,直接执行
strictrisk_level 为 medium / high / critical 的工具先返回 approval_required 信封

Skill 工具的策略:

工具risk_levelside_effectsstrict 下审批
list_skillslow
search_skillslow
load_skilllowskill_cache_write
execute_skillshighisolated_runtime_execution

审批信封结构(approval_required):

approval envelope
{
  "ok": False,
  "type": "approval_required",
  "approval_required": True,
  "approval_request": {
    "id": "appr_xxx",
    "tool_name": "execute_skills",
    "risk_level": "high",
    "side_effects": ["isolated_runtime_execution"],
  },
}

调用方把 approval={"approved": True} 回传即可放行;approved=False 则返回 approval_rejected。dispatcher 的 call 分支会把 dispatched_tool_name 附在 返回信封上,方便 UI 关联被调度的真实工具。

公开 Skill Runtime 契约

一个公开的 Skill Runtime 集成应记录:

  • skill 名称和用途。
  • 输入 schema 和必填字段。
  • 输出 schema 和错误形态。
  • 必需的 optional dependencies 或 extras。
  • skill 在本地执行还是在 sandbox backend 中执行。
  • 需要的环境变量名称。
  • 文件、网络和执行时间限制。

工具描述应足够精确,让 LLM 知道什么时候不该调用它。避免暗示工具拥有宽泛的文件系统、shell、网络或凭证访问能力。

Skills Mode 判定

ADK runner 会先解析 skills mode。显式设置 KSADK_SKILLS_MODE 时直接生效; auto 模式下按环境与本地目录推断。

Skills Mode 判定

Skill Runtime Backend

Skill Runtime Backend 是一个协议接口。核心方法 run_workflow() 接收 workflow prompt、技能空间、session id、可选技能名、环境变量、输入文件和 timeout,返回 SkillRuntimeResult

字段含义
runtime_id后端运行实例标识
exit_code子进程或 sandbox 命令退出码
stdout / stderr公开安全的输出摘要
duration_ms执行耗时
timed_out是否超时
error_type / error_message稳定错误分类和摘要
output_files工具生成的输出文件引用

ok 只在 exit_code == 0、没有 error_type 且未超时时为真。

业务代码应读结构化字段,不要解析本地 UI 文本。

Backend 选择

Backend选择条件执行位置Prompt 传递方式
disabled默认无 sandbox 配置或显式 disabled不执行调用即抛出 setup 错误
local_processKSADK_SKILL_RUNTIME_BACKEND=local_process本地 Python 子进程写入临时 prompt 文件
e2b显式 e2b、sandbox backend 或 template idE2B 沙箱写入 sandbox 内 prompt 文件

公开文档可以描述这些 backend 的契约,但不应发布私有镜像、私有 registry、内部控制面地址或团队运行手册。

远程技能 manifest

当配置了 KSADK_SKILL_SERVICE_URL 且存在 skill space id 时,KsADK 会加载 远程技能 manifest。只保留 active skills,并按技能名去重;如果某个公共技能 空间配置了 allowlist,则只暴露 allowlist 内的技能。

manifest instruction 会提醒模型:

  • 只有任务匹配时才调用 execute_skills
  • 调用时使用原始 workflow_prompt
  • skill_names 必须使用精确技能名。
  • 不要假设完整技能说明已经加载;execute_skills 会按需加载。

MCP 配置边界

MCP Runtime 以 KSADK_MCP_SERVERS 为入口,默认启用。只有 KSADK_ENABLE_MCP_TOOLS0falsenooff 时禁用。

配置项作用校验或默认行为
KSADK_ENABLE_MCP_TOOLSMCP 工具注入开关默认启用
KSADK_MCP_SERVERSJSON 数组形式的 MCP Server 配置必须是 JSON array
nameMCP Server 名称必须是非空字符串
urlStreamable HTTP MCP Endpoint必须是绝对 http(s) URL 且 path 以 /mcp 结尾
api_keyBearer Token可选字符串,生成 Authorization header
tool_filter限制暴露工具名可选非空字符串列表
tool_name_prefix工具名前缀与去重键组成部分可选字符串

示例可以展示变量名,但不能包含真实 token、私有 endpoint 或内部服务地址。

环境配置

密钥不要进源码。把本地开发值放在 .env 或 shell 环境中,公开文档只展示占位变量名。

如果后端需要凭证,文档只写变量名和配置步骤:

.env
export EXAMPLE_SANDBOX_API_KEY=...

不要提交 .env.pypirc、PyPI token、kubeconfig、私有 registry 凭证、云访问密钥或生成的运行状态文件。

本地 Backend

local backend 适合开发、测试和只操作已知输入的示例。应按可信本地执行处理:

  • 测试里使用临时目录。
  • 传入显式输入文件,不扫描整个仓库。
  • 使用较短 timeout。
  • 尽量返回结构化失败,而不是原始 traceback。
  • 避免执行任意用户提供 shell 的示例。

公开文档中的 local backend 示例应可复现,不依赖内部基础设施。

Sandbox Backend

sandbox backend 适合需要更强隔离、网络策略或依赖控制的场景。公开文档应描述 契约,而不是私有 provider wiring:

主题公开文档应说明
authentication需要的变量名,不写 token 值
limitstimeout、内存、文件大小和网络策略
files允许上传/下载路径和保留行为
errors稳定错误码或错误类别
cleanupsandbox 是否每次调用后销毁

内部账号 ID、私有镜像、registry host、托管控制面 URL 和支持 runbook 都应留在内部文档中。

Sandbox 工具参考

ksadk.toolsets.sandbox 暴露三个工具,全部经 ToolGateway。backend 由 KSADK_SANDBOX_BACKEND 决定,sandbox_status 报告实际生效的边界。

Sandbox 工具审批与执行边界

sandbox_status

报告 AgentEngine sandbox 状态和边界,不执行任何命令。

签名
sandbox_status() -> dict
风险等级low
是否需要审批
副作用无(只读)
所属组sandbox / focused

返回 ok / backend / enabled / isolated / template_bound / template_id / timeout_seconds / ttl_seconds / idle_ttl_seconds / max_sessions / sandbox_id / created_at / last_used_at / expires_at / idle_seconds / boundaryboundary 字段说明执行边界:隔离 backend 标注「只通过 sandbox backend 执行,不暴露主机 shell」, local_process / pod_process 标注共享本地/pod 文件系统与网络。

环境变量作用默认
KSADK_SANDBOX_BACKENDbackend(e2b / local_process / pod_process / none自动推断:有 template id 用 e2b,否则 none
KSADK_SANDBOX_TEMPLATE_IDE2B 模板 id
KSADK_SANDBOX_TIMEOUT单次命令超时秒数600
KSADK_SANDBOX_TTL_SECONDSsandbox 存活 TTL600
KSADK_SANDBOX_IDLE_TTL_SECONDS空闲回收秒数300
KSADK_SANDBOX_MAX_SESSIONS最大复用会话数(0 = 不限)0
agent.py
from ksadk.toolsets.sandbox import sandbox_status

info = sandbox_status()
if not info["enabled"]:
    raise RuntimeError(f"sandbox disabled: {info['boundary']}")

run_command

在隔离 sandbox 内执行 shell 命令。

签名
run_command(
    command: str,
    cwd: str | None = None,
    timeout: int | None = None,
    env: dict[str, str] | None = None,
    background: bool = False,
    approval: dict | None = None,  
) -> dict
风险等级high
是否需要审批strict 模式下是
副作用sandbox_command_execution
所属组sandbox

高风险 + 显式审批

  • KSADK_TOOL_APPROVAL_MODE=strict 下返回 approval_required,需回传 approval={"approved": True}
  • 命令策略 (check_command_policy) 默认拦截:递归 rmsudo/kubectl/dockergit reset/clean/push/checkout、云元数据端点(169.254.169.254metadata.google.internal)。
  • pod_process backend 必须显式 KSADK_ALLOW_POD_PROCESS_TOOLS=true 才会创建,否则 create_sandbox_backendSandboxError
  • background=True 在 P0 直接返回 background_not_supported

返回 ok / backend(形如 sandbox/e2b)/ sandbox_id / command / stdout / stderr / exit_code,以及预算字段 stdout_truncated / stdout_original_chars / stdout_preview_charsstderr 同),超阈值时附 persisted_outputs

agent.py
from ksadk.toolsets.sandbox import run_command

result = run_command(
    "python -c 'print(2+2)'",
    timeout=30,
    approval={"approved": True},
)
assert result["ok"]
print(result["stdout"], result["exit_code"])

run_code

在 sandbox 内执行代码片段,不是 shell 替代。写入 /tmp/ksadk-run-code-* 后调用对应解释器。

签名
run_code(
    code: str,
    language: str = "python",  # python | javascript | bash
    timeout: int | None = None,
    env: dict[str, str] | None = None,
    approval: dict | None = None,  
) -> dict
风险等级high
是否需要审批strict 模式下是
副作用sandbox_code_execution
所属组sandbox

高风险 + 仅隔离 backend

  • 只接受隔离 backend(e2b);local_process / pod_process 直接返回 isolated_sandbox_required——snippet 执行不允许落在共享 pod 文件系统上。
  • 支持 python(别名 py/python3)、javascriptjs/node/nodejs)、bashsh/shell), 其它返回 unsupported language

返回 ok / backend / sandbox_id / execution_model="snippet_runner" / boundary / language / path / command / stdout / stderr / exit_code

agent.py
from ksadk.toolsets.sandbox import run_code

result = run_code(
    "import math; print(math.sqrt(16))",
    language="python",
    approval={"approved": True},
)
assert result["ok"]

E2B Sandbox Backend

Sandbox Runtime 架构:调用方共用底座、backend 选择、E2B readiness 重试、Registry 复用与 TTL 回收

KSADK_SANDBOX_BACKEND=e2b(或配置了 template id 自动选 e2b)时, create_sandbox_backend() 返回 E2BSandboxBackend,通过 e2b.Sandbox.create(template=..., envs=..., allow_internet_access=...) 拉起隔离沙箱。 启动期重试 true 命令与就绪探测,最多 KSADK_SANDBOX_STARTUP_RETRY_ATTEMPTS (默认 6)次,指数退避 KSADK_SANDBOX_STARTUP_RETRY_DELAY(默认 0.2s)。

环境变量作用默认
KSADK_SANDBOX_TEMPLATE_ID / KSADK_SKILL_RUNTIME_TEMPLATE_IDE2B 模板 id(必填,缺失抛 SandboxError
KSADK_SANDBOX_TIMEOUTsandbox 存活秒数900
KSADK_SANDBOX_TYPEaio / threadSandboxTypeaio
KSADK_SANDBOX_ALLOW_INTERNET_ACCESS是否允许出网true
KSADK_SANDBOX_STARTUP_RETRY_ATTEMPTS启动重试次数6
KSADK_SANDBOX_STARTUP_RETRY_DELAY启动重试退避基数(秒)0.2
KSADK_SANDBOX_SESSION_ID固定复用 session keyToolExecutionContext.session_id 推断

E2B 需要 e2b>=2.0.0。未安装时 create_sessionSandboxError("e2b>=2.0.0 is required for KSADK_SANDBOX_BACKEND=e2b")

会话复用由 GLOBAL_SANDBOX_REGISTRY 管理:同一 session key 的 run_command/run_code 共享 sandbox,空闲超过 KSADK_SANDBOX_IDLE_TTL_SECONDS 回收。隔离 backend 还会把 workspace 根目录文件同步进 /workspace/,受 KSADK_SANDBOX_SYNC_MAX_FILES / KSADK_SANDBOX_SYNC_MAX_FILE_BYTES / KSADK_SANDBOX_SYNC_MAX_TOTAL_BYTES 限制,并跳过 .git / __pycache__ / .venv / node_modules / .pytest_cache

Web 工具参考

ksadk.toolsets.web 暴露 web_fetchweb_search,风险等级均为 low, 但仍走 Tool Gateway,可被 strict 模式统一拦截。

web_fetch

抓取公开 HTTP(S) URL,返回剥离 script/style/noscript 后的纯文本。

签名
web_fetch(
    url: str,
    max_chars: int | None = None,
    timeout: int = 30,
) -> dict
风险等级low
是否需要审批否(但受 SSRF 策略约束)
副作用无(只读 HTTP GET)
所属组web

SSRF 防护

_validate_public_http_url 在请求前做三层校验:

  1. scheme 必须是 http/https,host 必填。
  2. socket.getaddrinfo 解析所有 A/AAAA 记录。
  3. 每个 IP 用 ipaddress.ip_address 检查 is_loopback / is_private / is_link_local / is_multicast / is_reserved,命中即返回 blocked_by_ssrf_policy

重定向逐跳重新校验,最多 5 跳,超出返回 too_many_redirectsKSADK_WEB_SSRF_POLICY_JSON(或 OPENCLAW_BROWSER_SSRF_POLICY_JSON)可传 {"allow_private": true} 关闭私网拦截——仅在受信内网环境使用。

返回 ok / url / final_url / status_code / content_type / text / truncated / original_chars / preview_chars, 超阈值时附 persisted={"path": ..., "mime_type": ...}

agent.py
from ksadk.toolsets.web import web_fetch

doc = web_fetch("https://example.com/docs", max_chars=20000)
if doc["ok"]:
    print(doc["final_url"], doc["content_type"])
    if doc.get("truncated"):
        print("persisted at", doc["persisted"]["path"])

返回 title/url/snippet 搜索结果,不抓取整页。provider 由 KSADK_WEB_SEARCH_PROVIDER(或 OPENCLAW_WEB_SEARCH_PROVIDER)选择, 未配置时返回 provider_not_configured

签名
web_search(
    query: str,
    max_results: int = 5,
    recency_days: int | None = None,  # 仅 http provider
) -> dict
风险等级low
是否需要审批
副作用无(只读搜索 API)
所属组web
provider说明凭证变量
fake测试 provider,返回固定假结果,无需凭证
http通用 HTTP 搜索 API(GET,Bearer 鉴权,支持 recency_daysKSADK_WEB_SEARCH_API_KEY
ksyun金山云 AI 搜索(POST,scope=webpage/document/scholar/podcast/videoKSADK_WEB_SEARCH_API_KEY / KSADK_MCP_KEY / KSC_AIPRO_API_KEY

返回 ok / query / provider / results[](每项含 title / url / snippet / rank / providerksyun 额外有 date)。

环境变量作用
KSADK_WEB_SEARCH_PROVIDERfake / http / ksyun
KSADK_WEB_SEARCH_BASE_URLhttp/ksyun endpoint(ksyun 默认 https://search.aipro.ksyun.com/v1/aisearch/search
KSADK_WEB_SEARCH_API_KEYBearer token
KSADK_WEB_SEARCH_SCOPEksyun scope,默认 webpage
agent.py (ksyun)
import os
os.environ["KSADK_WEB_SEARCH_PROVIDER"] = "ksyun"
os.environ["KSADK_WEB_SEARCH_API_KEY"] = "..."

from ksadk.toolsets.web import web_search

res = web_search("ksadk agent runtime", max_results=5)
for item in res["results"]:
    print(item["rank"], item["title"], item["url"])
agent.py (http)
import os
os.environ["KSADK_WEB_SEARCH_PROVIDER"] = "http"
os.environ["KSADK_WEB_SEARCH_BASE_URL"] = "https://search.example.com/v1/search"
os.environ["KSADK_WEB_SEARCH_API_KEY"] = "..."

from ksadk.toolsets.web import web_search

res = web_search("agent sdk", max_results=3, recency_days=7)
agent.py (fake)
import os
os.environ["KSADK_WEB_SEARCH_PROVIDER"] = "fake"

from ksadk.toolsets.web import web_search

res = web_search("anything")  # 测试用,不发起真实请求

ToolResultBudget

ksadk.tools.result_budget 对长输出做截断 + 持久化,避免把超大 stdout/text 灌进模型上下文。 run_command / run_codestdout/stderr 调用 budget_text_fieldsweb_fetchtext 调用 budget_tool_output

ToolResultBudget 截断与持久化

环境变量作用默认
KSADK_TOOL_RESULT_MAX_CHARS单字段最大字符数50000
KSADK_TOOL_RESULT_PREVIEW_CHARS持久化时返回的预览长度8000
KSADK_TOOL_RESULT_PERSIST_THRESHOLD_CHARS超过此长度则落盘max_chars
KSADK_TOOL_RESULT_DIR持久化目录<session_dir>/tool-results

返回结构在原字段旁追加 <field>_truncated / <field>_original_chars / <field>_preview_chars,落盘字段放在 persisted_outputs[<field>] = {"path", "mime_type"}。 路径始终在 persist_dir 内,越界会抛 ValueError

测试工具集成

公开测试应覆盖边界,而不是私有服务账号:

  • 工具 schema 转换。
  • 使用确定性输入的本地成功执行。
  • timeout 和失败处理。
  • runner payload 字段。
  • Web UI 展示工具结果时的请求形态。
  • audit 检查 fixture 中没有凭证或私有 endpoint。

优先使用 fake client、临时文件和本地 HTTP server。provider-backed 测试必须放在显式环境变量开关之后。

发布前安全检查

发布工具或 skill 示例前,至少检查:

  • 示例不依赖内部账号。
  • 私有 endpoint 和客户数据已移除。
  • 生成文件已加入 ignore 或不在仓库内。
  • 必需 optional extras 已记录。
  • 执行时间和文件大小有限制。
  • 没有宽泛 shell、网络或文件系统访问。
  • 已运行开源审计。

账号边界与 invocation_id

0.6.7 新增

运行时统一写入 PlatformInvocationContext,memory、sandbox、workspace 和 skill 都按账号边界读取上下文。

进入 runner 的请求可以携带账号标识。OpenAI 兼容入口 /v1/responses/v1/chat/completions 和内部 RunAgentAction 都接受 account_id(A2A/gRPC 侧为 AccountId),运行时把它写入 PlatformInvocationContext,供下游按账号 边界隔离数据。

入口字段说明
/v1/responsesaccount_idOpenAI 兼容 Responses API
/v1/chat/completionsaccount_idOpenAI 兼容 Chat Completions API
RunAgentActionAccountIdrunner / A2A / gRPC 内部动作

下游模块通过运行时提供的上下文读取器拿当前账号,而不是各自从请求头解析:

agent.py
from ksadk.runtime_context import get_current_account_id, get_current_invocation_context

account_id = get_current_account_id()
ctx = get_current_invocation_context()

PlatformInvocationContext 至少包含:

  • account_id:当前账号边界。
  • agent_id / user_id / session_id:调用关联标识。
  • 可选的 attachment、memory、knowledge 等上下文字段。

单次运行的 invocation_idToolExecutionContext 携带,与上面上下文一起贯穿一次执行。

memory、sandbox、workspace 和 skill 在执行工具或读写状态时,会读取当前 PlatformInvocationContext,确保不同账号之间的数据、文件和技能执行互不 串扰。

invocation_id 作为单次运行的稳定 ID,贯穿整条链路:

invocation_id 贯穿链路

  • runner payload 带上 invocation_id,便于本地与远端对齐。
  • SubscribeRunEventsCancelRuninvocation_id 定位运行实例。
  • trace 与观测数据用同一个 invocation_id 串联,便于排查卡顿和失败。

账号边界优先用上下文读取器

工具和业务代码应调用 get_current_account_id() / get_current_invocation_context(),而不是手动解析请求头或自行缓存 account_id,避免在沙箱或异步分支里拿到错误的账号。

更多上下文字段和读取方式见 智能体上下文,HTTP 入口字段 见 OpenAI 兼容 API

相关指南

建议结合这些页面阅读:

本页导航