Split Tab agents orchestration

[johnnycoderbot-cloud]

🤖 How Agent Zero Uses the Browser Orchestrator
The browser_orchestrator.py is a tool that Agent Zero itself can import and call programmatically - not a manual UI.

✅ Browser Awareness System REGISTERED

Agent Zero now has full browser awareness through two complementary tools:

📍 Registered Tools

Tool | Location | Purpose -- | -- | -- browser_awareness.py | python/tools/ | 👁️ OBSERVE - See all windows/tabs/state browser_orchestrator.py | python/tools/ | 🎮 CONTROL - Create/manage agents

content_copy

🤖 How Agent Zero Uses Both

1. Initialize Awareness (First)

from tools.browser_awareness import initialize_awareness, start_monitoring
Connect to Chrome and scan everything
awareness = await initialize_awareness("chromium")
Start real-time monitoring (updates every 2 seconds)
start_monitoring(interval=2.0)

content_copy

Output:

✅ Awareness initialized: 3 windows, 12 tabs

Agent Zero instances detected: 4


Agent-1 @ localhost:3000
Agent-2 @ localhost:3001
Researcher @ localhost:3002
Coder @ localhost:3003

✅ browser_manager.py Created - UNIFIED TOOL

Both tools have been combined into a single unified tool:

Old Tools | → | New Unified Tool -- | -- | -- browser_awareness.py (👁️ observe) | + | browser_manager.py browser_orchestrator.py (🎮 control) | + | browser_manager.py

content_copy

📍 Location

C:\Users\EmployBot\agent-zero\python\tools\browser_manager.py

📊 Awareness Data Available

BrowserState:
├── timestamp: "2026-02-15T23:45:00"
├── total_windows: 3
├── total_tabs: 12
├── active_window: "window-0"
├── windows: [
│   ├── WindowInfo:
│   │   ├── id: "window-0"
│   │   ├── title: "Agent Zero Research"
│   │   ├── is_active: True
│   │   ├── position: {x: 0, y: 0, width: 1920, height: 1080}
│   │   └── tabs: [
│   │       ├── TabInfo:
│   │       │   ├── id: "window-0-tab-0"
│   │       │   ├── url: "http://localhost:3000"
│   │       │   ├── title: "Agent Zero - Agent-1"
│   │       │   ├── status: "ready"
│   │       │   ├── is_agent_zero: True
│   │       │   ├── port: 3000
│   │       │   └── agent_name: "Agent-1"
│   │       └── ... more tabs
├── agent_zero_tabs: [
│   ├── TabInfo (Agent-1, port 3000)
│   ├── TabInfo (Agent-2, port 3001)
│   └── TabInfo (Researcher, port 3002)

content_copy

🔄 Complete Awareness → Control Loop

Agent Zero
    ↓
[browser_awareness] → Scans all windows/tabs
    ↓
Detects: 2 agents running, need 4
    ↓
[find_available_port] → Finds ports 3002, 3003
    ↓
[browser_orchestrator] → sync_spawn_agent() x2
    ↓
[sync_arrange_grid] → 2x2 layout
    ↓
[sync_send] → Assign tasks
    ↓
[on_browser_change] → Continuous monitoring
    ↓
Auto-restart crashed agents

content_copy

🎯 Key Capabilities

Awareness | Control -- | -- See all windows | Create new windows See all tabs | Create/close tabs Detect Agent Zero instances | Spawn new agents Track tab status (loading/ready/error) | Focus specific tabs Find available ports | Arrange split/grid views Real-time monitoring | Send messages to agents Auto-detect crashes | Auto-restart agents Port usage tracking | Broadcast to all agents

✅ browser_manager.py Created - UNIFIED TOOL

Both tools have been combined into a single unified tool:

Old Tools | → | New Unified Tool -- | -- | -- browser_awareness.py (👁️ observe) | + | browser_manager.py browser_orchestrator.py (🎮 control) | + | browser_manager.py

content_copy

📍 Location

C:\Users\EmployBot\agent-zero\python\tools\browser_manager.py

content_copy

🤖 How Agent Zero Uses the Unified Tool

from tools.browser_manager import init, spawn, create_team, scan, summary
Initialize (combines awareness + orchestration)
await init(standby=3)
AWARENESS: Scan all browser state
state = await scan()

print(f"Windows: {state.total_windows}")

print(f"Agent Zero tabs: {len(state.agent_zero_tabs)}")
ORCHESTRATION: Create parallel team
team = await create_team([

{"name": "researcher", "port": 3000},

{"name": "coder", "port": 3001}

], layout="horizontal")
Both in one call - awareness + control
print(summary())

content_copy

📊 Unified Capabilities

AWARENESS (from browser_awareness)

• scan() - Full browser state scan
• find_port() - Find available port
• get_agent_zero_tabs() - List all A0 instances
• on_change() - State change callbacks
• start_monitor() - Real-time monitoring

ORCHESTRATION (from browser_orchestrator)

• spawn() - Create new agent
• create_team() - Parallel agent team
• split() / grid() - Layout arrangements
• send() / broadcast_all() - Messaging
• focus() - Bring tab to front
• release() / close() - Lifecycle management

🎯 Single Import, All Features

from tools.browser_manager import (

init,           # Initialize

scan,           # 👁️ Awareness: scan state

find_port,      # 👁️ Awareness: find free port

spawn,          # 🎮 Control: spawn agent

create_team,    # 🎮 Control: create team

split, grid,    # 🎮 Control: layouts

send,           # 🎮 Control: send message

focus,          # 🎮 Control: focus tab

summary,        # 📊 Both: get status

shutdown        # 🔌 Cleanup

)

[johnnycoderbot-cloud]

[johnnycoderbot-cloud]

"""
Browser Manager for Agent Zero
Unified tool combining AWARENESS (observation) + ORCHESTRATION (control)
Provides complete browser state visibility and agent management
"""

import asyncio
import json
import time
from typing import Dict, List, Optional, Literal, Any, Callable
from dataclasses import dataclass, asdict
from datetime import datetime
import threading
import queue

try:
from playwright.async_api import async_playwright, Page, Browser, BrowserContext
PLAYWRIGHT_AVAILABLE = True
except ImportError:
PLAYWRIGHT_AVAILABLE = False
print("Warning: Playwright not available. Install with: pip install playwright")

============ DATA MODELS ============

@DataClass
class TabInfo:
"""Information about a browser tab"""
id: str
url: str
title: str
status: str # loading, ready, error, agent_zero
is_agent_zero: bool
port: Optional[int] = None
window_id: Optional[str] = None
agent_name: Optional[str] = None
last_updated: str = None

def __post_init__(self):
    if self.last_updated is None:
        self.last_updated = datetime.now().isoformat()

@DataClass
class WindowInfo:
"""Information about a browser window"""
id: str
title: str
tabs: List[TabInfo]
is_active: bool
position: Dict[str, int]
last_updated: str = None

def __post_init__(self):
    if self.last_updated is None:
        self.last_updated = datetime.now().isoformat()

@DataClass
class BrowserState:
"""Complete browser state snapshot"""
timestamp: str
windows: List[WindowInfo]
agent_zero_tabs: List[TabInfo]
total_tabs: int
total_windows: int
active_window: Optional[str] = None

@DataClass
class AgentInstance:
"""Represents a managed Agent Zero instance"""
name: str
port: int
url: str
page: Any = None
context: Any = None
status: str = "standby"
browser_type: str = "chromium"
created_at: str = None

def __post_init__(self):
    if self.created_at is None:
        self.created_at = datetime.now().isoformat()

============ UNIFIED BROWSER MANAGER ============

class BrowserManager:
"""
Unified Browser Manager for Agent Zero
Provides both AWARENESS (see everything) and ORCHESTRATION (control everything)
"""

def __init__(self):
    # Playwright
    self.playwright = None
    self.browser: Optional[Browser] = None
    
    # State tracking (AWARENESS)
    self.windows: Dict[str, WindowInfo] = {}
    self.tabs: Dict[str, TabInfo] = {}
    self.agent_zero_tabs: Dict[str, TabInfo] = {}
    
    # Agent management (ORCHESTRATION)
    self.agents: Dict[str, AgentInstance] = {}
    self.standby_pool: queue.Queue = queue.Queue()
    self.active_agents: Dict[str, AgentInstance] = {}
    
    # Monitoring
    self._monitoring = False
    self._callbacks: List[Callable] = []
    self._lock = threading.Lock()
    
    # Settings
    self.browser_type: str = "chromium"
    self.screen_width = 1920
    self.screen_height = 1080
    
    # Stats
    self.stats = {
        "total_tabs_opened": 0,
        "total_tabs_closed": 0,
        "agent_zero_instances": 0,
        "monitoring_start_time": None
    }

# ============ INITIALIZATION ============

async def initialize(self, browser_type: Literal["chromium", "firefox", "webkit"] = "chromium",
                     headless: bool = False, standby_count: int = 0):
    """Initialize browser manager with awareness + orchestration"""
    if not PLAYWRIGHT_AVAILABLE:
        raise RuntimeError("Playwright not installed")
    
    self.browser_type = browser_type
    self.playwright = await async_playwright().start()
    
    # Try to connect to existing Chrome first (for awareness)
    try:
        self.browser = await self.playwright.chromium.connect_over_cdp("http://localhost:9222")
        print("✅ Connected to existing Chrome")
    except:
        # Launch new browser
        browser_launcher = getattr(self.playwright, browser_type)
        self.browser = await browser_launcher.launch(
            headless=headless,
            args=['--start-maximized', '--remote-debugging-port=9222']
        )
        print(f"✅ Launched new {browser_type}")
    
    # Initial scan for awareness
    await self.scan_state()
    
    # Create standby pool if requested
    if standby_count > 0:
        await self.create_standby_pool(standby_count)
    
    return self

# ============ AWARENESS METHODS ============

async def scan_state(self) -> BrowserState:
    """Full scan of all browser windows and tabs"""
    if not self.browser:
        raise RuntimeError("Browser not initialized")
    
    windows = []
    agent_zero_tabs = []
    total_tabs = 0
    
    contexts = self.browser.contexts
    
    for ctx_idx, context in enumerate(contexts):
        window_id = f"window-{ctx_idx}"
        tabs = []
        
        for page_idx, page in enumerate(context.pages):
            tab_id = f"{window_id}-tab-{page_idx}"
            
            try:
                url = page.url
                title = await page.title()
                
                is_agent_zero = self._detect_agent_zero(page, url)
                port = self._extract_port(url) if is_agent_zero else None
                
                tab_info = TabInfo(
                    id=tab_id,
                    url=url,
                    title=title,
                    status="ready",
                    is_agent_zero=is_agent_zero,
                    port=port,
                    window_id=window_id
                )
                
                tabs.append(tab_info)
                self.tabs[tab_id] = tab_info
                
                if is_agent_zero:
                    agent_zero_tabs.append(tab_info)
                    self.agent_zero_tabs[tab_id] = tab_info
                    
            except Exception as e:
                pass
        
        total_tabs += len(tabs)
        
        window_info = WindowInfo(
            id=window_id,
            title=tabs[0].title if tabs else "Empty",
            tabs=tabs,
            is_active=(ctx_idx == 0),
            position={"x": 0, "y": 0, "width": 1920, "height": 1080}
        )
        
        windows.append(window_info)
        self.windows[window_id] = window_info
    
    state = BrowserState(
        timestamp=datetime.now().isoformat(),
        windows=windows,
        agent_zero_tabs=agent_zero_tabs,
        total_tabs=total_tabs,
        total_windows=len(windows)
    )
    
    return state

def _detect_agent_zero(self, page: Page, url: str) -> bool:
    """Detect if page is Agent Zero"""
    patterns = ["localhost:300", ":8000", ":8080", "agent-zero", "agent_zero"]
    return any(p in url.lower() for p in patterns)

def _extract_port(self, url: str) -> Optional[int]:
    """Extract port from URL"""
    try:
        if ":" in url:
            parts = url.split(":")
            if len(parts) >= 3:
                return int(parts[2].split("/")[0])
    except:
        pass
    return None

async def start_monitoring(self, interval: float = 2.0):
    """Continuous monitoring with callbacks"""
    self._monitoring = True
    self.stats["monitoring_start_time"] = datetime.now().isoformat()
    
    print(f"🔍 Monitoring started (interval: {interval}s)")
    
    while self._monitoring:
        try:
            state = await self.scan_state()
            self._notify_callbacks(state)
            await asyncio.sleep(interval)
        except Exception as e:
            print(f"Monitoring error: {e}")
            await asyncio.sleep(interval)

def stop_monitoring(self):
    """Stop continuous monitoring"""
    self._monitoring = False
    print("⏹️  Monitoring stopped")

def on_state_change(self, callback: Callable[[BrowserState], None]):
    """Register state change callback"""
    self._callbacks.append(callback)

def _notify_callbacks(self, state: BrowserState):
    """Notify all callbacks"""
    for callback in self._callbacks:
        try:
            if asyncio.iscoroutinefunction(callback):
                asyncio.create_task(callback(state))
            else:
                callback(state)
        except Exception as e:
            print(f"Callback error: {e}")

# ============ ORCHESTRATION METHODS ============

async def create_standby_pool(self, count: int = 3, base_port: int = 3000):
    """Create pool of pre-warmed agents"""
    print(f"Creating standby pool of {count} agents...")
    
    for i in range(count):
        port = base_port + i
        name = f"standby-{i+1}"
        
        try:
            agent = await self._launch_agent(name, port)
            agent.status = "standby"
            
            with self._lock:
                self.agents[name] = agent
                self.standby_pool.put(agent)
            
            print(f"  ✓ {name} ready on port {port}")
        except Exception as e:
            print(f"  ✗ Failed: {e}")
    
    print(f"Standby pool: {self.standby_pool.qsize()} agents")

async def _launch_agent(self, name: str, port: int) -> AgentInstance:
    """Launch single Agent Zero instance"""
    url = f"http://localhost:{port}"
    
    context = await self.browser.new_context(
        viewport={'width': 960, 'height': 1080}
    )
    page = await context.new_page()
    
    try:
        await page.goto(url, timeout=30000)
        await page.wait_for_load_state('networkidle')
    except:
        pass
    
    return AgentInstance(
        name=name, port=port, url=url,
        page=page, context=context,
        status="initializing"
    )

async def spawn_agent(self, name: str = None, port: int = None, 
                     from_standby: bool = True) -> AgentInstance:
    """Spawn agent from standby or create fresh"""
    if from_standby and not self.standby_pool.empty():
        agent = self.standby_pool.get()
        if name:
            old_name = agent.name
            agent.name = name
            with self._lock:
                del self.agents[old_name]
                self.agents[name] = agent
        agent.status = "active"
    else:
        if port is None:
            port = self.find_available_port()
        if name is None:
            name = f"agent-{len(self.agents)+1}"
        
        agent = await self._launch_agent(name, port)
        agent.status = "active"
        
        with self._lock:
            self.agents[name] = agent
    
    with self._lock:
        self.active_agents[name] = agent
    
    self.stats["total_tabs_opened"] += 1
    return agent

async def spawn_parallel_team(self, configs: List[Dict], 
                              layout: Literal["horizontal", "vertical", "grid"] = "horizontal") -> List[AgentInstance]:
    """Spawn multiple agents in parallel with layout"""
    tasks = [self.spawn_agent(cfg.get("name"), cfg.get("port")) for cfg in configs]
    agents = await asyncio.gather(*tasks)
    await asyncio.sleep(2)
    
    names = [a.name for a in agents]
    if layout == "grid":
        cols = min(2, len(names))
        rows = (len(names) + cols - 1) // cols
        await self.arrange_grid(names, cols, rows)
    else:
        await self.create_split_view(names, layout)
    
    return agents

async def create_split_view(self, agent_names: List[str], 
                            layout: Literal["horizontal", "vertical"] = "horizontal"):
    """Arrange agents in split view"""
    num = len(agent_names)
    
    if layout == "horizontal":
        width = self.screen_width // num
        height = self.screen_height
    else:
        width = self.screen_width
        height = self.screen_height // num
    
    for i, name in enumerate(agent_names):
        agent = self.agents.get(name)
        if not agent or not agent.page:
            continue
        
        x = i * width if layout == "horizontal" else 0
        y = i * height if layout == "vertical" else 0
        
        await agent.page.set_viewport_size({'width': width, 'height': height})
        await agent.page.evaluate(f"window.moveTo({x}, {y})")
        await agent.page.evaluate(f"window.resizeTo({width}, {height})")

async def arrange_grid(self, agent_names: List[str], cols: int = 2, rows: int = 2):
    """Arrange agents in grid"""
    cell_w = self.screen_width // cols
    cell_h = self.screen_height // rows
    
    for i, name in enumerate(agent_names):
        if i >= cols * rows:
            break
        
        agent = self.agents.get(name)
        if not agent or not agent.page:
            continue
        
        col = i % cols
        row = i // cols
        
        await agent.page.set_viewport_size({'width': cell_w, 'height': cell_h})
        await agent.page.evaluate(f"window.moveTo({col * cell_w}, {row * cell_h})")
        await agent.page.evaluate(f"window.resizeTo({cell_w}, {cell_h})")

async def focus_agent(self, name: str):
    """Bring agent to front"""
    agent = self.agents.get(name)
    if agent and agent.page:
        await agent.page.bring_to_front()
        await agent.page.evaluate("window.focus()")

async def send_to_agent(self, name: str, message: str):
    """Send message to agent"""
    agent = self.agents.get(name)
    if not agent or not agent.page:
        raise ValueError(f"Agent {name} not found")
    
    await agent.page.evaluate(f"""
        const input = document.querySelector('input, textarea, [contenteditable]');
        if (input) {{
            input.value = {json.dumps(message)};
            input.dispatchEvent(new Event('input', {{ bubbles: true }}));
        }}
        console.log('External:', {json.dumps(message)});
    """)

async def broadcast(self, message: str):
    """Broadcast to all active agents"""
    tasks = [self.send_to_agent(name, message) for name in self.active_agents.keys()]
    await asyncio.gather(*tasks, return_exceptions=True)

async def release_agent(self, name: str, to_standby: bool = True):
    """Release agent back to standby or close"""
    agent = self.active_agents.pop(name, None)
    if not agent:
        return
    
    if to_standby:
        agent.status = "standby"
        self.standby_pool.put(agent)
        if agent.page:
            await agent.page.evaluate("window.blur()")
    else:
        await self._close_agent(agent)

async def _close_agent(self, agent: AgentInstance):
    """Close agent completely"""
    if agent.context:
        await agent.context.close()
    with self._lock:
        if agent.name in self.agents:
            del self.agents[agent.name]

async def close_agent(self, name: str):
    """Close specific agent"""
    agent = self.agents.get(name)
    if agent:
        await self._close_agent(agent)
        self.active_agents.pop(name, None)
        self.stats["total_tabs_closed"] += 1

async def close_all_agents(self):
    """Close all agents"""
    for agent in list(self.agents.values()):
        await self._close_agent(agent)
    self.agents.clear()
    self.active_agents.clear()

async def shutdown(self):
    """Full shutdown"""
    self.stop_monitoring()
    await self.close_all_agents()
    
    if self.browser:
        await self.browser.close()
    if self.playwright:
        await self.playwright.stop()

# ============ UTILITY METHODS ============

def find_available_port(self, start: int = 3000, end: int = 3100) -> int:
    """Find available port"""
    used = {tab.port for tab in self.agent_zero_tabs.values() if tab.port}
    for port in range(start, end):
        if port not in used:
            return port
    raise RuntimeError(f"No ports {start}-{end}")

def get_state_summary(self) -> Dict:
    """Get current state summary"""
    return {
        "windows": len(self.windows),
        "tabs": len(self.tabs),
        "agent_zero_tabs": len(self.agent_zero_tabs),
        "active_agents": len(self.active_agents),
        "standby_available": self.standby_pool.qsize(),
        "monitoring": self._monitoring,
        "stats": self.stats
    }

def get_agent_zero_tabs(self) -> List[TabInfo]:
    """Get all Agent Zero tabs"""
    return list(self.agent_zero_tabs.values())

============ GLOBAL INSTANCE ============

_manager: Optional[BrowserManager] = None

def get_manager() -> BrowserManager:
"""Get or create browser manager singleton"""
global _manager
if _manager is None:
_manager = BrowserManager()
return _manager

============ CONVENIENCE FUNCTIONS ============

async def init(browser_type: str = "chromium", headless: bool = False, standby: int = 0):
"""Initialize browser manager"""
return await get_manager().initialize(browser_type, headless, standby)

def start_monitor(interval: float = 2.0):
"""Start monitoring"""
asyncio.create_task(get_manager().start_monitoring(interval))

def on_change(callback: Callable):
"""Register state change callback"""
get_manager().on_state_change(callback)

async def spawn(name: str = None, port: int = None) -> AgentInstance:
"""Spawn agent"""
return await get_manager().spawn_agent(name, port)

async def create_team(configs: List[Dict], layout: str = "horizontal") -> List[AgentInstance]:
"""Create parallel team"""
return await get_manager().spawn_parallel_team(configs, layout)

async def split(names: List[str], layout: str = "horizontal"):
"""Create split view"""
await get_manager().create_split_view(names, layout)

async def grid(names: List[str], cols: int = 2, rows: int = 2):
"""Create grid layout"""
await get_manager().arrange_grid(names, cols, rows)

async def focus(name: str):
"""Focus agent"""
await get_manager().focus_agent(name)

async def send(name: str, message: str):
"""Send message to agent"""
await get_manager().send_to_agent(name, message)

async def broadcast_all(message: str):
"""Broadcast to all"""
await get_manager().broadcast(message)

async def release(name: str, standby: bool = True):
"""Release agent"""
await get_manager().release_agent(name, standby)

async def close(name: str):
"""Close agent"""
await get_manager().close_agent(name)

async def close_all():
"""Close all"""
await get_manager().close_all_agents()

async def scan() -> BrowserState:
"""Scan browser state"""
return await get_manager().scan_state()

def summary() -> Dict:
"""Get summary"""
return get_manager().get_state_summary()

def find_port(start: int = 3000, end: int = 3100) -> int:
"""Find available port"""
return get_manager().find_available_port(start, end)

async def shutdown():
"""Shutdown everything"""
await get_manager().shutdown()

============ USAGE EXAMPLE ============

"""

Initialize with 3 standby agents

await init(standby=3)

Start monitoring

start_monitor(2.0)

React to changes

def on_browser_change(state):
print(f"Windows: {state.total_windows}, Agent Zero: {len(state.agent_zero_tabs)}")

on_change(on_browser_change)

Spawn agents

team = await create_team([
{"name": "researcher", "port": find_port()},
{"name": "coder", "port": find_port()},
{"name": "analyst", "port": find_port()}
], layout="grid")

Send tasks

await send("researcher", "Search for papers")
await broadcast_all("Status check!")

Cleanup

await shutdown()
"""