Python 客户端

概述

我们提供了 agents-api 仓库，其中包含 Agents 资源的 Python Client 库。

前置条件

使用 Python Client 前，请注意以下事项：

• Python 版本：Python 3.7 或更高版本
• 依赖要求：须使用 Kubernetes Python 客户端库，kubernetes.client。

安装

通过 Git 安装

pip install git+https://github.com/openkruise/agents-api.git@master#subdirectory=clients/python/openkruise

使用示例

以下提供了两个 Demo 进行参考，分别演示了 SandboxSet 和 SandboxClaim 的基本操作。

以下代码仅作为简单使用的 Demo 示例。在生产环境中，你需要自行封装高可用的 Client（如连接池管理、超时重试机制等）以及完善的异常处理逻辑。

from agents import SandboxSet, SandboxSetClient, SandboxSetSpec
from kubernetes.client import V1ObjectMeta, V1Container, V1PodSpec, V1PodTemplateSpec


class SandboxSetExample:

    def __init__(self, namespace="default"):
        self.client = SandboxSetClient(namespace=namespace)

    def create_sandboxset(self, name: str, replicas: int, image: str) -> dict:
        """
        Create SandboxSet
        """
        try:
            sandboxset = self._build_sandboxset(name, replicas, image)
            result = self.client.create_sandboxset(sandboxset)
            print(f"SandboxSet created: {name}")
            return result
        except Exception as e:
            print(f"Failed to create sandboxset: {e}")
            return None

    def get_sandboxset(self, name: str) -> dict:
        """
        Get SandboxSet
        """
        try:
            result = self.client.get_sandboxset(name)
            print(f"SandboxSet retrieved: {name}")
            return result
        except Exception as e:
            print(f"Failed to get sandboxset: {e}")
            return None

    def delete_sandboxset(self, name: str):
        """
        Delete SandboxSet
        """
        try:
            self.client.delete_sandboxset(name)
            print(f"SandboxSet deleted: {name}")
        except Exception as e:
            print(f"Failed to delete sandboxset: {e}")

    def _build_sandboxset(self, name: str, replicas: int, image: str) -> SandboxSet:
        """
        Build SandboxSet
        """
        labels = {"sandboxset": "true"}
        container_name = "test-container"

        # Build Metadata
        metadata = V1ObjectMeta(
            name=name,
            labels=labels,
        )

        # Build Container
        container = V1Container(
            name=container_name,
            image=image,
        )

        # Build PodSpec
        pod_spec = V1PodSpec(
            containers=[container],
        )

        # Build PodTemplateSpec
        template = V1PodTemplateSpec(
            metadata=V1ObjectMeta(labels={"sandboxset": "true"}),
            spec=pod_spec,
        )

        # Build SandboxSetSpec
        spec = SandboxSetSpec(
            replicas=replicas,
            template=template,
        )

        return SandboxSet(
            metadata=metadata,
            spec=spec,
        )


if __name__ == '__main__':
    import time

    namespace = "default"
    name = f"test-sandboxset-{int(time.time() * 1000)}"
    replicas = 1
    image = "nginx:stable-alpine3.23"

    example = SandboxSetExample(namespace=namespace)

    # create
    example.create_sandboxset(name, replicas, image)

    # get
    example.get_sandboxset(name)

    # delete
    example.delete_sandboxset(name)

import time
from agents import (
    SandboxSet, SandboxSetClient, SandboxSetSpec,
    SandboxClaim, SandboxClaimClient, SandboxClaimSpec
)
from kubernetes.client import V1ObjectMeta, V1Container, V1PodSpec, V1PodTemplateSpec


class SandboxClaimExample:

    def __init__(self, namespace="default"):
        self.sandboxset_client = SandboxSetClient(namespace=namespace)
        self.sandboxclaim_client = SandboxClaimClient(namespace=namespace)

    def create_sandboxset(self, name: str, replicas: int, image: str) -> dict:
        """
        Create SandboxSet (pool)
        """
        try:
            sandboxset = self._build_sandboxset(name, replicas, image)
            result = self.sandboxset_client.create_sandboxset(sandboxset)
            print(f"SandboxSet created: {name}")
            return result
        except Exception as e:
            print(f"Failed to create sandboxset: {e}")
            return None

    def create_sandboxclaim(self, name: str, template_name: str, replicas: int) -> dict:
        """
        Create SandboxClaim
        """
        try:
            sandboxclaim = self._build_sandboxclaim(name, template_name, replicas)
            result = self.sandboxclaim_client.create_sandboxclaim(sandboxclaim)
            print(f"SandboxClaim created: {name}")
            return result
        except Exception as e:
            print(f"Failed to create sandboxclaim: {e}")
            return None

    def get_sandboxclaim(self, name: str) -> dict:
        """
        Get SandboxClaim
        """
        try:
            result = self.sandboxclaim_client.get_sandboxclaim(name)
            print(f"SandboxClaim retrieved: {name}")
            return result
        except Exception as e:
            print(f"Failed to get sandboxclaim: {e}")
            return None

    def delete_sandboxclaim(self, name: str):
        """
        Delete SandboxClaim
        """
        try:
            self.sandboxclaim_client.delete_sandboxclaim(name)
            print(f"SandboxClaim deleted: {name}")
        except Exception as e:
            print(f"Failed to delete sandboxclaim: {e}")

    def delete_sandboxset(self, name: str):
        """
        Delete SandboxSet
        """
        try:
            self.sandboxset_client.delete_sandboxset(name)
            print(f"SandboxSet deleted: {name}")
        except Exception as e:
            print(f"Failed to delete sandboxset: {e}")

    def _build_sandboxset(self, name: str, replicas: int, image: str) -> SandboxSet:
        """
        Build SandboxSet
        """

        metadata = V1ObjectMeta(name=name)

        # build Container
        container = V1Container(
            name="test-container",
            image=image,
        )

        # build PodSpec
        pod_spec = V1PodSpec(containers=[container])

        # build PodTemplateSpec
        template = V1PodTemplateSpec(spec=pod_spec)

        # build SandboxSetSpec
        spec = SandboxSetSpec(
            replicas=replicas,
            template=template,
        )

        return SandboxSet(metadata=metadata, spec=spec)

    def _build_sandboxclaim(self, name: str, template_name: str, replicas: int) -> SandboxClaim:
        """
        Build SandboxClaim Object
        """
        # build Metadata
        metadata = V1ObjectMeta(name=name)

        # build SandboxClaimSpec
        spec = SandboxClaimSpec(
            templateName=template_name,
            replicas=replicas,
        )

        return SandboxClaim(metadata=metadata, spec=spec)


if __name__ == '__main__':
    namespace = "default"

    sandboxset_name = f"test-pool-{int(time.time() * 1000)}"
    sandboxclaim_name = f"test-claim-{int(time.time() * 1000)}"

    pool_replicas = 5
    claim_replicas = 1
    image = "nginx:stable-alpine3.23"

    example = SandboxClaimExample(namespace=namespace)

    try:
        # Step 1: Creating SandboxSet (pool)
        print("=== Step 1: Creating SandboxSet (pool) ===")
        created_pool = example.create_sandboxset(sandboxset_name, pool_replicas, image)
        if not created_pool:
            print("Failed to create pool, exiting")
            exit(1)

        # Waiting SandboxSet ready
        print("Waiting for SandboxSet to be ready...")
        time.sleep(10)

        # Step 2: Creating SandboxClaim
        print("\n=== Step 2: Creating SandboxClaim ===")
        created_claim = example.create_sandboxclaim(sandboxclaim_name, sandboxset_name, claim_replicas)
        if not created_claim:
            print("Failed to create claim, cleaning up pool")
            example.delete_sandboxset(sandboxset_name)
            exit(1)

        # Waiting claim
        print("Waiting for claim to be processed...")
        time.sleep(5)

        # Step 3: Getting SandboxClaim
        print("\n=== Step 3: Getting SandboxClaim ===")
        example.get_sandboxclaim(sandboxclaim_name)

        # Step 4: Deleting SandboxClaim
        print("\n=== Step 4: Deleting SandboxClaim ===")
        example.delete_sandboxclaim(sandboxclaim_name)

        # Step 5: Cleaning SandboxSet
        print("\n=== Step 5: Cleaning up SandboxSet ===")
        example.delete_sandboxset(sandboxset_name)

        print("\n=== Example completed successfully ===")

    except KeyboardInterrupt:
        print("\nInterrupted by user")