Adapter Pattern Architecture for Production-Ready SDK¶

class ContainerRuntimeAdapter(ABC):
    @abstractmethod
    async def check_availability(self) -> bool:
        """Verify runtime is installed and accessible"""

    @abstractmethod
    async def provision_container(
        self,
        image: str,
        command: list[str],
        mounts: list[Mount],
        resources: ResourceLimits,
        network: NetworkConfig,
        security: SecurityConfig,
    ) -> ContainerHandle:
        """Create and start a container"""

    @abstractmethod
    async def exec_in_container(
        self,
        container_id: str,
        command: list[str],
        env: dict[str, str] | None = None,
    ) -> ExecResult:
        """Execute command inside container"""

    @abstractmethod
    async def copy_from_container(
        self,
        container_id: str,
        src_path: str,
        dest_path: str,
    ) -> None:
        """Copy file/directory from container to host"""

    @abstractmethod
    async def copy_to_container(
        self,
        container_id: str,
        src_path: str,
        dest_path: str,
    ) -> None:
        """Copy file/directory from host to container"""

    @abstractmethod
    async def get_container_stats(
        self,
        container_id: str,
    ) -> ContainerStats:
        """Get CPU/memory usage metrics"""

    @abstractmethod
    async def destroy_container(
        self,
        container_id: str,
        force: bool = True,
    ) -> None:
        """Stop and remove container"""

class StorageBackendAdapter(ABC):
    @abstractmethod
    async def store_artifact(
        self,
        session_id: str,
        artifact_name: str,
        content: bytes,
        metadata: dict[str, str] | None = None,
    ) -> str:
        """Store artifact and return URI"""

    @abstractmethod
    async def retrieve_artifact(
        self,
        uri: str,
    ) -> bytes:
        """Retrieve artifact by URI"""

    @abstractmethod
    async def list_artifacts(
        self,
        session_id: str,
    ) -> list[ArtifactInfo]:
        """List all artifacts for a session"""

    @abstractmethod
    async def delete_artifacts(
        self,
        session_id: str,
    ) -> None:
        """Delete all artifacts for a session"""

    @abstractmethod
    async def get_artifact_url(
        self,
        uri: str,
        expiry_seconds: int = 3600,
    ) -> str:
        """Generate signed URL for artifact access"""

class AuditLoggerAdapter(ABC):
    @abstractmethod
    async def emit_event(
        self,
        event: AuditEvent,
    ) -> None:
        """Emit a single audit event"""

    @abstractmethod
    async def emit_batch(
        self,
        events: list[AuditEvent],
    ) -> None:
        """Emit multiple events efficiently"""

    @abstractmethod
    async def query_events(
        self,
        session_id: str | None = None,
        event_type: str | None = None,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        limit: int = 100,
    ) -> list[AuditEvent]:
        """Query historical audit events"""

    @abstractmethod
    async def flush(self) -> None:
        """Ensure all buffered events are written"""

class PolicyValidatorAdapter(ABC):
    @abstractmethod
    async def validate_policy(
        self,
        policy: Policy,
    ) -> ValidationResult:
        """Validate policy and return errors/warnings"""

    @abstractmethod
    async def enforce_constraints(
        self,
        policy: Policy,
        context: EnforcementContext,
    ) -> Policy:
        """Apply organizational constraints and return modified policy"""

    @abstractmethod
    async def check_compliance(
        self,
        policy: Policy,
        compliance_rules: list[str],
    ) -> ComplianceReport:
        """Check policy against compliance requirements"""

from isolated_agents_sdk import configure_adapters
from isolated_agents_sdk.adapters.storage import S3StorageAdapter

configure_adapters(
    container_runtime="docker",
    storage_backend=S3StorageAdapter(
        bucket="my-bucket",
        region="us-east-1",
    ),
    audit_logger="cloudwatch",
)

class AdapterFactory:
    """Central factory for creating adapter instances"""

    _container_adapters: dict[str, type[ContainerRuntimeAdapter]] = {}
    _storage_adapters: dict[str, type[StorageBackendAdapter]] = {}
    _audit_adapters: dict[str, type[AuditLoggerAdapter]] = {}
    _policy_adapters: dict[str, type[PolicyValidatorAdapter]] = {}

    @classmethod
    def register_container_adapter(
        cls,
        name: str,
        adapter_class: type[ContainerRuntimeAdapter],
    ) -> None:
        """Register a container runtime adapter"""

    @classmethod
    def create_container_adapter(
        cls,
        name: str,
        config: dict[str, Any] | None = None,
    ) -> ContainerRuntimeAdapter:
        """Create container runtime adapter instance"""

    # Similar methods for other adapter types...

from isolated_agents_sdk import run_agent, Policy

result = run_agent(
    agent=my_agent,
    working_dir="./workspace",
    policy=Policy(memory_mb=1024),
)

from isolated_agents_sdk import run_agent, Policy, configure_adapters

# Optional: Configure adapters (uses defaults if not called)
configure_adapters(
    container_runtime="docker",
    storage_backend="s3",
)

result = run_agent(
    agent=my_agent,
    working_dir="./workspace",
    policy=Policy(memory_mb=1024),
)

class DockerAdapter(ContainerRuntimeAdapter):
    async def provision_container(self, ...):
        # Use Docker SDK instead of Podman CLI
        import docker
        client = docker.from_env()
        container = client.containers.run(
            image=image,
            command=command,
            detach=True,
            # Map parameters to Docker API...
        )
        return ContainerHandle(container_id=container.id)

class S3StorageAdapter(StorageBackendAdapter):
    async def store_artifact(self, session_id, artifact_name, content, metadata):
        import boto3
        s3 = boto3.client('s3')
        key = f"{self.prefix}{session_id}/{artifact_name}"
        s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=content,
            Metadata=metadata or {},
        )
        return f"s3://{self.bucket}/{key}"