YepCodeCodeExecutor(api_token=None, timeout=60, remove_on_done=False, sync_execution=True)
Bases: CodeExecutor
A code executor class that executes code using YepCode's serverless runtime.
This executor runs code in YepCode's secure, production-grade sandboxes. It supports Python and JavaScript execution with access to any external library with automatic discovery and installation.
The executor executes code blocks serially in the order they are received. Each code block is executed in a separate YepCode execution environment. Currently supports Python and JavaScript languages.
| PARAMETER | DESCRIPTION |
api_token | YepCode API token. If None, will try to get from YEPCODE_API_TOKEN environment variable. TYPE: Optional[str] DEFAULT: None |
timeout | The timeout for code execution in seconds. Default is 60. TYPE: int DEFAULT: 60 |
remove_on_done | Whether to remove the execution after completion. Default is False. TYPE: bool DEFAULT: False |
sync_execution | Whether to wait for execution to complete. Default is True. TYPE: bool DEFAULT: True |
| RAISES | DESCRIPTION |
ImportError | If yepcode-run package is not installed. |
ValueError | If YepCode API token is not provided or timeout is invalid. |
RuntimeError | If YepCode runner initialization fails. |
Source code in autogen/coding/yepcode_code_executor.py
| def __init__(
self,
api_token: str | None = None,
timeout: int = 60,
remove_on_done: bool = False,
sync_execution: bool = True,
):
if YepCodeRun is None or YepCodeApiConfig is None:
raise ImportError(
"Missing dependencies for YepCodeCodeExecutor. Please install with: pip install ag2[yepcode]"
)
if timeout < 1:
raise ValueError("Timeout must be greater than or equal to 1.")
# Load environment variables from .env file if dotenv is available
if _load_dotenv is not None:
_load_dotenv()
# Get API token from parameter or environment
self._api_token = api_token or os.getenv("YEPCODE_API_TOKEN")
if not self._api_token:
raise ValueError(
"YepCode API token is required. Provide it via api_token parameter or YEPCODE_API_TOKEN environment variable."
)
self._timeout = timeout
self._remove_on_done = remove_on_done
self._sync_execution = sync_execution
try:
config = YepCodeApiConfig(api_token=self._api_token)
self._runner = YepCodeRun(config)
except Exception as e:
raise RuntimeError(f"Failed to initialize YepCode runner: {str(e)}") from e
|
class-attribute
SUPPORTED_LANGUAGES = ['python', 'javascript']
(Experimental) Export a code extractor that can be used by an agent.
property
The timeout for code execution.
execute_code_blocks(code_blocks)
Execute the code blocks and return the result.
| PARAMETER | DESCRIPTION |
code_blocks | The code blocks to execute. TYPE: List[CodeBlock] |
| RETURNS | DESCRIPTION |
YepCodeCodeResult | The result of the code execution. TYPE: YepCodeCodeResult |
Source code in autogen/coding/yepcode_code_executor.py
| def execute_code_blocks(self, code_blocks: list[CodeBlock]) -> YepCodeCodeResult:
"""Execute the code blocks and return the result.
Args:
code_blocks (List[CodeBlock]): The code blocks to execute.
Returns:
YepCodeCodeResult: The result of the code execution.
"""
if not code_blocks:
return YepCodeCodeResult(exit_code=0, output="")
outputs: list[str] = []
last_execution_id: str | None = None
for code_block in code_blocks:
lang = self._normalize_language(code_block.language)
if lang not in ["python", "javascript"]:
return YepCodeCodeResult(
exit_code=1,
output=f"Unsupported language: {code_block.language}. Supported languages: {', '.join(self.SUPPORTED_LANGUAGES)}",
)
try:
# Execute code using YepCode
execution = self._runner.run(
code_block.code,
{
"language": lang,
"removeOnDone": self._remove_on_done,
"timeout": self._timeout * 1000, # Convert to milliseconds
},
)
last_execution_id = execution.id
if self._sync_execution:
# Wait for execution to complete
execution.wait_for_done()
logs_output = ""
# Get logs
if execution.logs:
logs_output = "\n\nExecution logs:\n" + "\n".join([
f"{log.timestamp} - {log.level}: {log.message}" for log in execution.logs
])
# Check if execution was successful
if execution.error:
output = f"Execution failed with error:\n{execution.error}{logs_output}"
return YepCodeCodeResult(exit_code=1, output=output, execution_id=execution.id)
# Get output
output = ""
if execution.return_value:
output = f"Execution result:\n{execution.return_value}"
output += logs_output
outputs.append(output)
else:
outputs.append(f"Execution started with ID: {execution.id}")
except Exception as e:
return YepCodeCodeResult(
exit_code=1,
output=f"Error executing code: {str(e)}",
execution_id=last_execution_id,
)
return YepCodeCodeResult(exit_code=0, output="\n===\n".join(outputs), execution_id=last_execution_id)
|
Restart the code executor.
Source code in autogen/coding/yepcode_code_executor.py
| def restart(self) -> None:
"""Restart the code executor."""
pass
|