Agent-Computer Interface (ACI)
Bridging agents with computer interfaces: desktop control, OS integration, and computer use.
ACI Fundamentals
Agent-Computer Interface (ACI)
├── Desktop Control
│ ├── Screen Observation
│ ├── Input Injection (Mouse/Keyboard)
│ ├── Window Management
│ └── Accessibility APIs
├── OS Integration
│ ├── File System Operations
│ ├── Process Management
│ ├── Shell/Terminal Access
│ └── System APIs
├── Application Control
│ ├── Browser Automation
│ ├── Office/IDE Control
│ ├── API Interaction
│ └── Clipboard/OS Services
├── Security & Isolation
│ ├── Sandboxing
│ ├── Permission Model
│ ├── Audit Logging
│ └── Remote Access
└── Observability
├── Screen Recording
├── Action Logging
├── Performance Metrics
└── Debugging Interface
Desktop Control
Screen Observation
class ScreenObserver:
def __init__(self, config: ObservationConfig = None):
self.config = config or ObservationConfig()
self.capture = ScreenCapture()
async def capture(self, region: Region = None) -> ScreenCapture:
"""Capture screen with configurable options"""
if region:
return await self.capture.capture_region(region)
return await self.capture.capture_full()
async def capture_stream(self, fps: int = 30) -> AsyncIterator[ScreenCapture]:
"""Stream screen captures at specified FPS"""
interval = 1.0 / fps
while True:
start = time.time()
capture = await self.capture()
yield capture
elapsed = time.time() - start
await asyncio.sleep(max(0, interval - elapsed))
class ScreenCapture:
def __init__(self):
# Platform-specific backends
self.backends = {
'linux': X11Capture(),
'darwin': QuartzCapture(),
'windows': WindowsCapture()
}
self.backend = self.backends[platform.system().lower()]
async def capture_full(self) -> ScreenCapture:
return await self.backend.capture_full()
async def capture_region(self, region: Region) -> ScreenCapture:
return await self.backend.capture_region(region)
async def capture_window(self, window_id: str) -> ScreenCapture:
return await self.backend.capture_window(window_id)
Input Injection
class InputInjector:
def __init__(self):
self.backends = {
'linux': X11Input(),
'darwin': QuartzInput(),
'windows': WindowsInput()
}
self.injector = self.backends[platform.system().lower()]
async def click(self, x: int, y: int, button: str = 'left', count: int = 1):
await self.injector.click(x, y, button, count)
async def type_text(self, text: str, interval: float = 0.01):
await self.injector.type_text(text, interval)
async def key_press(self, key: str, modifiers: list[str] = None):
await self.injector.key_press(key, modifiers)
async def key_combination(self, keys: list[str]):
await self.injector.key_combination(keys)
async def scroll(self, x: int, y: int, dx: int, dy: int):
await self.injector.scroll(x, y, dx, dy)
async def drag_drop(self, start: Point, end: Point):
await self.injector.drag_drop(start, end)
class X11Input:
def __init__(self):
self.display = Xlib.display.Display()
async def click(self, x: int, y: int, button: str, count: int):
# X11 button events
button_map = {'left': 1, 'middle': 2, 'right': 3, 'wheel_up': 4, 'wheel_down': 5}
btn = button_map.get(button, 1)
for _ in range(count):
self.display.screen().root.warp_pointer(x, y)
self.display.send_event(
Xlib.X.ButtonPress,
detail=btn,
root=self.display.screen().root,
time=Xlib.X.CurrentTime
)
self.display.send_event(
Xlib.X.ButtonRelease,
detail=btn,
root=self.display.screen().root,
time=Xlib.X.CurrentTime
)
self.display.flush()
async def type_text(self, text: str, interval: float):
for char in text:
keysym = Xlib.XK.string_to_keysym(char)
keycode = self.display.keysym_to_keycode(keysym)
self.inject_key(keycode, True)
self.inject_key(keycode, False)
await asyncio.sleep(interval)
Window Management
class WindowManager:
def __init__(self):
self.backends = {
'linux': X11WindowManager(),
'darwin': QuartzWindowManager(),
'windows': WindowsWindowManager()
}
self.wm = self.backends[platform.system().lower()]
async def list_windows(self) -> list[WindowInfo]:
return await self.wm.list_windows()
async def get_active_window(self) -> WindowInfo:
return await self.wm.get_active_window()
async def focus_window(self, window_id: str):
await self.wm.focus_window(window_id)
async def get_window_geometry(self, window_id: str) -> Geometry:
return await self.wm.get_geometry(window_id)
async def set_window_state(self, window_id: str, state: WindowState):
await self.wm.set_state(window_id, state)
async def capture_window(self, window_id: str) -> ScreenCapture:
return await self.wm.capture(window_id)
class X11WindowManager:
def __init__(self):
self.display = Xlib.display.Display()
async def list_windows(self) -> list[WindowInfo]:
root = self.display.screen().root
windows = []
def traverse(window):
try:
attrs = window.get_attributes()
if attrs.map_state == Xlib.X.IsViewable:
geom = window.get_geometry()
name = window.get_wm_name() or ""
windows.append(WindowInfo(
id=str(window.id),
title=name,
geometry=Geometry(geom.x, geom.y, geom.width, geom.height),
pid=self._get_pid(window)
))
except:
pass
for child in window.query_tree().children:
traverse(child)
traverse(root)
return windows
OS Integration
File System Operations
class FileSystemAgent:
def __init__(self, sandbox: Sandbox = None):
self.sandbox = sandbox or LocalSandbox()
async def read_file(self, path: str, encoding: str = 'utf-8') -> str:
path = self._resolve_path(path)
async with aiofiles.open(path, 'r', encoding=encoding) as f:
return await f.read()
async def write_file(self, path: str, content: str, encoding: str = 'utf-8'):
path = self._resolve_path(path)
async with aiofiles.open(path, 'w', encoding=encoding) as f:
await f.write(content)
async def list_directory(self, path: str, recursive: bool = False) -> list[FileInfo]:
path = self._resolve_path(path)
if recursive:
files = []
for root, dirs, files in os.walk(path):
for f in files:
path = os.path.join(root, f)
stat = os.stat(path)
files.append(FileInfo(path=path, size=stat.st_size, modified=stat.st_mtime))
return files
else:
entries = await aiofiles.os.scandir(path)
return [FileInfo.from_dirent(e) for e in entries]
async def search_files(self, pattern: str, path: str = '.') -> list[str]:
path = self._resolve_path(path)
return await asyncio.to_thread(glob.glob, os.path.join(path, pattern), recursive=True)
async def grep(self, pattern: str, path: str = '.', file_pattern: str = '*') -> list[GrepResult]:
path = self._resolve_path(path)
cmd = ['rg', '--json', pattern, path, '--glob', file_pattern]
result = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, _ = await cmd.communicate()
return [GrepResult.from_json(line) for line in stdout.decode().splitlines() if line]
class LocalSandbox:
def __init__(self, root: str = '/tmp/agent_sandbox'):
self.root = Path(root)
self.root.mkdir(parents=True, exist_ok=True)
def _resolve_path(self, path: str) -> Path:
path = Path(path).expanduser()
if not path.is_absolute():
path = self.root / path
# Ensure path is within sandbox
try:
path.resolve().relative_to(self.root.resolve())
except ValueError:
raise SecurityError(f"Path {path} escapes sandbox")
return path
Process Management
class ProcessManager:
def __init__(self):
self.processes: dict[str, ProcessInfo] = {}
async def spawn(self, command: str, args: list[str] = None,
env: dict = None, cwd: str = None) -> ProcessHandle:
proc = await asyncio.create_subprocess_exec(
command, *(args or []),
env={**os.environ, **(env or {})},
cwd=cwd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
handle = ProcessHandle(
pid=proc.pid,
command=command,
args=args,
process=proc,
stdout=asyncio.Queue(),
stderr=asyncio.Queue()
)
# Start output readers
asyncio.create_task(self._read_stream(proc.stdout, handle.stdout))
asyncio.create_task(self._read_stream(proc.stderr, handle.stderr))
self.processes[str(proc.pid)] = ProcessInfo(handle=handle, start_time=time.time())
return handle
async def execute(self, command: str, args: list[str] = None,
timeout: float = 30.0) -> ProcessResult:
handle = await self.spawn(command, args)
try:
stdout, stderr = await asyncio.wait_for(
handle.wait(), timeout=timeout
)
return ProcessResult(
exit_code=handle.process.returncode,
stdout=stdout,
stderr=stderr
)
except asyncio.TimeoutError:
handle.process.kill()
raise TimeoutError(f"Command timed out after {timeout}s")
async def kill(self, pid: int):
if pid in self.processes:
self.processes[pid].handle.process.terminate()
await self.processes[pid].handle.process.wait()
del self.processes[pid]
Shell/Terminal Access
class TerminalAgent:
def __init__(self):
self.sessions: dict[str, TerminalSession] = {}
async def create_session(self, session_id: str, shell: str = '/bin/bash') -> TerminalSession:
# Use pty for interactive terminal
master, slave = pty.openpty()
process = await asyncio.create_subprocess_exec(
shell,
stdin=slave,
stdout=slave,
stderr=slave,
preexec_fn=os.setsid
)
session = TerminalSession(
id=session_id,
master_fd=master,
slave_fd=slave,
process=process,
output_buffer=""
)
# Start output reader
asyncio.create_task(self._read_output(session))
self.sessions[session_id] = session
return session
async def send_input(self, session_id: str, data: str):
session = self.sessions[session_id]
os.write(session.master_fd, data.encode())
async def resize(self, session_id: str, rows: int, cols: int):
session = self.sessions[session_id]
fcntl.ioctl(session.master_fd, termios.TIOCSWINSZ,
struct.pack('HHHH', rows, cols, 0, 0))
async def close_session(self, session_id: str):
session = self.sessions.pop(session_id)
os.close(session.master_fd)
session.process.terminate()
await session.process.wait()
class WebTerminalAgent:
"""WebSocket-based terminal for web-based agents"""
def __init__(self):
self.terminal = None
async def start(self, websocket: WebSocket):
self.terminal = await TerminalSession.create_websocket(websocket)
async def handle_message(self, message: str):
if message['type'] == 'input':
await self.terminal.write(message['data'])
elif message['type'] == 'resize':
await self.terminal.resize(message['rows'], message['cols'])
Application Control
Browser Automation
class BrowserAgent:
def __init__(self, headless: bool = True):
self.playwright = None
self.browser = None
self.headless = headless
async def start(self):
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(
headless=self.headless,
args=['--no-sandbox', '--disable-dev-shm-usage']
)
async def new_page(self, url: str = None) -> Page:
context = await self.browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
)
page = await context.new_page()
if url:
await page.goto(url, wait_until='networkidle')
return page
async def close(self):
if self.browser:
await self.browser.close()
if self.playwright:
await self.playwright.stop()
class Page:
async def click(self, selector: str):
await self.page.click(selector)
async def fill(self, selector: str, value: str):
await self.page.fill(selector, value)
async def type(self, selector: str, text: str, delay: int = 50):
await self.page.type(selector, text, delay=delay)
async def wait_for_selector(self, selector: str, timeout: int = 30000):
await self.page.wait_for_selector(selector, timeout=timeout)
async def evaluate(self, script: str) -> Any:
return await self.page.evaluate(script)
async def screenshot(self, path: str = None, full_page: bool = True) -> bytes:
return await self.page.screenshot(path=path, full_page=full_page)
async def pdf(self, path: str = None) -> bytes:
return await self.page.pdf(path=path)
async def wait_for_navigation(self, timeout: int = 30000):
await self.page.wait_for_load_state('networkidle', timeout=timeout)
Office/IDE Control
class VSCodeAgent:
def __init__(self):
self.vscode = None
async def connect(self, port: int = 9222):
"""Connect to VS Code via Chrome DevTools Protocol"""
import websockets
self.ws = await websockets.connect(f'ws://localhost:{port}')
await self._initialize_cdp()
async def open_file(self, path: str):
await self.send_cdp('Runtime.evaluate', {
'expression': f'vscode.workspace.openTextDocument("{path}").then(doc => vscode.window.showTextDocument(doc))'
})
async def get_diagnostics(self, file_path: str) -> list[Diagnostic]:
result = await self.send_cdp('Runtime.evaluate', {
'expression': f'vscode.languages.getDiagnostics(vscode.Uri.file("{file_path}"))'
})
return result['result']['value']
async def apply_code_action(self, file_path: str, action: CodeAction):
await self.send_cdp('Runtime.evaluate', {
'expression': f'vscode.commands.executeCommand("editor.action.codeAction", "{action}")'
})
async def run_task(self, task_name: str):
await self.send_cdp('Runtime.evaluate', {
'expression': f'vscode.tasks.executeTask("{task_name}")'
})
class IntelliJAgent:
"""Control IntelliJ IDEA via HTTP API"""
def __init__(self, host: str = 'localhost', port: int = 63342):
self.base_url = f'http://{host}:{port}'
self.session = aiohttp.ClientSession()
async def open_file(self, path: str):
await self.session.post(f'{self.base_url}/api/file/open', json={'path': path})
async def run_inspection(self, scope: str = 'project') -> list[InspectionResult]:
async with self.session.post(f'{self.base_url}/api/inspect', json={'scope': scope}) as resp:
return await resp.json()
Security & Isolation
Sandboxing
class SandboxManager:
def __init__(self):
self.sandboxes: dict[str, Sandbox] = {}
def create_sandbox(self, sandbox_id: str, config: SandboxConfig) -> Sandbox:
if config.type == 'docker':
sandbox = DockerSandbox(sandbox_id, config)
elif config.type == 'gvisor':
sandbox = GVisorSandbox(sandbox_id, config)
elif config.type == 'firecracker':
sandbox = FirecrackerSandbox(sandbox_id, config)
elif config.type == 'wasm':
sandbox = WasmSandbox(sandbox_id, config)
else:
sandbox = LocalSandbox(sandbox_id, config)
self.sandboxes[sandbox_id] = sandbox
return sandbox
async def execute(self, sandbox_id: str, code: str,
language: str = 'python') -> ExecutionResult:
sandbox = self.sandboxes.get(sandbox_id)
if not sandbox:
raise ValueError(f"Sandbox {sandbox_id} not found")
return await sandbox.execute(code, language)
class DockerSandbox:
def __init__(self, sandbox_id: str, config: SandboxConfig):
self.sandbox_id = sandbox_id
self.config = config
self.container = None
async def start(self):
self.container = docker.run(
image=self.config.image or 'python:3.11-slim',
detach=True,
mem_limit=self.config.memory_limit,
cpu_quota=self.config.cpu_quota,
network_mode='none' if not self.config.network else 'bridge',
security_opt=['no-new-privileges:true'],
read_only=True,
tmpfs={'/tmp': 'rw,noexec,nosuid,size=100m'},
user='1000:1000'
)
async def execute(self, code: str, language: str) -> ExecutionResult:
if language == 'python':
cmd = ['python3', '-c', code]
elif language == 'javascript':
cmd = ['node', '-e', code]
elif language == 'bash':
cmd = ['bash', '-c', code]
else:
raise ValueError(f"Unsupported language: {language}")
exec_result = self.container.exec_run(
cmd,
demux=True,
timeout=30
)
return ExecutionResult(
exit_code=exec_result.exit_code,
stdout=exec_result.output[0].decode() if exec_result.output[0] else '',
stderr=exec_result.output[1].decode() if exec_result.output[1] else ''
)
class GVisorSandbox:
"""Stronger isolation using gVisor"""
async def start(self):
self.container = docker.run(
image='gcr.io/gvisor/containerd:latest',
runtime='runsc',
detach=True,
# gVisor provides stronger isolation
)
class FirecrackerSandbox:
"""MicroVM for strongest isolation"""
async def start(self):
# Use firecracker-containerd or firecracker directly
pass
class WasmSandbox:
"""WebAssembly sandbox for safest execution"""
async def execute(self, code: str, language: str) -> ExecutionResult:
# Compile to WASM and run in wasmtime/Wasmer
pass
Permission Model
class PermissionManager:
def __init__(self):
self.policies: dict[str, PermissionPolicy] = {}
def define_policy(self, agent_id: str, policy: PermissionPolicy):
self.policies[agent_id] = policy
async def check_permission(self, agent_id: str, action: Action) -> PermissionResult:
policy = self.policies.get(agent_id)
if not policy:
return PermissionResult(allowed=False, reason="No policy defined")
return await policy.evaluate(action)
class PermissionPolicy:
def __init__(self):
self.rules: list[PermissionRule] = []
self.default_deny = True
def allow(self, action: str, resource: str = '*', conditions: dict = None):
self.rules.append(PermissionRule(
action=action, resource=resource, effect='allow', conditions=conditions
))
def deny(self, action: str, resource: str = '*', conditions: dict = None):
self.rules.append(PermissionRule(
action=action, resource=resource, effect='deny', conditions=conditions
))
async def evaluate(self, action: Action) -> PermissionResult:
for rule in self.rules:
if rule.matches(action):
if rule.effect == 'allow':
return PermissionResult(allowed=True)
else:
return PermissionResult(allowed=False, reason=rule.reason)
return PermissionResult(allowed=self.default_allow)
# Default policies
DEFAULT_POLICIES = {
'web_agent': PermissionPolicy(
rules=[
PermissionRule(action='browser.*', effect='allow'),
PermissionRule(action='file.read', resource='~/downloads/*', effect='allow'),
PermissionRule(action='file.write', resource='~/downloads/*', effect='allow'),
PermissionRule(action='network.*', resource='api.*', effect='allow'),
PermissionRule(action='*', resource='*', effect='deny')
])
],
'coding_agent': PermissionPolicy(
rules=[
PermissionRule(action='file.*', resource='~/projects/*', effect='allow'),
PermissionRule(action='shell.*', resource='~/projects/*', effect='allow'),
PermissionRule(action='git.*', effect='allow'),
PermissionRule(action='network.*', effect='deny'),
]
),
'research_agent': PermissionPolicy(
rules=[
PermissionRule(action='web.search', effect='allow'),
PermissionRule(action='web.scrape', effect='allow'),
PermissionRule(action='file.*', resource='~/research/*', effect='allow'),
PermissionRule(action='network.*', resource='api.semanticscholar.org', effect='allow'),
]
)
}
Remote Access & Collaboration
Remote Desktop (VNC/RDP)
class RemoteDesktopAgent:
def __init__(self):
self.vnc_clients: dict[str, VNCClient] = {}
self.rdp_clients: dict[str, RDPClient] = {}
async def connect_vnc(self, host: str, port: int = 5900,
password: str = None) -> VNCSession:
client = VNCClient(host, port, password)
await client.connect()
session = VNCSession(client=client)
self.vnc_clients[session.id] = session
return session
async def connect_rdp(self, host: str, port: int = 3389,
username: str = None, password: str = None) -> RDPSession:
client = RDPClient(host, port, username, password)
await client.connect()
session = RDPSession(client=client)
self.rdp_clients[session.id] = session
return session
class VNCSession:
def __init__(self, client: VNCClient):
self.client = client
self.id = str(uuid4())
async def capture(self) -> ScreenCapture:
return await self.client.capture()
async def send_key(self, key: str):
await self.client.send_key(key)
async def click(self, x: int, y: int, button: str = 'left'):
await self.client.click(x, y, button)
async def type_text(self, text: str):
await self.client.type_text(text)
Collaborative Editing
class CollaborativeEditor:
def __init__(self):
self.documents: dict[str, CollaborativeDocument] = {}
async def create_document(self, doc_id: str, content: str = '') -> CollaborativeDocument:
doc = CollaborativeDocument(doc_id, content)
self.documents[doc_id] = doc
return doc
async def join_document(self, doc_id: str, user: User) -> DocumentSession:
doc = self.documents.get(doc_id)
if not doc:
raise ValueError(f"Document {doc_id} not found")
session = DocumentSession(doc, user)
await doc.join(user)
return session
class CollaborativeDocument:
def __init__(self, doc_id: str, content: str):
self.doc_id = doc_id
self.content = content
self.version = 0
self.operations: list[Operation] = []
self.cursors: dict[str, Cursor] = {}
self.lock = asyncio.Lock()
async def apply_operation(self, user_id: str, operation: Operation) -> OperationResult:
async with self.lock:
# Transform operation against concurrent ops
transformed = self.transform_operation(operation)
# Apply
self.apply_operation(transformed)
self.operations.append(transformed)
self.version += 1
# Broadcast to other users
await self.broadcast(transformed)
return OperationResult(version=self.version, success=True)
def transform_operation(self, op: Operation) -> Operation:
# Operational Transformation (OT)
for concurrent_op in self.operations[op.base_version:]:
op = self.transform(op, concurrent_op)
return op
Observability
Screen Recording
class ScreenRecorder:
def __init__(self, output_dir: str = './recordings'):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.recording: RecordingSession = None
async def start_recording(self, session_id: str,
fps: int = 30,
region: Region = None) -> RecordingSession:
if self.recording:
raise RuntimeError("Recording already in progress")
self.recording = RecordingSession(
id=str(uuid4()),
session_id=session_id,
output_path=self.output_dir / f"{session_id}_{datetime.utcnow().isoformat()}.mp4",
fps=fps,
region=region
)
# Start capture loop
self.recording.task = asyncio.create_task(self._capture_loop())
return self.recording
async def _capture_loop(self):
interval = 1.0 / self.recording.fps
while self.recording and self.recording.active:
start = time.time()
frame = await self.capture.capture(self.recording.region)
await self.recording.writer.write_frame(frame)
elapsed = time.time() - start
await asyncio.sleep(max(0, interval - elapsed))
async def stop_recording(self) -> RecordingResult:
if not self.recording:
raise RuntimeError("No recording in progress")
self.recording.active = False
await self.recording.task
await self.recording.writer.close()
result = RecordingResult(
path=self.recording.output_path,
duration=time.time() - self.recording.start_time,
frames=self.recording.frame_count
)
self.recording = None
return result
Action Logging
class ActionLogger:
def __init__(self, storage: LogStorage):
self.storage = storage
self.buffer: list[ActionLog] = []
self.flush_interval = 5.0
self._flush_task = asyncio.create_task(self._periodic_flush())
async def log(self, log: ActionLog):
self.buffer.append(log)
async def _periodic_flush(self):
while True:
await asyncio.sleep(self.flush_interval)
if self.buffer:
await self.storage.write_batch(self.buffer)
self.buffer.clear()
async def log_action(self, agent_id: str, action: Action,
result: ActionResult, context: dict = None):
log = ActionLog(
id=str(uuid4()),
timestamp=datetime.utcnow(),
agent_id=agent_id,
action=action,
result=result,
context=context or {}
)
await self.log(action_log)
async def query_logs(self, query: LogQuery) -> list[ActionLog]:
return await self.storage.query(query)
class ActionLog:
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
def to_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}
Performance Monitoring
Agent Metrics
class AgentMetrics:
def __init__(self, agent_id: str, prometheus: PrometheusClient = None):
self.agent_id = agent_id
self.prometheus = prometheus
# Prometheus metrics
self.request_duration = Histogram(
'agent_request_duration_seconds',
'Request duration',
['agent', 'operation']
)
self.tokens_used = Counter(
'agent_tokens_total',
'Tokens used',
['agent', 'type']
)
self.errors = Counter(
'agent_errors_total',
'Total errors',
['agent', 'error_type']
)
self.active_tasks = Gauge(
'agent_active_tasks',
'Active tasks',
['agent']
)
@contextmanager
def measure(self, operation: str):
start = time.time()
self.active_tasks.inc()
try:
yield
self.request_duration.labels(
agent=self.agent_id, operation=operation
).observe(time.time() - start)
except Exception as e:
self.errors.labels(
agent=self.agent_name,
error_type=type(e).__name__
).inc()
raise
finally:
self.active_tasks.dec()
Resource Monitoring
class ResourceMonitor:
def __init__(self, interval: float = 5.0):
self.interval = interval
self.metrics_history: dict[str, list[ResourceMetrics]] = defaultdict(list)
self._monitor_task = None
async def start(self):
self._monitor_task = asyncio.create_task(self._monitor_loop())
async def _monitor_loop(self):
while True:
metrics = await self.collect_metrics()
for agent_id, metrics in metrics.items():
self.metrics_history[agent_id].append(metrics)
# Keep last 1000 points
if len(self.metrics_history[agent_id]) > 1000:
self.metrics_history[agent_id] = self.metrics_history[agent_id][-1000:]
await asyncio.sleep(self.interval)
async def collect_metrics(self) -> dict[str, ResourceMetrics]:
metrics = {}
for agent_id, agent in self.agents.items():
process = psutil.Process(agent.pid) if agent.pid else None
if process:
metrics[agent_id] = ResourceMetrics(
cpu_percent=process.cpu_percent(),
memory_mb=process.memory_info().rss / 1024 / 1024,
memory_percent=process.memory_percent(),
threads=process.num_threads(),
open_files=len(process.open_files()),
connections=len(process.connections()),
timestamp=datetime.utcnow()
)
return metrics
Related Notes
- Function Calling — ACI as tool
- Browser Automation — Browser automation
- Code Execution — Code execution in sandbox
- MCP Protocol — MCP for ACI
- API Integration — API integration
- Secure Coding — Secure ACI
- Supply Chain Security — Supply chain security
- Security — Production security
- Reliability — Reliability patterns
- Observability — Observability
- Reliability — Reliability
- Deployment Architecture — Deployment architecture
- Coding Agents — Coding agents with ACI
- Customer Facing Agents — Customer-facing ACI
- Agent Societies — Multi-agent ACI
- Agent Societies — Multi-agent ACI
- Research Directions — ACI research
- Security — Security hardening
- Reliability — Reliability
- Observability — Observability
- Research Directions — ACI research
Sources: Playwright, Puppeteer, Selenium, Selenium, CDP, Chrome DevTools Protocol, VNC, RFB, RDP, FreeRDP, X11, Xlib, Quartz, Windows UI Automation, UI Automation, Docker, gVisor, Firecracker, Kata Containers, Kata Containers, Wasm, Wasmtime, Wasmer, WASI, WebAssembly, eBPF, bpftrace, strace, ltrace, perf, SystemTap, Frida, Frida, DynamoRIO, Pin, Intel PT, Intel PT, Intel PT, Frida, DynamoRIO, Pin, Intel PT, QEMU, QEMU, KVM, KVM, Xen, Xen, VirtualBox, VMware, Hyper-V, Parallels, UTM, Limactl, OrbStack, Colima, Rancher Desktop, Docker Desktop, Podman, Podman, Podman, Buildah, Buildah, BuildKit, BuildKit, BuildKit, Kaniko, Kaniko, img, img, img, BuildKit, BuildKit, BuildKit, Kaniko, Kaniko, img, img, img.