AgentCube Add-on

apiVersion: runtime.agentcube.volcano.sh/v1alpha1
kind: CodeInterpreter
metadata:
  name: my-codeinterpreter
  namespace: default
spec:
  template:
    # runtimeClassName: kata    # To use a secure container, set runtimeClassName to kata or kuasar-vmm. A node supporting secure runtime is required.
    # Use the PicoD image, which supports shell and Python code execution only.
    image: ghcr.io/volcano-sh/picod:latest
    # Resources
    resources:
      requests:
        cpu: "100m"
        memory: "128Mi"
      limits:
        cpu: "500m"
        memory: "512Mi"
  # Session configuration
  sessionTimeout: "15m"      # The session times out after it is idle for 15 minutes.
  maxSessionDuration: "8h"   # Maximum session duration: 8 hours
  # Key function: warm pool
  warmPoolSize: 5    # 5 pods will be warmed up (near-zero latency).

kubectl apply -f code-interpreter-demo.yaml

After the creation is complete, run the following command to check the status:

kubectl get codeinterpreter my-codeinterpreter -o yaml

# View the created SandboxWarmPool.
kubectl get sandboxwarmpool -n default
# View the pods in the warm pool.
kubectl get pods |grep my-codeinterpreter

1. Create the test_codeinterpreter folder.

   Create the test_codeinterpreter.py and requirements.txt files in the folder. The file content is as follows:

   • test_codeinterpreter.py

     import time
     from agentcube import CodeInterpreterClient
     import os
     
     WORKLOAD_MANAGER_URL = os.getenv('WORKLOAD_MANAGER_URL', 'http://workloadmanager.agentcube.svc.cluster.local:8080')
     ROUTER_URL = os.getenv('ROUTER_URL', 'http://agentcube-router.agentcube.svc.cluster.local:8080')
     CODEINTERPRETER_NAME = os.getenv('CODEINTERPRETER_NAME', 'my-codeinterpreter')
     CODEINTERPRETER_NAMESPACE = os.getenv('CODEINTERPRETER_NAMESPACE', 'default')
     
     def basic_operations():
         """Demonstrate basic operations"""
         print("=== basic operations ===\n")
     
         with CodeInterpreterClient(
                 name=CODEINTERPRETER_NAME,
                 namespace=CODEINTERPRETER_NAMESPACE,
                 workload_manager_url=WORKLOAD_MANAGER_URL,
                 router_url=ROUTER_URL,
                 verbose=True,
         ) as client:
             print(f"session ID: {client.session_id}\n")
     
             # 1. Run the shell command.
             print("1. Shell command (whoami):")
             output = client.execute_command("whoami")
             print(f"output: {output.strip()}\n")
     
             # 2. Run the Python code.
             print("2. Python code execution:")
             code = """
     import sys
     print(f"Python version: {sys.version}")
     import math
     print(f"π value: {math.pi:.10f}")
     """
             result = client.run_code("python", code)
             print(f"output:\n{result}\n")
     
             # 3. Complete file operations.
             print("3. file operate:")
             client.write_file("Hello, AgentCube!", "hello.txt")
             client.write_file("42", "number.txt")
     
             files = client.list_files(".")
             print(f"file list: {[f['name'] for f in files]}\n")
     
         print("session auto delete")
     
     def session_reuse_workflow():
         """
         Demonstrate session reuse that is important for AI workflows.
     
         Scenario: Data is processed step by step, and the file system status needs to be maintained between sessions.
         """
         print("=== session reuse workflow ===\n")
     
         # Step 1: Create a session and write data.
         print("step 1: Create a session and write initial data.")
         client1 = CodeInterpreterClient(
             name=CODEINTERPRETER_NAME,
             namespace=CODEINTERPRETER_NAMESPACE,
             workload_manager_url=WORKLOAD_MANAGER_URL,
             router_url=ROUTER_URL,
             verbose=True,
         )
     
         # Write the data to multiple files.
         client1.write_file("100", "counter.txt")
         client1.write_file("[]", "results.json")
     
         session_id = client1.session_id
         print(f"session ID: {session_id}")
         print("The file system status has been saved.\n")
     
         # Note: Do not call client1.stop() to keep the session active.
     
         # Step 2: Reuse the session to read and process data.
         print("step 2: Reusing sessions and processing data.")
         client2 = CodeInterpreterClient(
             name=CODEINTERPRETER_NAME,
             namespace=CODEINTERPRETER_NAMESPACE,
             workload_manager_url=WORKLOAD_MANAGER_URL,
             router_url=ROUTER_URL,
             session_id=session_id,  # Session reused
             verbose=True,
         )
     
         code = """
     import json
     import time
     
     # Read the counter.
     with open('counter.txt') as f:
         counter = int(f.read().strip())
     print(f"current count: {counter}")
     
     # Increase the count.
     counter += 1
     with open('counter.txt', 'w') as f:
         f.write(str(counter))
     
     # Read the result list.
     with open('results.json') as f:
         results = json.load(f)
     
     # Add a new result.
     results.append({
         'timestamp': time.time(),
         'counter': counter
     })
     
     # Save the result.
     with open('results.json', 'w') as f:
         json.dump(results, f, indent=2)
     
     print(f"new count: {counter}")
     print(f"result count: {len(results)}")
     """
     
         result = client2.run_code("python", code)
         print(f"{result}\n")
     
         # Step 3: Check the status of the file system.
         print("step 3: Verifying the File System Statuses")
         files = client2.list_files(".")
         print(f"Files in session: {[f['name'] for f in files]}\n")
     
         # Clear the session.
         client2.stop()
         print("session is deleted")
     
     if __name__ == "__main__":
         basic_operations()
         print("\n" + "="*50 + "\n")
         session_reuse_workflow()

   • requirements.txt

     agentcube_sdk

2. Go to the test_codeinterpreter folder and run the following command to create a Python virtual environment:

   python -m venv venv

3. Activate the virtual environment (on Windows, run the .\venv\Scripts\activate command in PowerShell).

   source venv/bin/activate

4. Install the dependency.

   pip install -r requirements.txt

1. Set environment variables. (If you want to access the workload from outside the cluster, set the Service type of workloadmanager and agentcube-router to NodePort and bind an EIP to a CCE cluster node. Then, you can access the workload using the EIP and node port.)
   • Linux:

     export WORKLOAD_MANAGER_URL="http://addr:workloadmanager-service-nodeport" 
     export ROUTER_URL="http://addr:agentcube-router-service-nodeport"

   • Windows (in PowerShell):

     $env:ROUTER_URL = "http://addr:agentcube-router-service-nodeport"
     $env:WORKLOAD_MANAGER_URL = "http://addr:workloadmanager-service-nodeport"

   In the command, addr indicates the EIP, agentcube-router-service-nodeport indicates the node port configured for the Service of agentcube-router, and workloadmanager-service-nodeport indicates the node port configured for the Service of workloadmanager.

2. Run the test.

   python test_codeinterpreter.py

   Output

   

   

   CodeInterpreter typically runs AI-generated code. For security, execute it in a secure container runtime.

docker build  -t my-agent-app:latest .

agent-runtime-demo.yaml file:

apiVersion: runtime.agentcube.volcano.sh/v1alpha1
kind: AgentRuntime
metadata:
  name: my-agent-app
  namespace: default
spec:
  targetPort:
    - pathPrefix: "/"
      port: 8080
      protocol: "HTTP"
  podTemplate:
    labels:
      app: my-agent-app
    spec:
      # runtimeClassName: kata    # To use a secure container, set runtimeClassName to kata or kuasar-vmm. A node supporting secure runtime is required.
      schedulerName: default-scheduler    # If Volcano agent-scheduler is installed, set this parameter to agent-scheduler.
      containers:
        - name: my-agent-app
          image: {{agent image}}    # Example agent image
          env:
            - name: WORKLOAD_MANAGER_URL
              value: http://workloadmanager.agentcube.svc.cluster.local:8080
            - name: ROUTER_URL
              value: http://agentcube-router.agentcube.svc.cluster.local:8080
            - name: CODEINTERPRETER_NAME         # Name of the code interpreter
              value: my-codeinterpreter
            - name: CODEINTERPRETER_NAMESPACE
              value: default
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            periodSeconds: 5
  sessionTimeout: "15m"             # The session times out after it is idle for 15 minutes.
  maxSessionDuration: "8h"          # Maximum session duration: 8 hours
status: {}

kubectl apply -f agent-runtime-demo.yaml
# Check the status.
kubectl get agentruntime my-agent-app -oyaml

1. Create the test_agentruntime folder.
   Create the test_agentruntime.py and requirements.txt files in the folder. The file content is as follows:

   • test_agentruntime.py

     from agentcube.agent_runtime import AgentRuntimeClient
     import os
     import time
     
     # Configure environment variables.
     ROUTER_URL = os.getenv('ROUTER_URL', 'http://agentcube-router.agentcube.svc.cluster.local:8080')
     AGENT_NAME = os.getenv('AGENT_NAME', 'my-agent-app')
     AGENT_NAMESPACE = os.getenv('AGENT_NAMESPACE', 'default')
     
     saved_session_id = ""
     
     def test_basic_invoke():
         """Test the basic calling function."""
         print("=== test basic invoke ===\n")
     
         # Create a client.
         agent_client = AgentRuntimeClient(
             agent_name=AGENT_NAME,
             namespace=AGENT_NAMESPACE,
             router_url=ROUTER_URL,
             verbose=True,
             timeout=500,
             connect_timeout=120.0
         )
     
         print(f"Session ID: {agent_client.session_id}\n")
     
         # Call for the first time (creates a new pod)
         print("first invoke (create new Pod): ")
         result = agent_client.invoke(
             payload={"prompt": "hello world!"},
         )
         print(f"response: {result}\n")
     
         # # Save the session ID.
         global saved_session_id
         saved_session_id = agent_client.session_id
     
     def test_session_reuse():
         """Test session reuse"""
         print(f"=== test session reuse ===\n")
     
         # Use the original session ID.
         agent_client = AgentRuntimeClient(
             agent_name=AGENT_NAME,
             namespace=AGENT_NAMESPACE,
             router_url=ROUTER_URL,
             session_id=saved_session_id, # Session reused
             verbose=True,
             timeout=500,
             connect_timeout=120.0
         )
     
         print(f"reuse session ID: {agent_client.session_id}\n")
     
         # Call for the second time (reuses the existing pod)
         print("second invoke(reuse Pod):")
         result = agent_client.invoke(
             payload={"prompt": "How's the weather today?"},
         )
         print(f"response: {result}\n")
     
     def test_conversation():
         """Test continuous conversation"""
         print("=== test conversation ===\n")
     
         agent_client = AgentRuntimeClient(
             agent_name=AGENT_NAME,
             namespace=AGENT_NAMESPACE,
             router_url=ROUTER_URL,
             verbose=True,
             timeout=500,
             connect_timeout=120.0
         )
     
         messages = [
             "Introduce yourself",
             "What can you do?",
             "Write a Python function for me",
         ]
     
         for i, msg in enumerate(messages, 1):
             print(f"message {i}: {msg}")
             result = agent_client.invoke(
                 payload={"prompt": msg},
             )
             print(f"response {i}: {result}\n")
             time.sleep(1)
     
     if __name__ == "__main__":
         test_basic_invoke()
         test_session_reuse()
         test_conversation()

   • requirements.txt

     agentcube_sdk

2. Go to the test_agentruntime folder and run the following command to create a Python virtual environment:

   python -m venv venv

3. Activate the virtual environment (on Windows, run the .\venv\Scripts\activate command in PowerShell).

   source venv/bin/activate

4. Install the dependency.

   pip install -r requirements.txt

1. Set environment variables. (To enable external access to the workload, configure the agentcube-router Service as NodePort and bind an EIP to a CCE cluster node. Access the workload using the EIP and NodePort.)
   • Linux:

     export ROUTER_URL="http://addr:agentcube-router-service-nodeport"

   • Windows (in PowerShell):

     $env:ROUTER_URL = "http://addr:agentcube-router-service-nodeport"

   In the command, addr indicates the EIP, and agentcube-router-service-nodeport indicates the node port configured for the Service of agentcube-router.

2. Run the test.

   python test_agentruntime.py

   Output