Skip to main content

coding.kubernetes.pod_commandline_code_executor

PodCommandLineCodeExecutor​

class PodCommandLineCodeExecutor(CodeExecutor)

__init__​

def __init__(image: str = "python:3-slim",
             pod_name: Optional[str] = None,
             namespace: Optional[str] = None,
             pod_spec: Optional[client.V1Pod] = None,
             container_name: Optional[str] = "autogen-code-exec",
             timeout: int = 60,
             work_dir: Union[Path, str] = Path("/workspace"),
             kube_config_file: Optional[str] = None,
             stop_container: bool = True,
             execution_policies: Optional[Dict[str, bool]] = None)

(Experimental) A code executor class that executes code through a command line environment in a kubernetes pod.

The executor first saves each code block in a file in the working directory, and then executes the code file in the container. The executor executes the code blocks in the order they are received. Currently, the executor only supports Python and shell scripts. For Python code, use the language "python" for the code block. For shell scripts, use the language "bash", "shell", or "sh" for the code block.

Arguments:

  • image type, optional - Docker image to use for code execution. Defaults to "python:3-slim".
  • pod_name Optional[str], optional - Name of the kubernetes pod which is created. If None, will autogenerate a name. Defaults to None.
  • namespace Optional[str], optional - Namespace of kubernetes pod which is created. If None, will use current namespace of this instance
  • pod_spec Optional[client.V1Pod], optional - Specification of kubernetes pod. custom pod spec can be provided with this param. if pod_spec is provided, params above(image, pod_name, namespace) are neglected.
  • container_name Optional[str], optional - Name of the container where code block will be executed. if pod_spec param is provided, container_name must be provided also.
  • timeout int, optional - The timeout for code execution. Defaults to 60.
  • work_dir Union[Path, str], optional - The working directory for the code execution. Defaults to Path("/workspace").
  • kube_config_file Optional[str], optional - kubernetes configuration file path. If None, will use KUBECONFIG environment variables or service account token(incluster config)
  • stop_container bool, optional - If true, will automatically stop the container when stop is called, when the context manager exits or when the Python process exits with atext. Defaults to True.
  • execution_policies dict[str, bool], optional - defines supported execution language

Raises:

  • ValueError - On argument error, or if the container fails to start.

timeout​

@property
def timeout() -> int

(Experimental) The timeout for code execution.

work_dir​

@property
def work_dir() -> Path

(Experimental) The working directory for the code execution.

code_extractor​

@property
def code_extractor() -> CodeExtractor

(Experimental) Export a code extractor that can be used by an agent.

execute_code_blocks​

def execute_code_blocks(code_blocks: List[CodeBlock]) -> CommandLineCodeResult

(Experimental) Execute the code blocks and return the result.

Arguments:

  • code_blocks List[CodeBlock] - The code blocks to execute.

Returns:

  • CommandlineCodeResult - The result of the code execution.

stop​

def stop() -> None

(Experimental) Stop the code executor.