Direct driver Quick start

This page provides a quick start guide for capability drivers based on DirectAgentDriver. Please refer to Universal Agent main terms for context.

Driver interface

A direct driver is based on:

• DirectAgentDriver
• backend client (AbstractBackendClient)
• target fields storage (AbstractTargetFieldsStorage)

DirectAgentDriver does not use a meta file. It gets resources directly from the backend and stores only target fields for hash calculation and reconciliation.

Quick start

Let's implement a direct driver that works with an HTTP REST API. Objects behind this API are files.

• Data plane: external service with files
• Capability kind: file_target
• Backend client: HTTP client for file API
• Target fields storage: persists target fields by (kind, uuid)

Step 1. Implement backend client

import typing as tp

from gcl_sdk.clients.http import base as http_base
from gcl_sdk.agents.universal.clients.backend import base as client_base
from gcl_sdk.agents.universal.clients.backend import exceptions as client_exc
from gcl_sdk.agents.universal.dm import models


FILE_TARGET_KIND = "file_target"


class FilesRestBackendClient(client_base.AbstractBackendClient):
    def __init__(self, client: http_base.CollectionBaseClient):
        self._client = client

    def _to_view(self, item: dict[str, tp.Any]) -> dict[str, tp.Any]:
        return {
            "uuid": item["uuid"],
            "name": item["name"],
            "path": item["path"],
            "size": item.get("size"),
            "checksum": item.get("checksum"),
        }

    def get(self, resource: models.Resource) -> dict:
        try:
            item = self._client.get(f"/v1/files/{resource.uuid}")
        except Exception:
            raise client_exc.ResourceNotFound(resource=resource)
        return self._to_view(item)

    def list(self, capability: str) -> list[dict]:
        # Full listing of file objects from backend API.
        # API response example:
        # {
        #   "items": [
        #     {"uuid": "...", "name": "...", "path": "...", "size": 128}
        #   ]
        # }
        response = self._client.get("/v1/files/")
        return [self._to_view(item) for item in response.get("items", [])]

    def create(self, resource: models.Resource) -> dict:
        payload = {
            "uuid": str(resource.uuid),
            "name": resource.value["name"],
            "path": resource.value["path"],
        }

        try:
            item = self._client.post("/v1/files/", data=payload)
        except Exception:
            raise client_exc.ResourceAlreadyExists(resource=resource)
        return self._to_view(item)

    def update(self, resource: models.Resource) -> dict:
        payload = {
            "name": resource.value["name"],
            "path": resource.value["path"],
        }

        try:
            item = self._client.patch(f"/v1/files/{resource.uuid}", data=payload)
        except Exception:
            raise client_exc.ResourceNotFound(resource=resource)
        return self._to_view(item)

    def delete(self, resource: models.Resource) -> None:
        try:
            self._client.delete(f"/v1/files/{resource.uuid}")
        except Exception:
            raise client_exc.ResourceNotFound(resource=resource)

Step 2. Implement driver based on DirectAgentDriver

import os

import bazooka

from gcl_sdk.clients.http import base as http_base
from gcl_sdk.agents.universal.drivers import direct
from gcl_sdk.agents.universal.storage import fs


class FilesDirectCapabilityDriver(direct.DirectAgentDriver):
    def __init__(self, api_url: str, work_dir: str) -> None:
        http_client = bazooka.Client()
        rest_client = http_base.CollectionBaseClient(
            http_client=http_client,
            base_url=api_url,
            auth=None,
        )

        storage_path = f"{work_dir}/file_target_fields.json"
        storage = fs.TargetFieldsFileStorage(storage_path)
        client = FilesRestBackendClient(rest_client)
        super().__init__(client=client, storage=storage)

    def get_capabilities(self) -> list[str]:
        return ["file_target"]

The base class already implements create/get/list/update/delete and maps backend/storage exceptions to driver exceptions.

Optioanal step. Transform bckend response

DirectAgentDriver supports transformer_map for per-kind transformations.

from gcl_sdk.agents.universal.drivers import direct


transformer_map = {
    "file_target": direct.ResourceTransformer(
        ignore_null_attributes=True,
        attributes={"owner", "group"},
    )
}

Pass this map into DirectAgentDriver if backend payload needs normalization.

Example target resource

{
  "kind": "file_target",
  "value": {
    "uuid": "a1b2c3d4-e5f6-7890-a1b2-c3d4e5f67890",
    "name": "config.yaml",
    "path": "/opt/example/config.yaml"
  }
}

After create:

• file object is created via REST API
• target fields for this (kind, uuid) are saved in storage

Register the driver

[project.entry-points.gcl_sdk_universal_agent]
FilesDirectCapabilityDriver = "your_package.drivers.files_direct:FilesDirectCapabilityDriver"

Usage

[universal_agent]
caps_drivers = ...,FilesDirectCapabilityDriver

systemctl restart genesis-universal-agent