1  # Copyright 2025 Alibaba Group Holding Ltd.
  2  #
  3  # Licensed under the Apache License, Version 2.0 (the "License");
  4  # you may not use this file except in compliance with the License.
  5  # You may obtain a copy of the License at
  6  #
  7  #     http://www.apache.org/licenses/LICENSE-2.0
  8  #
  9  # Unless required by applicable law or agreed to in writing, software
 10  # distributed under the License is distributed on an "AS IS" BASIS,
 11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12  # See the License for the specific language governing permissions and
 13  # limitations under the License.
 14  
 15  import os
 16  from datetime import timedelta
 17  
 18  from google.adk.agents import Agent
 19  from google.adk.apps import App
 20  from google.adk.runners import Runner
 21  from google.adk.sessions.in_memory_session_service import InMemorySessionService
 22  from google.adk.utils._debug_output import print_event
 23  from google.adk.utils.context_utils import Aclosing
 24  from google.genai import types
 25  from opensandbox import Sandbox
 26  from opensandbox.config import ConnectionConfig
 27  
 28  
 29  def _required_env(name: str) -> str:
 30      value = os.getenv(name)
 31      if not value:
 32          raise RuntimeError(f"{name} is required")
 33      return value
 34  
 35  
 36  async def main() -> None:
 37      _required_env("GOOGLE_API_KEY")
 38      domain = os.getenv("SANDBOX_DOMAIN", "localhost:8080")
 39      api_key = os.getenv("SANDBOX_API_KEY")
 40      image = os.getenv(
 41          "SANDBOX_IMAGE",
 42          "sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/code-interpreter:v1.0.2",
 43      )
 44      model_name = os.getenv("GOOGLE_ADK_MODEL", "gemini-2.5-flash")
 45  
 46      config = ConnectionConfig(
 47          domain=domain,
 48          api_key=api_key,
 49          request_timeout=timedelta(seconds=120),
 50      )
 51  
 52      sandbox = await Sandbox.create(
 53          image,
 54          connection_config=config,
 55      )
 56  
 57      async def run_in_sandbox(command: str) -> str:
 58          """Run a shell command in OpenSandbox and return the output."""
 59  
 60          execution = await sandbox.commands.run(command)
 61          stdout = "\n".join(msg.text for msg in execution.logs.stdout)
 62          stderr = "\n".join(msg.text for msg in execution.logs.stderr)
 63          if execution.error:
 64              stderr = "\n".join(
 65                  [
 66                      stderr,
 67                      f"[error] {execution.error.name}: {execution.error.value}",
 68                  ]
 69              ).strip()
 70  
 71          output = stdout.strip()
 72          if stderr:
 73              output = "\n".join([output, f"[stderr]\n{stderr}"]).strip()
 74          return output or "(no output)"
 75  
 76      async def write_file(path: str, content: str) -> str:
 77          """Write a file inside the sandbox."""
 78  
 79          await sandbox.files.write_file(path, content)
 80          return f"wrote {len(content)} bytes to {path}"
 81  
 82      async def read_file(path: str) -> str:
 83          """Read a file from the sandbox."""
 84  
 85          return await sandbox.files.read_file(path)
 86  
 87      agent = Agent(
 88          name="opensandbox_adk",
 89          model=model_name,
 90          instruction=(
 91              "You have access to OpenSandbox tools. Use write_file to create or "
 92              "update files, read_file to read files, and run_in_sandbox to run "
 93              "commands."
 94          ),
 95          tools=[run_in_sandbox, write_file, read_file],
 96      )
 97  
 98      app = App(name="opensandbox_adk", root_agent=agent)
 99      session_service = InMemorySessionService()
100      runner = Runner(app=app, session_service=session_service)
101      session = await session_service.create_session(
102          app_name=app.name,
103          user_id="local-user",
104      )
105  
106      prompts = [
107          "Use write_file to save /tmp/math.py that prints 137 * 42.",
108          "Run the script using run_in_sandbox and report the result.",
109          "Write /tmp/notes.txt with 'ADK + OpenSandbox', then read it back.",
110      ]
111  
112      try:
113          for prompt in prompts:
114              content = types.Content(
115                  role="user",
116                  parts=[types.Part(text=prompt)],
117              )
118              async with Aclosing(
119                  runner.run_async(
120                      user_id=session.user_id,
121                      session_id=session.id,
122                      new_message=content,
123                  )
124              ) as agen:
125                  async for event in agen:
126                      print_event(event, verbose=True)
127      finally:
128          await sandbox.kill()
129          await sandbox.close()
130  
131  
132  if __name__ == "__main__":
133      import asyncio
134  
135      asyncio.run(main())