实现 LangChain 集成

from __future__ import annotations

from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
)
from deepagents.backends.sandbox import BaseSandbox  


class MySandbox(BaseSandbox):
    def __init__(self, client: MySandboxSdkClient) -> None:
        self._client = client

    @property
    def id(self) -> str:
        return self._client.sandbox_id

    def execute(
        self,
        command: str,
        *,
        timeout: int | None = None,
    ) -> ExecuteResponse:
        # Execute `command` in your sandbox and map the provider response
        # into ExecuteResponse.
        result = self._client.run(command=command, timeout=timeout)
        output = result.stdout or ""
        if result.stderr:
            output += f"\n<stderr>{result.stderr}</stderr>"
        return ExecuteResponse(
            output=output,
            exit_code=result.exit_code,
            truncated=False,
        )

    def upload_files(
        self,
        files: list[tuple[str, bytes]],
    ) -> list[FileUploadResponse]:
        # Validate paths, batch requests where possible, and map provider
        # results back into FileUploadResponse objects in input order.
        # Only catch and normalize errors that an LLM can plausibly retry
        # or fix, such as invalid_path or file_not_found.
        return self._client.upload_files(files)

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        # Validate paths, batch requests where possible, and map provider
        # results back into FileDownloadResponse objects in input order.
        # Only catch and normalize errors that an LLM can plausibly retry
        # or fix, such as invalid_path or file_not_found.
        return self._client.download_files(paths)

    async def aexecute(
        self,
        command: str,
        *,
        timeout: int | None = None,
    ) -> ExecuteResponse:
        ...

    async def aupload_files(
        self,
        files: list[tuple[str, bytes]],
    ) -> list[FileUploadResponse]:
        ...

    async def adownload_files(
        self,
        paths: list[str],
    ) -> list[FileDownloadResponse]:
        ...

from __future__ import annotations

from collections.abc import Iterator

import pytest
from deepagents.backends.protocol import SandboxBackendProtocol
from langchain_tests.integration_tests import SandboxIntegrationTests

from langchain_myprovider import MySandbox
from myprovider_sdk import MySandboxSdkClient


class TestMySandboxStandard(SandboxIntegrationTests):
    @pytest.fixture(scope="class")
    def sandbox(self) -> Iterator[SandboxBackendProtocol]:
        client = MySandboxSdkClient()
        backend = MySandbox(client=client)
        try:
            yield backend
        finally:
            # Replace this with your provider's cleanup logic.
            client.delete_sandbox(backend.id)