autogen_ext.code_executors.docker_jupyter#

class DockerJupyterCodeExecutor(jupyter_server: JupyterConnectable | JupyterConnectionInfo, kernel_name: str = 'python3', timeout: int = 60, output_dir: Path | None = None)[source]#

Bases: CodeExecutor, Component[DockerJupyterCodeExecutorConfig]

(Experimental) A code executor class that executes code statefully using a Jupyter server supplied to this class.

Each execution is stateful and can access variables created from previous executions in the same session.

To use this, you need to install the following dependencies:

pip install "autogen-ext[docker-jupyter-executor]"

Parameters:

jupyter_server (Union[JupyterConnectable, JupyterConnectionInfo]) – The Jupyter server to use.
kernel_name (str) – The kernel name to use. Make sure it is installed. By default, it is “python3”.
timeout (int) – The timeout for code execution, by default 60.
output_dir (str) – The directory to save output files, by default None.

Example of using it directly:

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer


async def main() -> None:
    async with DockerJupyterServer() as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) as executor:
            code_blocks = [CodeBlock(code="print('hello world!')", language="python")]
            code_result = await executor.execute_code_blocks(code_blocks, cancellation_token=CancellationToken())
            print(code_result)


asyncio.run(main())

Example of using it with your own jupyter image:

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer


async def main() -> None:
    async with DockerJupyterServer(custom_image_name="your_custom_images_name", expose_port=8888) as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) as executor:
            code_blocks = [CodeBlock(code="print('hello world!')", language="python")]
            code_result = await executor.execute_code_blocks(code_blocks, cancellation_token=CancellationToken())
            print(code_result)


asyncio.run(main())

Example of using it with PythonCodeExecutionTool:

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.code_execution import PythonCodeExecutionTool


async def main() -> None:
    async with DockerJupyterServer() as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) as executor:
            tool = PythonCodeExecutionTool(executor)
            model_client = OpenAIChatCompletionClient(model="gpt-4o")
            agent = AssistantAgent("assistant", model_client=model_client, tools=[tool])
            result = await agent.run(task="What is the 10th Fibonacci number? Use Python to calculate it.")
            print(result)


asyncio.run(main())

Example of using it inside a CodeExecutorAgent:

import asyncio
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer
from autogen_core import CancellationToken


async def main() -> None:
    async with DockerJupyterServer() as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) as executor:
            code_executor_agent = CodeExecutorAgent("code_executor", code_executor=executor)
            task = TextMessage(
                content='''Here is some code
        ```python
        print('Hello world')
        ```
        ''',
                source="user",
            )
            response = await code_executor_agent.on_messages([task], CancellationToken())
            print(response.chat_message)


asyncio.run(main())

component_config_schema#: alias of DockerJupyterCodeExecutorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.code_executors.docker_jupyter.DockerJupyterCodeExecutor'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → DockerJupyterCodeResult[source]#

(Experimental) Execute a list of code blocks and return the result.

This method executes a list of code blocks as cells in the Jupyter kernel. See: https://jupyter-client.readthedocs.io/en/stable/messaging.html for the message protocol.

Parameters:: code_blocks (List[CodeBlock]) – A list of code blocks to execute.
Returns:: DockerJupyterCodeResult – The result of the code execution.

async restart() → None[source]#: (Experimental) Restart a new session.

async start() → None[source]#: (Experimental) Start a new session.

async stop() → None[source]#: Stop the kernel.

class DockerJupyterCodeResult(exit_code: int, output: str, output_files: list[Path])[source]#

Bases: CodeResult

(Experimental) A code result class for IPython code executor.

output_files: list[Path]#

Bases: JupyterConnectable

DEFAULT_DOCKERFILE = 'FROM quay.io/jupyter/docker-stacks-foundation\n\n SHELL ["/bin/bash", "-o", "pipefail", "-c"]\n\n USER ${NB_UID}\n RUN mamba install --yes jupyter_kernel_gateway ipykernel && mamba clean --all -f -y && fix-permissions "${CONDA_DIR}" && fix-permissions "/home/${NB_USER}"\n\n ENV TOKEN="UNSET"\n CMD python -m jupyter kernelgateway --KernelGatewayApp.ip=0.0.0.0 --KernelGatewayApp.port=8888 --KernelGatewayApp.auth_token="${TOKEN}" --JupyterApp.answer_yes=true --JupyterWebsocketPersonality.list_kernels=true\n\n EXPOSE 8888\n\n WORKDIR "${HOME}"\n '#

class GenerateToken[source]#: Bases: object

property connection_info: JupyterConnectionInfo#: Return the connection information for this connectable.

async get_client() → JupyterClient[source]#

async stop() → None[source]#

class JupyterClient(connection_info: JupyterConnectionInfo)[source]#

Bases: object

async close() → None[source]#: Close the async session

async delete_kernel(kernel_id: str) → None[source]#

async get_kernel_client(kernel_id: str) → JupyterKernelClient[source]#

async list_kernel_specs() → Dict[str, Dict[str, str]][source]#

async list_kernels() → List[Dict[str, str]][source]#

async restart_kernel(kernel_id: str) → None[source]#

async start_kernel(kernel_spec_name: str) → str[source]#

Start a new kernel asynchronously.

Parameters:: kernel_spec_name (str) – Name of the kernel spec to start
Returns:: str – ID of the started kernel

class JupyterKernelClient(websocket: ClientConnection)[source]#

Bases: object

An asynchronous client for communicating with a Jupyter kernel.

async execute(code: str, timeout_seconds: float | None = None) → ExecutionResult[source]#

async stop() → None[source]#

async wait_for_ready(timeout_seconds: float | None = None) → bool[source]#