コンテンツにスキップ

Sandbox clients

BaseSandboxClient

Bases: ABC, Generic[ClientOptionsT]

Source code in src/agents/sandbox/session/sandbox_client.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class BaseSandboxClient(abc.ABC, Generic[ClientOptionsT]):
    backend_id: str
    supports_default_options: bool = False
    _dependencies: Dependencies | None = None

    def _resolve_dependencies(self) -> Dependencies | None:
        if self._dependencies is None:
            return None
        # Sessions get clones instead of the shared template so per-session factory caches and
        # owned resources do not leak across unrelated sandboxes.
        return self._dependencies.clone()

    def _wrap_session(
        self,
        inner: BaseSandboxSession,
        *,
        instrumentation: Instrumentation | None = None,
    ) -> SandboxSession:
        # Always return the instrumented wrapper so callers get consistent events and dependency
        # lifecycle handling regardless of which backend created the inner session.
        return SandboxSession(
            inner,
            instrumentation=instrumentation,
            dependencies=self._resolve_dependencies(),
        )

    @abc.abstractmethod
    async def create(
        self,
        *,
        snapshot: SnapshotSpec | SnapshotBase | None = None,
        manifest: Manifest | None = None,
        options: ClientOptionsT,
    ) -> SandboxSession:
        """Create a new session.

        Args:
            snapshot: Snapshot or spec used to create a snapshot instance for
                the session. If omitted, the session uses a no-op snapshot.
            manifest: Optional manifest to materialize into the workspace when
                the session starts.
            options: Sandbox-specific settings. For example, Docker expects
                ``DockerSandboxClientOptions(image="...")``.
        Returns:
            A `SandboxSession` that can be entered with `async with` or closed explicitly with
            `await session.aclose()`.
        """

    @abc.abstractmethod
    async def delete(self, session: SandboxSession) -> SandboxSession:
        """Delete a session and release sandbox resources."""

    @abc.abstractmethod
    async def resume(
        self,
        state: SandboxSessionState,
    ) -> SandboxSession:
        """Resume an owning session from a previously persisted `SandboxSessionState`.

        Providers should first try to reattach to the backend sandbox identified
        by `state`. If that resource still exists, including after unclean
        process/client shutdown where `delete()` was never called, the returned
        session should target the same backend sandbox and be able to clean it
        up later.

        If the original backend sandbox is unavailable, providers may create a
        replacement and should hydrate its workspace from `state.snapshot`
        during `SandboxSession.start()`.

        The returned session owns its provider lifecycle; pass a live
        `session=` when you want to reuse an already-running sandbox session.
        """

    def serialize_session_state(self, state: SandboxSessionState) -> dict[str, object]:
        """Serialize backend-specific sandbox state into a JSON-compatible payload."""
        return state.model_dump(mode="json")

    @abc.abstractmethod
    def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
        """Deserialize backend-specific sandbox state from a JSON-compatible payload."""

create abstractmethod async 

create(
    *,
    snapshot: SnapshotSpec | SnapshotBase | None = None,
    manifest: Manifest | None = None,
    options: ClientOptionsT,
) -> SandboxSession

Create a new session.

Parameters:

Name Type Description Default
snapshot SnapshotSpec | SnapshotBase | None

Snapshot or spec used to create a snapshot instance for the session. If omitted, the session uses a no-op snapshot.

 None
manifest Manifest | None

Optional manifest to materialize into the workspace when the session starts.

 None
options ClientOptionsT

Sandbox-specific settings. For example, Docker expects DockerSandboxClientOptions(image="...").

 required

Returns: A SandboxSession that can be entered with async with or closed explicitly with await session.aclose().

Source code in src/agents/sandbox/session/sandbox_client.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@abc.abstractmethod
async def create(
    self,
    *,
    snapshot: SnapshotSpec | SnapshotBase | None = None,
    manifest: Manifest | None = None,
    options: ClientOptionsT,
) -> SandboxSession:
    """Create a new session.

    Args:
        snapshot: Snapshot or spec used to create a snapshot instance for
            the session. If omitted, the session uses a no-op snapshot.
        manifest: Optional manifest to materialize into the workspace when
            the session starts.
        options: Sandbox-specific settings. For example, Docker expects
            ``DockerSandboxClientOptions(image="...")``.
    Returns:
        A `SandboxSession` that can be entered with `async with` or closed explicitly with
        `await session.aclose()`.
    """

delete abstractmethod async 

delete(session: SandboxSession) -> SandboxSession

Delete a session and release sandbox resources.

Source code in src/agents/sandbox/session/sandbox_client.py 
148
149
150
@abc.abstractmethod
async def delete(self, session: SandboxSession) -> SandboxSession:
    """Delete a session and release sandbox resources."""

resume abstractmethod async 

resume(state: SandboxSessionState) -> SandboxSession

Resume an owning session from a previously persisted SandboxSessionState.

Providers should first try to reattach to the backend sandbox identified by state. If that resource still exists, including after unclean process/client shutdown where delete() was never called, the returned session should target the same backend sandbox and be able to clean it up later.

If the original backend sandbox is unavailable, providers may create a replacement and should hydrate its workspace from state.snapshot during SandboxSession.start().

The returned session owns its provider lifecycle; pass a live session= when you want to reuse an already-running sandbox session.

Source code in src/agents/sandbox/session/sandbox_client.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@abc.abstractmethod
async def resume(
    self,
    state: SandboxSessionState,
) -> SandboxSession:
    """Resume an owning session from a previously persisted `SandboxSessionState`.

    Providers should first try to reattach to the backend sandbox identified
    by `state`. If that resource still exists, including after unclean
    process/client shutdown where `delete()` was never called, the returned
    session should target the same backend sandbox and be able to clean it
    up later.

    If the original backend sandbox is unavailable, providers may create a
    replacement and should hydrate its workspace from `state.snapshot`
    during `SandboxSession.start()`.

    The returned session owns its provider lifecycle; pass a live
    `session=` when you want to reuse an already-running sandbox session.
    """

serialize_session_state 

serialize_session_state(
    state: SandboxSessionState,
) -> dict[str, object]

Serialize backend-specific sandbox state into a JSON-compatible payload.

Source code in src/agents/sandbox/session/sandbox_client.py 
173
174
175
def serialize_session_state(self, state: SandboxSessionState) -> dict[str, object]:
    """Serialize backend-specific sandbox state into a JSON-compatible payload."""
    return state.model_dump(mode="json")

deserialize_session_state abstractmethod 

deserialize_session_state(
    payload: dict[str, object],
) -> SandboxSessionState

Deserialize backend-specific sandbox state from a JSON-compatible payload.

Source code in src/agents/sandbox/session/sandbox_client.py 
177
178
179
@abc.abstractmethod
def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
    """Deserialize backend-specific sandbox state from a JSON-compatible payload."""

BaseSandboxClientOptions

Bases: BaseModel

Polymorphic base for sandbox client options that need JSON round-trips.

Source code in src/agents/sandbox/session/sandbox_client.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class BaseSandboxClientOptions(BaseModel):
    """Polymorphic base for sandbox client options that need JSON round-trips."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    type: str
    _subclass_registry: ClassVar[dict[str, SandboxClientOptionsClass]] = {}

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        if args:
            positional_fields = [name for name in type(self).model_fields if name != "type"]
            if len(args) > len(positional_fields):
                raise TypeError(
                    f"{type(self).__name__}() takes at most {len(positional_fields)} positional "
                    f"arguments but {len(args)} were given"
                )
            for field_name, value in zip(positional_fields, args, strict=False):
                if field_name in kwargs:
                    raise TypeError(
                        f"{type(self).__name__}() got multiple values for argument {field_name!r}"
                    )
                kwargs[field_name] = value
        super().__init__(**kwargs)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: object) -> None:
        super().__pydantic_init_subclass__(**kwargs)

        type_field = cls.model_fields.get("type")
        type_default = type_field.default if type_field is not None else None
        if not isinstance(type_default, str) or type_default == "":
            raise TypeError(f"{cls.__name__} must define a non-empty string default for `type`")

        existing = BaseSandboxClientOptions._subclass_registry.get(type_default)
        if (
            existing is not None
            and existing is not cls
            and (existing.__module__, existing.__qualname__) != (cls.__module__, cls.__qualname__)
        ):
            raise TypeError(
                f"sandbox client options type `{type_default}` is already registered by "
                f"{existing.__name__}"
            )
        if existing is not None:
            return
        BaseSandboxClientOptions._subclass_registry[type_default] = cls

    @classmethod
    def parse(cls, payload: object) -> BaseSandboxClientOptions:
        if isinstance(payload, BaseSandboxClientOptions):
            return payload

        if isinstance(payload, dict):
            options_type = payload.get("type")
            if isinstance(options_type, str):
                options_class = cls._options_class_for_type(options_type)
                if options_class is not None:
                    return options_class.model_validate(payload)

            raise ValueError(f"unknown sandbox client options type `{options_type}`")

        raise TypeError(
            "sandbox client options payload must be a BaseSandboxClientOptions or object payload"
        )

    @model_serializer(mode="wrap")
    def _serialize_always_include_type(self, handler: Any) -> dict[str, Any]:
        data = handler(self)
        if isinstance(data, dict):
            data["type"] = self.type
        return cast(dict[str, Any], data)

    @classmethod
    def _options_class_for_type(
        cls,
        options_type: str,
    ) -> SandboxClientOptionsClass | None:
        return BaseSandboxClientOptions._subclass_registry.get(options_type)