ExtSkill API (extended skills)

⚠️ Not yet released: in-development v1.5.0-SNAPSHOT capability; interface and protocol may change. Do not use in production before the official release.

New in v1.5.0. All classes are re-exported from the top-level package: from linksoul_agentsdk import ExtSkillRequest, ExtSkillQueryCallback, ExtSkillInvokeCallback. See the ExtSkill guide for semantics.

Class diagram

text

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

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

ExtSkillQueryCallback            # abstract base
  └── on_query_result(agent_id, request_id, event_id, code, msg, result: AgentParam | None) -> None

ExtSkillInvokeCallback           # abstract base
  ├── 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 random chars>"

`AgentSdk` (new methods)

Method	Description
`register_ext_skill(request: ExtSkillRequest) -> None`	Binds the request to the current SDK's `LinkskyClient` and adds it to the global `AgentSdkExtSkillMgr` index. Must be called before `query` / `invoke`.
`unregister_ext_skill(request_id: str) -> None`	Removes it from the index. Call when done.

`ExtSkillRequest`

python

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

Method	Semantics
`query(query_callback, timeout) -> str`	Query the extended skills currently available on the agent. `timeout < 50` is clamped to 50 (stored but currently inert). Returns the message `event_id`. Result comes back via `query_callback.on_query_result`.
`invoke(ext_skill_id, input_param, invoke_callback, timeout) -> str`	Trigger a skill to run. `ext_skill_id` is the skill ID; `input_param` is structured input (`AgentParam`); `timeout` handled as in `query`. Returns the message `event_id`. `invoke_callback.on_invoke_ack` fires first (accept), then `invoke_callback.on_state` (execution-time progress, zero or more times), then `invoke_callback.on_invoke_result` (output).
`update_ts` (property)	Last-send timestamp in ms. Idle > 7200 s gets swept.
`agent_id` / `request_id` (property)	Session identity.
`query_callback` / `invoke_callback` (property)	Most recently bound callbacks.

Callback codes

`code`	Meaning
`0`	Success (gateway business ACK; `result` / `param` carries content)
`-1`	Server returned failure; `msg` carries `errorMsg`
`1000`	Local failure: `LinkskyClient` was closed at send time; the SDK invokes this code synchronously on the calling thread

`ExtSkillQueryCallback`

Abstract base; subclass and override 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 success, result carries the skill list; -1 server reject; 1000 local close (result is None)
        ...

`ExtSkillInvokeCallback`

Abstract base; invoke reports in three stages. on_state is execution-time state reporting (zero or more times) with a default no-op implementation, so it is optional to override:

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:
        # stage 1: gateway accepted this 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:
        # stage 2: execution-time progress reported via agentsdk.ext_skill.report_state; zero or more times (optional)
        ...

    def on_invoke_result(self, agent_id: str, request_id: str, event_id: str,
                         code: int, msg: str, param: AgentParam | None) -> None:
        # stage 3: skill execution output; param carries the result when code == 0
        ...

Both callback base classes also carry a timeout field (SDK-injected via set_timeout(...), clamped placeholder >=50 ms).

`IdGenerator.generate_ext_skill_request_id()`

Returns "extskill_<21 secrets chars>". Distinct from other prefixes (flow_, event_, teleop_).

Event types (`AgentEventType`)

See the AGENTSDK_EXT_SKILL_* block in Enums.