Skip to content

Tools

Tools and utilities for Bedrock AgentCore SDK including browser and code interpreter tools.

bedrock_agentcore.tools.code_interpreter_client

Client for interacting with the Code Interpreter sandbox service.

This module provides a client for the AWS Code Interpreter sandbox, allowing applications to start, stop, and invoke code execution in a managed sandbox environment.

CodeInterpreter

Client for interacting with the AWS Code Interpreter sandbox service.

This client handles the session lifecycle and method invocation for Code Interpreter sandboxes, providing an interface to execute code in a secure, managed environment.

Attributes:

Name Type Description
data_plane_service_name str

AWS service name for the data plane.
client

The boto3 client for interacting with the service.
identifier str

The code interpreter identifier.
session_id str

The active session ID.
Source code in bedrock_agentcore/tools/code_interpreter_client.py 
 19
 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
 98
 99
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
class CodeInterpreter:
    """Client for interacting with the AWS Code Interpreter sandbox service.

    This client handles the session lifecycle and method invocation for
    Code Interpreter sandboxes, providing an interface to execute code
    in a secure, managed environment.

    Attributes:
        data_plane_service_name (str): AWS service name for the data plane.
        client: The boto3 client for interacting with the service.
        identifier (str, optional): The code interpreter identifier.
        session_id (str, optional): The active session ID.
    """

    def __init__(self, region: str) -> None:
        """Initialize a Code Interpreter client for the specified AWS region.

        Args:
            region (str): The AWS region to use for the Code Interpreter service.
        """
        self.data_plane_service_name = "bedrock-agentcore"
        self.client = boto3.client(
            self.data_plane_service_name, region_name=region, endpoint_url=get_data_plane_endpoint(region)
        )
        self._identifier = None
        self._session_id = None

    @property
    def identifier(self) -> Optional[str]:
        """Get the current code interpreter identifier.

        Returns:
            Optional[str]: The current identifier or None if not set.
        """
        return self._identifier

    @identifier.setter
    def identifier(self, value: Optional[str]):
        """Set the code interpreter identifier.

        Args:
            value (Optional[str]): The identifier to set.
        """
        self._identifier = value

    @property
    def session_id(self) -> Optional[str]:
        """Get the current session ID.

        Returns:
            Optional[str]: The current session ID or None if not set.
        """
        return self._session_id

    @session_id.setter
    def session_id(self, value: Optional[str]):
        """Set the session ID.

        Args:
            value (Optional[str]): The session ID to set.
        """
        self._session_id = value

    def start(
        self,
        identifier: Optional[str] = DEFAULT_IDENTIFIER,
        name: Optional[str] = None,
        session_timeout_seconds: Optional[int] = DEFAULT_TIMEOUT,
    ) -> str:
        """Start a code interpreter sandbox session.

        This method initializes a new code interpreter session with the provided parameters.

        Args:
            identifier (Optional[str]): The code interpreter sandbox identifier to use.
                Defaults to DEFAULT_IDENTIFIER.
            name (Optional[str]): A name for this session. If not provided, a name
                will be generated using a UUID.
            session_timeout_seconds (Optional[int]): The timeout for the session in seconds.
                Defaults to DEFAULT_TIMEOUT.
            description (Optional[str]): A description for this session.
                Defaults to an empty string.

        Returns:
            str: The session ID of the newly created session.
        """
        response = self.client.start_code_interpreter_session(
            codeInterpreterIdentifier=identifier,
            name=name or f"code-session-{uuid.uuid4().hex[:8]}",
            sessionTimeoutSeconds=session_timeout_seconds,
        )

        self.identifier = response["codeInterpreterIdentifier"]
        self.session_id = response["sessionId"]

        return self.session_id

    def stop(self):
        """Stop the current code interpreter session if one is active.

        This method stops any active session and clears the session state.
        If no session is active, this method does nothing.

        Returns:
            bool: True if no session was active or the session was successfully stopped.
        """
        if not self.session_id or not self.identifier:
            return True

        self.client.stop_code_interpreter_session(
            **{"codeInterpreterIdentifier": self.identifier, "sessionId": self.session_id}
        )

        self.identifier = None
        self.session_id = None

    def invoke(self, method: str, params: Optional[Dict] = None):
        """Invoke a method in the code interpreter sandbox.

        If no session is active, this method automatically starts a new session
        before invoking the requested method.

        Args:
            method (str): The name of the method to invoke in the sandbox.
            params (Optional[Dict]): Parameters to pass to the method. Defaults to None.
            request_id (Optional[str]): A custom request ID. If not provided, a unique ID is generated.

        Returns:
            dict: The response from the code interpreter service.
        """
        if not self.session_id or not self.identifier:
            self.start()

        return self.client.invoke_code_interpreter(
            **{
                "codeInterpreterIdentifier": self.identifier,
                "sessionId": self.session_id,
                "name": method,
                "arguments": params or {},
            }
        )

identifier property writable

Get the current code interpreter identifier.

Returns:

Type Description
Optional[str]

Optional[str]: The current identifier or None if not set.

session_id property writable

Get the current session ID.

Returns:

Type Description
Optional[str]

Optional[str]: The current session ID or None if not set.

__init__(region)

Initialize a Code Interpreter client for the specified AWS region.

Parameters:

Name Type Description Default
region str

The AWS region to use for the Code Interpreter service.

 required
Source code in bedrock_agentcore/tools/code_interpreter_client.py 
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, region: str) -> None:
    """Initialize a Code Interpreter client for the specified AWS region.

    Args:
        region (str): The AWS region to use for the Code Interpreter service.
    """
    self.data_plane_service_name = "bedrock-agentcore"
    self.client = boto3.client(
        self.data_plane_service_name, region_name=region, endpoint_url=get_data_plane_endpoint(region)
    )
    self._identifier = None
    self._session_id = None

invoke(method, params=None)

Invoke a method in the code interpreter sandbox.

If no session is active, this method automatically starts a new session before invoking the requested method.

Parameters:

Name Type Description Default
method str

The name of the method to invoke in the sandbox.

 required
params Optional[Dict]

Parameters to pass to the method. Defaults to None.

 None
request_id Optional[str]

A custom request ID. If not provided, a unique ID is generated.

 required

Returns:

Name Type Description
dict

The response from the code interpreter service.
Source code in bedrock_agentcore/tools/code_interpreter_client.py 
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
def invoke(self, method: str, params: Optional[Dict] = None):
    """Invoke a method in the code interpreter sandbox.

    If no session is active, this method automatically starts a new session
    before invoking the requested method.

    Args:
        method (str): The name of the method to invoke in the sandbox.
        params (Optional[Dict]): Parameters to pass to the method. Defaults to None.
        request_id (Optional[str]): A custom request ID. If not provided, a unique ID is generated.

    Returns:
        dict: The response from the code interpreter service.
    """
    if not self.session_id or not self.identifier:
        self.start()

    return self.client.invoke_code_interpreter(
        **{
            "codeInterpreterIdentifier": self.identifier,
            "sessionId": self.session_id,
            "name": method,
            "arguments": params or {},
        }
    )

start(identifier=DEFAULT_IDENTIFIER, name=None, session_timeout_seconds=DEFAULT_TIMEOUT)

Start a code interpreter sandbox session.

This method initializes a new code interpreter session with the provided parameters.

Parameters:

Name Type Description Default
identifier Optional[str]

The code interpreter sandbox identifier to use. Defaults to DEFAULT_IDENTIFIER.

 DEFAULT_IDENTIFIER
name Optional[str]

A name for this session. If not provided, a name will be generated using a UUID.

 None
session_timeout_seconds Optional[int]

The timeout for the session in seconds. Defaults to DEFAULT_TIMEOUT.

 DEFAULT_TIMEOUT
description Optional[str]

A description for this session. Defaults to an empty string.

 required

Returns:

Name Type Description
str str

The session ID of the newly created session.
Source code in bedrock_agentcore/tools/code_interpreter_client.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def start(
    self,
    identifier: Optional[str] = DEFAULT_IDENTIFIER,
    name: Optional[str] = None,
    session_timeout_seconds: Optional[int] = DEFAULT_TIMEOUT,
) -> str:
    """Start a code interpreter sandbox session.

    This method initializes a new code interpreter session with the provided parameters.

    Args:
        identifier (Optional[str]): The code interpreter sandbox identifier to use.
            Defaults to DEFAULT_IDENTIFIER.
        name (Optional[str]): A name for this session. If not provided, a name
            will be generated using a UUID.
        session_timeout_seconds (Optional[int]): The timeout for the session in seconds.
            Defaults to DEFAULT_TIMEOUT.
        description (Optional[str]): A description for this session.
            Defaults to an empty string.

    Returns:
        str: The session ID of the newly created session.
    """
    response = self.client.start_code_interpreter_session(
        codeInterpreterIdentifier=identifier,
        name=name or f"code-session-{uuid.uuid4().hex[:8]}",
        sessionTimeoutSeconds=session_timeout_seconds,
    )

    self.identifier = response["codeInterpreterIdentifier"]
    self.session_id = response["sessionId"]

    return self.session_id

stop()

Stop the current code interpreter session if one is active.

This method stops any active session and clears the session state. If no session is active, this method does nothing.

Returns:

Name Type Description
bool

True if no session was active or the session was successfully stopped.
Source code in bedrock_agentcore/tools/code_interpreter_client.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def stop(self):
    """Stop the current code interpreter session if one is active.

    This method stops any active session and clears the session state.
    If no session is active, this method does nothing.

    Returns:
        bool: True if no session was active or the session was successfully stopped.
    """
    if not self.session_id or not self.identifier:
        return True

    self.client.stop_code_interpreter_session(
        **{"codeInterpreterIdentifier": self.identifier, "sessionId": self.session_id}
    )

    self.identifier = None
    self.session_id = None

code_session(region)

Context manager for creating and managing a code interpreter session.

This context manager handles creating a client, starting a session, and ensuring the session is properly cleaned up when the context exits.

Parameters:

Name Type Description Default
region str

The AWS region to use for the Code Interpreter service.

 required

Yields:

Name Type Description
CodeInterpreterClient CodeInterpreter

An initialized and started code interpreter client.
Example

with code_session('us-west-2') as client: ... result = client.invoke('listFiles') ... # Process result here

Source code in bedrock_agentcore/tools/code_interpreter_client.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@contextmanager
def code_session(region: str) -> Generator[CodeInterpreter, None, None]:
    """Context manager for creating and managing a code interpreter session.

    This context manager handles creating a client, starting a session, and
    ensuring the session is properly cleaned up when the context exits.

    Args:
        region (str): The AWS region to use for the Code Interpreter service.

    Yields:
        CodeInterpreterClient: An initialized and started code interpreter client.

    Example:
        >>> with code_session('us-west-2') as client:
        ...     result = client.invoke('listFiles')
        ...     # Process result here
    """
    client = CodeInterpreter(region)
    client.start()

    try:
        yield client
    finally:
        client.stop()

bedrock_agentcore.tools.browser_client

Client for interacting with the Browser sandbox service.

This module provides a client for the AWS Browser sandbox, allowing applications to start, stop, and automate browser interactions in a managed sandbox environment using Playwright.

BrowserClient

Client for interacting with the AWS Browser sandbox service.

This client handles the session lifecycle and browser automation for Browser sandboxes, providing an interface to perform web automation tasks in a secure, managed environment.

Attributes:

Name Type Description
region str

The AWS region being used.
data_plane_service_name str

AWS service name for the data plane.
client

The boto3 client for interacting with the service.
identifier str

The browser identifier.
session_id str

The active session ID.
Source code in bedrock_agentcore/tools/browser_client.py 
 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
 98
 99
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class BrowserClient:
    """Client for interacting with the AWS Browser sandbox service.

    This client handles the session lifecycle and browser automation for
    Browser sandboxes, providing an interface to perform web automation
    tasks in a secure, managed environment.

    Attributes:
        region (str): The AWS region being used.
        data_plane_service_name (str): AWS service name for the data plane.
        client: The boto3 client for interacting with the service.
        identifier (str, optional): The browser identifier.
        session_id (str, optional): The active session ID.
    """

    def __init__(self, region: str) -> None:
        """Initialize a Browser client for the specified AWS region.

        Args:
            region (str): The AWS region to use for the Browser service.
        """
        self.region = region
        self.data_plane_service_name = "bedrock-agentcore"
        self.client = boto3.client(
            self.data_plane_service_name, region_name=region, endpoint_url=get_data_plane_endpoint(region)
        )
        self._identifier = None
        self._session_id = None
        self.logger = logging.getLogger(__name__)

    @property
    def identifier(self) -> Optional[str]:
        """Get the current browser identifier.

        Returns:
            Optional[str]: The current identifier or None if not set.
        """
        return self._identifier

    @identifier.setter
    def identifier(self, value: Optional[str]):
        """Set the browser identifier.

        Args:
            value (Optional[str]): The identifier to set.
        """
        self._identifier = value

    @property
    def session_id(self) -> Optional[str]:
        """Get the current session ID.

        Returns:
            Optional[str]: The current session ID or None if not set.
        """
        return self._session_id

    @session_id.setter
    def session_id(self, value: Optional[str]):
        """Set the session ID.

        Args:
            value (Optional[str]): The session ID to set.
        """
        self._session_id = value

    def start(
        self,
        identifier: Optional[str] = DEFAULT_IDENTIFIER,
        name: Optional[str] = None,
        session_timeout_seconds: Optional[int] = DEFAULT_SESSION_TIMEOUT,
    ) -> str:
        """Start a browser sandbox session.

        This method initializes a new browser session with the provided parameters.

        Args:
            identifier (Optional[str]): The browser sandbox identifier to use.
                Defaults to DEFAULT_IDENTIFIER.
            name (Optional[str]): A name for this session. If not provided, a name
                will be generated using a UUID.
            session_timeout_seconds (Optional[int]): The timeout for the session in seconds.
                Defaults to DEFAULT_TIMEOUT.
            description (Optional[str]): A description for this session.
                Defaults to an empty string.

        Returns:
            str: The session ID of the newly created session.
        """
        self.logger.info("Starting browser session...")

        response = self.client.start_browser_session(
            browserIdentifier=identifier,
            name=name or f"browser-session-{uuid.uuid4().hex[:8]}",
            sessionTimeoutSeconds=session_timeout_seconds,
        )

        self.identifier = response["browserIdentifier"]
        self.session_id = response["sessionId"]

        return self.session_id

    def stop(self):
        """Stop the current browser session if one is active.

        This method stops any active session and clears the session state.
        If no session is active, this method does nothing.

        Returns:
            bool: True if no session was active or the session was successfully stopped.
        """
        self.logger.info("Stopping browser session...")

        if not self.session_id or not self.identifier:
            return True

        self.client.stop_browser_session(**{"browserIdentifier": self.identifier, "sessionId": self.session_id})

        self.identifier = None
        self.session_id = None

    def generate_ws_headers(self) -> Tuple[str, Dict[str, str]]:
        """Generate the WebSocket headers needed for connecting to the browser sandbox.

        This method creates properly signed WebSocket headers for connecting to
        the browser automation endpoint.

        Returns:
            Tuple[str, Dict[str, str]]: A tuple containing the WebSocket URL and
                the headers dictionary.

        Raises:
            RuntimeError: If no AWS credentials are found.
        """
        self.logger.info("Generating websocket headers...")

        if not self.identifier or not self.session_id:
            self.start()

        host = get_data_plane_endpoint(self.region).replace("https://", "")
        path = f"/browser-streams/{self.identifier}/sessions/{self.session_id}/automation"
        ws_url = f"wss://{host}{path}"

        boto_session = boto3.Session()
        credentials = boto_session.get_credentials()
        if not credentials:
            raise RuntimeError("No AWS credentials found")

        frozen_credentials = credentials.get_frozen_credentials()

        request = AWSRequest(
            method="GET",
            url=f"https://{host}{path}",
            headers={
                "host": host,
                "x-amz-date": datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
            },
        )

        auth = SigV4Auth(frozen_credentials, self.data_plane_service_name, self.region)
        auth.add_auth(request)

        headers = {
            "Host": host,
            "X-Amz-Date": request.headers["x-amz-date"],
            "Authorization": request.headers["Authorization"],
            "Upgrade": "websocket",
            "Connection": "Upgrade",
            "Sec-WebSocket-Version": "13",
            "Sec-WebSocket-Key": base64.b64encode(secrets.token_bytes(16)).decode(),
            "User-Agent": f"BrowserSandbox-Client/1.0 (Session: {self.session_id})",
        }

        if frozen_credentials.token:
            headers["X-Amz-Security-Token"] = frozen_credentials.token

        return ws_url, headers

    def generate_live_view_url(self, expires: int = DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT) -> str:
        """Generate a pre-signed URL for viewing the browser session.

        Creates a pre-signed URL that can be used to view the current browser session.
        If no session is active, a new session will be started.

        Args:
            expires (int, optional): The number of seconds until the pre-signed URL expires.
                Defaults to DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT (300 seconds).

        Returns:
            str: The pre-signed URL for viewing the browser session.

        Raises:
            RuntimeError: If the URL generation fails.
        """
        self.logger.info("Generating live view url...")

        if not self.identifier or not self.session_id:
            self.start()

        url = urlparse(
            f"{get_data_plane_endpoint(self.region)}/browser-streams/{self.identifier}/sessions/{self.session_id}/live-view"
        )
        boto_session = boto3.Session()
        credentials = boto_session.get_credentials().get_frozen_credentials()
        request = AWSRequest(method="GET", url=url.geturl(), headers={"host": url.hostname})
        signer = SigV4QueryAuth(
            credentials=credentials, service_name=self.data_plane_service_name, region_name=self.region, expires=expires
        )
        signer.add_auth(request)

        if not request.url:
            raise RuntimeError("Failed to generate live view url")

        return request.url

    def take_control(self):
        """Take control of the browser session by disabling the automation stream.

        This method disables external automation capabilities of the browser session,
        giving this client exclusive control. If no session is active, a new session
        will be started.

        Raises:
            RuntimeError: If a session could not be found or started.
        """
        self.logger.info("Taking control of browser session...")

        if not self.identifier or not self.session_id:
            self.start()

        if not self.identifier or not self.session_id:
            raise RuntimeError("Could not find or start a browser session")

        self._update_browser_stream(self.identifier, self.session_id, "DISABLED")

    def release_control(self):
        """Release control of the browser session by enabling the automation stream.

        This method enables external automation capabilities of the browser session,
        relinquishing exclusive control. If no session exists, a warning is logged
        and the method returns without taking action.
        """
        self.logger.info("Releasing control of browser session...")

        if not self.identifier or not self.session_id:
            self.logger.warning("Could not find a browser session when releasing control")
            return

        self._update_browser_stream(self.identifier, self.session_id, "ENABLED")

    def _update_browser_stream(self, identifier: str, session_id: str, stream_status: str) -> None:
        """Update the browser stream status.

        This private helper method updates the status of the browser automation stream.

        Args:
            identifier (str): The browser identifier.
            session_id (str): The session ID.
            stream_status (str): The status to set for the automation stream.
                Valid values are "ENABLED" or "DISABLED".
        """
        self.client.update_browser_stream(
            **{
                "browserIdentifier": identifier,
                "sessionId": session_id,
                "streamUpdate": {"automationStreamUpdate": {"streamStatus": stream_status}},
            }
        )

identifier property writable

Get the current browser identifier.

Returns:

Type Description
Optional[str]

Optional[str]: The current identifier or None if not set.

session_id property writable

Get the current session ID.

Returns:

Type Description
Optional[str]

Optional[str]: The current session ID or None if not set.

__init__(region)

Initialize a Browser client for the specified AWS region.

Parameters:

Name Type Description Default
region str

The AWS region to use for the Browser service.

 required
Source code in bedrock_agentcore/tools/browser_client.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, region: str) -> None:
    """Initialize a Browser client for the specified AWS region.

    Args:
        region (str): The AWS region to use for the Browser service.
    """
    self.region = region
    self.data_plane_service_name = "bedrock-agentcore"
    self.client = boto3.client(
        self.data_plane_service_name, region_name=region, endpoint_url=get_data_plane_endpoint(region)
    )
    self._identifier = None
    self._session_id = None
    self.logger = logging.getLogger(__name__)

generate_live_view_url(expires=DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT)

Generate a pre-signed URL for viewing the browser session.

Creates a pre-signed URL that can be used to view the current browser session. If no session is active, a new session will be started.

Parameters:

Name Type Description Default
expires int

The number of seconds until the pre-signed URL expires. Defaults to DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT (300 seconds).

 DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT

Returns:

Name Type Description
str str

The pre-signed URL for viewing the browser session.

Raises:

Type Description
RuntimeError

If the URL generation fails.
Source code in bedrock_agentcore/tools/browser_client.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def generate_live_view_url(self, expires: int = DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT) -> str:
    """Generate a pre-signed URL for viewing the browser session.

    Creates a pre-signed URL that can be used to view the current browser session.
    If no session is active, a new session will be started.

    Args:
        expires (int, optional): The number of seconds until the pre-signed URL expires.
            Defaults to DEFAULT_LIVE_VIEW_PRESIGNED_URL_TIMEOUT (300 seconds).

    Returns:
        str: The pre-signed URL for viewing the browser session.

    Raises:
        RuntimeError: If the URL generation fails.
    """
    self.logger.info("Generating live view url...")

    if not self.identifier or not self.session_id:
        self.start()

    url = urlparse(
        f"{get_data_plane_endpoint(self.region)}/browser-streams/{self.identifier}/sessions/{self.session_id}/live-view"
    )
    boto_session = boto3.Session()
    credentials = boto_session.get_credentials().get_frozen_credentials()
    request = AWSRequest(method="GET", url=url.geturl(), headers={"host": url.hostname})
    signer = SigV4QueryAuth(
        credentials=credentials, service_name=self.data_plane_service_name, region_name=self.region, expires=expires
    )
    signer.add_auth(request)

    if not request.url:
        raise RuntimeError("Failed to generate live view url")

    return request.url

generate_ws_headers()

Generate the WebSocket headers needed for connecting to the browser sandbox.

This method creates properly signed WebSocket headers for connecting to the browser automation endpoint.

Returns:

Type Description
Tuple[str, Dict[str, str]]

Tuple[str, Dict[str, str]]: A tuple containing the WebSocket URL and the headers dictionary.

Raises:

Type Description
RuntimeError

If no AWS credentials are found.
Source code in bedrock_agentcore/tools/browser_client.py 
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def generate_ws_headers(self) -> Tuple[str, Dict[str, str]]:
    """Generate the WebSocket headers needed for connecting to the browser sandbox.

    This method creates properly signed WebSocket headers for connecting to
    the browser automation endpoint.

    Returns:
        Tuple[str, Dict[str, str]]: A tuple containing the WebSocket URL and
            the headers dictionary.

    Raises:
        RuntimeError: If no AWS credentials are found.
    """
    self.logger.info("Generating websocket headers...")

    if not self.identifier or not self.session_id:
        self.start()

    host = get_data_plane_endpoint(self.region).replace("https://", "")
    path = f"/browser-streams/{self.identifier}/sessions/{self.session_id}/automation"
    ws_url = f"wss://{host}{path}"

    boto_session = boto3.Session()
    credentials = boto_session.get_credentials()
    if not credentials:
        raise RuntimeError("No AWS credentials found")

    frozen_credentials = credentials.get_frozen_credentials()

    request = AWSRequest(
        method="GET",
        url=f"https://{host}{path}",
        headers={
            "host": host,
            "x-amz-date": datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
        },
    )

    auth = SigV4Auth(frozen_credentials, self.data_plane_service_name, self.region)
    auth.add_auth(request)

    headers = {
        "Host": host,
        "X-Amz-Date": request.headers["x-amz-date"],
        "Authorization": request.headers["Authorization"],
        "Upgrade": "websocket",
        "Connection": "Upgrade",
        "Sec-WebSocket-Version": "13",
        "Sec-WebSocket-Key": base64.b64encode(secrets.token_bytes(16)).decode(),
        "User-Agent": f"BrowserSandbox-Client/1.0 (Session: {self.session_id})",
    }

    if frozen_credentials.token:
        headers["X-Amz-Security-Token"] = frozen_credentials.token

    return ws_url, headers

release_control()

Release control of the browser session by enabling the automation stream.

This method enables external automation capabilities of the browser session, relinquishing exclusive control. If no session exists, a warning is logged and the method returns without taking action.

Source code in bedrock_agentcore/tools/browser_client.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def release_control(self):
    """Release control of the browser session by enabling the automation stream.

    This method enables external automation capabilities of the browser session,
    relinquishing exclusive control. If no session exists, a warning is logged
    and the method returns without taking action.
    """
    self.logger.info("Releasing control of browser session...")

    if not self.identifier or not self.session_id:
        self.logger.warning("Could not find a browser session when releasing control")
        return

    self._update_browser_stream(self.identifier, self.session_id, "ENABLED")

start(identifier=DEFAULT_IDENTIFIER, name=None, session_timeout_seconds=DEFAULT_SESSION_TIMEOUT)

Start a browser sandbox session.

This method initializes a new browser session with the provided parameters.

Parameters:

Name Type Description Default
identifier Optional[str]

The browser sandbox identifier to use. Defaults to DEFAULT_IDENTIFIER.

 DEFAULT_IDENTIFIER
name Optional[str]

A name for this session. If not provided, a name will be generated using a UUID.

 None
session_timeout_seconds Optional[int]

The timeout for the session in seconds. Defaults to DEFAULT_TIMEOUT.

 DEFAULT_SESSION_TIMEOUT
description Optional[str]

A description for this session. Defaults to an empty string.

 required

Returns:

Name Type Description
str str

The session ID of the newly created session.
Source code in bedrock_agentcore/tools/browser_client.py 
 96
 97
 98
 99
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
def start(
    self,
    identifier: Optional[str] = DEFAULT_IDENTIFIER,
    name: Optional[str] = None,
    session_timeout_seconds: Optional[int] = DEFAULT_SESSION_TIMEOUT,
) -> str:
    """Start a browser sandbox session.

    This method initializes a new browser session with the provided parameters.

    Args:
        identifier (Optional[str]): The browser sandbox identifier to use.
            Defaults to DEFAULT_IDENTIFIER.
        name (Optional[str]): A name for this session. If not provided, a name
            will be generated using a UUID.
        session_timeout_seconds (Optional[int]): The timeout for the session in seconds.
            Defaults to DEFAULT_TIMEOUT.
        description (Optional[str]): A description for this session.
            Defaults to an empty string.

    Returns:
        str: The session ID of the newly created session.
    """
    self.logger.info("Starting browser session...")

    response = self.client.start_browser_session(
        browserIdentifier=identifier,
        name=name or f"browser-session-{uuid.uuid4().hex[:8]}",
        sessionTimeoutSeconds=session_timeout_seconds,
    )

    self.identifier = response["browserIdentifier"]
    self.session_id = response["sessionId"]

    return self.session_id

stop()

Stop the current browser session if one is active.

This method stops any active session and clears the session state. If no session is active, this method does nothing.

Returns:

Name Type Description
bool

True if no session was active or the session was successfully stopped.
Source code in bedrock_agentcore/tools/browser_client.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def stop(self):
    """Stop the current browser session if one is active.

    This method stops any active session and clears the session state.
    If no session is active, this method does nothing.

    Returns:
        bool: True if no session was active or the session was successfully stopped.
    """
    self.logger.info("Stopping browser session...")

    if not self.session_id or not self.identifier:
        return True

    self.client.stop_browser_session(**{"browserIdentifier": self.identifier, "sessionId": self.session_id})

    self.identifier = None
    self.session_id = None

take_control()

Take control of the browser session by disabling the automation stream.

This method disables external automation capabilities of the browser session, giving this client exclusive control. If no session is active, a new session will be started.

Raises:

Type Description
RuntimeError

If a session could not be found or started.
Source code in bedrock_agentcore/tools/browser_client.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def take_control(self):
    """Take control of the browser session by disabling the automation stream.

    This method disables external automation capabilities of the browser session,
    giving this client exclusive control. If no session is active, a new session
    will be started.

    Raises:
        RuntimeError: If a session could not be found or started.
    """
    self.logger.info("Taking control of browser session...")

    if not self.identifier or not self.session_id:
        self.start()

    if not self.identifier or not self.session_id:
        raise RuntimeError("Could not find or start a browser session")

    self._update_browser_stream(self.identifier, self.session_id, "DISABLED")

browser_session(region)

Context manager for creating and managing a browser sandbox session.

This context manager handles creating a client, starting a session, and ensuring the session is properly cleaned up when the context exits.

Parameters:

Name Type Description Default
region str

The AWS region to use for the Browser service.

 required

Yields:

Name Type Description
BrowserClient BrowserClient

An initialized and started browser client.
Example

with browser_session('us-west-2') as client: ... browser = client.get_browser_obj() ... page = browser.new_page() ... page.goto('https://example.com')

Source code in bedrock_agentcore/tools/browser_client.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
@contextmanager
def browser_session(region: str) -> Generator[BrowserClient, None, None]:
    """Context manager for creating and managing a browser sandbox session.

    This context manager handles creating a client, starting a session, and
    ensuring the session is properly cleaned up when the context exits.

    Args:
        region (str): The AWS region to use for the Browser service.

    Yields:
        BrowserClient: An initialized and started browser client.

    Example:
        >>> with browser_session('us-west-2') as client:
        ...     browser = client.get_browser_obj()
        ...     page = browser.new_page()
        ...     page.goto('https://example.com')
    """
    client = BrowserClient(region)
    client.start()

    try:
        yield client
    finally:
        client.stop()