LangChain agent tutorial

K7_ENDPOINT=https://your-k7-endpoint:31007
K7_API_KEY=your-api-key
K7_API_CA=./k7-ca.crt
K7_SANDBOX_NAME=lc-agent
K7_SANDBOX_IMAGE=alpine:latest
K7_NAMESPACE=default
OPENAI_API_KEY=sk-your-openai-key
OPENAI_MODEL=gpt-4o-mini

pip install "langchain>=1.0" langchain-openai python-dotenv k7-sdk==0.4.0

import os, time
from typing import Optional
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from k7_sdk import Client, SandboxProxy

load_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] = None

def 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")

@tool
def 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)