Skip to content

code_executor

Class info

Classes

Name Children Inherits
RemoteCodeExecutor
llmling_agent.resource_providers.codemode.code_executor
Provides secure code execution with tool access via FastAPI server.
    ToolServerLifecycleHandler
    llmling_agent.resource_providers.codemode.code_executor
    Manages FastAPI server lifecycle for tool access.

      🛈 DocStrings

      Code execution provider with secure tool isolation via FastAPI server.

      RemoteCodeExecutor dataclass

      Provides secure code execution with tool access via FastAPI server.

      Architecture: - FastAPI server runs in HOST environment with tool routes - User code runs in SANDBOX environment (Docker, E2B, etc.) - Sandbox makes HTTP calls to server for tool execution

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
       31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
      @dataclass
class RemoteCodeExecutor:
    """Provides secure code execution with tool access via FastAPI server.

    Architecture:
    - FastAPI server runs in HOST environment with tool routes
    - User code runs in SANDBOX environment (Docker, E2B, etc.)
    - Sandbox makes HTTP calls to server for tool execution
    """

    toolset_generator: ToolsetCodeGenerator
    """Code generator for tools."""

    execution_env: ExecutionEnvironment
    """Execution environment for running code."""

    @classmethod
    def from_tools(
        cls,
        tools: Sequence[Tool],
        env_config: ExecutionEnvironmentConfig,
        server_host: str = "localhost",
        server_port: int = 8000,
        include_docstrings: bool = True,
    ) -> RemoteCodeExecutor:
        """Create provider from tools and environment configuration.

        Args:
            tools: Tools to make available for code execution
            env_config: Execution environment configuration
            server_host: Host for FastAPI server
            server_port: Port for FastAPI server
            include_docstrings: Include function docstrings in documentation

        Returns:
            RemoteCodeExecutor instance
        """
        from llmling_agent.resource_providers.codemode.helpers import tools_to_codegen

        toolset_gen = tools_to_codegen(tools, include_docstrings)
        server_handler = ToolServerLifecycleHandler(toolset_gen, server_host, server_port)
        execution_env = env_config.get_provider(server_handler)
        return cls(toolset_gen, execution_env)

    def get_tool_description(self) -> str:
        """Get comprehensive description of available tools."""
        return self.toolset_generator.generate_tool_description()

    async def execute_code(self, code: str) -> Any:
        """Execute code with tools available via HTTP API.

        Args:
            code: Python code to execute

        Returns:
            Execution result from the environment
        """
        return await self.execution_env.execute(code)

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

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Async context manager exit."""
        return await self.execution_env.__aexit__(exc_type, exc_val, exc_tb)

      execution_env instance-attribute 

      execution_env: ExecutionEnvironment

      Execution environment for running code.

      toolset_generator instance-attribute 

      toolset_generator: ToolsetCodeGenerator

      Code generator for tools.

      __aenter__ async 

      __aenter__() -> Self

      Async context manager entry.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      90
91
92
93
      async def __aenter__(self) -> Self:
    """Async context manager entry."""
    await self.execution_env.__aenter__()
    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/code_executor.py 
       95
 96
 97
 98
 99
100
101
102
      async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context manager exit."""
    return await self.execution_env.__aexit__(exc_type, exc_val, exc_tb)

      execute_code async 

      execute_code(code: str) -> Any

      Execute code with tools available via HTTP API.

      Parameters:

      Name Type Description Default
      code str

      Python code to execute

      		 required

      Returns:

      Type Description
      Any

      Execution result from the environment
      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      79
80
81
82
83
84
85
86
87
88
      async def execute_code(self, code: str) -> Any:
    """Execute code with tools available via HTTP API.

    Args:
        code: Python code to execute

    Returns:
        Execution result from the environment
    """
    return await self.execution_env.execute(code)

      from_tools classmethod 

      from_tools(
    tools: Sequence[Tool],
    env_config: ExecutionEnvironmentConfig,
    server_host: str = "localhost",
    server_port: int = 8000,
    include_docstrings: bool = True,
) -> RemoteCodeExecutor

      Create provider from tools and environment configuration.

      Parameters:

      Name Type Description Default
      tools Sequence[Tool]

      Tools to make available for code execution

      		 required
      env_config ExecutionEnvironmentConfig

      Execution environment configuration

      		 required
      server_host str

      Host for FastAPI server

      		 'localhost'
      server_port int

      Port for FastAPI server

      		 8000
      include_docstrings bool

      Include function docstrings in documentation

      		 True

      Returns:

      Type Description
      RemoteCodeExecutor

      RemoteCodeExecutor instance
      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      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
      @classmethod
def from_tools(
    cls,
    tools: Sequence[Tool],
    env_config: ExecutionEnvironmentConfig,
    server_host: str = "localhost",
    server_port: int = 8000,
    include_docstrings: bool = True,
) -> RemoteCodeExecutor:
    """Create provider from tools and environment configuration.

    Args:
        tools: Tools to make available for code execution
        env_config: Execution environment configuration
        server_host: Host for FastAPI server
        server_port: Port for FastAPI server
        include_docstrings: Include function docstrings in documentation

    Returns:
        RemoteCodeExecutor instance
    """
    from llmling_agent.resource_providers.codemode.helpers import tools_to_codegen

    toolset_gen = tools_to_codegen(tools, include_docstrings)
    server_handler = ToolServerLifecycleHandler(toolset_gen, server_host, server_port)
    execution_env = env_config.get_provider(server_handler)
    return cls(toolset_gen, execution_env)

      get_tool_description 

      get_tool_description() -> str

      Get comprehensive description of available tools.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      75
76
77
      def get_tool_description(self) -> str:
    """Get comprehensive description of available tools."""
    return self.toolset_generator.generate_tool_description()

      ToolServerLifecycleHandler

      Manages FastAPI server lifecycle for tool access.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
      class ToolServerLifecycleHandler:
    """Manages FastAPI server lifecycle for tool access."""

    def __init__(
        self,
        toolset_generator: ToolsetCodeGenerator,
        host: str = "localhost",
        port: int = 8000,
    ) -> None:
        self.toolset_generator = toolset_generator
        self.host = host
        self.port = port  # Will be set when socket is created
        self.app: FastAPI | None = None
        self.server: uvicorn.Server | None = None
        self._server_task: asyncio.Task[None] | None = None
        self._socket: socket.socket | None = None

    async def __aenter__(self) -> ServerInfo:
        """Start FastAPI server with tool routes."""
        from anyenv.code_execution.models import ServerInfo
        from fastapi import FastAPI

        from llmling_agent.utils.network import _create_socket

        if self.server is not None:
            return ServerInfo(url=f"http://{self.host}:{self.port}", port=self.port)
        # Create socket and get actual port
        self._socket, self.port = _create_socket(self.port)
        # Create FastAPI app
        self.app = FastAPI(title="Tool Server", description="Generated tool endpoints")

        # Add tool routes
        self.toolset_generator.add_all_routes(self.app, "/tools")

        import uvicorn

        config = uvicorn.Config(
            self.app,
            log_level="error",  # Reduce log noise
            access_log=False,
            host=self.host,
            port=self.port,
            ws="websockets-sansio",
        )
        self.server = uvicorn.Server(config)

        # Start server in background with our socket
        self._server_task = asyncio.create_task(self.server.serve([self._socket]))
        logger.info("Started tool server", host=self.host, port=self.port)
        await asyncio.sleep(0.1)  # Wait for server to start properly
        # Verify server is running
        max_retries = 10
        for _ in range(max_retries):
            if self.server.started:
                break
            await asyncio.sleep(0.1)

        return ServerInfo(url=f"http://{self.host}:{self.port}", port=self.port)

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Stop FastAPI server."""
        # Stop server gracefully
        if self.server:
            with contextlib.suppress(Exception):
                self.server.should_exit = True
                if hasattr(self.server, "force_exit"):
                    self.server.force_exit = True

        # Cancel server task
        if self._server_task and not self._server_task.done():
            self._server_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await asyncio.wait_for(self._server_task, timeout=1.0)

        # Close socket
        if self._socket:
            with contextlib.suppress(Exception):
                self._socket.close()

        # Reset state
        self.server = None
        self._server_task = None
        self._socket = None
        self.app = None

      __aenter__ async 

      __aenter__() -> ServerInfo

      Start FastAPI server with tool routes.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
      async def __aenter__(self) -> ServerInfo:
    """Start FastAPI server with tool routes."""
    from anyenv.code_execution.models import ServerInfo
    from fastapi import FastAPI

    from llmling_agent.utils.network import _create_socket

    if self.server is not None:
        return ServerInfo(url=f"http://{self.host}:{self.port}", port=self.port)
    # Create socket and get actual port
    self._socket, self.port = _create_socket(self.port)
    # Create FastAPI app
    self.app = FastAPI(title="Tool Server", description="Generated tool endpoints")

    # Add tool routes
    self.toolset_generator.add_all_routes(self.app, "/tools")

    import uvicorn

    config = uvicorn.Config(
        self.app,
        log_level="error",  # Reduce log noise
        access_log=False,
        host=self.host,
        port=self.port,
        ws="websockets-sansio",
    )
    self.server = uvicorn.Server(config)

    # Start server in background with our socket
    self._server_task = asyncio.create_task(self.server.serve([self._socket]))
    logger.info("Started tool server", host=self.host, port=self.port)
    await asyncio.sleep(0.1)  # Wait for server to start properly
    # Verify server is running
    max_retries = 10
    for _ in range(max_retries):
        if self.server.started:
            break
        await asyncio.sleep(0.1)

    return ServerInfo(url=f"http://{self.host}:{self.port}", port=self.port)

      __aexit__ async 

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

      Stop FastAPI server.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
      async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Stop FastAPI server."""
    # Stop server gracefully
    if self.server:
        with contextlib.suppress(Exception):
            self.server.should_exit = True
            if hasattr(self.server, "force_exit"):
                self.server.force_exit = True

    # Cancel server task
    if self._server_task and not self._server_task.done():
        self._server_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await asyncio.wait_for(self._server_task, timeout=1.0)

    # Close socket
    if self._socket:
        with contextlib.suppress(Exception):
            self._socket.close()

    # Reset state
    self.server = None
    self._server_task = None
    self._socket = None
    self.app = None

      add_numbers 

      add_numbers(x: int, y: int) -> int

      Add two numbers.

      Source code in src/llmling_agent/resource_providers/codemode/code_executor.py 
      201
202
203
      def add_numbers(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y