Architecture Overview

← Home

The Python SDK shares the same architecture as the Java SDK. This page focuses on Python-specific notes.

System architecture

text

┌─────────────────────────────────────────────────────────────────────────────────────────────────┐
│                                   LinkSoul Platform (灵心平台)                                     │
│                                                                                                 │
│  ┌─────────────────────────┐       Bridge (signaling)     ┌─────────────────────────┐           │
│  │ LinkskyGateway          │ ◄───────────────────────────►│  Interaction Gateway    │           │
│  │ (二开网关 / Secondary    │   Boss Channel (control)     │  (交互网关)              │           │
│  │  Dev Gateway)           │   Worker Channel (data)       │                         │           │
│  │ Accepts SDK connections │   Bidirectional signaling     │ Connects to robots      │           │
│  │ Auth & message routing  │   handshake                   │ Audio/video/cmd relay    │           │
│  └────────────┬────────────┘                              └────────────┬────────────┘           │
│               │                                                        │                        │
└───────────────│────────────────────────────────────────────────────────│────────────────────────┘
                │                                                        │
                │ WebSocket (wss)                                        │ WebSocket / TCP
                │ wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk                  │
                │                                                        │
┌───────────────▼────────────────┐                          ┌────────────▼─────────────┐
│  Your Application (AgentSdk)   │                          │   Robot / Agent Endpoint  │
│                                │                          │                          │
│  ┌──────────────────────────┐  │                          │  ┌────────────────────┐  │
│  │  LinkskyClient           │  │                          │  │  Microphone (MIC)  │  │
│  │  (websocket-client)      │  │                          │  │  Camera            │  │
│  └───────────┬──────────────┘  │                          │  │  Speaker (TTS)     │  │
│              │                 │                          │  │  Motor (Actions)   │  │
│  ┌───────────▼──────────────┐  │                          │  └────────────────────┘  │
│  │  MessageRouter           │  │                          └──────────────────────────┘
│  │  (handlers.py dispatch)  │  │
│  └───────────┬──────────────┘  │
│              │                 │
│  ┌───────────▼──────────────┐  │
│  │  PassiveCallback         │  │     Passive: Robot initiates → SDK processes → result
│  │  (8 business subclasses) │  │     Audio → ASR → LLM → TTS
│  ├──────────────────────────┤  │
│  │  TaskFlowRequest         │  │     Active: SDK initiates → gateway → robot responds
│  │  (active task flow)      │  │     register_task_flow / start_request / task_request / end_request
│  └──────────────────────────┘  │
└────────────────────────────────┘

Data flow

Passive (Robot → SDK)

text

Robot Mic → Interaction GW → Bridge Worker → LinkskyGateway → AgentSdk → callback.on_request()
                                                                              │
Callback returns → AgentSdk → LinkskyGateway → Bridge Worker → Interaction GW → Robot Speaker

Active (SDK → Robot)

text

agent_sdk.register_task_flow() → LinkskyGateway → Bridge Boss → Interaction GW → Robot
                                                                          │
Robot responds → Interaction GW → Bridge Boss → LinkskyGateway → AgentSdk → callback.on_*_ack()

Module layout

text

linksoul_agentsdk/
├── __init__.py                 # exposes 11 top-level symbols (AgentSdk, AgentParam, ...)
├── agent_sdk.py                # main entry (singleton factory, lifecycle)
├── agent_param.py / agent_meta.py
├── agent_auth_callback.py
├── id_generator.py             # secrets-backed unique-id generator
├── client/linksky_client.py    # HMAC-SHA256 gateway signing
├── enums.py                    # 5 enums
├── client/                     # WebSocket layer
│   ├── linksky_client.py       # sync blocking WS client + reconnect + heartbeat
│   ├── message_router.py       # dispatch JSON by event_type
│   └── handlers.py             # per-event handler functions
├── passive/                    # 8 Callback + Response pairs + GreetResponse
├── active/                     # TaskFlowRequest + 4 callbacks
└── mgr/                        # 4 managers (callback type / response / task flow / agent thread)

Communication protocol

Auth headers

AGIBOT_APP_ID: <app_id>
AGIBOT_AUTHORIZATION: Bearer <JWT>

JWT payload: {"appId": "...", "callbackTypes": "[\"asr2llm\"]", "exp": <epoch+3600>}

Message format (JSON)

json

{
    "type": "agentsdk.audio_request.start",
    "agentId": "AGENT_0000001",
    "eventId": "event_xxxxx",
    "robotCid": "cid_xxx",
    "cid": "cid_xxx"
}

Threading model

Thread	Count	Responsibility
`linksky-recv-<app_id>`	1	WebSocket receive thread running `ws.run_forever`
`agent-worker-N`	1..50	One serial worker per `agent_id` (the agent ID configured on the LinkSoul platform); ordered per-agent message processing
`threading.Timer` daemon	0..1	Reconnect scheduler

All response.on_* calls inside on_request run on the agent worker. LinkskyClient.send_text uses an internal threading.Lock, so it is safe to call from any thread.

Reconnection

Auto-reconnect on disconnect
Reconnect delay: 3 s (reconnect_delay, overridable in constructor)
Unlimited attempts by default (max_reconnect_attempts=-1)
Heartbeat: ping every 10 s (built-in to websocket-client)

Typical sequence

Below is the end-to-end flow of one conversation using Audio2Tts (audio in → ASR + LLM + TTS).

mermaid

sequenceDiagram
    participant Robot
    participant IGW as Interaction GW
    participant Bridge as Bridge Relay
    participant Linksky as LinkskyGateway
    participant SDK as AgentSdk

    Note over SDK,Linksky: Startup
    SDK->>Linksky: WebSocket connect + JWT
    Linksky-->>SDK: on_auth_state(code=0)
    Robot->>IGW: robot online
    IGW->>Bridge: online event
    Bridge->>Linksky: online event
    Linksky-->>SDK: on_robot_online(agent_id, agent_meta)

    Note over Robot,SDK: One conversation
    Robot->>IGW: user starts speaking (VAD start)
    IGW->>Linksky: audio_request.start
    Linksky-->>SDK: on_request(flag=0)
    loop audio frames
        Robot->>IGW: PCM frame
        IGW->>Linksky: audio_request.append
        Linksky-->>SDK: on_request(flag=1, buf=audio)
    end
    Robot->>IGW: VAD end
    IGW->>Linksky: audio_request.commit
    Linksky-->>SDK: on_request(flag=2)

    Note over SDK: business ASR → LLM → TTS
    SDK->>Linksky: response.on_asr_middle / on_asr_final
    SDK->>Linksky: response.on_interrupt(event_id, "chat", None)
    SDK->>Linksky: response.on_llm_item_delta × N
    SDK->>Linksky: response.on_llm_item_done / on_llm_done
    SDK->>Linksky: response.on_tts_item_delta × N
    SDK->>Linksky: response.on_tts_item_done / on_tts_done
    Linksky->>IGW: forward
    IGW->>Robot: render + speak

Key points:

A whole conversation is tied together by a single event_id.
ASR/LLM/VLM/TTS are streaming — emit *_delta zero-or-more times and close with *_done.
For valid intents you MUST call on_interrupt before on_skill / on_llm_* / on_vlm_* / on_tts_*.
Skill dispatch (on_skill) can run in parallel with text/audio output. See Semantic Skills.
Report business failures via response.on_error(event_id, code, msg).