autogen_ext.code_executors.jupyter#

class JupyterCodeExecutor(kernel_name: str = 'python3', timeout: int = 60, output_dir: Path = Path('.'))[source]#

Bases: CodeExecutor, Component[JupyterCodeExecutorConfig]

A code executor class that executes code statefully using [nbclient](jupyter/nbclient).

Danger

This will execute code on the local machine. If being used with LLM generated code, caution should be used.

Example of using it directly:

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.jupyter import JupyterCodeExecutor


async def main() -> None:
    async with JupyterCodeExecutor() as executor:
        cancel_token = CancellationToken()
        code_blocks = [CodeBlock(code="print('hello world!')", language="python")]
        code_result = await executor.execute_code_blocks(code_blocks, cancel_token)
        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.jupyter import JupyterCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.code_execution import PythonCodeExecutionTool


async def main() -> None:
    async with JupyterCodeExecutor() 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.jupyter import JupyterCodeExecutor
from autogen_core import CancellationToken


async def main() -> None:
    async with JupyterCodeExecutor() 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())

Parameters:

kernel_name (str) – The kernel name to use. By default, “python3”.
timeout (int) – The timeout for code execution, by default 60.
output_dir (Path) – The directory to save output files, by default “.”.

component_config_schema#: alias of JupyterCodeExecutorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.code_executors.jupyter.JupyterCodeExecutor'#: 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) → JupyterCodeResult[source]#

Execute code blocks and return the result.

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

async restart() → None[source]#: Restart the code executor.

async start() → None[source]#

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

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

Bases: CodeResult

A code result class for Jupyter code executor.

output_files: list[Path]#