Appearance
Session Management Examples
This directory contains examples demonstrating the complete lifecycle of AGB sessions.
Key Concepts
Session Creation
- Use
CreateSessionParamsto configure your session (image, timeouts). - Always check
result.successbefore proceeding.
Session Retrieval
- You can reconnect to any active session using
agb.get(session_id). - This is useful for stateless applications (e.g., web servers) that need to resume control of a session.
Session Pooling
- For high-throughput applications, creating a new session for every request is inefficient.
- Maintaining a pool of "warm" sessions allows for faster execution and better resource utilization.
Session Metrics
- Use
session.get_metrics()to retrieve real-time resource usage information. - Metrics include CPU usage, memory usage, disk usage, and network statistics.
- Useful for monitoring session performance and resource consumption.
Cleanup
- Always call
agb.delete(session)in afinallyblock or when the session is no longer needed to avoid unnecessary charges.
Examples
Basic Session Lifecycle
Basic session creation, listing, and deletion.
py
"""
This example demonstrates how to create a session in AGB.
"""
import os
import time
from typing import Dict, Optional
from agb import AGB
from agb.session_params import CreateSessionParams
def create_session_with_default_params() -> None:
"""Create a session with default parameters."""
# Initialize the AGB client
api_key = os.environ.get("AGB_API_KEY", "")
agb = AGB(api_key=api_key)
# Create a session with default parameters
params = CreateSessionParams(image_id="agb-code-space-1")
result = agb.create(params)
if result.success and result.session:
session = result.session
print(f"Session created successfully with ID: {session.session_id}")
print(f"Request ID: {result.request_id}")
# Clean up
delete_result = agb.delete(session)
if delete_result.success:
print("Session deleted successfully")
else:
print(f"Failed to delete session: {delete_result.error_message}")
else:
print(f"Failed to create session: {result.error_message}")
def create_session_with_custom_image() -> None:
"""Create a session with custom image."""
# Initialize the AGB client
api_key = os.environ.get("AGB_API_KEY", "")
agb = AGB(api_key=api_key)
# Create session parameters with custom image
params = CreateSessionParams(image_id="agb-code-space-1")
# Create a session with the parameters
result = agb.create(params)
if result.success and result.session:
session = result.session
print(
f"Session with custom image created successfully with ID: {session.session_id}"
)
print(f"Request ID: {result.request_id}")
# Clean up
delete_result = agb.delete(session)
if delete_result.success:
print("Session deleted successfully")
else:
print(f"Failed to delete session: {delete_result.error_message}")
else:
print(f"Failed to create session with custom image: {result.error_message}")
def create_session_with_labels() -> None:
"""Create a session with labels."""
# Initialize the AGB client
api_key = os.environ.get("AGB_API_KEY", "")
agb = AGB(api_key=api_key)
# Define labels
labels: Dict[str, str] = {
"environment": "development",
"project": "example",
"owner": "user123",
"team": "backend",
"version": "v1.0.0",
}
# Create session parameters with labels
params = CreateSessionParams(image_id="agb-browser-use-1", labels=labels)
# Create a session with the parameters
result = agb.create(params)
if result.success and result.session:
session = result.session
print(f"Session with labels created successfully with ID: {session.session_id}")
print(f"Request ID: {result.request_id}")
# Verify the labels were set
label_result = session.get_labels()
if label_result.success:
retrieved_labels = label_result.data
print("Retrieved labels:")
for key, value in retrieved_labels.items():
print(f" {key}: {value}")
else:
print(f"Failed to get labels: {label_result.error_message}")
# Update labels during session lifecycle
updated_labels = {
**labels,
"status": "active",
"last_updated": str(int(time.time())),
}
update_result = session.set_labels(updated_labels)
if update_result.success:
print("Labels updated successfully")
# Get updated labels
final_label_result = session.get_labels()
if final_label_result.success:
final_labels = final_label_result.data
print("Final labels:")
for key, value in final_labels.items():
print(f" {key}: {value}")
else:
print(f"Failed to update labels: {update_result.error_message}")
# Clean up
delete_result = agb.delete(session)
if delete_result.success:
print("Session deleted successfully")
else:
print(f"Failed to delete session: {delete_result.error_message}")
else:
print(f"Failed to create session with labels: {result.error_message}")
def main() -> None:
"""Run all examples."""
print("1. Creating session with default parameters...")
create_session_with_default_params()
print("\n2. Creating session with custom image...")
create_session_with_custom_image()
print("\n3. Creating session with labels...")
create_session_with_labels()
if __name__ == "__main__":
main()Reconnecting to Sessions
How to retrieve an existing session by ID.
py
"""
Example demonstrating how to use the Get API to retrieve a session by its ID.
This example shows:
1. Creating a session
2. Retrieving the session using the Get API
3. Using the session for operations
4. Cleaning up resources
"""
import os
import sys
# Add the project root to Python path
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
sys.path.insert(0, project_root)
from agb import AGB
from agb.session_params import CreateSessionParams
def main():
# Get API key from environment variable
api_key = os.getenv("AGB_API_KEY")
if not api_key:
raise ValueError("AGB_API_KEY environment variable is not set")
# Initialize AGB client
agb = AGB(api_key=api_key)
# For demonstration, first create a session
print("Creating a session...")
params = CreateSessionParams(image_id="agb-browser-use-1")
create_result = agb.create(params)
if not create_result.success:
raise RuntimeError(f"Failed to create session: {create_result.error_message}")
session_id = create_result.session.session_id
print(f"Created session with ID: {session_id}")
# Retrieve the session by ID using Get API
print("\nRetrieving session using Get API...")
get_result = agb.get(session_id)
if not get_result.success:
raise RuntimeError(f"Failed to get session: {get_result.error_message}")
session = get_result.session
# Display session information
print("Successfully retrieved session:")
print(f" Session ID: {session.session_id}")
print(f" Request ID: {get_result.request_id}")
print(f" Resource URL: {session.resource_url}")
print(f" Resource ID: {session.resource_id}")
print(f" App Instance ID: {session.app_instance_id}")
# The session object can be used for further operations
# For example, you can execute commands, work with files, etc.
print("\nSession is ready for use")
# Clean up: Delete the session when done
print("\nCleaning up...")
delete_result = agb.delete(session)
if delete_result.success:
print(f"Session {session_id} deleted successfully")
else:
print(f"Failed to delete session {session_id}")
if __name__ == "__main__":
main()Session Pooling
Implementation of a thread-safe session pool for high-concurrency applications.
py
"""
AGB Session Pool Example
This example demonstrates how to implement a thread-safe session pool.
Pooling sessions is critical for applications that need to execute many short-lived operations
without incurring the latency of creating a new session for every request.
"""
import os
import threading
import time
from contextlib import contextmanager
from typing import Dict, Optional
from agb import AGB
from agb.session_params import CreateSessionParams
class SessionPool:
"""Thread-safe session pool for high-throughput applications"""
def __init__(self, api_key: str, max_sessions: int = 5, session_timeout: int = 300):
self.agb = AGB(api_key=api_key)
self.max_sessions = max_sessions
self.session_timeout = session_timeout
self.sessions: Dict[str, dict] = {}
self.lock = threading.Lock()
print(f"🎱 Session Pool initialized (max={max_sessions})")
@contextmanager
def get_session(self):
"""Context manager to acquire and release a session"""
session_info = self._acquire_session()
try:
print(f"🟢 Acquired session: {session_info['id']}")
yield session_info["session"]
finally:
print(f"🟡 Released session: {session_info['id']}")
self._release_session(session_info["id"])
def _acquire_session(self):
with self.lock:
# 1. Cleanup expired sessions
self._cleanup_expired_sessions()
# 2. Try to reuse an idle session
for session_id, info in self.sessions.items():
if not info["in_use"]:
info["in_use"] = True
info["last_used"] = time.time()
return info
# 3. Create new session if under limit
if len(self.sessions) < self.max_sessions:
print("✨ Creating new session for pool...")
params = CreateSessionParams(image_id="agb-code-space-1")
result = self.agb.create(params)
if result.success:
session_info = {
"id": result.session.session_id,
"session": result.session,
"created": time.time(),
"last_used": time.time(),
"in_use": True,
}
self.sessions[session_info["id"]] = session_info
return session_info
else:
raise Exception(f"Failed to create session: {result.error_message}")
raise Exception("No sessions available and pool is at maximum capacity")
def _release_session(self, session_id: str):
with self.lock:
if session_id in self.sessions:
self.sessions[session_id]["in_use"] = False
def _cleanup_expired_sessions(self):
current_time = time.time()
expired_ids = []
for session_id, info in self.sessions.items():
# If idle for too long
if (
not info["in_use"]
and (current_time - info["last_used"]) > self.session_timeout
):
expired_ids.append(session_id)
for session_id in expired_ids:
print(f"⌛ Cleaning up expired session: {session_id}")
session_info = self.sessions.pop(session_id)
try:
self.agb.delete(session_info["session"])
except Exception as e:
print(f"Error deleting session {session_id}: {e}")
def destroy_all(self):
"""Clean up all sessions in the pool"""
print("💥 Destroying pool...")
with self.lock:
for session_info in self.sessions.values():
try:
self.agb.delete(session_info["session"])
print(f"Deleted {session_info['id']}")
except Exception as e:
print(f"Error deleting session: {e}")
self.sessions.clear()
def main():
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("Error: AGB_API_KEY environment variable is not set")
return
# 1. Initialize pool
pool = SessionPool(api_key=api_key, max_sessions=2)
try:
# 2. Run tasks using the pool
# We'll simulate 4 tasks with a pool of size 2
# This means sessions will be reused
def run_task(task_name):
try:
with pool.get_session() as session:
print(f"▶️ Running {task_name}...")
result = session.code.run(
f"print('Hello from {task_name}')", "python"
)
# Simulate work holding the session
time.sleep(1)
output = result.results[0].text if result.success and result.results else ""
print(f"✅ {task_name} Result: {output.strip()}")
except Exception as e:
print(f"❌ {task_name} failed: {e}")
threads = []
for i in range(4):
t = threading.Thread(target=run_task, args=(f"Task-{i}",))
threads.append(t)
t.start()
for t in threads:
t.join()
finally:
# 3. Cleanup
pool.destroy_all()
if __name__ == "__main__":
main()Session Metrics
Demonstrates how to retrieve real-time session metrics including CPU, memory, disk, and network usage.
py
"""
This example demonstrates how to retrieve session metrics in AGB using a shared session.
Session metrics provide real-time information about resource usage including:
- CPU usage and core count
- Memory usage (total and used)
- Disk usage (total and used)
- Network statistics (RX/TX rate and total usage)
- Timestamp of the metrics snapshot
This version uses a single shared session across all examples for better resource efficiency.
"""
import os
import time
from agb import AGB
from agb.session_params import CreateSessionParams
def get_basic_metrics(session) -> None:
"""Get basic session metrics."""
print(f"Getting metrics for session: {session.session_id}")
# Get session metrics
metrics_result = session.get_metrics()
if metrics_result.success and metrics_result.metrics:
metrics = metrics_result.metrics
print("\n=== Session Metrics ===")
print(f"Request ID: {metrics_result.request_id}")
# Display CPU information
if hasattr(metrics, 'cpu_count') and hasattr(metrics, 'cpu_used_pct'):
print(f"\nCPU:")
print(f" CPU Count: {metrics.cpu_count}")
print(f" CPU Used: {metrics.cpu_used_pct:.2f}%")
# Display Memory information
if hasattr(metrics, 'mem_total') and hasattr(metrics, 'mem_used'):
mem_total_gb = metrics.mem_total / (1024**3) # Convert bytes to GB
mem_used_gb = metrics.mem_used / (1024**3)
mem_usage_pct = (metrics.mem_used / metrics.mem_total) * 100 if metrics.mem_total > 0 else 0
print(f"\nMemory:")
print(f" Total: {mem_total_gb:.2f} GB")
print(f" Used: {mem_used_gb:.2f} GB")
print(f" Usage: {mem_usage_pct:.2f}%")
# Display Disk information
if hasattr(metrics, 'disk_total') and hasattr(metrics, 'disk_used'):
disk_total_gb = metrics.disk_total / (1024**3) # Convert bytes to GB
disk_used_gb = metrics.disk_used / (1024**3)
disk_usage_pct = (metrics.disk_used / metrics.disk_total) * 100 if metrics.disk_total > 0 else 0
print(f"\nDisk:")
print(f" Total: {disk_total_gb:.2f} GB")
print(f" Used: {disk_used_gb:.2f} GB")
print(f" Usage: {disk_usage_pct:.2f}%")
# Display Network information
if hasattr(metrics, 'rx_rate_kbps') and hasattr(metrics, 'tx_rate_kbps'):
print(f"\nNetwork:")
print(f" RX Rate: {metrics.rx_rate_kbps:.2f} KB/s")
print(f" TX Rate: {metrics.tx_rate_kbps:.2f} KB/s")
if hasattr(metrics, 'rx_used_kb') and hasattr(metrics, 'tx_used_kb'):
print(f" RX Total: {metrics.rx_used_kb:.2f} KB")
print(f" TX Total: {metrics.tx_used_kb:.2f} KB")
# Display timestamp
if hasattr(metrics, 'timestamp'):
print(f"\nTimestamp: {metrics.timestamp}")
else:
print(f"Failed to get session metrics: {metrics_result.error_message}")
def monitor_metrics_over_time(session) -> None:
"""Monitor session metrics over time using the shared session."""
print(f"Monitoring metrics for session: {session.session_id}")
print("Monitoring metrics over time (5 snapshots, 10 seconds apart)...")
for i in range(5):
print(f"\n--- Snapshot {i + 1} ---")
# Get session metrics
metrics_result = session.get_metrics()
if metrics_result.success and metrics_result.metrics:
metrics = metrics_result.metrics
# Display key metrics
if hasattr(metrics, 'cpu_used_pct'):
print(f"CPU Usage: {metrics.cpu_used_pct:.2f}%")
if hasattr(metrics, 'mem_used') and hasattr(metrics, 'mem_total'):
mem_usage_pct = (metrics.mem_used / metrics.mem_total) * 100 if metrics.mem_total > 0 else 0
print(f"Memory Usage: {mem_usage_pct:.2f}%")
if hasattr(metrics, 'rx_rate_kbps') and hasattr(metrics, 'tx_rate_kbps'):
print(f"Network: RX {metrics.rx_rate_kbps:.2f} KB/s, TX {metrics.tx_rate_kbps:.2f} KB/s")
if hasattr(metrics, 'timestamp'):
print(f"Time: {metrics.timestamp}")
else:
print(f"Failed to get metrics: {metrics_result.error_message}")
# Wait before next snapshot (except for the last one)
if i < 4:
time.sleep(10)
def main() -> None:
"""Run all metrics examples using a shared session."""
# Initialize the AGB client
api_key = os.environ.get("AGB_API_KEY", "")
if not api_key:
raise ValueError("AGB_API_KEY environment variable is not set")
agb = AGB(api_key=api_key)
session = None
try:
# Create a single shared session
print("Creating shared session...")
params = CreateSessionParams(image_id="agb-computer-use-ubuntu-2204")
result = agb.create(params)
if not result.success or not result.session:
raise RuntimeError(f"Failed to create session: {result.error_message}")
session = result.session
print(f"Shared session created successfully with ID: {session.session_id}")
# Run all examples using the shared session
print("\n" + "="*60)
print("1. Getting basic session metrics...")
get_basic_metrics(session)
print("\n" + "="*60)
print("2. Monitoring metrics over time...")
monitor_metrics_over_time(session)
except Exception as e:
print(f"Error occurred: {e}")
finally:
# Clean up: Delete the shared session when done
if session:
print("\n" + "="*60)
print("Cleaning up shared session...")
delete_result = agb.delete(session)
if delete_result.success:
print(f"Shared session {session.session_id} deleted successfully")
else:
print(f"Failed to delete session: {delete_result.error_message}")
if __name__ == "__main__":
main()MCP Tools
Demonstrates how to list and call MCP (Model Context Protocol) tools available in a session.
py
"""
Example demonstrating how to use MCP tool calling methods in Session.
This example shows:
1. Listing available MCP tools for a session
2. Calling MCP tools with different arguments
3. Handling tool call results and errors
"""
import os
import sys
import json
# Add the project root to Python path
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
sys.path.insert(0, project_root)
from agb import AGB
from agb.session_params import CreateSessionParams
def list_mcp_tools_example(session):
"""Example: List available MCP tools"""
print("=" * 60)
print("Example 1: List Available MCP Tools")
print("=" * 60)
result = session.list_mcp_tools()
if not result.success:
print(f"Failed to list MCP tools: {result.error_message}")
return
print(f"\nFound {len(result.tools)} available tools:")
print(f"Request ID: {result.request_id}\n")
for i, tool in enumerate(result.tools, 1):
print(f"{i}. {tool.name}")
print(f" Description: {tool.description}")
print(f" Server: {tool.server}")
print(f" Type: {tool.tool}")
if tool.input_schema:
print(f" Input Schema: {json.dumps(tool.input_schema, indent=2)}")
print()
def list_mcp_tools_with_image_id_example(session):
"""Example: List MCP tools for a specific image"""
print("=" * 60)
print("Example 2: List MCP Tools for Specific Image")
print("=" * 60)
# List tools for a specific image (e.g., code space)
result = session.list_mcp_tools(image_id="agb-code-space-1")
if not result.success:
print(f"Failed to list MCP tools: {result.error_message}")
return
print(f"\nFound {len(result.tools)} tools for image 'agb-code-space-1':")
for tool in result.tools:
print(f" - {tool.name}: {tool.description}")
def call_mcp_tool_example(session):
"""Example: Call an MCP tool"""
print("=" * 60)
print("Example 3: Call MCP Tool")
print("=" * 60)
# First, list available tools to see what we can call
list_result = session.list_mcp_tools()
if not list_result.success or not list_result.tools:
print("No tools available or failed to list tools")
return
# Use the first available tool as an example
# Note: In practice, you should know which tool you want to call
if list_result.tools:
tool = list_result.tools[0]
print(f"\nCalling tool: {tool.name}")
print(f"Description: {tool.description}")
# Example: Call tool with empty args (adjust based on actual tool requirements)
# In real usage, you would provide appropriate arguments based on the tool's input_schema
result = session.call_mcp_tool(tool.name, {})
if result.success:
print(f"\nTool call succeeded!")
print(f"Request ID: {result.request_id}")
if result.data:
try:
# Try to parse and pretty-print the result
data = json.loads(result.data)
print(f"Result: {json.dumps(data, indent=2)}")
except json.JSONDecodeError:
print(f"Result (raw): {result.data}")
else:
print(f"\nTool call failed: {result.error_message}")
print(f"Request ID: {result.request_id}")
def call_mcp_tool_with_timeouts_example(session):
"""Example: Call MCP tool with custom timeouts"""
print("=" * 60)
print("Example 4: Call MCP Tool with Custom Timeouts")
print("=" * 60)
list_result = session.list_mcp_tools()
if not list_result.success or not list_result.tools:
print("No tools available")
return
if list_result.tools:
tool = list_result.tools[0]
print(f"\nCalling tool '{tool.name}' with custom timeouts...")
# Call with custom read and connect timeouts (in milliseconds)
result = session.call_mcp_tool(
tool.name,
{},
read_timeout=30000, # 30 seconds
connect_timeout=5000, # 5 seconds
)
if result.success:
print("Tool call succeeded with custom timeouts!")
else:
print(f"Tool call failed: {result.error_message}")
def main():
"""Main function demonstrating MCP tool usage"""
# Get API key from environment variable
api_key = os.getenv("AGB_API_KEY")
if not api_key:
raise ValueError("AGB_API_KEY environment variable is not set")
# Initialize AGB client
agb = AGB(api_key=api_key)
try:
# Create a session
print("Creating session...")
params = CreateSessionParams(image_id="agb-code-space-1")
create_result = agb.create(params)
if not create_result.success:
raise RuntimeError(f"Failed to create session: {create_result.error_message}")
session = create_result.session
print(f"Session created: {session.session_id}\n")
# Run examples
list_mcp_tools_example(session)
list_mcp_tools_with_image_id_example(session)
call_mcp_tool_example(session)
call_mcp_tool_with_timeouts_example(session)
finally:
# Clean up: Delete the session when done
print("\n" + "=" * 60)
print("Cleaning up...")
if "session" in locals():
delete_result = agb.delete(session)
if delete_result.success:
print("Session deleted successfully")
else:
print(f"Failed to delete session: {delete_result.error_message}")
if __name__ == "__main__":
main()