ShydaSHYDAPLATFORM
Vega
Capability Development

Complete example: unitree_go2_speak

A complete Capability that exposes one Action, speech.speak, for a Unitree Go2 deployment. It demonstrates the parts needed for VCP-compatible development: manifest, schemas, configuration, Provider lifecycle, adapter boundary, progress, cancellation, health, recovery declaration, and packaging.

Concepts

  • Public contract: manifest, schemas, and Provider result/error behavior observed by the Runtime.
  • Backend adapter: the Capability-owned interface that hides hardware or service implementation details.
  • Mock backend: a deterministic adapter used for development and tests.
  • Physical termination: what the Provider knows about an underlying physical action after completion or cancellation.

Package structure

text
unitree_go2_speak/
├── capability.yaml
├── pyproject.toml
├── environment.yml
├── requirements-go2.txt
├── config/default.yaml
├── schemas/
│   ├── speak.input.schema.json
│   ├── speak.output.schema.json
│   └── speak.progress.schema.json
└── src/unitree_go2_speak/
    ├── provider.py
    ├── speech_backend.py
    ├── audio_bridge.py
    └── gemini_tts.py
capability.yaml
↓
Vega Runtime
↓
UnitreeGo2SpeakProvider
↓
SpeechBackend interface
MockSpeechBackend
Local backend
Go2 backend

1. Package metadata and dependencies

toml
[project]
name = "unitree-go2-speak"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["vega-capability-sdk>=0.1.0,<0.2.0"]


[project.optional-dependencies]
go2 = ["google-genai>=2.3.0,<3.0"]

The SDK is the portable contract dependency; robot-specific dependencies remain owned by the Capability. A Runtime installation does not need to carry every robot vendor's SDK.

2. Manifest: declare the VCP contract first

yaml
apiVersion: vcp.shyda.ai/v1alpha1
kind: Capability
metadata:
  id: shyda.speech.unitree_go2
  name: Unitree Go2 Speak
  version: 0.1.0
vega:
  sdkVersion: ">=0.1.0,<0.2.0"
provider:
  id: unitree_go2_speak
  runtime: python
  entrypoint: unitree_go2_speak.provider:UnitreeGo2SpeakProvider
compatibility:
  manufacturers: [Unitree]
  models: [Go2]
  modelVariants: [EDU]
  robotTypes: [quadruped]
  architectures: [aarch64]
yaml
resources:
  - id: speaker
    mode: exclusive
actions:
  - id: speech.speak
    version: 1.0.0
    inputSchema: schemas/speak.input.schema.json
    outputSchema: schemas/speak.output.schema.json
    progressSchema: schemas/speak.progress.schema.json
    resources: [speaker]
    execution:
      defaultTimeoutSeconds: 60
      cancellation: true
      pause: false
      resume: false
      progress: true
      idempotency: non_idempotent
      restartRecovery: not_recoverable

speaker is exclusive because overlapping speech output would be ambiguous. The Capability declares non-idempotency and no restart recovery because replaying an interrupted announcement could produce an unintended second audible message.

3. Action self-description

The contract section is a planner-safe, human-readable statement. It says the Action can render text as speech, cannot move or manipulate the robot, requires non-empty text and an available backend, and does not guarantee that a person heard the output.

Speak supplied text
→
Text & backend available?
→
Render speech
→
Structured playback result

4. Schemas

json
{
  "type": "object",
  "additionalProperties": false,
  "required": ["text"],
  "properties": {
    "text": {"type": "string", "minLength": 1, "maxLength": 2000},
    "language": {"type": "string", "default": "en-US"},
    "volume": {"type": "number", "minimum": 0, "maximum": 1}
  }
}

The output schema requires spoken and durationSeconds; backend is optional. The progress schema constrains phase to synthesizing, speaking, or completed, and bounds percentComplete from 0 to 100.

5. Backend interface

python
class SpeechBackend(Protocol):
    async def connect(self) -> None: ...
    async def speak(self, command: SpeakCommand) -> SpeakBackendResult: ...
    async def stop(self) -> None: ...
    async def close(self) -> None: ...
    async def health(self) -> bool: ...

SpeakCommand normalizes text, language, and volume. This boundary makes Provider tests independent from Go2 hardware and lets a mock backend simulate duration, cancellation, health, and failure.

6. Initialization

python
async def on_initialize(self, context: ProviderInitializationContext) -> None:
    if self._backend is None:
        backend_name = str(context.configuration.get("backend", "mock"))
        if backend_name == "mock":
            self._backend = MockSpeechBackend()
        else:
            raise UnsupportedOperationError("Unsupported speech backend")
    await self._backend.connect()

Select dependencies in initialization, not at module import, and fail with a specific error when an unsupported configuration is requested.

7. Execution and cancellation

python
if request.action_id != "speech.speak":
    raise UnsupportedOperationError(f"Unsupported action: {request.action_id}")
text = request.inputs.get("text")
if not isinstance(text, str) or not text.strip():
    raise InvalidInputError("text must be a non-empty string")


speak_task = asyncio.create_task(self._backend.speak(command))
cancel_task = asyncio.create_task(context.cancellation.wait_for_cancel())
done, pending = await asyncio.wait(
    {speak_task, cancel_task}, return_when=asyncio.FIRST_COMPLETED
)
if cancel_task in done:
    await self._backend.stop()
    speak_task.cancel()
    raise CancellationCapabilityError("Speak cancelled")

Always clear transient Provider state — such as the active execution ID — in a finally block, even when the backend raises or cancellation wins.

8. Result and physical termination

python
return CapabilityExecutionResult(
    output={
        "spoken": True,
        "durationSeconds": backend_result.duration_seconds,
        "backend": backend_result.backend_name,
    },
    diagnostics={"language": language, "volume": volume, "characterCount": len(text)},
    physical_termination=PhysicalTerminationStatus.NOT_APPLICABLE,
)

The real package uses UNKNOWN, rather than NOT_APPLICABLE, for its Go2 backend because the bridge completion is not a hard playback acknowledgement. Returning from Python does not prove a physical action completed or stopped.

9. Cancellation, health, shutdown, recovery

The Provider implements cancel(execution_id) to stop only the active matching execution. on_health_check asks the backend for a side-effect-free health fact. on_shutdown closes the backend and clears the reference. For restart recovery, the speech example reports NOT_FOUND, no reattach, no post-restart cancellation, and NON_IDEMPOTENT safety class — correct for an action declared not_recoverable.

10. Validate and test the package

bash
vega-capability validate . --json

Inject MockSpeechBackend and SDK fake contexts for Provider tests. Assert output, progress phases, cancellation behavior, health state, and cleanup. Then run ProviderContractHarness to exercise initialization, health, execution, and shutdown. Hardware-in-the-loop tests must be separate and verify actual robot behavior and emergency-stop procedures.