Aws bedrock agentcore

AWS Bedrock AgentCore tools and runtime.

AgentCoreBrowserToolSpec

Bases: BaseToolSpec

AWS Bedrock AgentCore Browser Tool Spec.

This toolkit provides a set of tools for working with a remote browser environment:

• navigate_browser - Navigate to a URL
• click_element - Click on an element using CSS selectors
• extract_text - Extract all text from the current webpage
• extract_hyperlinks - Extract all hyperlinks from the current webpage
• get_elements - Get elements matching a CSS selector
• navigate_back - Navigate to the previous page
• current_webpage - Get information about the current webpage

The toolkit supports multiple threads by maintaining separate browser sessions for each thread ID.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

class AgentCoreBrowserToolSpec(BaseToolSpec):
    """
    AWS Bedrock AgentCore Browser Tool Spec.

    This toolkit provides a set of tools for working with a remote browser environment:

    * navigate_browser - Navigate to a URL
    * click_element - Click on an element using CSS selectors
    * extract_text - Extract all text from the current webpage
    * extract_hyperlinks - Extract all hyperlinks from the current webpage
    * get_elements - Get elements matching a CSS selector
    * navigate_back - Navigate to the previous page
    * current_webpage - Get information about the current webpage

    The toolkit supports multiple threads by maintaining separate browser sessions for each thread ID.
    """

    spec_functions = [
        ("navigate_browser", "anavigate_browser"),
        ("click_element", "aclick_element"),
        ("extract_text", "aextract_text"),
        ("extract_hyperlinks", "aextract_hyperlinks"),
        ("get_elements", "aget_elements"),
        ("navigate_back", "anavigate_back"),
        ("current_webpage", "acurrent_webpage"),
        ("generate_live_view_url", "agenerate_live_view_url"),
        ("take_control", "atake_control"),
        ("release_control", "arelease_control"),
    ]

    def __init__(
        self,
        region: Optional[str] = None,
        identifier: Optional[str] = None,
    ) -> None:
        """
        Initialize the AWS Bedrock AgentCore Browser Tool Spec.

        Args:
            region (Optional[str]): AWS region to use for Bedrock AgentCore services.
                If not provided, will try to get it from environment variables.
            identifier (Optional[str]): Custom browser identifier for VPC-enabled
                resources. If not provided, uses the default identifier.

        """
        self.region = region if region is not None else get_aws_region()
        self._identifier = identifier
        self._browser_clients: Dict[str, BrowserClient] = {}
        self._cp_browser_client: Optional[BrowserClient] = None
        self._session_manager = BrowserSessionManager(
            region=self.region, identifier=self._identifier
        )

    def _get_or_create_browser_client(
        self, thread_id: str = "default"
    ) -> BrowserClient:
        """
        Get or create a browser client for the specified thread.

        Args:
            thread_id: Thread ID for the browser session

        Returns:
            BrowserClient instance

        """
        if thread_id in self._browser_clients:
            return self._browser_clients[thread_id]

        # Create a new browser client for this thread
        browser_client = BrowserClient(self.region, integration_source="llamaindex")
        self._browser_clients[thread_id] = browser_client
        return browser_client

    def _get_control_plane_client(self) -> BrowserClient:
        """
        Get or create a browser client for control-plane operations only.

        This client is used for account-level operations (list, create, delete, get)
        that do not require a browser session.
        """
        if self._cp_browser_client is None:
            self._cp_browser_client = BrowserClient(
                self.region, integration_source="llamaindex"
            )
        return self._cp_browser_client

    def navigate_browser(
        self,
        url: str,
        thread_id: str = "default",
    ) -> str:
        """
        Navigate to a URL (synchronous version).

        Args:
            url (str): URL to navigate to.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Validate URL scheme
            parsed_url = urlparse(url)
            if parsed_url.scheme not in ("http", "https"):
                return f"URL scheme must be 'http' or 'https', got: {parsed_url.scheme}"

            # Get browser and navigate to URL
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)
                response = page.goto(url)
                status = response.status if response else "unknown"
                return f"Navigated to {url} with status code {status}"
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error navigating to URL: {e!s}"

    async def anavigate_browser(
        self,
        url: str,
        thread_id: str = "default",
    ) -> str:
        """
        Navigate to a URL (asynchronous version).

        Args:
            url (str): URL to navigate to.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Validate URL scheme
            parsed_url = urlparse(url)
            if parsed_url.scheme not in ("http", "https"):
                return f"URL scheme must be 'http' or 'https', got: {parsed_url.scheme}"

            # Get browser and navigate to URL
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)
            response = await page.goto(url)
            status = response.status if response else "unknown"

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return f"Navigated to {url} with status code {status}"
        except Exception as e:
            return f"Error navigating to URL: {e!s}"

    def click_element(
        self,
        selector: str,
        thread_id: str = "default",
    ) -> str:
        """
        Click on an element with the given CSS selector (synchronous version).

        Args:
            selector (str): CSS selector for the element to click on.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Get browser and click on element
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                try:
                    page.click(selector, timeout=5000)
                    return f"Clicked on element with selector '{selector}'"
                except Exception as click_error:
                    return f"Unable to click on element with selector '{selector}': {click_error!s}"
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error clicking on element: {e!s}"

    async def aclick_element(
        self,
        selector: str,
        thread_id: str = "default",
    ) -> str:
        """
        Click on an element with the given CSS selector (asynchronous version).

        Args:
            selector (str): CSS selector for the element to click on.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Get browser and click on element
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            try:
                await page.click(selector, timeout=5000)
                result = f"Clicked on element with selector '{selector}'"
            except Exception as click_error:
                result = f"Unable to click on element with selector '{selector}': {click_error!s}"

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return result
        except Exception as e:
            return f"Error clicking on element: {e!s}"

    def extract_text(
        self,
        selector: Optional[str] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Extract text from the current page (synchronous version).

        Args:
            selector (Optional[str]): CSS selector for the element to extract text from. If not provided, extracts text from the entire page.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: The extracted text.

        """
        try:
            # Get browser and extract text
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                if selector:
                    try:
                        element = page.query_selector(selector)
                        if element:
                            text = element.text_content()
                            result = (
                                text if text else "Element found but contains no text"
                            )
                        else:
                            result = f"No element found with selector '{selector}'"
                    except Exception as selector_error:
                        result = f"Error extracting text from selector '{selector}': {selector_error!s}"
                else:
                    # Extract text from the entire page
                    result = page.content()

                return result
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error extracting text: {e!s}"

    async def aextract_text(
        self,
        selector: Optional[str] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Extract text from the current page (asynchronous version).

        Args:
            selector (Optional[str]): CSS selector for the element to extract text from. If not provided, extracts text from the entire page.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: The extracted text.

        """
        try:
            # Get browser and extract text
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            if selector:
                try:
                    element = await page.query_selector(selector)
                    if element:
                        text = await element.text_content()
                        result = text if text else "Element found but contains no text"
                    else:
                        result = f"No element found with selector '{selector}'"
                except Exception as selector_error:
                    result = f"Error extracting text from selector '{selector}': {selector_error!s}"
            else:
                # Extract text from the entire page
                result = await page.content()

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return result
        except Exception as e:
            return f"Error extracting text: {e!s}"

    def extract_hyperlinks(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Extract hyperlinks from the current page (synchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: The extracted hyperlinks.

        """
        try:
            # Get browser and extract hyperlinks
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                # Extract all hyperlinks from the page
                links = page.eval_on_selector_all(
                    "a[href]",
                    """
                    (elements) => {
                        return elements.map(el => {
                            return {
                                text: el.innerText || el.textContent,
                                href: el.href
                            };
                        });
                    }
                """,
                )

                # Format the links
                formatted_links = []
                for i, link in enumerate(links):
                    formatted_links.append(
                        f"{i + 1}. {link.get('text', 'No text')}: {link.get('href', 'No href')}"
                    )

                return (
                    "\n".join(formatted_links)
                    if formatted_links
                    else "No hyperlinks found on the page"
                )
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error extracting hyperlinks: {e!s}"

    async def aextract_hyperlinks(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Extract hyperlinks from the current page (asynchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: The extracted hyperlinks.

        """
        try:
            # Get browser and extract hyperlinks
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            # Extract all hyperlinks from the page
            links = await page.eval_on_selector_all(
                "a[href]",
                """
                (elements) => {
                    return elements.map(el => {
                        return {
                            text: el.innerText || el.textContent,
                            href: el.href
                        };
                    });
                }
            """,
            )

            # Format the links
            formatted_links = []
            for i, link in enumerate(links):
                formatted_links.append(
                    f"{i + 1}. {link.get('text', 'No text')}: {link.get('href', 'No href')}"
                )

            result = (
                "\n".join(formatted_links)
                if formatted_links
                else "No hyperlinks found on the page"
            )

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return result
        except Exception as e:
            return f"Error extracting hyperlinks: {e!s}"

    def get_elements(
        self,
        selector: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get elements matching a CSS selector (synchronous version).

        Args:
            selector (str): CSS selector for the elements to get.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Information about the matching elements.

        """
        try:
            # Get browser and find elements
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                # Find elements matching the selector
                elements = page.query_selector_all(selector)

                if not elements:
                    result = f"No elements found matching selector '{selector}'"
                else:
                    # Extract information about the elements
                    elements_info = []
                    for i, element in enumerate(elements):
                        tag_name = element.evaluate("el => el.tagName.toLowerCase()")
                        text = element.text_content() or ""
                        attributes = element.evaluate(
                            """
                            (el) => {
                                const attrs = {};
                                for (const attr of el.attributes) {
                                    attrs[attr.name] = attr.value;
                                }
                                return attrs;
                            }
                        """
                        )

                        # Format element info
                        attr_str = ", ".join(
                            [f'{k}="{v}"' for k, v in attributes.items()]
                        )
                        elements_info.append(
                            f"{i + 1}. <{tag_name} {attr_str}>{text}</{tag_name}>"
                        )

                    result = (
                        f"Found {len(elements)} element(s) matching selector '{selector}':\n"
                        + "\n".join(elements_info)
                    )

                return result
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error getting elements: {e!s}"

    async def aget_elements(
        self,
        selector: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get elements matching a CSS selector (asynchronous version).

        Args:
            selector (str): CSS selector for the elements to get.
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Information about the matching elements.

        """
        try:
            # Get browser and find elements
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            # Find elements matching the selector
            elements = await page.query_selector_all(selector)

            if not elements:
                result = f"No elements found matching selector '{selector}'"
            else:
                # Extract information about the elements
                elements_info = []
                for i, element in enumerate(elements):
                    tag_name = await element.evaluate("el => el.tagName.toLowerCase()")
                    text = await element.text_content() or ""
                    attributes = await element.evaluate(
                        """
                        (el) => {
                            const attrs = {};
                            for (const attr of el.attributes) {
                                attrs[attr.name] = attr.value;
                            }
                            return attrs;
                        }
                    """
                    )

                    # Format element info
                    attr_str = ", ".join([f'{k}="{v}"' for k, v in attributes.items()])
                    elements_info.append(
                        f"{i + 1}. <{tag_name} {attr_str}>{text}</{tag_name}>"
                    )

                result = (
                    f"Found {len(elements)} element(s) matching selector '{selector}':\n"
                    + "\n".join(elements_info)
                )

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return result
        except Exception as e:
            return f"Error getting elements: {e!s}"

    def navigate_back(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Navigate to the previous page (synchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Get browser and navigate back
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                # Navigate back
                response = page.go_back()

                # Get the current URL after navigating back
                current_url = page.url if response else "unknown"

                if response:
                    return f"Navigated back to {current_url}"
                else:
                    return "Could not navigate back (no previous page in history)"
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error navigating back: {e!s}"

    async def anavigate_back(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Navigate to the previous page (asynchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Confirmation message.

        """
        try:
            # Get browser and navigate back
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            # Navigate back
            response = await page.go_back()

            # Get the current URL after navigating back
            current_url = page.url if response else "unknown"

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            if response:
                return f"Navigated back to {current_url}"
            else:
                return "Could not navigate back (no previous page in history)"
        except Exception as e:
            return f"Error navigating back: {e!s}"

    def current_webpage(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Get information about the current webpage (synchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Information about the current webpage.

        """
        try:
            # Get browser and get current webpage info
            browser = self._session_manager.get_sync_browser(thread_id)
            try:
                page = get_current_page(browser)

                # Get the current URL
                url = page.url

                # Get the page title
                title = page.title()

                # Get basic page metrics
                metrics = page.evaluate(
                    """
                    () => {
                        return {
                            width: document.documentElement.clientWidth,
                            height: document.documentElement.clientHeight,
                            links: document.querySelectorAll('a').length,
                            images: document.querySelectorAll('img').length,
                            forms: document.querySelectorAll('form').length
                        }
                    }
                """
                )

                # Format the result
                result = f"Current webpage information:\n"
                result += f"URL: {url}\n"
                result += f"Title: {title}\n"
                result += f"Viewport size: {metrics['width']}x{metrics['height']}\n"
                result += f"Links: {metrics['links']}\n"
                result += f"Images: {metrics['images']}\n"
                result += f"Forms: {metrics['forms']}"

                return result
            finally:
                self._session_manager.release_sync_browser(thread_id)
        except Exception as e:
            return f"Error getting current webpage information: {e!s}"

    async def acurrent_webpage(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Get information about the current webpage (asynchronous version).

        Args:
            thread_id (str): Thread ID for the browser session.

        Returns:
            str: Information about the current webpage.

        """
        try:
            # Get browser and get current webpage info
            browser = await self._session_manager.get_async_browser(thread_id)
            page = await aget_current_page(browser)

            # Get the current URL
            url = page.url

            # Get the page title
            title = await page.title()

            # Get basic page metrics
            metrics = await page.evaluate(
                """
                () => {
                    return {
                        width: document.documentElement.clientWidth,
                        height: document.documentElement.clientHeight,
                        links: document.querySelectorAll('a').length,
                        images: document.querySelectorAll('img').length,
                        forms: document.querySelectorAll('form').length
                    }
                }
            """
            )

            # Format the result
            result = f"Current webpage information:\n"
            result += f"URL: {url}\n"
            result += f"Title: {title}\n"
            result += f"Viewport size: {metrics['width']}x{metrics['height']}\n"
            result += f"Links: {metrics['links']}\n"
            result += f"Images: {metrics['images']}\n"
            result += f"Forms: {metrics['forms']}"

            # Release the browser
            await self._session_manager.release_async_browser(thread_id)

            return result
        except Exception as e:
            return f"Error getting current webpage information: {e!s}"

    def generate_live_view_url(
        self,
        expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
        thread_id: str = "default",
    ) -> str:
        """
        Generate a presigned URL for live viewing a browser session (synchronous version).

        This URL allows a human to observe the browser session in real-time for oversight.
        A browser session must already exist for the given thread_id (e.g., by navigating
        to a URL first).

        Args:
            expires (int): Seconds until the URL expires. Maximum 300. Default is 300.
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: The presigned URL for viewing the browser session.

        """
        try:
            browser_client = self._session_manager.get_browser_client(thread_id)
            if browser_client is None:
                return (
                    f"No browser session found for thread '{thread_id}'. "
                    "Navigate to a URL first to start a session."
                )
            return browser_client.generate_live_view_url(expires=expires)
        except Exception as e:
            return f"Error generating live view URL: {e!s}"

    async def agenerate_live_view_url(
        self,
        expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
        thread_id: str = "default",
    ) -> str:
        """
        Generate a presigned URL for live viewing a browser session (asynchronous version).

        This URL allows a human to observe the browser session in real-time for oversight.
        A browser session must already exist for the given thread_id (e.g., by navigating
        to a URL first).

        Args:
            expires (int): Seconds until the URL expires. Maximum 300. Default is 300.
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: The presigned URL for viewing the browser session.

        """
        return await asyncio.to_thread(
            self.generate_live_view_url, expires=expires, thread_id=thread_id
        )

    def list_browsers(
        self,
        browser_type: Optional[str] = None,
        max_results: int = 10,
        thread_id: str = "default",
    ) -> str:
        """
        List all browsers in the account (synchronous version).

        Args:
            browser_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
            max_results (int): Maximum results to return (1-100). Default is 10.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: JSON-formatted list of browser summaries.

        """
        try:
            browser_client = self._get_control_plane_client()
            response = browser_client.list_browsers(
                browser_type=browser_type, max_results=max_results
            )
            summaries = response.get("browserSummaries", [])
            if not summaries:
                return "No browsers found."
            lines = []
            for b in summaries:
                lines.append(
                    f"- {b.get('name', 'N/A')} (ID: {b.get('browserId', 'N/A')}, "
                    f"Status: {b.get('status', 'N/A')}, Type: {b.get('type', 'N/A')})"
                )
            return f"Found {len(summaries)} browser(s):\n" + "\n".join(lines)
        except Exception as e:
            return f"Error listing browsers: {e!s}"

    async def alist_browsers(
        self,
        browser_type: Optional[str] = None,
        max_results: int = 10,
        thread_id: str = "default",
    ) -> str:
        """
        List all browsers in the account (asynchronous version).

        Args:
            browser_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
            max_results (int): Maximum results to return (1-100). Default is 10.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: JSON-formatted list of browser summaries.

        """
        return await asyncio.to_thread(
            self.list_browsers,
            browser_type=browser_type,
            max_results=max_results,
            thread_id=thread_id,
        )

    def create_browser(
        self,
        name: str,
        execution_role_arn: str,
        network_mode: str = "PUBLIC",
        description: str = "",
        subnet_ids: Optional[List[str]] = None,
        security_group_ids: Optional[List[str]] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Create a custom browser with specific configuration (synchronous version).

        Args:
            name (str): Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
            execution_role_arn (str): IAM role ARN with permissions for browser operations.
            network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
            description (str): Description of the browser. Default is "".
            subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
            security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation with browser ID and status.

        """
        try:
            browser_client = self._get_control_plane_client()
            network_config: Dict[str, Any] = {"networkMode": network_mode}
            if subnet_ids or security_group_ids:
                vpc_config: Dict[str, Any] = {}
                if subnet_ids:
                    vpc_config["subnets"] = subnet_ids
                if security_group_ids:
                    vpc_config["securityGroups"] = security_group_ids
                network_config["vpcConfig"] = vpc_config
            kwargs: Dict[str, Any] = {
                "name": name,
                "execution_role_arn": execution_role_arn,
                "network_configuration": network_config,
            }
            if description:
                kwargs["description"] = description
            response = browser_client.create_browser(**kwargs)
            browser_id = response.get("browserId", "unknown")
            status = response.get("status", "unknown")
            return f"Browser created (ID: {browser_id}, Status: {status})"
        except Exception as e:
            return f"Error creating browser: {e!s}"

    async def acreate_browser(
        self,
        name: str,
        execution_role_arn: str,
        network_mode: str = "PUBLIC",
        description: str = "",
        subnet_ids: Optional[List[str]] = None,
        security_group_ids: Optional[List[str]] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Create a custom browser with specific configuration (asynchronous version).

        Args:
            name (str): Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
            execution_role_arn (str): IAM role ARN with permissions for browser operations.
            network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
            description (str): Description of the browser. Default is "".
            subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
            security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation with browser ID and status.

        """
        return await asyncio.to_thread(
            self.create_browser,
            name=name,
            execution_role_arn=execution_role_arn,
            network_mode=network_mode,
            description=description,
            subnet_ids=subnet_ids,
            security_group_ids=security_group_ids,
            thread_id=thread_id,
        )

    def delete_browser(
        self,
        browser_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Delete a custom browser (synchronous version).

        Args:
            browser_id (str): The browser identifier to delete.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation of deletion.

        """
        try:
            browser_client = self._get_control_plane_client()
            response = browser_client.delete_browser(browser_id=browser_id)
            status = response.get("status", "unknown")
            return f"Browser '{browser_id}' deleted (Status: {status})"
        except Exception as e:
            return f"Error deleting browser: {e!s}"

    async def adelete_browser(
        self,
        browser_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Delete a custom browser (asynchronous version).

        Args:
            browser_id (str): The browser identifier to delete.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation of deletion.

        """
        return await asyncio.to_thread(
            self.delete_browser, browser_id=browser_id, thread_id=thread_id
        )

    def get_browser(
        self,
        browser_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get detailed information about a browser (synchronous version).

        Args:
            browser_id (str): The browser identifier.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Browser details including name, status, and configuration.

        """
        try:
            browser_client = self._get_control_plane_client()
            response = browser_client.get_browser(browser_id=browser_id)
            name = response.get("name", "N/A")
            status = response.get("status", "N/A")
            desc = response.get("description", "")
            result = f"Browser '{browser_id}':\n"
            result += f"  Name: {name}\n"
            result += f"  Status: {status}\n"
            if desc:
                result += f"  Description: {desc}\n"
            network = response.get("networkConfiguration", {})
            if network:
                result += f"  Network mode: {network.get('networkMode', 'N/A')}"
            return result
        except Exception as e:
            return f"Error getting browser: {e!s}"

    async def aget_browser(
        self,
        browser_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get detailed information about a browser (asynchronous version).

        Args:
            browser_id (str): The browser identifier.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Browser details including name, status, and configuration.

        """
        return await asyncio.to_thread(
            self.get_browser, browser_id=browser_id, thread_id=thread_id
        )

    def take_control(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Take manual control of a browser session by disabling the automation stream (synchronous version).

        This allows a human to interact with the browser via the live view URL while
        preventing the automation agent from making changes.

        Args:
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: Confirmation message.

        """
        try:
            browser_client = self._session_manager.get_browser_client(thread_id)
            if browser_client is None:
                return (
                    f"No browser session found for thread '{thread_id}'. "
                    "Navigate to a URL first to start a session."
                )
            browser_client.take_control()
            return "Took manual control of the browser session. Automation stream disabled."
        except Exception as e:
            return f"Error taking control: {e!s}"

    async def atake_control(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Take manual control of a browser session by disabling the automation stream (asynchronous version).

        Args:
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: Confirmation message.

        """
        return await asyncio.to_thread(self.take_control, thread_id=thread_id)

    def release_control(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Release manual control and re-enable the automation stream (synchronous version).

        This returns control to the automation agent after manual interaction.

        Args:
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: Confirmation message.

        """
        try:
            browser_client = self._session_manager.get_browser_client(thread_id)
            if browser_client is None:
                return (
                    f"No browser session found for thread '{thread_id}'. "
                    "Navigate to a URL first to start a session."
                )
            browser_client.release_control()
            return "Released manual control. Automation stream re-enabled."
        except Exception as e:
            return f"Error releasing control: {e!s}"

    async def arelease_control(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Release manual control and re-enable the automation stream (asynchronous version).

        Args:
            thread_id (str): Thread ID for the browser session. Default is "default".

        Returns:
            str: Confirmation message.

        """
        return await asyncio.to_thread(self.release_control, thread_id=thread_id)

    async def cleanup(self, thread_id: Optional[str] = None) -> None:
        """
        Clean up resources

        Args:
            thread_id: Optional thread ID to clean up. If None, cleans up all sessions.

        """
        if thread_id:
            # Clean up a specific thread's session
            if thread_id in self._browser_clients:
                try:
                    self._browser_clients[thread_id].stop()
                    del self._browser_clients[thread_id]
                    logger.info(f"Browser session for thread {thread_id} cleaned up")
                except Exception as e:
                    logger.warning(
                        f"Error stopping browser for thread {thread_id}: {e}"
                    )
        else:
            # Clean up all sessions
            thread_ids = list(self._browser_clients.keys())
            for tid in thread_ids:
                try:
                    self._browser_clients[tid].stop()
                except Exception as e:
                    logger.warning(f"Error stopping browser for thread {tid}: {e}")

            self._browser_clients = {}
            logger.info("All browser sessions cleaned up")

navigate_browser

navigate_browser(
    url: str, thread_id: str = "default"
) -> str

Navigate to a URL (synchronous version).

Parameters:

Name	Type	Description	Default
url	str	URL to navigate to.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def navigate_browser(
    self,
    url: str,
    thread_id: str = "default",
) -> str:
    """
    Navigate to a URL (synchronous version).

    Args:
        url (str): URL to navigate to.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Validate URL scheme
        parsed_url = urlparse(url)
        if parsed_url.scheme not in ("http", "https"):
            return f"URL scheme must be 'http' or 'https', got: {parsed_url.scheme}"

        # Get browser and navigate to URL
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)
            response = page.goto(url)
            status = response.status if response else "unknown"
            return f"Navigated to {url} with status code {status}"
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error navigating to URL: {e!s}"

anavigate_browser async

anavigate_browser(
    url: str, thread_id: str = "default"
) -> str

Navigate to a URL (asynchronous version).

Parameters:

Name	Type	Description	Default
url	str	URL to navigate to.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def anavigate_browser(
    self,
    url: str,
    thread_id: str = "default",
) -> str:
    """
    Navigate to a URL (asynchronous version).

    Args:
        url (str): URL to navigate to.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Validate URL scheme
        parsed_url = urlparse(url)
        if parsed_url.scheme not in ("http", "https"):
            return f"URL scheme must be 'http' or 'https', got: {parsed_url.scheme}"

        # Get browser and navigate to URL
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)
        response = await page.goto(url)
        status = response.status if response else "unknown"

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return f"Navigated to {url} with status code {status}"
    except Exception as e:
        return f"Error navigating to URL: {e!s}"

click_element

click_element(
    selector: str, thread_id: str = "default"
) -> str

Click on an element with the given CSS selector (synchronous version).

Parameters:

Name	Type	Description	Default
selector	str	CSS selector for the element to click on.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def click_element(
    self,
    selector: str,
    thread_id: str = "default",
) -> str:
    """
    Click on an element with the given CSS selector (synchronous version).

    Args:
        selector (str): CSS selector for the element to click on.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Get browser and click on element
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            try:
                page.click(selector, timeout=5000)
                return f"Clicked on element with selector '{selector}'"
            except Exception as click_error:
                return f"Unable to click on element with selector '{selector}': {click_error!s}"
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error clicking on element: {e!s}"

aclick_element async

aclick_element(
    selector: str, thread_id: str = "default"
) -> str

Click on an element with the given CSS selector (asynchronous version).

Parameters:

Name	Type	Description	Default
selector	str	CSS selector for the element to click on.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def aclick_element(
    self,
    selector: str,
    thread_id: str = "default",
) -> str:
    """
    Click on an element with the given CSS selector (asynchronous version).

    Args:
        selector (str): CSS selector for the element to click on.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Get browser and click on element
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        try:
            await page.click(selector, timeout=5000)
            result = f"Clicked on element with selector '{selector}'"
        except Exception as click_error:
            result = f"Unable to click on element with selector '{selector}': {click_error!s}"

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return result
    except Exception as e:
        return f"Error clicking on element: {e!s}"

extract_text

extract_text(
    selector: Optional[str] = None,
    thread_id: str = "default",
) -> str

Extract text from the current page (synchronous version).

Parameters:

Name	Type	Description	Default
selector	Optional[str]	CSS selector for the element to extract text from. If not provided, extracts text from the entire page.	None
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	The extracted text.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def extract_text(
    self,
    selector: Optional[str] = None,
    thread_id: str = "default",
) -> str:
    """
    Extract text from the current page (synchronous version).

    Args:
        selector (Optional[str]): CSS selector for the element to extract text from. If not provided, extracts text from the entire page.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: The extracted text.

    """
    try:
        # Get browser and extract text
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            if selector:
                try:
                    element = page.query_selector(selector)
                    if element:
                        text = element.text_content()
                        result = (
                            text if text else "Element found but contains no text"
                        )
                    else:
                        result = f"No element found with selector '{selector}'"
                except Exception as selector_error:
                    result = f"Error extracting text from selector '{selector}': {selector_error!s}"
            else:
                # Extract text from the entire page
                result = page.content()

            return result
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error extracting text: {e!s}"

aextract_text async

aextract_text(
    selector: Optional[str] = None,
    thread_id: str = "default",
) -> str

Extract text from the current page (asynchronous version).

Parameters:

Name	Type	Description	Default
selector	Optional[str]	CSS selector for the element to extract text from. If not provided, extracts text from the entire page.	None
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	The extracted text.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def aextract_text(
    self,
    selector: Optional[str] = None,
    thread_id: str = "default",
) -> str:
    """
    Extract text from the current page (asynchronous version).

    Args:
        selector (Optional[str]): CSS selector for the element to extract text from. If not provided, extracts text from the entire page.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: The extracted text.

    """
    try:
        # Get browser and extract text
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        if selector:
            try:
                element = await page.query_selector(selector)
                if element:
                    text = await element.text_content()
                    result = text if text else "Element found but contains no text"
                else:
                    result = f"No element found with selector '{selector}'"
            except Exception as selector_error:
                result = f"Error extracting text from selector '{selector}': {selector_error!s}"
        else:
            # Extract text from the entire page
            result = await page.content()

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return result
    except Exception as e:
        return f"Error extracting text: {e!s}"

extract_hyperlinks

extract_hyperlinks(thread_id: str = 'default') -> str

Extract hyperlinks from the current page (synchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	The extracted hyperlinks.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def extract_hyperlinks(
    self,
    thread_id: str = "default",
) -> str:
    """
    Extract hyperlinks from the current page (synchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: The extracted hyperlinks.

    """
    try:
        # Get browser and extract hyperlinks
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            # Extract all hyperlinks from the page
            links = page.eval_on_selector_all(
                "a[href]",
                """
                (elements) => {
                    return elements.map(el => {
                        return {
                            text: el.innerText || el.textContent,
                            href: el.href
                        };
                    });
                }
            """,
            )

            # Format the links
            formatted_links = []
            for i, link in enumerate(links):
                formatted_links.append(
                    f"{i + 1}. {link.get('text', 'No text')}: {link.get('href', 'No href')}"
                )

            return (
                "\n".join(formatted_links)
                if formatted_links
                else "No hyperlinks found on the page"
            )
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error extracting hyperlinks: {e!s}"

aextract_hyperlinks async

aextract_hyperlinks(thread_id: str = 'default') -> str

Extract hyperlinks from the current page (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	The extracted hyperlinks.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def aextract_hyperlinks(
    self,
    thread_id: str = "default",
) -> str:
    """
    Extract hyperlinks from the current page (asynchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: The extracted hyperlinks.

    """
    try:
        # Get browser and extract hyperlinks
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        # Extract all hyperlinks from the page
        links = await page.eval_on_selector_all(
            "a[href]",
            """
            (elements) => {
                return elements.map(el => {
                    return {
                        text: el.innerText || el.textContent,
                        href: el.href
                    };
                });
            }
        """,
        )

        # Format the links
        formatted_links = []
        for i, link in enumerate(links):
            formatted_links.append(
                f"{i + 1}. {link.get('text', 'No text')}: {link.get('href', 'No href')}"
            )

        result = (
            "\n".join(formatted_links)
            if formatted_links
            else "No hyperlinks found on the page"
        )

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return result
    except Exception as e:
        return f"Error extracting hyperlinks: {e!s}"

get_elements

get_elements(
    selector: str, thread_id: str = "default"
) -> str

Get elements matching a CSS selector (synchronous version).

Parameters:

Name	Type	Description	Default
selector	str	CSS selector for the elements to get.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Information about the matching elements.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def get_elements(
    self,
    selector: str,
    thread_id: str = "default",
) -> str:
    """
    Get elements matching a CSS selector (synchronous version).

    Args:
        selector (str): CSS selector for the elements to get.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Information about the matching elements.

    """
    try:
        # Get browser and find elements
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            # Find elements matching the selector
            elements = page.query_selector_all(selector)

            if not elements:
                result = f"No elements found matching selector '{selector}'"
            else:
                # Extract information about the elements
                elements_info = []
                for i, element in enumerate(elements):
                    tag_name = element.evaluate("el => el.tagName.toLowerCase()")
                    text = element.text_content() or ""
                    attributes = element.evaluate(
                        """
                        (el) => {
                            const attrs = {};
                            for (const attr of el.attributes) {
                                attrs[attr.name] = attr.value;
                            }
                            return attrs;
                        }
                    """
                    )

                    # Format element info
                    attr_str = ", ".join(
                        [f'{k}="{v}"' for k, v in attributes.items()]
                    )
                    elements_info.append(
                        f"{i + 1}. <{tag_name} {attr_str}>{text}</{tag_name}>"
                    )

                result = (
                    f"Found {len(elements)} element(s) matching selector '{selector}':\n"
                    + "\n".join(elements_info)
                )

            return result
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error getting elements: {e!s}"

aget_elements async

aget_elements(
    selector: str, thread_id: str = "default"
) -> str

Get elements matching a CSS selector (asynchronous version).

Parameters:

Name	Type	Description	Default
selector	str	CSS selector for the elements to get.	required
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Information about the matching elements.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def aget_elements(
    self,
    selector: str,
    thread_id: str = "default",
) -> str:
    """
    Get elements matching a CSS selector (asynchronous version).

    Args:
        selector (str): CSS selector for the elements to get.
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Information about the matching elements.

    """
    try:
        # Get browser and find elements
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        # Find elements matching the selector
        elements = await page.query_selector_all(selector)

        if not elements:
            result = f"No elements found matching selector '{selector}'"
        else:
            # Extract information about the elements
            elements_info = []
            for i, element in enumerate(elements):
                tag_name = await element.evaluate("el => el.tagName.toLowerCase()")
                text = await element.text_content() or ""
                attributes = await element.evaluate(
                    """
                    (el) => {
                        const attrs = {};
                        for (const attr of el.attributes) {
                            attrs[attr.name] = attr.value;
                        }
                        return attrs;
                    }
                """
                )

                # Format element info
                attr_str = ", ".join([f'{k}="{v}"' for k, v in attributes.items()])
                elements_info.append(
                    f"{i + 1}. <{tag_name} {attr_str}>{text}</{tag_name}>"
                )

            result = (
                f"Found {len(elements)} element(s) matching selector '{selector}':\n"
                + "\n".join(elements_info)
            )

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return result
    except Exception as e:
        return f"Error getting elements: {e!s}"

navigate_back

navigate_back(thread_id: str = 'default') -> str

Navigate to the previous page (synchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def navigate_back(
    self,
    thread_id: str = "default",
) -> str:
    """
    Navigate to the previous page (synchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Get browser and navigate back
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            # Navigate back
            response = page.go_back()

            # Get the current URL after navigating back
            current_url = page.url if response else "unknown"

            if response:
                return f"Navigated back to {current_url}"
            else:
                return "Could not navigate back (no previous page in history)"
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error navigating back: {e!s}"

anavigate_back async

anavigate_back(thread_id: str = 'default') -> str

Navigate to the previous page (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def anavigate_back(
    self,
    thread_id: str = "default",
) -> str:
    """
    Navigate to the previous page (asynchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Confirmation message.

    """
    try:
        # Get browser and navigate back
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        # Navigate back
        response = await page.go_back()

        # Get the current URL after navigating back
        current_url = page.url if response else "unknown"

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        if response:
            return f"Navigated back to {current_url}"
        else:
            return "Could not navigate back (no previous page in history)"
    except Exception as e:
        return f"Error navigating back: {e!s}"

current_webpage

current_webpage(thread_id: str = 'default') -> str

Get information about the current webpage (synchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Information about the current webpage.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def current_webpage(
    self,
    thread_id: str = "default",
) -> str:
    """
    Get information about the current webpage (synchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Information about the current webpage.

    """
    try:
        # Get browser and get current webpage info
        browser = self._session_manager.get_sync_browser(thread_id)
        try:
            page = get_current_page(browser)

            # Get the current URL
            url = page.url

            # Get the page title
            title = page.title()

            # Get basic page metrics
            metrics = page.evaluate(
                """
                () => {
                    return {
                        width: document.documentElement.clientWidth,
                        height: document.documentElement.clientHeight,
                        links: document.querySelectorAll('a').length,
                        images: document.querySelectorAll('img').length,
                        forms: document.querySelectorAll('form').length
                    }
                }
            """
            )

            # Format the result
            result = f"Current webpage information:\n"
            result += f"URL: {url}\n"
            result += f"Title: {title}\n"
            result += f"Viewport size: {metrics['width']}x{metrics['height']}\n"
            result += f"Links: {metrics['links']}\n"
            result += f"Images: {metrics['images']}\n"
            result += f"Forms: {metrics['forms']}"

            return result
        finally:
            self._session_manager.release_sync_browser(thread_id)
    except Exception as e:
        return f"Error getting current webpage information: {e!s}"

acurrent_webpage async

acurrent_webpage(thread_id: str = 'default') -> str

Get information about the current webpage (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session.	'default'

Returns:

Name	Type	Description
str	str	Information about the current webpage.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def acurrent_webpage(
    self,
    thread_id: str = "default",
) -> str:
    """
    Get information about the current webpage (asynchronous version).

    Args:
        thread_id (str): Thread ID for the browser session.

    Returns:
        str: Information about the current webpage.

    """
    try:
        # Get browser and get current webpage info
        browser = await self._session_manager.get_async_browser(thread_id)
        page = await aget_current_page(browser)

        # Get the current URL
        url = page.url

        # Get the page title
        title = await page.title()

        # Get basic page metrics
        metrics = await page.evaluate(
            """
            () => {
                return {
                    width: document.documentElement.clientWidth,
                    height: document.documentElement.clientHeight,
                    links: document.querySelectorAll('a').length,
                    images: document.querySelectorAll('img').length,
                    forms: document.querySelectorAll('form').length
                }
            }
        """
        )

        # Format the result
        result = f"Current webpage information:\n"
        result += f"URL: {url}\n"
        result += f"Title: {title}\n"
        result += f"Viewport size: {metrics['width']}x{metrics['height']}\n"
        result += f"Links: {metrics['links']}\n"
        result += f"Images: {metrics['images']}\n"
        result += f"Forms: {metrics['forms']}"

        # Release the browser
        await self._session_manager.release_async_browser(thread_id)

        return result
    except Exception as e:
        return f"Error getting current webpage information: {e!s}"

generate_live_view_url

generate_live_view_url(
    expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
    thread_id: str = "default",
) -> str

Generate a presigned URL for live viewing a browser session (synchronous version).

This URL allows a human to observe the browser session in real-time for oversight. A browser session must already exist for the given thread_id (e.g., by navigating to a URL first).

Parameters:

Name	Type	Description	Default
expires	int	Seconds until the URL expires. Maximum 300. Default is 300.	DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The presigned URL for viewing the browser session.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def generate_live_view_url(
    self,
    expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
    thread_id: str = "default",
) -> str:
    """
    Generate a presigned URL for live viewing a browser session (synchronous version).

    This URL allows a human to observe the browser session in real-time for oversight.
    A browser session must already exist for the given thread_id (e.g., by navigating
    to a URL first).

    Args:
        expires (int): Seconds until the URL expires. Maximum 300. Default is 300.
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: The presigned URL for viewing the browser session.

    """
    try:
        browser_client = self._session_manager.get_browser_client(thread_id)
        if browser_client is None:
            return (
                f"No browser session found for thread '{thread_id}'. "
                "Navigate to a URL first to start a session."
            )
        return browser_client.generate_live_view_url(expires=expires)
    except Exception as e:
        return f"Error generating live view URL: {e!s}"

agenerate_live_view_url async

agenerate_live_view_url(
    expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
    thread_id: str = "default",
) -> str

Generate a presigned URL for live viewing a browser session (asynchronous version).

This URL allows a human to observe the browser session in real-time for oversight. A browser session must already exist for the given thread_id (e.g., by navigating to a URL first).

Parameters:

Name	Type	Description	Default
expires	int	Seconds until the URL expires. Maximum 300. Default is 300.	DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The presigned URL for viewing the browser session.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def agenerate_live_view_url(
    self,
    expires: int = DEFAULT_BROWSER_LIVE_VIEW_PRESIGNED_URL_TIMEOUT,
    thread_id: str = "default",
) -> str:
    """
    Generate a presigned URL for live viewing a browser session (asynchronous version).

    This URL allows a human to observe the browser session in real-time for oversight.
    A browser session must already exist for the given thread_id (e.g., by navigating
    to a URL first).

    Args:
        expires (int): Seconds until the URL expires. Maximum 300. Default is 300.
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: The presigned URL for viewing the browser session.

    """
    return await asyncio.to_thread(
        self.generate_live_view_url, expires=expires, thread_id=thread_id
    )

list_browsers

list_browsers(
    browser_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str

List all browsers in the account (synchronous version).

Parameters:

Name	Type	Description	Default
browser_type	Optional[str]	Filter by type: "SYSTEM" or "CUSTOM".	None
max_results	int	Maximum results to return (1-100). Default is 10.	10
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	JSON-formatted list of browser summaries.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def list_browsers(
    self,
    browser_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str:
    """
    List all browsers in the account (synchronous version).

    Args:
        browser_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
        max_results (int): Maximum results to return (1-100). Default is 10.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: JSON-formatted list of browser summaries.

    """
    try:
        browser_client = self._get_control_plane_client()
        response = browser_client.list_browsers(
            browser_type=browser_type, max_results=max_results
        )
        summaries = response.get("browserSummaries", [])
        if not summaries:
            return "No browsers found."
        lines = []
        for b in summaries:
            lines.append(
                f"- {b.get('name', 'N/A')} (ID: {b.get('browserId', 'N/A')}, "
                f"Status: {b.get('status', 'N/A')}, Type: {b.get('type', 'N/A')})"
            )
        return f"Found {len(summaries)} browser(s):\n" + "\n".join(lines)
    except Exception as e:
        return f"Error listing browsers: {e!s}"

alist_browsers async

alist_browsers(
    browser_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str

List all browsers in the account (asynchronous version).

Parameters:

Name	Type	Description	Default
browser_type	Optional[str]	Filter by type: "SYSTEM" or "CUSTOM".	None
max_results	int	Maximum results to return (1-100). Default is 10.	10
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	JSON-formatted list of browser summaries.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def alist_browsers(
    self,
    browser_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str:
    """
    List all browsers in the account (asynchronous version).

    Args:
        browser_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
        max_results (int): Maximum results to return (1-100). Default is 10.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: JSON-formatted list of browser summaries.

    """
    return await asyncio.to_thread(
        self.list_browsers,
        browser_type=browser_type,
        max_results=max_results,
        thread_id=thread_id,
    )

create_browser

create_browser(
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str

Create a custom browser with specific configuration (synchronous version).

Parameters:

Name	Type	Description	Default
name	str	Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_].	required
execution_role_arn	str	IAM role ARN with permissions for browser operations.	required
network_mode	str	Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".	'PUBLIC'
description	str	Description of the browser. Default is "".	''
subnet_ids	Optional[List[str]]	Subnet IDs for VPC mode.	None
security_group_ids	Optional[List[str]]	Security group IDs for VPC mode.	None
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation with browser ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def create_browser(
    self,
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str:
    """
    Create a custom browser with specific configuration (synchronous version).

    Args:
        name (str): Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
        execution_role_arn (str): IAM role ARN with permissions for browser operations.
        network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
        description (str): Description of the browser. Default is "".
        subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
        security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation with browser ID and status.

    """
    try:
        browser_client = self._get_control_plane_client()
        network_config: Dict[str, Any] = {"networkMode": network_mode}
        if subnet_ids or security_group_ids:
            vpc_config: Dict[str, Any] = {}
            if subnet_ids:
                vpc_config["subnets"] = subnet_ids
            if security_group_ids:
                vpc_config["securityGroups"] = security_group_ids
            network_config["vpcConfig"] = vpc_config
        kwargs: Dict[str, Any] = {
            "name": name,
            "execution_role_arn": execution_role_arn,
            "network_configuration": network_config,
        }
        if description:
            kwargs["description"] = description
        response = browser_client.create_browser(**kwargs)
        browser_id = response.get("browserId", "unknown")
        status = response.get("status", "unknown")
        return f"Browser created (ID: {browser_id}, Status: {status})"
    except Exception as e:
        return f"Error creating browser: {e!s}"

acreate_browser async

acreate_browser(
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str

Create a custom browser with specific configuration (asynchronous version).

Parameters:

Name	Type	Description	Default
name	str	Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_].	required
execution_role_arn	str	IAM role ARN with permissions for browser operations.	required
network_mode	str	Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".	'PUBLIC'
description	str	Description of the browser. Default is "".	''
subnet_ids	Optional[List[str]]	Subnet IDs for VPC mode.	None
security_group_ids	Optional[List[str]]	Security group IDs for VPC mode.	None
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation with browser ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def acreate_browser(
    self,
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str:
    """
    Create a custom browser with specific configuration (asynchronous version).

    Args:
        name (str): Name for the browser. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
        execution_role_arn (str): IAM role ARN with permissions for browser operations.
        network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
        description (str): Description of the browser. Default is "".
        subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
        security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation with browser ID and status.

    """
    return await asyncio.to_thread(
        self.create_browser,
        name=name,
        execution_role_arn=execution_role_arn,
        network_mode=network_mode,
        description=description,
        subnet_ids=subnet_ids,
        security_group_ids=security_group_ids,
        thread_id=thread_id,
    )

delete_browser

delete_browser(
    browser_id: str, thread_id: str = "default"
) -> str

Delete a custom browser (synchronous version).

Parameters:

Name	Type	Description	Default
browser_id	str	The browser identifier to delete.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation of deletion.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def delete_browser(
    self,
    browser_id: str,
    thread_id: str = "default",
) -> str:
    """
    Delete a custom browser (synchronous version).

    Args:
        browser_id (str): The browser identifier to delete.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation of deletion.

    """
    try:
        browser_client = self._get_control_plane_client()
        response = browser_client.delete_browser(browser_id=browser_id)
        status = response.get("status", "unknown")
        return f"Browser '{browser_id}' deleted (Status: {status})"
    except Exception as e:
        return f"Error deleting browser: {e!s}"

adelete_browser async

adelete_browser(
    browser_id: str, thread_id: str = "default"
) -> str

Delete a custom browser (asynchronous version).

Parameters:

Name	Type	Description	Default
browser_id	str	The browser identifier to delete.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation of deletion.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def adelete_browser(
    self,
    browser_id: str,
    thread_id: str = "default",
) -> str:
    """
    Delete a custom browser (asynchronous version).

    Args:
        browser_id (str): The browser identifier to delete.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation of deletion.

    """
    return await asyncio.to_thread(
        self.delete_browser, browser_id=browser_id, thread_id=thread_id
    )

get_browser

get_browser(
    browser_id: str, thread_id: str = "default"
) -> str

Get detailed information about a browser (synchronous version).

Parameters:

Name	Type	Description	Default
browser_id	str	The browser identifier.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Browser details including name, status, and configuration.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def get_browser(
    self,
    browser_id: str,
    thread_id: str = "default",
) -> str:
    """
    Get detailed information about a browser (synchronous version).

    Args:
        browser_id (str): The browser identifier.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Browser details including name, status, and configuration.

    """
    try:
        browser_client = self._get_control_plane_client()
        response = browser_client.get_browser(browser_id=browser_id)
        name = response.get("name", "N/A")
        status = response.get("status", "N/A")
        desc = response.get("description", "")
        result = f"Browser '{browser_id}':\n"
        result += f"  Name: {name}\n"
        result += f"  Status: {status}\n"
        if desc:
            result += f"  Description: {desc}\n"
        network = response.get("networkConfiguration", {})
        if network:
            result += f"  Network mode: {network.get('networkMode', 'N/A')}"
        return result
    except Exception as e:
        return f"Error getting browser: {e!s}"

aget_browser async

aget_browser(
    browser_id: str, thread_id: str = "default"
) -> str

Get detailed information about a browser (asynchronous version).

Parameters:

Name	Type	Description	Default
browser_id	str	The browser identifier.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Browser details including name, status, and configuration.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def aget_browser(
    self,
    browser_id: str,
    thread_id: str = "default",
) -> str:
    """
    Get detailed information about a browser (asynchronous version).

    Args:
        browser_id (str): The browser identifier.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Browser details including name, status, and configuration.

    """
    return await asyncio.to_thread(
        self.get_browser, browser_id=browser_id, thread_id=thread_id
    )

take_control

take_control(thread_id: str = 'default') -> str

Take manual control of a browser session by disabling the automation stream (synchronous version).

This allows a human to interact with the browser via the live view URL while preventing the automation agent from making changes.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def take_control(
    self,
    thread_id: str = "default",
) -> str:
    """
    Take manual control of a browser session by disabling the automation stream (synchronous version).

    This allows a human to interact with the browser via the live view URL while
    preventing the automation agent from making changes.

    Args:
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: Confirmation message.

    """
    try:
        browser_client = self._session_manager.get_browser_client(thread_id)
        if browser_client is None:
            return (
                f"No browser session found for thread '{thread_id}'. "
                "Navigate to a URL first to start a session."
            )
        browser_client.take_control()
        return "Took manual control of the browser session. Automation stream disabled."
    except Exception as e:
        return f"Error taking control: {e!s}"

atake_control async

atake_control(thread_id: str = 'default') -> str

Take manual control of a browser session by disabling the automation stream (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def atake_control(
    self,
    thread_id: str = "default",
) -> str:
    """
    Take manual control of a browser session by disabling the automation stream (asynchronous version).

    Args:
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: Confirmation message.

    """
    return await asyncio.to_thread(self.take_control, thread_id=thread_id)

release_control

release_control(thread_id: str = 'default') -> str

Release manual control and re-enable the automation stream (synchronous version).

This returns control to the automation agent after manual interaction.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

def release_control(
    self,
    thread_id: str = "default",
) -> str:
    """
    Release manual control and re-enable the automation stream (synchronous version).

    This returns control to the automation agent after manual interaction.

    Args:
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: Confirmation message.

    """
    try:
        browser_client = self._session_manager.get_browser_client(thread_id)
        if browser_client is None:
            return (
                f"No browser session found for thread '{thread_id}'. "
                "Navigate to a URL first to start a session."
            )
        browser_client.release_control()
        return "Released manual control. Automation stream re-enabled."
    except Exception as e:
        return f"Error releasing control: {e!s}"

arelease_control async

arelease_control(thread_id: str = 'default') -> str

Release manual control and re-enable the automation stream (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the browser session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def arelease_control(
    self,
    thread_id: str = "default",
) -> str:
    """
    Release manual control and re-enable the automation stream (asynchronous version).

    Args:
        thread_id (str): Thread ID for the browser session. Default is "default".

    Returns:
        str: Confirmation message.

    """
    return await asyncio.to_thread(self.release_control, thread_id=thread_id)

cleanup async

cleanup(thread_id: Optional[str] = None) -> None

Clean up resources

Parameters:

Name	Type	Description	Default
thread_id	Optional[str]	Optional thread ID to clean up. If None, cleans up all sessions.	None

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/browser/base.py

async def cleanup(self, thread_id: Optional[str] = None) -> None:
    """
    Clean up resources

    Args:
        thread_id: Optional thread ID to clean up. If None, cleans up all sessions.

    """
    if thread_id:
        # Clean up a specific thread's session
        if thread_id in self._browser_clients:
            try:
                self._browser_clients[thread_id].stop()
                del self._browser_clients[thread_id]
                logger.info(f"Browser session for thread {thread_id} cleaned up")
            except Exception as e:
                logger.warning(
                    f"Error stopping browser for thread {thread_id}: {e}"
                )
    else:
        # Clean up all sessions
        thread_ids = list(self._browser_clients.keys())
        for tid in thread_ids:
            try:
                self._browser_clients[tid].stop()
            except Exception as e:
                logger.warning(f"Error stopping browser for thread {tid}: {e}")

        self._browser_clients = {}
        logger.info("All browser sessions cleaned up")

AgentCoreCodeInterpreterToolSpec

Bases: BaseToolSpec

AWS Bedrock AgentCore Code Interpreter Tool Spec.

This toolkit provides a set of tools for working with a remote code interpreter environment:

• execute_code - Run code in various languages (primarily Python)
• execute_command - Run shell commands
• read_files - Read content of files in the environment
• list_files - List files in directories
• delete_files - Remove files from the environment
• write_files - Create or update files
• start_command - Start long-running commands asynchronously
• get_task - Check status of async tasks
• stop_task - Stop running tasks

The toolkit lazily initializes the code interpreter session on first use. It supports multiple threads by maintaining separate code interpreter sessions for each thread ID.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

class AgentCoreCodeInterpreterToolSpec(BaseToolSpec):
    """
    AWS Bedrock AgentCore Code Interpreter Tool Spec.

    This toolkit provides a set of tools for working with a remote code interpreter environment:

    * execute_code - Run code in various languages (primarily Python)
    * execute_command - Run shell commands
    * read_files - Read content of files in the environment
    * list_files - List files in directories
    * delete_files - Remove files from the environment
    * write_files - Create or update files
    * start_command - Start long-running commands asynchronously
    * get_task - Check status of async tasks
    * stop_task - Stop running tasks

    The toolkit lazily initializes the code interpreter session on first use.
    It supports multiple threads by maintaining separate code interpreter sessions for each thread ID.
    """

    spec_functions = [
        ("execute_code", "aexecute_code"),
        ("execute_command", "aexecute_command"),
        ("read_files", "aread_files"),
        ("list_files", "alist_files"),
        ("delete_files", "adelete_files"),
        ("write_files", "awrite_files"),
        ("start_command", "astart_command"),
        ("get_task", "aget_task"),
        ("stop_task", "astop_task"),
        ("upload_file", "aupload_file"),
        ("upload_files", "aupload_files"),
        ("install_packages", "ainstall_packages"),
        ("download_file", "adownload_file"),
        ("download_files", "adownload_files"),
        ("clear_context", "aclear_context"),
    ]

    def __init__(
        self,
        region: Optional[str] = None,
        identifier: Optional[str] = None,
    ) -> None:
        """
        Initialize the AWS Bedrock AgentCore Code Interpreter Tool Spec.

        Args:
            region (Optional[str]): AWS region to use for Bedrock AgentCore services.
                If not provided, will try to get it from environment variables.
            identifier (Optional[str]): Custom code interpreter identifier for
                VPC-enabled resources. If not provided, uses the default identifier.

        """
        self.region = region if region is not None else get_aws_region()
        self._identifier = identifier
        self._code_interpreters: Dict[str, CodeInterpreter] = {}
        self._cp_ci_client: Optional[CodeInterpreter] = None

    def _get_or_create_interpreter(self, thread_id: str = "default") -> CodeInterpreter:
        """
        Get or create a code interpreter for the specified thread.

        Args:
            thread_id: Thread ID for the code interpreter session

        Returns:
            CodeInterpreter instance

        """
        if thread_id in self._code_interpreters:
            return self._code_interpreters[thread_id]

        # Create a new code interpreter for this thread
        code_interpreter = CodeInterpreter(
            region=self.region, integration_source="llamaindex"
        )
        start_kwargs = {}
        if self._identifier is not None:
            start_kwargs["identifier"] = self._identifier
        code_interpreter.start(**start_kwargs)
        logger.info(
            f"Started code interpreter with session_id:{code_interpreter.session_id} for thread:{thread_id}"
        )

        # Store the interpreter
        self._code_interpreters[thread_id] = code_interpreter
        return code_interpreter

    def _get_control_plane_client(self) -> CodeInterpreter:
        """
        Get or create a code interpreter client for control-plane operations only.

        This client is used for account-level operations (list, create, delete, get)
        that do not require starting a session.
        """
        if self._cp_ci_client is None:
            self._cp_ci_client = CodeInterpreter(
                region=self.region, integration_source="llamaindex"
            )
        return self._cp_ci_client

    def execute_code(
        self,
        code: str,
        language: str = "python",
        clear_context: bool = False,
        thread_id: str = "default",
    ) -> str:
        """
        Execute code in the code interpreter sandbox (synchronous version).

        Args:
            code (str): The code to execute.
            language (str): The programming language of the code. Default is "python".
            clear_context (bool): Whether to clear execution context. Default is False.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the code execution.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Execute code
            response = code_interpreter.invoke(
                method="executeCode",
                params={
                    "code": code,
                    "language": language,
                    "clearContext": clear_context,
                },
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error executing code: {e!s}"

    async def aexecute_code(
        self,
        code: str,
        language: str = "python",
        clear_context: bool = False,
        thread_id: str = "default",
    ) -> str:
        """
        Execute code in the code interpreter sandbox (asynchronous version).

        Args:
            code (str): The code to execute.
            language (str): The programming language of the code. Default is "python".
            clear_context (bool): Whether to clear execution context. Default is False.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the code execution.

        """
        return await asyncio.to_thread(
            self.execute_code,
            code=code,
            language=language,
            clear_context=clear_context,
            thread_id=thread_id,
        )

    def execute_command(
        self,
        command: str,
        thread_id: str = "default",
    ) -> str:
        """
        Execute a shell command in the code interpreter sandbox (synchronous version).

        Args:
            command (str): The command to execute.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the command execution.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Execute command
            response = code_interpreter.invoke(
                method="executeCommand", params={"command": command}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error executing command: {e!s}"

    async def aexecute_command(
        self,
        command: str,
        thread_id: str = "default",
    ) -> str:
        """
        Execute a shell command in the code interpreter sandbox (asynchronous version).

        Args:
            command (str): The command to execute.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the command execution.

        """
        return await asyncio.to_thread(
            self.execute_command, command=command, thread_id=thread_id
        )

    def read_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Read content of files in the environment (synchronous version).

        Args:
            paths (List[str]): List of file paths to read.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The content of the files.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Read files
            response = code_interpreter.invoke(
                method="readFiles", params={"paths": paths}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error reading files: {e!s}"

    async def aread_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Read content of files in the environment (asynchronous version).

        Args:
            paths (List[str]): List of file paths to read.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The content of the files.

        """
        return await asyncio.to_thread(
            self.read_files, paths=paths, thread_id=thread_id
        )

    def list_files(
        self,
        directory_path: str = "",
        thread_id: str = "default",
    ) -> str:
        """
        List files in directories in the environment (synchronous version).

        Args:
            directory_path (str): Path to the directory to list. Default is current directory.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The list of files.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # List files
            response = code_interpreter.invoke(
                method="listFiles", params={"directoryPath": directory_path}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error listing files: {e!s}"

    async def alist_files(
        self,
        directory_path: str = "",
        thread_id: str = "default",
    ) -> str:
        """
        List files in directories in the environment (asynchronous version).

        Args:
            directory_path (str): Path to the directory to list. Default is current directory.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The list of files.

        """
        return await asyncio.to_thread(
            self.list_files, directory_path=directory_path, thread_id=thread_id
        )

    def delete_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Remove files from the environment (synchronous version).

        Args:
            paths (List[str]): List of file paths to delete.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the delete operation.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Remove files
            response = code_interpreter.invoke(
                method="removeFiles", params={"paths": paths}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error deleting files: {e!s}"

    async def adelete_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Remove files from the environment (asynchronous version).

        Args:
            paths (List[str]): List of file paths to delete.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the delete operation.

        """
        return await asyncio.to_thread(
            self.delete_files, paths=paths, thread_id=thread_id
        )

    def write_files(
        self,
        files: List[Dict[str, str]],
        thread_id: str = "default",
    ) -> str:
        """
        Create or update files in the environment (synchronous version).

        Args:
            files (List[Dict[str, str]]): List of dictionaries with path and text fields.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the write operation.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Write files
            response = code_interpreter.invoke(
                method="writeFiles", params={"content": files}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error writing files: {e!s}"

    async def awrite_files(
        self,
        files: List[Dict[str, str]],
        thread_id: str = "default",
    ) -> str:
        """
        Create or update files in the environment (asynchronous version).

        Args:
            files (List[Dict[str, str]]): List of dictionaries with path and text fields.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the write operation.

        """
        return await asyncio.to_thread(
            self.write_files, files=files, thread_id=thread_id
        )

    def start_command(
        self,
        command: str,
        thread_id: str = "default",
    ) -> str:
        """
        Start a long-running command asynchronously (synchronous version).

        Args:
            command (str): The command to execute asynchronously.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The task ID and status.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Start command execution
            response = code_interpreter.invoke(
                method="startCommandExecution", params={"command": command}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error starting command: {e!s}"

    async def astart_command(
        self,
        command: str,
        thread_id: str = "default",
    ) -> str:
        """
        Start a long-running command asynchronously (asynchronous version).

        Args:
            command (str): The command to execute asynchronously.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The task ID and status.

        """
        return await asyncio.to_thread(
            self.start_command, command=command, thread_id=thread_id
        )

    def get_task(
        self,
        task_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Check status of an async task (synchronous version).

        Args:
            task_id (str): The ID of the task to check.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The task status.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Get task status
            response = code_interpreter.invoke(
                method="getTask", params={"taskId": task_id}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error getting task status: {e!s}"

    async def aget_task(
        self,
        task_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Check status of an async task (asynchronous version).

        Args:
            task_id (str): The ID of the task to check.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The task status.

        """
        return await asyncio.to_thread(
            self.get_task, task_id=task_id, thread_id=thread_id
        )

    def stop_task(
        self,
        task_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Stop a running task (synchronous version).

        Args:
            task_id (str): The ID of the task to stop.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the stop operation.

        """
        try:
            # Get or create code interpreter
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

            # Stop task
            response = code_interpreter.invoke(
                method="stopTask", params={"taskId": task_id}
            )

            return extract_output_from_stream(response)
        except Exception as e:
            return f"Error stopping task: {e!s}"

    async def astop_task(
        self,
        task_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Stop a running task (asynchronous version).

        Args:
            task_id (str): The ID of the task to stop.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The result of the stop operation.

        """
        return await asyncio.to_thread(
            self.stop_task, task_id=task_id, thread_id=thread_id
        )

    def upload_file(
        self,
        path: str,
        content: str,
        description: str = "",
        thread_id: str = "default",
    ) -> str:
        """
        Upload a file to the code interpreter sandbox (synchronous version).

        Args:
            path (str): Relative file path in the sandbox.
            content (str): File content as a string.
            description (str): Semantic description of the file. Default is "".
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation message with the uploaded file path.

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            code_interpreter.upload_file(
                path=path, content=content, description=description
            )
            return f"Uploaded file to {path}"
        except Exception as e:
            return f"Error uploading file: {e!s}"

    async def aupload_file(
        self,
        path: str,
        content: str,
        description: str = "",
        thread_id: str = "default",
    ) -> str:
        """
        Upload a file to the code interpreter sandbox (asynchronous version).

        Args:
            path (str): Relative file path in the sandbox.
            content (str): File content as a string.
            description (str): Semantic description of the file. Default is "".
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation message with the uploaded file path.

        """
        return await asyncio.to_thread(
            self.upload_file,
            path=path,
            content=content,
            description=description,
            thread_id=thread_id,
        )

    def upload_files(
        self,
        files: List[Dict[str, str]],
        thread_id: str = "default",
    ) -> str:
        """
        Upload multiple files to the code interpreter sandbox (synchronous version).

        Args:
            files (List[Dict[str, str]]): List of file specifications, each with
                'path', 'content', and optional 'description' keys.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation message with the number of files uploaded.

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            code_interpreter.upload_files(files=files)
            return f"Uploaded {len(files)} file(s)"
        except Exception as e:
            return f"Error uploading files: {e!s}"

    async def aupload_files(
        self,
        files: List[Dict[str, str]],
        thread_id: str = "default",
    ) -> str:
        """
        Upload multiple files to the code interpreter sandbox (asynchronous version).

        Args:
            files (List[Dict[str, str]]): List of file specifications, each with
                'path', 'content', and optional 'description' keys.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation message with the number of files uploaded.

        """
        return await asyncio.to_thread(
            self.upload_files, files=files, thread_id=thread_id
        )

    def install_packages(
        self,
        packages: List[str],
        upgrade: bool = False,
        thread_id: str = "default",
    ) -> str:
        """
        Install Python packages in the code interpreter sandbox (synchronous version).

        Args:
            packages (List[str]): List of package names to install. Can include version
                specifiers (e.g., 'pandas>=2.0').
            upgrade (bool): Whether to upgrade existing packages. Default is False.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The pip install output (stdout/stderr).

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            result = code_interpreter.install_packages(
                packages=packages, upgrade=upgrade
            )
            return str(result)
        except Exception as e:
            return f"Error installing packages: {e!s}"

    async def ainstall_packages(
        self,
        packages: List[str],
        upgrade: bool = False,
        thread_id: str = "default",
    ) -> str:
        """
        Install Python packages in the code interpreter sandbox (asynchronous version).

        Args:
            packages (List[str]): List of package names to install. Can include version
                specifiers (e.g., 'pandas>=2.0').
            upgrade (bool): Whether to upgrade existing packages. Default is False.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The pip install output (stdout/stderr).

        """
        return await asyncio.to_thread(
            self.install_packages,
            packages=packages,
            upgrade=upgrade,
            thread_id=thread_id,
        )

    def download_file(
        self,
        path: str,
        thread_id: str = "default",
    ) -> str:
        """
        Download a file from the code interpreter sandbox (synchronous version).

        Args:
            path (str): Path to the file in the sandbox.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The file content as text, or base64-encoded string for binary files.

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            content = code_interpreter.download_file(path=path)
            if isinstance(content, bytes):
                encoded = base64.b64encode(content).decode("utf-8")
                return f"[base64 encoded binary file: {path}]\n{encoded}"
            return content
        except Exception as e:
            return f"Error downloading file: {e!s}"

    async def adownload_file(
        self,
        path: str,
        thread_id: str = "default",
    ) -> str:
        """
        Download a file from the code interpreter sandbox (asynchronous version).

        Args:
            path (str): Path to the file in the sandbox.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: The file content as text, or base64-encoded string for binary files.

        """
        return await asyncio.to_thread(
            self.download_file, path=path, thread_id=thread_id
        )

    def download_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Download multiple files from the code interpreter sandbox (synchronous version).

        Args:
            paths (List[str]): List of file paths in the sandbox.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Formatted output with each file's content.

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            results = code_interpreter.download_files(paths=paths)
            output = []
            for file_path, content in results.items():
                if isinstance(content, bytes):
                    encoded = base64.b64encode(content).decode("utf-8")
                    output.append(
                        f"==== File: {file_path} (binary, base64) ====\n{encoded}"
                    )
                else:
                    output.append(f"==== File: {file_path} ====\n{content}")
            return "\n\n".join(output)
        except Exception as e:
            return f"Error downloading files: {e!s}"

    async def adownload_files(
        self,
        paths: List[str],
        thread_id: str = "default",
    ) -> str:
        """
        Download multiple files from the code interpreter sandbox (asynchronous version).

        Args:
            paths (List[str]): List of file paths in the sandbox.
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Formatted output with each file's content.

        """
        return await asyncio.to_thread(
            self.download_files, paths=paths, thread_id=thread_id
        )

    def list_code_interpreters(
        self,
        interpreter_type: Optional[str] = None,
        max_results: int = 10,
        thread_id: str = "default",
    ) -> str:
        """
        List all code interpreters in the account (synchronous version).

        Args:
            interpreter_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
            max_results (int): Maximum results to return (1-100). Default is 10.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Formatted list of code interpreter summaries.

        """
        try:
            code_interpreter = self._get_control_plane_client()
            response = code_interpreter.list_code_interpreters(
                interpreter_type=interpreter_type, max_results=max_results
            )
            summaries = response.get("codeInterpreterSummaries", [])
            if not summaries:
                return "No code interpreters found."
            lines = []
            for ci in summaries:
                lines.append(
                    f"- {ci.get('name', 'N/A')} (ID: {ci.get('codeInterpreterId', 'N/A')}, "
                    f"Status: {ci.get('status', 'N/A')}, Type: {ci.get('type', 'N/A')})"
                )
            return f"Found {len(summaries)} code interpreter(s):\n" + "\n".join(lines)
        except Exception as e:
            return f"Error listing code interpreters: {e!s}"

    async def alist_code_interpreters(
        self,
        interpreter_type: Optional[str] = None,
        max_results: int = 10,
        thread_id: str = "default",
    ) -> str:
        """
        List all code interpreters in the account (asynchronous version).

        Args:
            interpreter_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
            max_results (int): Maximum results to return (1-100). Default is 10.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Formatted list of code interpreter summaries.

        """
        return await asyncio.to_thread(
            self.list_code_interpreters,
            interpreter_type=interpreter_type,
            max_results=max_results,
            thread_id=thread_id,
        )

    def create_code_interpreter(
        self,
        name: str,
        execution_role_arn: str,
        network_mode: str = "PUBLIC",
        description: str = "",
        subnet_ids: Optional[List[str]] = None,
        security_group_ids: Optional[List[str]] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Create a custom code interpreter with specific configuration (synchronous version).

        Args:
            name (str): Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
            execution_role_arn (str): IAM role ARN with permissions for interpreter operations.
            network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
            description (str): Description of the interpreter. Default is "".
            subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
            security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation with interpreter ID and status.

        """
        try:
            code_interpreter = self._get_control_plane_client()
            network_config: Dict[str, Any] = {"networkMode": network_mode}
            if subnet_ids or security_group_ids:
                vpc_config: Dict[str, Any] = {}
                if subnet_ids:
                    vpc_config["subnets"] = subnet_ids
                if security_group_ids:
                    vpc_config["securityGroups"] = security_group_ids
                network_config["vpcConfig"] = vpc_config
            kwargs: Dict[str, Any] = {
                "name": name,
                "execution_role_arn": execution_role_arn,
                "network_configuration": network_config,
            }
            if description:
                kwargs["description"] = description
            response = code_interpreter.create_code_interpreter(**kwargs)
            interpreter_id = response.get("codeInterpreterId", "unknown")
            status = response.get("status", "unknown")
            return f"Code interpreter created (ID: {interpreter_id}, Status: {status})"
        except Exception as e:
            return f"Error creating code interpreter: {e!s}"

    async def acreate_code_interpreter(
        self,
        name: str,
        execution_role_arn: str,
        network_mode: str = "PUBLIC",
        description: str = "",
        subnet_ids: Optional[List[str]] = None,
        security_group_ids: Optional[List[str]] = None,
        thread_id: str = "default",
    ) -> str:
        """
        Create a custom code interpreter with specific configuration (asynchronous version).

        Args:
            name (str): Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
            execution_role_arn (str): IAM role ARN with permissions for interpreter operations.
            network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
            description (str): Description of the interpreter. Default is "".
            subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
            security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation with interpreter ID and status.

        """
        return await asyncio.to_thread(
            self.create_code_interpreter,
            name=name,
            execution_role_arn=execution_role_arn,
            network_mode=network_mode,
            description=description,
            subnet_ids=subnet_ids,
            security_group_ids=security_group_ids,
            thread_id=thread_id,
        )

    def delete_code_interpreter(
        self,
        interpreter_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Delete a custom code interpreter (synchronous version).

        Args:
            interpreter_id (str): The code interpreter identifier to delete.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation of deletion.

        """
        try:
            code_interpreter = self._get_control_plane_client()
            response = code_interpreter.delete_code_interpreter(
                interpreter_id=interpreter_id
            )
            status = response.get("status", "unknown")
            return f"Code interpreter '{interpreter_id}' deleted (Status: {status})"
        except Exception as e:
            return f"Error deleting code interpreter: {e!s}"

    async def adelete_code_interpreter(
        self,
        interpreter_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Delete a custom code interpreter (asynchronous version).

        Args:
            interpreter_id (str): The code interpreter identifier to delete.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Confirmation of deletion.

        """
        return await asyncio.to_thread(
            self.delete_code_interpreter,
            interpreter_id=interpreter_id,
            thread_id=thread_id,
        )

    def get_code_interpreter(
        self,
        interpreter_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get detailed information about a code interpreter (synchronous version).

        Args:
            interpreter_id (str): The code interpreter identifier.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Interpreter details including name, status, and configuration.

        """
        try:
            code_interpreter = self._get_control_plane_client()
            response = code_interpreter.get_code_interpreter(
                interpreter_id=interpreter_id
            )
            name = response.get("name", "N/A")
            status = response.get("status", "N/A")
            desc = response.get("description", "")
            result = f"Code interpreter '{interpreter_id}':\n"
            result += f"  Name: {name}\n"
            result += f"  Status: {status}\n"
            if desc:
                result += f"  Description: {desc}\n"
            network = response.get("networkConfiguration", {})
            if network:
                result += f"  Network mode: {network.get('networkMode', 'N/A')}"
            return result
        except Exception as e:
            return f"Error getting code interpreter: {e!s}"

    async def aget_code_interpreter(
        self,
        interpreter_id: str,
        thread_id: str = "default",
    ) -> str:
        """
        Get detailed information about a code interpreter (asynchronous version).

        Args:
            interpreter_id (str): The code interpreter identifier.
            thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

        Returns:
            str: Interpreter details including name, status, and configuration.

        """
        return await asyncio.to_thread(
            self.get_code_interpreter,
            interpreter_id=interpreter_id,
            thread_id=thread_id,
        )

    def clear_context(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Clear all variable state in the Python execution context (synchronous version).

        This resets the interpreter to a fresh state, removing all previously defined
        variables, imports, and function definitions.

        Args:
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation that the context was cleared.

        """
        try:
            code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
            code_interpreter.clear_context()
            return "Python execution context cleared successfully."
        except Exception as e:
            return f"Error clearing context: {e!s}"

    async def aclear_context(
        self,
        thread_id: str = "default",
    ) -> str:
        """
        Clear all variable state in the Python execution context (asynchronous version).

        Args:
            thread_id (str): Thread ID for the code interpreter session. Default is "default".

        Returns:
            str: Confirmation that the context was cleared.

        """
        return await asyncio.to_thread(self.clear_context, thread_id=thread_id)

    async def cleanup(self, thread_id: Optional[str] = None) -> None:
        """
        Clean up resources

        Args:
            thread_id: Optional thread ID to clean up. If None, cleans up all sessions.

        """
        if thread_id:
            # Clean up a specific thread's session
            if thread_id in self._code_interpreters:
                try:
                    self._code_interpreters[thread_id].stop()
                    del self._code_interpreters[thread_id]
                    logger.info(
                        f"Code interpreter session for thread {thread_id} cleaned up"
                    )
                except Exception as e:
                    logger.warning(
                        f"Error stopping code interpreter for thread {thread_id}: {e}"
                    )
        else:
            # Clean up all sessions
            thread_ids = list(self._code_interpreters.keys())
            for tid in thread_ids:
                try:
                    self._code_interpreters[tid].stop()
                except Exception as e:
                    logger.warning(
                        f"Error stopping code interpreter for thread {tid}: {e}"
                    )

            self._code_interpreters = {}
            logger.info("All code interpreter sessions cleaned up")

execute_code

execute_code(
    code: str,
    language: str = "python",
    clear_context: bool = False,
    thread_id: str = "default",
) -> str

Execute code in the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
code	str	The code to execute.	required
language	str	The programming language of the code. Default is "python".	'python'
clear_context	bool	Whether to clear execution context. Default is False.	False
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the code execution.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def execute_code(
    self,
    code: str,
    language: str = "python",
    clear_context: bool = False,
    thread_id: str = "default",
) -> str:
    """
    Execute code in the code interpreter sandbox (synchronous version).

    Args:
        code (str): The code to execute.
        language (str): The programming language of the code. Default is "python".
        clear_context (bool): Whether to clear execution context. Default is False.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the code execution.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Execute code
        response = code_interpreter.invoke(
            method="executeCode",
            params={
                "code": code,
                "language": language,
                "clearContext": clear_context,
            },
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error executing code: {e!s}"

aexecute_code async

aexecute_code(
    code: str,
    language: str = "python",
    clear_context: bool = False,
    thread_id: str = "default",
) -> str

Execute code in the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
code	str	The code to execute.	required
language	str	The programming language of the code. Default is "python".	'python'
clear_context	bool	Whether to clear execution context. Default is False.	False
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the code execution.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aexecute_code(
    self,
    code: str,
    language: str = "python",
    clear_context: bool = False,
    thread_id: str = "default",
) -> str:
    """
    Execute code in the code interpreter sandbox (asynchronous version).

    Args:
        code (str): The code to execute.
        language (str): The programming language of the code. Default is "python".
        clear_context (bool): Whether to clear execution context. Default is False.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the code execution.

    """
    return await asyncio.to_thread(
        self.execute_code,
        code=code,
        language=language,
        clear_context=clear_context,
        thread_id=thread_id,
    )

execute_command

execute_command(
    command: str, thread_id: str = "default"
) -> str

Execute a shell command in the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
command	str	The command to execute.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the command execution.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def execute_command(
    self,
    command: str,
    thread_id: str = "default",
) -> str:
    """
    Execute a shell command in the code interpreter sandbox (synchronous version).

    Args:
        command (str): The command to execute.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the command execution.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Execute command
        response = code_interpreter.invoke(
            method="executeCommand", params={"command": command}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error executing command: {e!s}"

aexecute_command async

aexecute_command(
    command: str, thread_id: str = "default"
) -> str

Execute a shell command in the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
command	str	The command to execute.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the command execution.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aexecute_command(
    self,
    command: str,
    thread_id: str = "default",
) -> str:
    """
    Execute a shell command in the code interpreter sandbox (asynchronous version).

    Args:
        command (str): The command to execute.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the command execution.

    """
    return await asyncio.to_thread(
        self.execute_command, command=command, thread_id=thread_id
    )

read_files

read_files(
    paths: List[str], thread_id: str = "default"
) -> str

Read content of files in the environment (synchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths to read.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The content of the files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def read_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Read content of files in the environment (synchronous version).

    Args:
        paths (List[str]): List of file paths to read.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The content of the files.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Read files
        response = code_interpreter.invoke(
            method="readFiles", params={"paths": paths}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error reading files: {e!s}"

aread_files async

aread_files(
    paths: List[str], thread_id: str = "default"
) -> str

Read content of files in the environment (asynchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths to read.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The content of the files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aread_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Read content of files in the environment (asynchronous version).

    Args:
        paths (List[str]): List of file paths to read.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The content of the files.

    """
    return await asyncio.to_thread(
        self.read_files, paths=paths, thread_id=thread_id
    )

list_files

list_files(
    directory_path: str = "", thread_id: str = "default"
) -> str

List files in directories in the environment (synchronous version).

Parameters:

Name	Type	Description	Default
directory_path	str	Path to the directory to list. Default is current directory.	''
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The list of files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def list_files(
    self,
    directory_path: str = "",
    thread_id: str = "default",
) -> str:
    """
    List files in directories in the environment (synchronous version).

    Args:
        directory_path (str): Path to the directory to list. Default is current directory.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The list of files.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # List files
        response = code_interpreter.invoke(
            method="listFiles", params={"directoryPath": directory_path}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error listing files: {e!s}"

alist_files async

alist_files(
    directory_path: str = "", thread_id: str = "default"
) -> str

List files in directories in the environment (asynchronous version).

Parameters:

Name	Type	Description	Default
directory_path	str	Path to the directory to list. Default is current directory.	''
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The list of files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def alist_files(
    self,
    directory_path: str = "",
    thread_id: str = "default",
) -> str:
    """
    List files in directories in the environment (asynchronous version).

    Args:
        directory_path (str): Path to the directory to list. Default is current directory.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The list of files.

    """
    return await asyncio.to_thread(
        self.list_files, directory_path=directory_path, thread_id=thread_id
    )

delete_files

delete_files(
    paths: List[str], thread_id: str = "default"
) -> str

Remove files from the environment (synchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths to delete.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the delete operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def delete_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Remove files from the environment (synchronous version).

    Args:
        paths (List[str]): List of file paths to delete.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the delete operation.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Remove files
        response = code_interpreter.invoke(
            method="removeFiles", params={"paths": paths}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error deleting files: {e!s}"

adelete_files async

adelete_files(
    paths: List[str], thread_id: str = "default"
) -> str

Remove files from the environment (asynchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths to delete.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the delete operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def adelete_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Remove files from the environment (asynchronous version).

    Args:
        paths (List[str]): List of file paths to delete.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the delete operation.

    """
    return await asyncio.to_thread(
        self.delete_files, paths=paths, thread_id=thread_id
    )

write_files

write_files(
    files: List[Dict[str, str]], thread_id: str = "default"
) -> str

Create or update files in the environment (synchronous version).

Parameters:

Name	Type	Description	Default
files	List[Dict[str, str]]	List of dictionaries with path and text fields.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the write operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def write_files(
    self,
    files: List[Dict[str, str]],
    thread_id: str = "default",
) -> str:
    """
    Create or update files in the environment (synchronous version).

    Args:
        files (List[Dict[str, str]]): List of dictionaries with path and text fields.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the write operation.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Write files
        response = code_interpreter.invoke(
            method="writeFiles", params={"content": files}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error writing files: {e!s}"

awrite_files async

awrite_files(
    files: List[Dict[str, str]], thread_id: str = "default"
) -> str

Create or update files in the environment (asynchronous version).

Parameters:

Name	Type	Description	Default
files	List[Dict[str, str]]	List of dictionaries with path and text fields.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the write operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def awrite_files(
    self,
    files: List[Dict[str, str]],
    thread_id: str = "default",
) -> str:
    """
    Create or update files in the environment (asynchronous version).

    Args:
        files (List[Dict[str, str]]): List of dictionaries with path and text fields.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the write operation.

    """
    return await asyncio.to_thread(
        self.write_files, files=files, thread_id=thread_id
    )

start_command

start_command(
    command: str, thread_id: str = "default"
) -> str

Start a long-running command asynchronously (synchronous version).

Parameters:

Name	Type	Description	Default
command	str	The command to execute asynchronously.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The task ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def start_command(
    self,
    command: str,
    thread_id: str = "default",
) -> str:
    """
    Start a long-running command asynchronously (synchronous version).

    Args:
        command (str): The command to execute asynchronously.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The task ID and status.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Start command execution
        response = code_interpreter.invoke(
            method="startCommandExecution", params={"command": command}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error starting command: {e!s}"

astart_command async

astart_command(
    command: str, thread_id: str = "default"
) -> str

Start a long-running command asynchronously (asynchronous version).

Parameters:

Name	Type	Description	Default
command	str	The command to execute asynchronously.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The task ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def astart_command(
    self,
    command: str,
    thread_id: str = "default",
) -> str:
    """
    Start a long-running command asynchronously (asynchronous version).

    Args:
        command (str): The command to execute asynchronously.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The task ID and status.

    """
    return await asyncio.to_thread(
        self.start_command, command=command, thread_id=thread_id
    )

get_task

get_task(task_id: str, thread_id: str = 'default') -> str

Check status of an async task (synchronous version).

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to check.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The task status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def get_task(
    self,
    task_id: str,
    thread_id: str = "default",
) -> str:
    """
    Check status of an async task (synchronous version).

    Args:
        task_id (str): The ID of the task to check.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The task status.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Get task status
        response = code_interpreter.invoke(
            method="getTask", params={"taskId": task_id}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error getting task status: {e!s}"

aget_task async

aget_task(task_id: str, thread_id: str = 'default') -> str

Check status of an async task (asynchronous version).

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to check.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The task status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aget_task(
    self,
    task_id: str,
    thread_id: str = "default",
) -> str:
    """
    Check status of an async task (asynchronous version).

    Args:
        task_id (str): The ID of the task to check.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The task status.

    """
    return await asyncio.to_thread(
        self.get_task, task_id=task_id, thread_id=thread_id
    )

stop_task

stop_task(task_id: str, thread_id: str = 'default') -> str

Stop a running task (synchronous version).

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to stop.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the stop operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def stop_task(
    self,
    task_id: str,
    thread_id: str = "default",
) -> str:
    """
    Stop a running task (synchronous version).

    Args:
        task_id (str): The ID of the task to stop.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the stop operation.

    """
    try:
        # Get or create code interpreter
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)

        # Stop task
        response = code_interpreter.invoke(
            method="stopTask", params={"taskId": task_id}
        )

        return extract_output_from_stream(response)
    except Exception as e:
        return f"Error stopping task: {e!s}"

astop_task async

astop_task(task_id: str, thread_id: str = 'default') -> str

Stop a running task (asynchronous version).

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to stop.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The result of the stop operation.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def astop_task(
    self,
    task_id: str,
    thread_id: str = "default",
) -> str:
    """
    Stop a running task (asynchronous version).

    Args:
        task_id (str): The ID of the task to stop.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The result of the stop operation.

    """
    return await asyncio.to_thread(
        self.stop_task, task_id=task_id, thread_id=thread_id
    )

upload_file

upload_file(
    path: str,
    content: str,
    description: str = "",
    thread_id: str = "default",
) -> str

Upload a file to the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
path	str	Relative file path in the sandbox.	required
content	str	File content as a string.	required
description	str	Semantic description of the file. Default is "".	''
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message with the uploaded file path.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def upload_file(
    self,
    path: str,
    content: str,
    description: str = "",
    thread_id: str = "default",
) -> str:
    """
    Upload a file to the code interpreter sandbox (synchronous version).

    Args:
        path (str): Relative file path in the sandbox.
        content (str): File content as a string.
        description (str): Semantic description of the file. Default is "".
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation message with the uploaded file path.

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        code_interpreter.upload_file(
            path=path, content=content, description=description
        )
        return f"Uploaded file to {path}"
    except Exception as e:
        return f"Error uploading file: {e!s}"

aupload_file async

aupload_file(
    path: str,
    content: str,
    description: str = "",
    thread_id: str = "default",
) -> str

Upload a file to the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
path	str	Relative file path in the sandbox.	required
content	str	File content as a string.	required
description	str	Semantic description of the file. Default is "".	''
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message with the uploaded file path.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aupload_file(
    self,
    path: str,
    content: str,
    description: str = "",
    thread_id: str = "default",
) -> str:
    """
    Upload a file to the code interpreter sandbox (asynchronous version).

    Args:
        path (str): Relative file path in the sandbox.
        content (str): File content as a string.
        description (str): Semantic description of the file. Default is "".
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation message with the uploaded file path.

    """
    return await asyncio.to_thread(
        self.upload_file,
        path=path,
        content=content,
        description=description,
        thread_id=thread_id,
    )

upload_files

upload_files(
    files: List[Dict[str, str]], thread_id: str = "default"
) -> str

Upload multiple files to the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
files	List[Dict[str, str]]	List of file specifications, each with 'path', 'content', and optional 'description' keys.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message with the number of files uploaded.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def upload_files(
    self,
    files: List[Dict[str, str]],
    thread_id: str = "default",
) -> str:
    """
    Upload multiple files to the code interpreter sandbox (synchronous version).

    Args:
        files (List[Dict[str, str]]): List of file specifications, each with
            'path', 'content', and optional 'description' keys.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation message with the number of files uploaded.

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        code_interpreter.upload_files(files=files)
        return f"Uploaded {len(files)} file(s)"
    except Exception as e:
        return f"Error uploading files: {e!s}"

aupload_files async

aupload_files(
    files: List[Dict[str, str]], thread_id: str = "default"
) -> str

Upload multiple files to the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
files	List[Dict[str, str]]	List of file specifications, each with 'path', 'content', and optional 'description' keys.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation message with the number of files uploaded.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aupload_files(
    self,
    files: List[Dict[str, str]],
    thread_id: str = "default",
) -> str:
    """
    Upload multiple files to the code interpreter sandbox (asynchronous version).

    Args:
        files (List[Dict[str, str]]): List of file specifications, each with
            'path', 'content', and optional 'description' keys.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation message with the number of files uploaded.

    """
    return await asyncio.to_thread(
        self.upload_files, files=files, thread_id=thread_id
    )

install_packages

install_packages(
    packages: List[str],
    upgrade: bool = False,
    thread_id: str = "default",
) -> str

Install Python packages in the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
packages	List[str]	List of package names to install. Can include version specifiers (e.g., 'pandas>=2.0').	required
upgrade	bool	Whether to upgrade existing packages. Default is False.	False
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The pip install output (stdout/stderr).

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def install_packages(
    self,
    packages: List[str],
    upgrade: bool = False,
    thread_id: str = "default",
) -> str:
    """
    Install Python packages in the code interpreter sandbox (synchronous version).

    Args:
        packages (List[str]): List of package names to install. Can include version
            specifiers (e.g., 'pandas>=2.0').
        upgrade (bool): Whether to upgrade existing packages. Default is False.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The pip install output (stdout/stderr).

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        result = code_interpreter.install_packages(
            packages=packages, upgrade=upgrade
        )
        return str(result)
    except Exception as e:
        return f"Error installing packages: {e!s}"

ainstall_packages async

ainstall_packages(
    packages: List[str],
    upgrade: bool = False,
    thread_id: str = "default",
) -> str

Install Python packages in the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
packages	List[str]	List of package names to install. Can include version specifiers (e.g., 'pandas>=2.0').	required
upgrade	bool	Whether to upgrade existing packages. Default is False.	False
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The pip install output (stdout/stderr).

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def ainstall_packages(
    self,
    packages: List[str],
    upgrade: bool = False,
    thread_id: str = "default",
) -> str:
    """
    Install Python packages in the code interpreter sandbox (asynchronous version).

    Args:
        packages (List[str]): List of package names to install. Can include version
            specifiers (e.g., 'pandas>=2.0').
        upgrade (bool): Whether to upgrade existing packages. Default is False.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The pip install output (stdout/stderr).

    """
    return await asyncio.to_thread(
        self.install_packages,
        packages=packages,
        upgrade=upgrade,
        thread_id=thread_id,
    )

download_file

download_file(path: str, thread_id: str = 'default') -> str

Download a file from the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
path	str	Path to the file in the sandbox.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The file content as text, or base64-encoded string for binary files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def download_file(
    self,
    path: str,
    thread_id: str = "default",
) -> str:
    """
    Download a file from the code interpreter sandbox (synchronous version).

    Args:
        path (str): Path to the file in the sandbox.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The file content as text, or base64-encoded string for binary files.

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        content = code_interpreter.download_file(path=path)
        if isinstance(content, bytes):
            encoded = base64.b64encode(content).decode("utf-8")
            return f"[base64 encoded binary file: {path}]\n{encoded}"
        return content
    except Exception as e:
        return f"Error downloading file: {e!s}"

adownload_file async

adownload_file(
    path: str, thread_id: str = "default"
) -> str

Download a file from the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
path	str	Path to the file in the sandbox.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	The file content as text, or base64-encoded string for binary files.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def adownload_file(
    self,
    path: str,
    thread_id: str = "default",
) -> str:
    """
    Download a file from the code interpreter sandbox (asynchronous version).

    Args:
        path (str): Path to the file in the sandbox.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: The file content as text, or base64-encoded string for binary files.

    """
    return await asyncio.to_thread(
        self.download_file, path=path, thread_id=thread_id
    )

download_files

download_files(
    paths: List[str], thread_id: str = "default"
) -> str

Download multiple files from the code interpreter sandbox (synchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths in the sandbox.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Formatted output with each file's content.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def download_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Download multiple files from the code interpreter sandbox (synchronous version).

    Args:
        paths (List[str]): List of file paths in the sandbox.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Formatted output with each file's content.

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        results = code_interpreter.download_files(paths=paths)
        output = []
        for file_path, content in results.items():
            if isinstance(content, bytes):
                encoded = base64.b64encode(content).decode("utf-8")
                output.append(
                    f"==== File: {file_path} (binary, base64) ====\n{encoded}"
                )
            else:
                output.append(f"==== File: {file_path} ====\n{content}")
        return "\n\n".join(output)
    except Exception as e:
        return f"Error downloading files: {e!s}"

adownload_files async

adownload_files(
    paths: List[str], thread_id: str = "default"
) -> str

Download multiple files from the code interpreter sandbox (asynchronous version).

Parameters:

Name	Type	Description	Default
paths	List[str]	List of file paths in the sandbox.	required
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Formatted output with each file's content.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def adownload_files(
    self,
    paths: List[str],
    thread_id: str = "default",
) -> str:
    """
    Download multiple files from the code interpreter sandbox (asynchronous version).

    Args:
        paths (List[str]): List of file paths in the sandbox.
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Formatted output with each file's content.

    """
    return await asyncio.to_thread(
        self.download_files, paths=paths, thread_id=thread_id
    )

list_code_interpreters

list_code_interpreters(
    interpreter_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str

List all code interpreters in the account (synchronous version).

Parameters:

Name	Type	Description	Default
interpreter_type	Optional[str]	Filter by type: "SYSTEM" or "CUSTOM".	None
max_results	int	Maximum results to return (1-100). Default is 10.	10
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Formatted list of code interpreter summaries.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def list_code_interpreters(
    self,
    interpreter_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str:
    """
    List all code interpreters in the account (synchronous version).

    Args:
        interpreter_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
        max_results (int): Maximum results to return (1-100). Default is 10.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Formatted list of code interpreter summaries.

    """
    try:
        code_interpreter = self._get_control_plane_client()
        response = code_interpreter.list_code_interpreters(
            interpreter_type=interpreter_type, max_results=max_results
        )
        summaries = response.get("codeInterpreterSummaries", [])
        if not summaries:
            return "No code interpreters found."
        lines = []
        for ci in summaries:
            lines.append(
                f"- {ci.get('name', 'N/A')} (ID: {ci.get('codeInterpreterId', 'N/A')}, "
                f"Status: {ci.get('status', 'N/A')}, Type: {ci.get('type', 'N/A')})"
            )
        return f"Found {len(summaries)} code interpreter(s):\n" + "\n".join(lines)
    except Exception as e:
        return f"Error listing code interpreters: {e!s}"

alist_code_interpreters async

alist_code_interpreters(
    interpreter_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str

List all code interpreters in the account (asynchronous version).

Parameters:

Name	Type	Description	Default
interpreter_type	Optional[str]	Filter by type: "SYSTEM" or "CUSTOM".	None
max_results	int	Maximum results to return (1-100). Default is 10.	10
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Formatted list of code interpreter summaries.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def alist_code_interpreters(
    self,
    interpreter_type: Optional[str] = None,
    max_results: int = 10,
    thread_id: str = "default",
) -> str:
    """
    List all code interpreters in the account (asynchronous version).

    Args:
        interpreter_type (Optional[str]): Filter by type: "SYSTEM" or "CUSTOM".
        max_results (int): Maximum results to return (1-100). Default is 10.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Formatted list of code interpreter summaries.

    """
    return await asyncio.to_thread(
        self.list_code_interpreters,
        interpreter_type=interpreter_type,
        max_results=max_results,
        thread_id=thread_id,
    )

create_code_interpreter

create_code_interpreter(
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str

Create a custom code interpreter with specific configuration (synchronous version).

Parameters:

Name	Type	Description	Default
name	str	Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_].	required
execution_role_arn	str	IAM role ARN with permissions for interpreter operations.	required
network_mode	str	Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".	'PUBLIC'
description	str	Description of the interpreter. Default is "".	''
subnet_ids	Optional[List[str]]	Subnet IDs for VPC mode.	None
security_group_ids	Optional[List[str]]	Security group IDs for VPC mode.	None
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation with interpreter ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def create_code_interpreter(
    self,
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str:
    """
    Create a custom code interpreter with specific configuration (synchronous version).

    Args:
        name (str): Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
        execution_role_arn (str): IAM role ARN with permissions for interpreter operations.
        network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
        description (str): Description of the interpreter. Default is "".
        subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
        security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation with interpreter ID and status.

    """
    try:
        code_interpreter = self._get_control_plane_client()
        network_config: Dict[str, Any] = {"networkMode": network_mode}
        if subnet_ids or security_group_ids:
            vpc_config: Dict[str, Any] = {}
            if subnet_ids:
                vpc_config["subnets"] = subnet_ids
            if security_group_ids:
                vpc_config["securityGroups"] = security_group_ids
            network_config["vpcConfig"] = vpc_config
        kwargs: Dict[str, Any] = {
            "name": name,
            "execution_role_arn": execution_role_arn,
            "network_configuration": network_config,
        }
        if description:
            kwargs["description"] = description
        response = code_interpreter.create_code_interpreter(**kwargs)
        interpreter_id = response.get("codeInterpreterId", "unknown")
        status = response.get("status", "unknown")
        return f"Code interpreter created (ID: {interpreter_id}, Status: {status})"
    except Exception as e:
        return f"Error creating code interpreter: {e!s}"

acreate_code_interpreter async

acreate_code_interpreter(
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str

Create a custom code interpreter with specific configuration (asynchronous version).

Parameters:

Name	Type	Description	Default
name	str	Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_].	required
execution_role_arn	str	IAM role ARN with permissions for interpreter operations.	required
network_mode	str	Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".	'PUBLIC'
description	str	Description of the interpreter. Default is "".	''
subnet_ids	Optional[List[str]]	Subnet IDs for VPC mode.	None
security_group_ids	Optional[List[str]]	Security group IDs for VPC mode.	None
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation with interpreter ID and status.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def acreate_code_interpreter(
    self,
    name: str,
    execution_role_arn: str,
    network_mode: str = "PUBLIC",
    description: str = "",
    subnet_ids: Optional[List[str]] = None,
    security_group_ids: Optional[List[str]] = None,
    thread_id: str = "default",
) -> str:
    """
    Create a custom code interpreter with specific configuration (asynchronous version).

    Args:
        name (str): Name for the interpreter. Must match pattern [a-zA-Z][a-zA-Z0-9_]{0,47}.
        execution_role_arn (str): IAM role ARN with permissions for interpreter operations.
        network_mode (str): Network mode: "PUBLIC" or "VPC". Default is "PUBLIC".
        description (str): Description of the interpreter. Default is "".
        subnet_ids (Optional[List[str]]): Subnet IDs for VPC mode.
        security_group_ids (Optional[List[str]]): Security group IDs for VPC mode.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation with interpreter ID and status.

    """
    return await asyncio.to_thread(
        self.create_code_interpreter,
        name=name,
        execution_role_arn=execution_role_arn,
        network_mode=network_mode,
        description=description,
        subnet_ids=subnet_ids,
        security_group_ids=security_group_ids,
        thread_id=thread_id,
    )

delete_code_interpreter

delete_code_interpreter(
    interpreter_id: str, thread_id: str = "default"
) -> str

Delete a custom code interpreter (synchronous version).

Parameters:

Name	Type	Description	Default
interpreter_id	str	The code interpreter identifier to delete.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation of deletion.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def delete_code_interpreter(
    self,
    interpreter_id: str,
    thread_id: str = "default",
) -> str:
    """
    Delete a custom code interpreter (synchronous version).

    Args:
        interpreter_id (str): The code interpreter identifier to delete.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation of deletion.

    """
    try:
        code_interpreter = self._get_control_plane_client()
        response = code_interpreter.delete_code_interpreter(
            interpreter_id=interpreter_id
        )
        status = response.get("status", "unknown")
        return f"Code interpreter '{interpreter_id}' deleted (Status: {status})"
    except Exception as e:
        return f"Error deleting code interpreter: {e!s}"

adelete_code_interpreter async

adelete_code_interpreter(
    interpreter_id: str, thread_id: str = "default"
) -> str

Delete a custom code interpreter (asynchronous version).

Parameters:

Name	Type	Description	Default
interpreter_id	str	The code interpreter identifier to delete.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Confirmation of deletion.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def adelete_code_interpreter(
    self,
    interpreter_id: str,
    thread_id: str = "default",
) -> str:
    """
    Delete a custom code interpreter (asynchronous version).

    Args:
        interpreter_id (str): The code interpreter identifier to delete.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Confirmation of deletion.

    """
    return await asyncio.to_thread(
        self.delete_code_interpreter,
        interpreter_id=interpreter_id,
        thread_id=thread_id,
    )

get_code_interpreter

get_code_interpreter(
    interpreter_id: str, thread_id: str = "default"
) -> str

Get detailed information about a code interpreter (synchronous version).

Parameters:

Name	Type	Description	Default
interpreter_id	str	The code interpreter identifier.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Interpreter details including name, status, and configuration.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def get_code_interpreter(
    self,
    interpreter_id: str,
    thread_id: str = "default",
) -> str:
    """
    Get detailed information about a code interpreter (synchronous version).

    Args:
        interpreter_id (str): The code interpreter identifier.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Interpreter details including name, status, and configuration.

    """
    try:
        code_interpreter = self._get_control_plane_client()
        response = code_interpreter.get_code_interpreter(
            interpreter_id=interpreter_id
        )
        name = response.get("name", "N/A")
        status = response.get("status", "N/A")
        desc = response.get("description", "")
        result = f"Code interpreter '{interpreter_id}':\n"
        result += f"  Name: {name}\n"
        result += f"  Status: {status}\n"
        if desc:
            result += f"  Description: {desc}\n"
        network = response.get("networkConfiguration", {})
        if network:
            result += f"  Network mode: {network.get('networkMode', 'N/A')}"
        return result
    except Exception as e:
        return f"Error getting code interpreter: {e!s}"

aget_code_interpreter async

aget_code_interpreter(
    interpreter_id: str, thread_id: str = "default"
) -> str

Get detailed information about a code interpreter (asynchronous version).

Parameters:

Name	Type	Description	Default
interpreter_id	str	The code interpreter identifier.	required
thread_id	str	Deprecated. Ignored. Kept for backward compatibility.	'default'

Returns:

Name	Type	Description
str	str	Interpreter details including name, status, and configuration.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aget_code_interpreter(
    self,
    interpreter_id: str,
    thread_id: str = "default",
) -> str:
    """
    Get detailed information about a code interpreter (asynchronous version).

    Args:
        interpreter_id (str): The code interpreter identifier.
        thread_id (str): Deprecated. Ignored. Kept for backward compatibility.

    Returns:
        str: Interpreter details including name, status, and configuration.

    """
    return await asyncio.to_thread(
        self.get_code_interpreter,
        interpreter_id=interpreter_id,
        thread_id=thread_id,
    )

clear_context

clear_context(thread_id: str = 'default') -> str

Clear all variable state in the Python execution context (synchronous version).

This resets the interpreter to a fresh state, removing all previously defined variables, imports, and function definitions.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation that the context was cleared.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

def clear_context(
    self,
    thread_id: str = "default",
) -> str:
    """
    Clear all variable state in the Python execution context (synchronous version).

    This resets the interpreter to a fresh state, removing all previously defined
    variables, imports, and function definitions.

    Args:
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation that the context was cleared.

    """
    try:
        code_interpreter = self._get_or_create_interpreter(thread_id=thread_id)
        code_interpreter.clear_context()
        return "Python execution context cleared successfully."
    except Exception as e:
        return f"Error clearing context: {e!s}"

aclear_context async

aclear_context(thread_id: str = 'default') -> str

Clear all variable state in the Python execution context (asynchronous version).

Parameters:

Name	Type	Description	Default
thread_id	str	Thread ID for the code interpreter session. Default is "default".	'default'

Returns:

Name	Type	Description
str	str	Confirmation that the context was cleared.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def aclear_context(
    self,
    thread_id: str = "default",
) -> str:
    """
    Clear all variable state in the Python execution context (asynchronous version).

    Args:
        thread_id (str): Thread ID for the code interpreter session. Default is "default".

    Returns:
        str: Confirmation that the context was cleared.

    """
    return await asyncio.to_thread(self.clear_context, thread_id=thread_id)

cleanup async

cleanup(thread_id: Optional[str] = None) -> None

Clean up resources

Parameters:

Name	Type	Description	Default
thread_id	Optional[str]	Optional thread ID to clean up. If None, cleans up all sessions.	None

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/code_interpreter/base.py

async def cleanup(self, thread_id: Optional[str] = None) -> None:
    """
    Clean up resources

    Args:
        thread_id: Optional thread ID to clean up. If None, cleans up all sessions.

    """
    if thread_id:
        # Clean up a specific thread's session
        if thread_id in self._code_interpreters:
            try:
                self._code_interpreters[thread_id].stop()
                del self._code_interpreters[thread_id]
                logger.info(
                    f"Code interpreter session for thread {thread_id} cleaned up"
                )
            except Exception as e:
                logger.warning(
                    f"Error stopping code interpreter for thread {thread_id}: {e}"
                )
    else:
        # Clean up all sessions
        thread_ids = list(self._code_interpreters.keys())
        for tid in thread_ids:
            try:
                self._code_interpreters[tid].stop()
            except Exception as e:
                logger.warning(
                    f"Error stopping code interpreter for thread {tid}: {e}"
                )

        self._code_interpreters = {}
        logger.info("All code interpreter sessions cleaned up")

AgentCoreRuntime

Serves a LlamaIndex agent via BedrockAgentCoreApp (POST /invocations, GET /ping).

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/runtime/base.py

class AgentCoreRuntime:
    """Serves a LlamaIndex agent via BedrockAgentCoreApp (POST /invocations, GET /ping)."""

    def __init__(
        self,
        agent: Any,
        stream: bool = True,
        port: int = 8080,
        host: Optional[str] = None,
        debug: bool = False,
        memory: Optional[Any] = None,
        lifespan: Optional[Lifespan] = None,
        middleware: Optional[Sequence[Middleware]] = None,
    ):
        self._agent = agent
        self._stream = stream
        self._port = port
        self._host = host
        self._memory = memory
        self._app = BedrockAgentCoreApp(
            debug=debug, lifespan=lifespan, middleware=middleware
        )

        # Register entrypoint using closure wrappers (not bound methods).
        # entrypoint() sets func.run attr which fails on bound methods.
        # Closures also preserve isasyncgenfunction() detection for streaming.
        runtime = self
        if stream:

            async def streaming_entrypoint(
                payload: dict, context: RequestContext
            ) -> AsyncGenerator[dict, None]:
                # Validate eagerly (before the first yield) so that
                # HTTPException propagates before streaming begins.
                prompt = runtime._extract_prompt(payload)
                memory = runtime._get_memory(context)
                async for chunk in runtime._stream_events(prompt, memory):
                    yield chunk

            self._app.entrypoint(streaming_entrypoint)
        else:

            async def non_streaming_entrypoint(
                payload: dict, context: RequestContext
            ) -> dict:
                return await runtime._non_streaming_handler(payload, context)

            self._app.entrypoint(non_streaming_entrypoint)

    @classmethod
    def serve(cls, agent: Any, **kwargs: Any) -> None:
        """Create runtime and start server in one call."""
        runtime = cls(agent=agent, **kwargs)
        runtime.run()

    def run(self, **kwargs: Any) -> None:
        """Start uvicorn server."""
        self._app.run(port=self._port, host=self._host, **kwargs)

    @property
    def app(self) -> BedrockAgentCoreApp:
        """Expose for ASGI mounting or testing."""
        return self._app

    @staticmethod
    def _extract_prompt(payload: dict) -> str:
        """Normalize payload to user message string."""
        prompt = payload.get("prompt") or payload.get("message") or payload.get("input")
        if isinstance(prompt, dict):
            prompt = prompt.get("prompt")
        if isinstance(prompt, str):
            prompt = prompt.strip()
        if not prompt or not isinstance(prompt, (str, int, float)):
            raise HTTPException(
                status_code=400,
                detail="Request must include 'prompt', 'message', or 'input' field"
                " with a string value",
            )
        return str(prompt)

    def _get_memory(self, context: Optional[RequestContext] = None) -> Optional[Any]:
        """Return per-request memory copy with session_id from AgentCore context."""
        if self._memory is None:
            return None
        if context and context.session_id and hasattr(self._memory, "_context"):
            if hasattr(self._memory._context, "session_id"):
                memory = copy.copy(self._memory)
                memory._context = copy.copy(self._memory._context)
                memory._context.session_id = context.session_id
                return memory
        return self._memory

    async def _non_streaming_handler(
        self, payload: dict, context: RequestContext
    ) -> dict:
        """Handle non-streaming invocation. Returns JSON response."""
        prompt = self._extract_prompt(payload)
        memory = self._get_memory(context)
        handler = self._agent.run(user_msg=prompt, memory=memory)
        result = await handler
        return {"response": str(result)}

    async def _stream_events(
        self, prompt: str, memory: Optional[Any]
    ) -> AsyncGenerator[dict, None]:
        """
        Yield SSE event dicts for a validated prompt.

        Callers should validate the prompt (via _extract_prompt) before
        calling this method so that validation errors are raised eagerly.
        """
        handler = self._agent.run(user_msg=prompt, memory=memory)

        try:
            async for event in handler.stream_events():
                if isinstance(event, AgentStream):
                    ev: Dict[str, Any] = {
                        "event": "agent_stream",
                        "delta": event.delta,
                        "response": event.response,
                    }
                    if event.thinking_delta:
                        ev["thinking_delta"] = event.thinking_delta
                    yield ev
                elif isinstance(event, ToolCall):
                    yield {
                        "event": "tool_call",
                        "tool_name": event.tool_name,
                        "tool_kwargs": event.tool_kwargs,
                    }
                elif isinstance(event, ToolCallResult):
                    yield {
                        "event": "tool_result",
                        "tool_name": event.tool_name,
                        "tool_output": str(event.tool_output),
                    }
                elif isinstance(event, AgentOutput):
                    yield {"event": "done", "response": str(event.response)}
                else:
                    logger.debug(
                        "Ignoring unknown event type: %s", type(event).__name__
                    )
        except Exception as e:
            logger.exception("Error during streaming")
            yield {"event": "error", "message": str(e)}
            return

        # Await handler to ensure background tasks complete (memory flush, etc.)
        try:
            await handler
        except Exception:
            logger.exception("Error awaiting handler completion")

app property

app: BedrockAgentCoreApp

Expose for ASGI mounting or testing.

serve classmethod

serve(agent: Any, **kwargs: Any) -> None

Create runtime and start server in one call.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/runtime/base.py

@classmethod
def serve(cls, agent: Any, **kwargs: Any) -> None:
    """Create runtime and start server in one call."""
    runtime = cls(agent=agent, **kwargs)
    runtime.run()

run

run(**kwargs: Any) -> None

Start uvicorn server.

Source code in llama-index-integrations/tools/llama-index-tools-aws-bedrock-agentcore/llama_index/tools/aws_bedrock_agentcore/runtime/base.py

def run(self, **kwargs: Any) -> None:
    """Start uvicorn server."""
    self._app.run(port=self._port, host=self._host, **kwargs)

options: members: - AgentCoreBrowserToolSpec - AgentCoreCodeInterpreterToolSpec - AgentCoreRuntime