Edge SDK (Python) — Edge Adapter

Implementing an edge adapter in Python boils down to subclassing EdgeAdapter and overriding the methods your hardware supports. Methods you don't override automatically return EdgeResponse.not_supported(...) and are reported as unavailable in get_capabilities.

For Java, see edge-sdk-adapter.md.

The contract

from edge_sdk import EdgeAdapter

class MyAdapter(EdgeAdapter):
    async def get_capabilities(self, sn: str, asset_id: str | None) -> Capabilities:
        ...

get_capabilities is the only abstract method. Use the helper:

return self._auto_capabilities(sn, AssetType.DOCK)

_auto_capabilities introspects which methods you've overridden and produces a Capabilities payload that mirrors that exactly. You only maintain the implementation list — the capability list stays in sync automatically.

RequestContext

Every command receives a RequestContext as the first argument:

@dataclass
class RequestContext:
    tid: str        # transaction id (correlate response/progress to request)
    sn: str         # asset serial number
    timestamp: datetime
    metadata: dict[str, str]

Always pass ctx.tid and ctx.sn back into your EdgeResponse to keep the platform's correlation working.

EdgeResponse

The unified response type used by every unary command:

EdgeResponse.success(tid, sn, message="Takeoff initiated")
EdgeResponse.error(tid, sn, message="Hardware fault", code=ErrorCode.HARDWARE_ERROR)
EdgeResponse.not_supported(tid, sn)            # default for un-overridden methods
EdgeResponse.in_progress(tid, sn, progress=CommandProgress(percent=42, status="climbing"))

For long-running commands you can stream multiple EdgeResponse objects via the streaming variants (see below).

Method groups

The base class organises methods into groups. You only override what your hardware supports.

Capability management (required)

Flight control

Manual control

Dock and asset operations

Camera

Live streaming

Tasks

Streaming responses

For commands that produce progress updates (typically tasks), override the streaming variants:

from collections.abc import AsyncIterator

async def start_task_stream(
    self, ctx: RequestContext, task: Task,
) -> AsyncIterator[EdgeResponse]:
    yield EdgeResponse.in_progress(ctx.tid, ctx.sn, CommandProgress(percent=0))
    async for update in self._run(task):
        yield EdgeResponse.in_progress(ctx.tid, ctx.sn, update)
    yield EdgeResponse.success(ctx.tid, ctx.sn, "Task complete")

Best practices

• Keep methods async-friendly. Wrap blocking SDK calls with asyncio.to_thread(...) or use the vendor SDK's async API.
• Always honour ctx.tid and ctx.sn when constructing responses; the platform correlates by these.
• Don't catch broad exceptions silently. Convert known hardware errors to EdgeResponse.error(...) with a meaningful ErrorCode; let the gRPC server surface the rest.
• Don't keep state on the adapter for per-request lifetimes. Use the tid as a key into a per-command map if you must.
• Use _auto_capabilities rather than maintaining the capability list manually.

Testing your adapter

EdgeServer works in-process with a real grpc.aio server, so a typical test:

import pytest
from edge_sdk import EdgeServer
from edge_sdk.generated import edge_pb2_grpc, common_pb2

@pytest.mark.asyncio
async def test_takeoff(monkeypatch):
    server = EdgeServer(adapter=MyDeviceAdapter(), port=0)
    addr = await server.start()
    # ... use a generated stub to call take_off and assert on the response
    await server.stop()

Or use the lower-level dispatcher with a MagicMock for purely unit-style tests; see the SDK's own tests/ directory for examples.