ExtSkill（扩展技能）API

⚠️ 尚未上线：本能力为 v1.5.0-SNAPSHOT 开发中特性，接口与协议可能调整，正式发布前请勿用于生产环境。

v1.5.0 新增。所有类均从 linksoul_agentsdk 顶层重新导出：from linksoul_agentsdk import ExtSkillRequest, ExtSkillQueryCallback, ExtSkillInvokeCallback。使用流程见 ExtSkill 指南。

类图

text

AgentSdk
  ├── register_ext_skill(ExtSkillRequest)               # 关联 LinkskyClient
  └── unregister_ext_skill(request_id: str) -> None

ExtSkillRequest
  ├── __init__(agent_sdk, agent_id, request_id)
  ├── query(query_callback: ExtSkillQueryCallback, timeout: int) -> str         # 返回 event_id
  └── invoke(ext_skill_id: str, input_param: AgentParam,
             invoke_callback: ExtSkillInvokeCallback, timeout: int) -> str      # 返回 event_id

ExtSkillQueryCallback            # 抽象基类
  └── on_query_result(agent_id, request_id, event_id, code, msg, result: AgentParam | None) -> None

ExtSkillInvokeCallback           # 抽象基类
  ├── on_invoke_ack(agent_id, request_id, event_id, code, msg) -> None
  ├── on_state(agent_id, request_id, event_id, state_name, state_value) -> None
  └── on_invoke_result(agent_id, request_id, event_id, code, msg, param: AgentParam | None) -> None

IdGenerator
  └── generate_ext_skill_request_id() -> str    # "extskill_<21 位随机字符>"

`AgentSdk`（新增方法）

方法	说明
`register_ext_skill(request: ExtSkillRequest) -> None`	将 request 与当前 SDK 的 `LinkskyClient` 绑定，并加入全局 `AgentSdkExtSkillMgr` 索引。必须在调用 `query` / `invoke` 之前调用。
`unregister_ext_skill(request_id: str) -> None`	从索引中移除。用完或不再需要该会话时调用。

`ExtSkillRequest`

python

ExtSkillRequest(agent_sdk: AgentSdk, agent_id: str, request_id: str)

方法	语义
`query(query_callback, timeout) -> str`	查询该智能体当前可用的扩展技能清单。`timeout < 50` 会被 clamp 到 50（当前仅存储不生效）。返回本次消息 `event_id`。结果通过 `query_callback.on_query_result` 回传。
`invoke(ext_skill_id, input_param, invoke_callback, timeout) -> str`	触发指定扩展技能执行。`ext_skill_id` 为技能 ID；`input_param` 为结构化入参（`AgentParam`）；`timeout` 处理规则同 `query`。返回本次消息 `event_id`。先经 `invoke_callback.on_invoke_ack` 回受理，再经 `invoke_callback.on_invoke_result` 回执行输出。
`update_ts` (property)	最近一次消息发送时间（毫秒）。7200s 无活动会被后台清理。
`agent_id` / `request_id` (property)	会话身份。
`query_callback` / `invoke_callback` (property)	最近一次绑定的回调对象。

回调错误码

`code`	含义
`0`	成功（网关业务层 ACK；`result` / `param` 承载返回内容）
`-1`	服务端返回失败，`msg` 携带 `errorMsg` 详情
`1000`	本地失败：发送时 SDK 侧 `LinkskyClient` 已 closed；SDK 同步在原线程回调该 code

`ExtSkillQueryCallback`

抽象基类，继承并覆盖 on_query_result：

python

from linksoul_agentsdk import ExtSkillQueryCallback, AgentParam


class MyQueryCallback(ExtSkillQueryCallback):
    def on_query_result(self, agent_id: str, request_id: str, event_id: str,
                        code: int, msg: str, result: AgentParam | None) -> None:
        # code == 0 成功，result 承载技能清单；-1 服务端拒绝；1000 本地连接关闭（result == None）
        ...

`ExtSkillInvokeCallback`

抽象基类，invoke 分三段回调：on_invoke_ack（受理）→ on_state（执行期间 0 到多次状态上报，默认空实现、可不覆盖）→ on_invoke_result（执行输出）。

python

from linksoul_agentsdk import ExtSkillInvokeCallback, AgentParam


class MyInvokeCallback(ExtSkillInvokeCallback):
    def on_invoke_ack(self, agent_id: str, request_id: str, event_id: str,
                      code: int, msg: str) -> None:
        # 第一段：网关已受理本次 invoke（code == 0）
        ...

    def on_state(self, agent_id: str, request_id: str, event_id: str,
                 state_name: str | None, state_value: str | None) -> None:
        # 第二段：执行期间技能侧通过 agentsdk.ext_skill.report_state 上报的进度，0 到多次（可选覆盖）
        ...

    def on_invoke_result(self, agent_id: str, request_id: str, event_id: str,
                         code: int, msg: str, param: AgentParam | None) -> None:
        # 第三段：技能执行输出，code == 0 时 param 承载出参
        ...

两个回调基类还包含 timeout 字段（由 set_timeout(...) 在 SDK 内部注入，clamp 到 ≥50ms 的占位值）。

`IdGenerator.generate_ext_skill_request_id()`

生成形如 extskill_ABCDEFGHIJKLMNOPQRSTU 的会话 ID（前缀 + 21 位随机字符）。与 flow_ / event_ / teleop_ 等其他前缀互不冲突。

事件类型（`AgentEventType`）

见枚举参考中的 AGENTSDK_EXT_SKILL_* 段。