Skip to content

remote_provider

Class info

Classes

Name Children Inherits
AgentContext
llmling_agent.agent.context
Runtime context for agent execution.
    CodeModeResourceProvider
    llmling_agent.resource_providers.codemode.provider
    Provider that wraps tools into a single Python execution environment.
    RemoteCodeExecutor
    llmling_agent.resource_providers.codemode.code_executor
    Provides secure code execution with tool access via FastAPI server.
      RemoteCodeModeResourceProvider
      llmling_agent.resource_providers.codemode.remote_provider
      Provider that executes code in secure isolation with tool access via server.
        Tool
        llmling_agent.tools.base
        Information about a registered tool.

          🛈 DocStrings

          Secure code execution provider using isolated execution environments.

          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 
           42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
          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,
    ):
        """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):
        """Async context manager entry."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | 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__()

          Async context manager entry.

          Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py 
          121
122
123
          async def __aenter__(self):
    """Async context manager entry."""
    return self

          __aexit__ async 

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

          Async context manager exit.

          Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py 
          125
126
127
128
129
130
131
132
133
134
135
          async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | 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,
)

          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 
          45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
          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,
):
    """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 
           78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
          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}"

          open_browser 

          open_browser(url: str) -> bool

          Use this to open url in the default browser.

          Source code in src/llmling_agent/resource_providers/codemode/remote_provider.py 
          149
150
151
          def open_browser(url: str) -> bool:
    """Use this to open url in the default browser."""
    return webbrowser.open(url)