Active Interaction Development Guide

← Home

Active interaction lets the SDK issue requests to the robot. v1.3.0 first exposed the active surface — alongside task flow it also shipped skill-query, state-query, pull-stream and push-listen interfaces, but those were never enabled at runtime; only task flow was ever functional. v1.4.0 therefore removes the never-enabled entries from the public API and narrows the surface to Task Flow. The corresponding classes remain in the source tree but are no longer exposed via AgentSdk.register*.

State push, greeting signals, video passthrough, face / voiceprint UID have all been folded into the passive callbacks.

Task flow

Task flow dispatches a multi-step task sequence to a robot. Two modes are supported:

SIMPLE: single-step (start_request carries the full payload; no task_request needed)
COMPLEX: multi-step (start → task × N → end)

Full lifecycle (COMPLEX mode)

python

from linksoul_agentsdk import AgentSdk, IdGenerator
from linksoul_agentsdk.enums import FlowMode
from linksoul_agentsdk.active import (
    FlowEndCallback,
    FlowStartCallback,
    FlowTaskCallback,
    TaskFlowRequest,
)


# 1) Build a flow (specify the target agent and flow id)
flow_id = IdGenerator.generate_flow_id()
flow = TaskFlowRequest(agent_sdk, "AGENT_001", flow_id, FlowMode.COMPLEX)

# 2) Register with the SDK
agent_sdk.register_task_flow(flow)


# 3) Start (payload is a JSON array; see taskflow_payload.md)
class _Start(FlowStartCallback):
    def on_start_request_ack(self, flow_id, event_id, code, msg):
        print(f"start ack => code={code}, msg={msg}")

    def on_start_execute_ack(self, flow_id, event_id, code, msg):
        print(f"start execute => code={code}, msg={msg}")


flow.start_request('[{"goal":"navigate_to_kitchen"}]', _Start(), timeout=10000)


# 4) Dispatch task steps (any number)
class _Task(FlowTaskCallback):
    def on_task_request_ack(self, flow_id, event_id, code, msg):
        print(f"task ack => code={code}, msg={msg}")

    def on_task_execute_ack(self, flow_id, event_id, code, msg):
        print(f"task execute => code={code}, msg={msg}")


flow.task_request('[{"action":"turn_left","angle":90}]', _Task(), timeout=30000)


# 5) End
class _End(FlowEndCallback):
    def on_end_request_ack(self, flow_id, event_id, code, msg):
        print(f"end ack => code={code}, msg={msg}")


flow.end_request(_End(), timeout=10000)


# 6) Unregister when done
agent_sdk.unregister_task_flow(flow_id)

Interrupt

python

from linksoul_agentsdk.active import FlowInterruptCallback


class _Interrupt(FlowInterruptCallback):
    def on_interrupt_request_ack(self, flow_id, event_id, code, msg):
        print(f"interrupt ack => code={code}, msg={msg}")


flow.interrupt_request(_Interrupt(), timeout=10000)

Callback parameters

Parameter	Description
`flow_id`	Unique flow identifier
`event_id`	Unique id for this specific request
`code`	0 = success, non-zero = failure
`msg`	`"success"` on success, error description otherwise

Payload protocol (skill orchestration)

The payload argument of start_request / task_request is a JSON array string of one or more action_group items, each containing a list of action entries (TTS, motion, expression, navigation, settings…). Even when you dispatch a single action_group, wrap it in an array ('[{...}]'); a bare object is not accepted. For the full protocol see:

👉 Task-flow payload protocol

State listen

v1.4.0 change: state listening has been folded into the passive callback's on_state(...). There is no separate registration.

Override the method on whichever passive callback you already register:

python

from linksoul_agentsdk.passive import Audio2TtsCallback


class MyCb(Audio2TtsCallback):
    def on_state(self, agent_id, event_id, state_name, state_value):
        print(f"state change: {state_name} = {state_value}")
        # state_value may be JSON — parse if you need structured data

    def on_request(self, *a, **k):
        ...

Timeouts

Each TaskFlowRequest method takes a timeout in milliseconds; if no response arrives in time the callback is not invoked and resources are auto-cleaned:

Operation	Recommended timeout	Lifecycle
`start_request` / `task_request` / `end_request` / `interrupt_request`	Application-defined	The whole task flow is auto-cleaned after 2 hours

timeout < 50 is clamped to 50 ms.

APIs not exposed in this release

Skill query (query_skill), state query (query_state), pull stream (register_pull_stream) and push listen (register_push_listen) have no public registration entry in v1.4.0. The underlying classes (SkillQueryRequest, StateQueryRequest, PullStreamRequest, PushListenCallback) are not exposed either.