import os, timefrom typing import Optionalfrom dotenv import load_dotenvfrom langchain.agents import create_agentfrom langchain.tools import toolfrom langchain_openai import ChatOpenAIfrom k7_sdk import Client, SandboxProxyload_dotenv()K7_ENDPOINT = os.getenv("K7_ENDPOINT")K7_API_KEY = os.getenv("K7_API_KEY")SANDBOX_NAME = os.getenv("K7_SANDBOX_NAME", "lc-agent")SANDBOX_IMAGE = os.getenv("K7_SANDBOX_IMAGE", "alpine:latest")SANDBOX_NAMESPACE = os.getenv("K7_NAMESPACE", "default")K7_API_CA = os.getenv("K7_API_CA", "./k7-ca.crt")k7 = Client(endpoint=K7_ENDPOINT, api_key=K7_API_KEY, verify_ssl=K7_API_CA)_sb: Optional[SandboxProxy] = Nonedef ensure_sandbox_ready(timeout_seconds: int = 120) -> SandboxProxy: try: sb = k7.create({ "name": SANDBOX_NAME, "image": SANDBOX_IMAGE, "namespace": SANDBOX_NAMESPACE, "backend": "k7d", "limits": {"cpu": "1", "memory": "1Gi"}, }) except Exception: sb = SandboxProxy(SANDBOX_NAME, SANDBOX_NAMESPACE, k7) deadline = time.time() + timeout_seconds while time.time() < deadline: for info in k7.list(namespace=SANDBOX_NAMESPACE): if info.get("name") == SANDBOX_NAME and info.get("ready") == "True": return sb time.sleep(2) raise RuntimeError("Sandbox did not become Running in time")@tooldef sandbox_exec(command: str) -> str: """Execute a shell command inside an isolated K7 sandbox.""" global _sb if _sb is None: _sb = ensure_sandbox_ready() result = _sb.exec(command) if result.get("exit_code", 1) != 0: return f"[stderr]\n{result.get('stderr','')}\n[stdout]\n{result.get('stdout','')}" return result.get("stdout", "")agent = create_agent( model=ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0), tools=[sandbox_exec], system_prompt="You run shell commands in an isolated K7 sandbox via the sandbox_exec tool.",)print("Ask me to run a command in a sandbox, e.g.: 'List files in /'\\n")while True: try: user = input("You: ") except (EOFError, KeyboardInterrupt): break if not user.strip(): continue resp = agent.invoke({"messages": [{"role": "user", "content": user}]}) print("Agent:", resp["messages"][-1].content)
