codemode

Class info

Classes

Name	Children	Inherits
CodeModeResourceProvider llmling_agent.resource_providers.codemode.provider Provider that wraps tools into a single Python execution environment.	RemoteCodeModeResourceProvider	AggregatingResourceProvider
RemoteCodeModeResourceProvider llmling_agent.resource_providers.codemode.remote_provider Provider that executes code in secure isolation with tool access via server.		CodeModeResourceProvider

🛈 DocStrings

Meta-resource provider that exposes tools through Python execution.

CodeModeResourceProvider

Bases: AggregatingResourceProvider

Provider that wraps tools into a single Python execution environment.

Source code in src/llmling_agent/resource_providers/codemode/provider.py

class CodeModeResourceProvider(AggregatingResourceProvider):
    """Provider that wraps tools into a single Python execution environment."""

    def __init__(
        self,
        providers: list[ResourceProvider],
        name: str = "meta_tools",
        include_docstrings: bool = True,
        usage_notes: str = USAGE,
    ) -> None:
        """Initialize meta provider.

        Args:
            providers: Providers whose tools to wrap
            name: Provider name
            include_docstrings: Include function docstrings in documentation
            usage_notes: Usage notes for the codemode tool
        """
        super().__init__(providers=providers, name=name)
        self.include_docstrings = include_docstrings
        self._cached_tool: Tool | None = None
        self.usage_notes = usage_notes

    async def get_tools(self) -> list[Tool]:
        """Return single meta-tool for Python execution with available tools."""
        # Always generate fresh toolset to reflect current tools
        toolset_generator = await self._get_code_generator()
        desc = toolset_generator.generate_tool_description()
        desc += self.usage_notes

        if self._cached_tool is None:
            # Create a closure that captures self but isn't a bound method
            async def execute_tool(
                ctx: AgentContext,
                python_code: str,
            ) -> Any:
                """These docstings are overriden by description_override."""
                return await self.execute(ctx, python_code)

            self._cached_tool = Tool.from_callable(execute_tool, description_override=desc)
        else:
            # Update the description on existing cached tool
            self._cached_tool.description = desc

        return [self._cached_tool]

    async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
        """Execute Python code with all wrapped tools available as functions.

        Args:
            python_code: Python code to execute

        Returns:
            Result of the last expression or explicit return value
        """
        toolset_generator = await self._get_code_generator()
        namespace = toolset_generator.generate_execution_namespace()

        # async def report_progress(current: int, total: int, message: str = ""):
        #     """Report progress during code execution."""
        #     await ctx.report_progress(current, total, message)

        # namespace["report_progress"] = NamespaceCallable(report_progress)

        validate_code(python_code)
        try:
            exec(python_code, namespace)
            result = await namespace["main"]()
            # Handle edge cases with coroutines and return values
            if inspect.iscoroutine(result):
                result = await result
            if not result:  # in order to not confuse the model, return a success message.
                return "Code executed successfully"
        except Exception as e:  # noqa: BLE001
            return f"Error executing code: {e!s}"
        else:
            return result

    def invalidate_cache(self) -> None:
        """Invalidate cached tool when providers change."""
        self._cached_tool = None
        # Note: We no longer cache the toolset generator, so no need to clear it

    async def _get_code_generator(self) -> ToolsetCodeGenerator:
        """Get fresh toolset generator with current tools."""
        return tools_to_codegen(
            tools=await super().get_tools(),
            include_docstrings=self.include_docstrings,
        )

__init__

__init__(
    providers: list[ResourceProvider],
    name: str = "meta_tools",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
) -> None

Initialize meta provider.

Parameters:

Name	Type	Description	Default
providers	list[ResourceProvider]	Providers whose tools to wrap	required
name	str	Provider name	'meta_tools'
include_docstrings	bool	Include function docstrings in documentation	True
usage_notes	str	Usage notes for the codemode tool	USAGE

Source code in src/llmling_agent/resource_providers/codemode/provider.py

def __init__(
    self,
    providers: list[ResourceProvider],
    name: str = "meta_tools",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
) -> None:
    """Initialize meta provider.

    Args:
        providers: Providers whose tools to wrap
        name: Provider name
        include_docstrings: Include function docstrings in documentation
        usage_notes: Usage notes for the codemode tool
    """
    super().__init__(providers=providers, name=name)
    self.include_docstrings = include_docstrings
    self._cached_tool: Tool | None = None
    self.usage_notes = usage_notes

execute async

execute(ctx: AgentContext, python_code: str) -> Any

Execute Python code with all wrapped tools available as functions.

Parameters:

Name	Type	Description	Default
python_code	str	Python code to execute	required

Returns:

Type	Description
Any	Result of the last expression or explicit return value

Source code in src/llmling_agent/resource_providers/codemode/provider.py

async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
    """Execute Python code with all wrapped tools available as functions.

    Args:
        python_code: Python code to execute

    Returns:
        Result of the last expression or explicit return value
    """
    toolset_generator = await self._get_code_generator()
    namespace = toolset_generator.generate_execution_namespace()

    # async def report_progress(current: int, total: int, message: str = ""):
    #     """Report progress during code execution."""
    #     await ctx.report_progress(current, total, message)

    # namespace["report_progress"] = NamespaceCallable(report_progress)

    validate_code(python_code)
    try:
        exec(python_code, namespace)
        result = await namespace["main"]()
        # Handle edge cases with coroutines and return values
        if inspect.iscoroutine(result):
            result = await result
        if not result:  # in order to not confuse the model, return a success message.
            return "Code executed successfully"
    except Exception as e:  # noqa: BLE001
        return f"Error executing code: {e!s}"
    else:
        return result

get_tools async

get_tools() -> list[Tool]

Return single meta-tool for Python execution with available tools.

Source code in src/llmling_agent/resource_providers/codemode/provider.py

async def get_tools(self) -> list[Tool]:
    """Return single meta-tool for Python execution with available tools."""
    # Always generate fresh toolset to reflect current tools
    toolset_generator = await self._get_code_generator()
    desc = toolset_generator.generate_tool_description()
    desc += self.usage_notes

    if self._cached_tool is None:
        # Create a closure that captures self but isn't a bound method
        async def execute_tool(
            ctx: AgentContext,
            python_code: str,
        ) -> Any:
            """These docstings are overriden by description_override."""
            return await self.execute(ctx, python_code)

        self._cached_tool = Tool.from_callable(execute_tool, description_override=desc)
    else:
        # Update the description on existing cached tool
        self._cached_tool.description = desc

    return [self._cached_tool]

invalidate_cache

invalidate_cache() -> None

Invalidate cached tool when providers change.

Source code in src/llmling_agent/resource_providers/codemode/provider.py

def invalidate_cache(self) -> None:
    """Invalidate cached tool when providers change."""
    self._cached_tool = None

RemoteCodeModeResourceProvider

Bases: CodeModeResourceProvider

Provider that executes code in secure isolation with tool access via server.

Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py

class RemoteCodeModeResourceProvider(CodeModeResourceProvider):
    """Provider that executes code in secure isolation with tool access via server."""

    def __init__(
        self,
        providers: list[ResourceProvider],
        execution_config: ExecutionEnvironmentConfig | None = None,
        name: str = "secure_code_executor",
        include_docstrings: bool = True,
        usage_notes: str = USAGE,
        server_host: str = "localhost",
        server_port: int = 8000,
    ) -> None:
        """Initialize secure code execution provider.

        Args:
            providers: Providers whose tools to expose
            execution_config: Execution environment configuration
            name: Provider name
            include_docstrings: Include function docstrings in documentation
            usage_notes: Usage notes for the provider
            server_host: Host for tool server
            server_port: Port for tool server
        """
        super().__init__(
            providers=providers,
            name=name,
            include_docstrings=include_docstrings,
            usage_notes=usage_notes,
        )
        self.execution_config = execution_config or LocalExecutionEnvironmentConfig()
        self.server_host = server_host
        self.server_port = server_port
        self._code_executor: RemoteCodeExecutor | None = None
        self._provider_lock = asyncio.Lock()

    async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
        """Execute Python code in secure environment with tools available via HTTP.

        Args:
            python_code: Python code to execute

        Returns:
            Result of the code execution
        """
        code_provider = await self._get_code_executor()
        logger.info("Validating code", code=python_code)
        validate_code(python_code)
        full_code = f"{PROGRESS_HELPER}\n\n{python_code}"
        logger.info("Complete code", code=full_code)
        try:
            result = await code_provider.execution_env.execute(full_code)
            if result.success:
                logger.info("Code executed successfully")
                if result.result is None:
                    return "Code executed successfully"
                return result.result
        except Exception as e:  # noqa: BLE001
            return f"Error in secure execution: {e!s}"
        else:
            return f"Error executing code: {result.error}"

    async def _get_code_executor(self) -> RemoteCodeExecutor:
        """Get cached code execution provider with thread-safe initialization."""
        async with self._provider_lock:
            if self._code_executor is None:
                all_tools = await super().get_tools()
                self._code_executor = RemoteCodeExecutor.from_tools(
                    all_tools,
                    self.execution_config,
                    server_host=self.server_host,
                    server_port=self.server_port,
                    include_docstrings=self.include_docstrings,
                )
                # Initialize the provider and start server
                await self._code_executor.__aenter__()

            return self._code_executor

    async def __aenter__(self) -> Self:
        """Async context manager entry."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Async context manager exit."""
        if self._code_executor is not None:
            with contextlib.suppress(Exception):
                await self._code_executor.__aexit__(exc_type, exc_val, exc_tb)
            self._code_executor = None

__aenter__ async

__aenter__() -> Self

Async context manager entry.

Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py

async def __aenter__(self) -> Self:
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Async context manager exit.

Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context manager exit."""
    if self._code_executor is not None:
        with contextlib.suppress(Exception):
            await self._code_executor.__aexit__(exc_type, exc_val, exc_tb)
        self._code_executor = None

__init__

__init__(
    providers: list[ResourceProvider],
    execution_config: ExecutionEnvironmentConfig | None = None,
    name: str = "secure_code_executor",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
    server_host: str = "localhost",
    server_port: int = 8000,
) -> None

Initialize secure code execution provider.

Parameters:

Name	Type	Description	Default
providers	list[ResourceProvider]	Providers whose tools to expose	required
execution_config	ExecutionEnvironmentConfig | None	Execution environment configuration	None
name	str	Provider name	'secure_code_executor'
include_docstrings	bool	Include function docstrings in documentation	True
usage_notes	str	Usage notes for the provider	USAGE
server_host	str	Host for tool server	'localhost'
server_port	int	Port for tool server	8000

Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py

def __init__(
    self,
    providers: list[ResourceProvider],
    execution_config: ExecutionEnvironmentConfig | None = None,
    name: str = "secure_code_executor",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
    server_host: str = "localhost",
    server_port: int = 8000,
) -> None:
    """Initialize secure code execution provider.

    Args:
        providers: Providers whose tools to expose
        execution_config: Execution environment configuration
        name: Provider name
        include_docstrings: Include function docstrings in documentation
        usage_notes: Usage notes for the provider
        server_host: Host for tool server
        server_port: Port for tool server
    """
    super().__init__(
        providers=providers,
        name=name,
        include_docstrings=include_docstrings,
        usage_notes=usage_notes,
    )
    self.execution_config = execution_config or LocalExecutionEnvironmentConfig()
    self.server_host = server_host
    self.server_port = server_port
    self._code_executor: RemoteCodeExecutor | None = None
    self._provider_lock = asyncio.Lock()

execute async

execute(ctx: AgentContext, python_code: str) -> Any

Execute Python code in secure environment with tools available via HTTP.

Parameters:

Name	Type	Description	Default
python_code	str	Python code to execute	required

Returns:

Type	Description
Any	Result of the code execution

Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py

async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
    """Execute Python code in secure environment with tools available via HTTP.

    Args:
        python_code: Python code to execute

    Returns:
        Result of the code execution
    """
    code_provider = await self._get_code_executor()
    logger.info("Validating code", code=python_code)
    validate_code(python_code)
    full_code = f"{PROGRESS_HELPER}\n\n{python_code}"
    logger.info("Complete code", code=full_code)
    try:
        result = await code_provider.execution_env.execute(full_code)
        if result.success:
            logger.info("Code executed successfully")
            if result.result is None:
                return "Code executed successfully"
            return result.result
    except Exception as e:  # noqa: BLE001
        return f"Error in secure execution: {e!s}"
    else:
        return f"Error executing code: {result.error}"