Active Interaction Examples

← Home
Source: agentsdk_for_python/examples/task_flow_example.py

v1.3.0 first exposed the active surface (alongside task flow it also shipped skill-query, state-query, pull-stream and push-listen) — but those companion entries were never functional at runtime, only task flow worked. v1.4.0 therefore removes the never-enabled entries and narrows the public surface to Task Flow, so that's the single active-side example shipped.

Active interaction skeleton

python

import time
from linksoul_agentsdk import AgentAuthCallback, AgentSdk


class _Auth(AgentAuthCallback):
    def on_auth_state(self, app_id, code, msg):
        print(f"on_auth_state => app_id={app_id} code={code} msg={msg}")


sdk = AgentSdk.create(
    url="wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk",
    app_id="<your appId>",
    app_key="<your appKey>",
    app_secret="<your appSecret>",
)
sdk.register_auth(_Auth())

# Active requests must be issued AFTER initialize()
sdk.initialize()

time.sleep(2)   # wait for auth + agent online

Note: unlike passive callbacks, active requests must run after initialize().

TaskFlowExample — start / task / end / interrupt

Source: task_flow_example.py
TaskFlowRequest models a multi-step robot task as a pipeline: start_request (required) → task_request (optional, repeatable) → end_request (optional) → interrupt_request (optional). Each call has two-phase acknowledgement: RequestAck = received, ExecuteAck = finished.
The placeholder "[{}]" in this example just illustrates the call shape; real payloads should follow the Task-flow payload protocol — populate action_group / action_type entries there.

1) Register the flow

python

from linksoul_agentsdk import IdGenerator
from linksoul_agentsdk.active import TaskFlowRequest

agent_id = "<the target agentId>"
flow_id = IdGenerator.generate_flow_id()
flow = TaskFlowRequest(sdk, agent_id, flow_id)
sdk.register_task_flow(flow)

2) Interrupt (optional, callable at any stage)

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 => {flow_id} {event_id} code={code} msg={msg}")


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

3) Start request (required)

python

from linksoul_agentsdk.active import FlowStartCallback


class _Start(FlowStartCallback):
    def on_start_request_ack(self, flow_id, event_id, code, msg):
        print(f"start ack => {flow_id} {event_id} code={code} msg={msg}")

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


flow.start_request("[{}]", _Start(), timeout=2000)

4) Task request (optional, repeatable)

python

from linksoul_agentsdk.active import FlowTaskCallback


class _Task(FlowTaskCallback):
    def on_task_request_ack(self, flow_id, event_id, code, msg):
        print(f"task ack => {flow_id} {event_id} code={code} msg={msg}")

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


flow.task_request("[{}]", _Task(), timeout=2000)

5) End request (optional)

python

from linksoul_agentsdk.active import FlowEndCallback


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


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

Callback summary

Callback	Trigger	Meaning
`on_start_request_ack`	Start request reached the robot	`code=0` means accepted
`on_start_execute_ack`	Start phase finished	Ready for `task_request` calls
`on_task_request_ack`	A task step reached the robot	The step request was accepted
`on_task_execute_ack`	The task step finished	Move on to the next step
`on_end_request_ack`	End request processed	The pipeline is closed
`on_interrupt_request_ack`	Interrupt processed	The robot was told to abort

SIMPLE vs COMPLEX

python

from linksoul_agentsdk.enums import FlowMode

# COMPLEX: start → task × N → end (multi-step)
flow = TaskFlowRequest(sdk, agent_id, flow_id)
flow = TaskFlowRequest(sdk, agent_id, flow_id, FlowMode.COMPLEX)

# SIMPLE: start_request carries the full payload, no task_request
flow = TaskFlowRequest(sdk, agent_id, flow_id, FlowMode.SIMPLE)

State listening (via the passive callback)

State pushes have been folded into PassiveCallback.on_state(...). Override it on any registered passive callback — see Passive Callbacks Guide — shared base callbacks.

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}")

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