📊 AgentProfile - Comprehensive Agent Representation

The AgentProfile is a multi-source data structure that consolidates administrator configuration, service-level capabilities, and real-time client telemetry into a unified, dynamically updated representation of each constellation agent.

📋 Overview

The AgentProfile is the primary data structure representing a registered constellation agent. It aggregates information from three distinct sources to provide a comprehensive view of each agent's identity, capabilities, operational status, and hardware characteristics.

For a complete understanding of how agents work in the constellation system, see:

Constellation Overview - Architecture and multi-device coordination
Constellation Agent - Agent behavior and lifecycle

Function	Description
Identity Management	Unique identification and endpoint tracking
Capability Advertisement	Declare supported features and tools
Status Monitoring	Real-time operational state tracking
Resource Profiling	Hardware and system information
Task Assignment	Enable intelligent task routing decisions

🏗️ Structure Definition

Core Dataclass

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime
from enum import Enum

class DeviceStatus(Enum):
    """Device connection status"""
    DISCONNECTED = "disconnected"
    CONNECTING = "connecting"
    CONNECTED = "connected"
    FAILED = "failed"
    REGISTERING = "registering"
    BUSY = "busy"
    IDLE = "idle"

@dataclass
class AgentProfile:
    """
    Device information and capabilities.
    
    Consolidates information from three sources:
    1. User-specified registration (devices.yaml)
    2. Service-level manifest (AIP registration)
    3. Client-side telemetry (DeviceInfoProvider)
    """
    
    # === Identity ===
    device_id: str                          # Unique device identifier
    server_url: str                         # WebSocket endpoint URL
    
    # === Platform & Capabilities ===
    os: Optional[str] = None                # Operating system (windows, linux, darwin)
    capabilities: List[str] = field(default_factory=list)  # Advertised capabilities
    metadata: Dict[str, Any] = field(default_factory=dict) # Multi-source metadata
    
    # === Operational Status ===
    status: DeviceStatus = DeviceStatus.DISCONNECTED  # Current state
    last_heartbeat: Optional[datetime] = None         # Last heartbeat timestamp
    
    # === Connection Management ===
    connection_attempts: int = 0            # Connection retry counter
    max_retries: int = 5                    # Maximum retry attempts
    
    # === Task Execution ===
    current_task_id: Optional[str] = None   # Currently executing task ID

🔍 Field Reference

Identity Fields

Field	Type	Source	Description	Example
`device_id`	`str`	User Config	Unique identifier for the device	`"windowsagent"`, `"linux_gpu_01"`
`server_url`	`str`	User Config	WebSocket endpoint of device agent server	`"ws://localhost:5005/ws"`

The device_id must be unique across the entire constellation. Attempting to register a duplicate device_id will fail.

Platform & Capabilities

Field	Type	Source	Description	Example
`os`	`Optional[str]`	User Config + Telemetry	Operating system type	`"windows"`, `"linux"`, `"darwin"`
`capabilities`	`List[str]`	User Config + Telemetry	Advertised capabilities/features	`["gui", "browser", "office"]`
`metadata`	`Dict[str, Any]`	All Sources	Multi-source metadata aggregation	See Metadata Structure

Capabilities Merging:

# Initial capabilities from user config
capabilities = ["web_browsing", "office_applications"]

# After telemetry collection, auto-detected features are merged
# Result: ["web_browsing", "office_applications", "gui", "cli", "browser", "file_system"]

Operational Status

Field	Type	Source	Description	Example
`status`	`DeviceStatus`	Runtime	Current connection/operational state	`DeviceStatus.IDLE`
`last_heartbeat`	`Optional[datetime]`	Runtime	Timestamp of last heartbeat	`2025-11-06T10:30:45Z`

Status Values:

DeviceStatus.DISCONNECTED  # Not connected
DeviceStatus.CONNECTING    # Connection in progress
DeviceStatus.CONNECTED     # WebSocket established
DeviceStatus.REGISTERING   # Performing AIP registration
DeviceStatus.IDLE          # Ready for tasks
DeviceStatus.BUSY          # Executing a task
DeviceStatus.FAILED        # Connection or execution failed

Connection Management

Field	Type	Source	Description	Example
`connection_attempts`	`int`	Runtime	Number of connection attempts made	`0`, `3`
`max_retries`	`int`	User Config	Maximum reconnection attempts before giving up	`5`, `10`

When a device disconnects, the system automatically retries connection up to max_retries times with exponential backoff.

Task Execution

Field	Type	Source	Description	Example
`current_task_id`	`Optional[str]`	Runtime	ID of task currently being executed	`"task_12345"`, `None`

Usage in Task Queue:

# When task is assigned
profile.status = DeviceStatus.BUSY
profile.current_task_id = "task_12345"

# When task completes
profile.status = DeviceStatus.IDLE
profile.current_task_id = None

🗂️ Metadata Structure

The metadata dictionary is a flexible container that aggregates information from all three profiling sources:

Metadata Schema

metadata = {
    # ===== Source 1: User Configuration =====
    "location": str,                    # Physical location
    "performance": str,                 # Performance tier
    "description": str,                 # Human-readable description
    "operation_engineer_email": str,    # Contact information
    "tags": List[str],                  # Custom tags
    # ... any custom user-defined fields
    
    # ===== Source 2: Service Manifest =====
    "platform": str,                    # Platform type (from registration)
    "registration_time": str,           # ISO timestamp of registration
    
    # ===== Source 3: Client Telemetry =====
    "system_info": {
        "platform": str,                # OS platform (windows, linux, darwin)
        "os_version": str,              # OS version string
        "cpu_count": int,               # Number of CPU cores
        "memory_total_gb": float,       # Total RAM in GB
        "hostname": str,                # Device hostname
        "ip_address": str,              # Device IP address
        "platform_type": str,           # Device category (computer, mobile, etc.)
        "schema_version": str           # Telemetry schema version
    },
    "custom_metadata": {                # Optional custom metadata from config
        "datacenter": str,
        "tier": str,
        # ... server-configured metadata
    }
}

Example Metadata

# Complete metadata example from a Windows GPU workstation
metadata = {
    # User Configuration
    "location": "office_desktop",
    "performance": "very_high",
    "description": "Primary Windows workstation with GPU",
    "operation_engineer_email": "admin@example.com",
    "tags": ["production", "gpu-enabled", "high-priority"],
    
    # Service Manifest
    "platform": "windows",
    "registration_time": "2025-11-06T10:30:00.000Z",
    
    # Client Telemetry
    "system_info": {
        "platform": "windows",
        "os_version": "10.0.22631",
        "cpu_count": 16,
        "memory_total_gb": 32.0,
        "hostname": "DESKTOP-GPU01",
        "ip_address": "192.168.1.100",
        "platform_type": "computer",
        "schema_version": "1.0"
    },
    "custom_metadata": {
        "datacenter": "us-west-2",
        "tier": "premium",
        "gpu_type": "NVIDIA RTX 4090",
        "gpu_count": 1
    }
}

🔄 Multi-Source Construction

Three-Source Architecture

graph LR
    A[User Config<br/>devices.yaml]
    B[AIP Registration<br/>Service Manifest]
    C[Device Telemetry<br/>DeviceInfoProvider]
    
    A -->|device_id, server_url<br/>capabilities, metadata| D[AgentProfile]
    B -->|platform, registration_time| D
    C -->|system_info, features| D
    
    style A fill:#e1f5ff
    style B fill:#fff4e1
    style C fill:#e8f5e9
    style D fill:#f3e5f5

Construction Timeline

sequenceDiagram
    participant Config as devices.yaml
    participant Manager as DeviceManager
    participant Server as UFO Server
    participant Telemetry as DeviceInfoProvider
    
    Note over Config,Telemetry: Phase 1: Initial Registration
    Config->>Manager: Load device config
    Manager->>Manager: Create AgentProfile<br/>(device_id, server_url, capabilities)
    
    Note over Config,Telemetry: Phase 2: Service Registration
    Manager->>Server: WebSocket REGISTER
    Server-->>Manager: Add platform, registration_time
    
    Note over Config,Telemetry: Phase 3: Telemetry Collection
    Manager->>Server: request_device_info()
    Server->>Telemetry: collect_system_info()
    Telemetry-->>Server: system_info
    Server-->>Manager: system_info
    Manager->>Manager: Update AgentProfile<br/>(merge system_info & features)

Merging Strategy

1. User Configuration (Priority: Baseline)

# Initial AgentProfile creation
profile = AgentProfile(
    device_id="windowsagent",
    server_url="ws://localhost:5005/ws",
    os="windows",
    capabilities=["web_browsing", "office_applications"],
    metadata={
        "location": "office_desktop",
        "performance": "high"
    }
)

2. Service Manifest (Priority: Override os, Add registration data)

# During AIP registration
profile.metadata.update({
    "platform": "windows",  # From registration message
    "registration_time": "2025-11-06T10:30:00Z"
})

3. Client Telemetry (Priority: Merge capabilities, Add system_info)

# After DeviceInfoProvider collects data
system_info = {
    "platform": "windows",
    "os_version": "10.0.22631",
    "cpu_count": 16,
    "memory_total_gb": 32.0,
    "hostname": "DESKTOP-GPU01",
    "ip_address": "192.168.1.100",
    "supported_features": ["gui", "cli", "browser", "file_system", "office", "windows_apps"],
    "platform_type": "computer"
}

# Update OS if not already set
if not profile.os:
    profile.os = system_info["platform"]

# Merge capabilities (avoid duplicates)
existing_caps = set(profile.capabilities)
new_caps = set(system_info["supported_features"])
profile.capabilities = list(existing_caps.union(new_caps))
# Result: ["web_browsing", "office_applications", "gui", "cli", "browser", "file_system", "windows_apps"]

# Add system_info to metadata
profile.metadata["system_info"] = system_info

📊 Example Profiles

Example 1: Windows GPU Workstation

AgentProfile(
    # Identity
    device_id="gpu_workstation_01",
    server_url="ws://192.168.1.100:5005/ws",
    
    # Platform & Capabilities
    os="windows",
    capabilities=[
        # User-configured
        "web_browsing",
        "office_applications",
        "gpu_computation",
        "model_training",
        # Auto-detected
        "gui",
        "cli",
        "browser",
        "file_system",
        "windows_apps"
    ],
    
    # Metadata
    metadata={
        # User Configuration
        "location": "office_desktop",
        "performance": "very_high",
        "description": "Primary GPU workstation for ML training",
        "operation_engineer_email": "ml-team@example.com",
        "tags": ["production", "gpu", "ml"],
        
        # Service Manifest
        "platform": "windows",
        "registration_time": "2025-11-06T10:30:00Z",
        
        # Client Telemetry
        "system_info": {
            "platform": "windows",
            "os_version": "10.0.22631",
            "cpu_count": 16,
            "memory_total_gb": 64.0,
            "hostname": "DESKTOP-GPU01",
            "ip_address": "192.168.1.100",
            "platform_type": "computer",
            "schema_version": "1.0"
        },
        "custom_metadata": {
            "gpu_type": "NVIDIA RTX 4090",
            "gpu_count": 2,
            "gpu_memory_gb": 48
        }
    },
    
    # Status
    status=DeviceStatus.IDLE,
    last_heartbeat=datetime(2025, 11, 6, 10, 45, 30),
    
    # Connection
    connection_attempts=0,
    max_retries=5,
    
    # Task
    current_task_id=None
)

Profile Summary

graph TB
    subgraph "AgentProfile: gpu_workstation_01"
        A["Status: IDLE<br/>Last Heartbeat: 10:45:30"]
        
        B["System<br/>━━━━━<br/>OS: Windows 10.0.22631<br/>CPU: 16 cores<br/>Memory: 64.0 GB<br/>Host: DESKTOP-GPU01<br/>IP: 192.168.1.100"]
        
        C["Capabilities<br/>━━━━━<br/>• web_browsing<br/>• office_applications<br/>• gpu_computation<br/>• model_training<br/>• gui, cli, browser<br/>• file_system"]
        
        D["Metadata<br/>━━━━━<br/>Location: office_desktop<br/>Performance: very_high<br/>Tags: production, gpu, ml<br/>GPU: 2× NVIDIA RTX 4090"]
    end
    
    style A fill:#e3f2fd
    style B fill:#f3e5f5
    style C fill:#e8f5e9
    style D fill:#fff3e0

Example 2: Linux Server

AgentProfile(
    # Identity
    device_id="linux_server_01",
    server_url="ws://10.0.0.50:5001/ws",
    
    # Platform & Capabilities
    os="linux",
    capabilities=[
        # User-configured
        "server_management",
        "log_monitoring",
        "database_operations",
        # Auto-detected
        "cli",
        "file_system",
        "linux_apps"
    ],
    
    # Metadata
    metadata={
        # User Configuration
        "location": "datacenter_rack_a42",
        "performance": "medium",
        "description": "Production Linux server for backend services",
        "logs_file_path": "/var/log/application.log",
        "dev_path": "/home/deploy/",
        
        # Service Manifest
        "platform": "linux",
        "registration_time": "2025-11-06T09:15:00Z",
        
        # Client Telemetry
        "system_info": {
            "platform": "linux",
            "os_version": "#1 SMP PREEMPT_DYNAMIC Wed Nov 1 15:36:23 UTC 2023",
            "cpu_count": 8,
            "memory_total_gb": 16.0,
            "hostname": "prod-server-01",
            "ip_address": "10.0.0.50",
            "platform_type": "computer",
            "schema_version": "1.0"
        }
    },
    
    # Status
    status=DeviceStatus.BUSY,
    last_heartbeat=datetime(2025, 11, 6, 10, 44, 15),
    
    # Connection
    connection_attempts=0,
    max_retries=3,
    
    # Task
    current_task_id="task_monitoring_567"
)

🔄 Lifecycle Operations

Creation

from galaxy.client.components import DeviceRegistry, AgentProfile, DeviceStatus

registry = DeviceRegistry()

# Create AgentProfile during registration
profile = registry.register_device(
    device_id="windowsagent",
    server_url="ws://localhost:5005/ws",
    os="windows",
    capabilities=["web_browsing", "office"],
    metadata={"location": "office"},
    max_retries=5
)

print(f"Created: {profile.device_id}")
print(f"Status: {profile.status.value}")  # "disconnected"

Status Updates

# Update connection status
registry.update_device_status("windowsagent", DeviceStatus.CONNECTING)
registry.update_device_status("windowsagent", DeviceStatus.CONNECTED)
registry.update_device_status("windowsagent", DeviceStatus.IDLE)

# Set device busy with task
registry.set_device_busy("windowsagent", task_id="task_123")
profile = registry.get_device("windowsagent")
print(f"Status: {profile.status.value}")       # "busy"
print(f"Current Task: {profile.current_task_id}")  # "task_123"

# Set device idle (task complete)
registry.set_device_idle("windowsagent")
profile = registry.get_device("windowsagent")
print(f"Status: {profile.status.value}")       # "idle"
print(f"Current Task: {profile.current_task_id}")  # None

System Info Updates

# Update with telemetry data
system_info = {
    "platform": "windows",
    "os_version": "10.0.22631",
    "cpu_count": 16,
    "memory_total_gb": 32.0,
    "hostname": "DESKTOP-DEV01",
    "ip_address": "192.168.1.100",
    "supported_features": ["gui", "cli", "browser", "file_system", "office"],
    "platform_type": "computer",
    "schema_version": "1.0"
}

registry.update_device_system_info("windowsagent", system_info)

# Verify update
profile = registry.get_device("windowsagent")
print(f"OS: {profile.os}")  # "windows"
print(f"CPU Cores: {profile.metadata['system_info']['cpu_count']}")  # 16
print(f"Memory: {profile.metadata['system_info']['memory_total_gb']} GB")  # 32.0
print(f"Capabilities: {profile.capabilities}")
# ["web_browsing", "office", "gui", "cli", "browser", "file_system"]

Heartbeat Tracking

from datetime import datetime, timezone

# Update heartbeat
registry.update_heartbeat("windowsagent")

profile = registry.get_device("windowsagent")
print(f"Last Heartbeat: {profile.last_heartbeat}")
# 2025-11-06 10:45:30.123456+00:00

Connection Retry Management

# Increment connection attempts
attempts = registry.increment_connection_attempts("windowsagent")
print(f"Attempts: {attempts}/{profile.max_retries}")

# Reset after successful connection
registry.reset_connection_attempts("windowsagent")
profile = registry.get_device("windowsagent")
print(f"Attempts: {profile.connection_attempts}")  # 0

🎯 Usage Patterns

The following patterns demonstrate how AgentProfile is used for intelligent task routing and device management. For more details on task constellation concepts, see Constellation Overview.

Task Assignment Decision

def can_assign_task(profile: AgentProfile, required_capabilities: List[str]) -> bool:
    """
    Check if device can handle a task based on its profile.
    """
    # Check if device is available
    if profile.status != DeviceStatus.IDLE:
        return False
    
    # Check if all required capabilities are supported
    device_caps = set(profile.capabilities)
    required_caps = set(required_capabilities)
    
    if not required_caps.issubset(device_caps):
        return False
    
    # Optional: Check system resources
    system_info = profile.metadata.get("system_info", {})
    if system_info.get("memory_total_gb", 0) < 8:  # Require at least 8GB
        return False
    
    return True

# Usage
profile = registry.get_device("windowsagent")
if can_assign_task(profile, ["browser", "gui"]):
    await manager.assign_task_to_device(
        task_id="task_web_001",
        device_id="windowsagent",
        task_description="Navigate to website and extract data",
        task_data={"url": "https://example.com"}
    )

Device Selection

def select_best_device(
    all_devices: Dict[str, AgentProfile],
    required_capabilities: List[str],
    prefer_high_performance: bool = True
) -> Optional[str]:
    """
    Select the best available device for a task.
    """
    candidates = []
    
    for device_id, profile in all_devices.items():
        # Must be idle
        if profile.status != DeviceStatus.IDLE:
            continue
        
        # Must have required capabilities
        device_caps = set(profile.capabilities)
        if not set(required_capabilities).issubset(device_caps):
            continue
        
        # Calculate score
        score = 0
        if profile.metadata.get("performance") == "very_high":
            score += 10
        elif profile.metadata.get("performance") == "high":
            score += 5
        
        # Prefer devices with more memory
        system_info = profile.metadata.get("system_info", {})
        score += system_info.get("memory_total_gb", 0) / 10
        
        candidates.append((device_id, score))
    
    if not candidates:
        return None
    
    # Sort by score (descending)
    candidates.sort(key=lambda x: x[1], reverse=True)
    return candidates[0][0]

# Usage
all_devices = registry.get_all_devices(connected=True)
best_device = select_best_device(
    all_devices,
    required_capabilities=["gpu_computation", "model_training"],
    prefer_high_performance=True
)
print(f"Selected device: {best_device}")

Health Monitoring

from datetime import datetime, timezone, timedelta

def check_device_health(profile: AgentProfile) -> Dict[str, Any]:
    """
    Check device health based on profile data.
    """
    health = {
        "device_id": profile.device_id,
        "healthy": True,
        "warnings": [],
        "errors": []
    }
    
    # Check heartbeat freshness
    if profile.last_heartbeat:
        age = datetime.now(timezone.utc) - profile.last_heartbeat
        if age > timedelta(minutes=5):
            health["warnings"].append(
                f"No heartbeat for {age.total_seconds():.0f} seconds"
            )
            if age > timedelta(minutes=10):
                health["errors"].append("Heartbeat timeout")
                health["healthy"] = False
    
    # Check connection attempts
    if profile.connection_attempts > profile.max_retries / 2:
        health["warnings"].append(
            f"High connection attempts: {profile.connection_attempts}/{profile.max_retries}"
        )
    
    # Check if device is stuck in BUSY state
    if profile.status == DeviceStatus.BUSY and profile.current_task_id:
        # Would need to check task age here
        health["warnings"].append(f"Device busy with task {profile.current_task_id}")
    
    return health

# Usage
profile = registry.get_device("windowsagent")
health = check_device_health(profile)
print(f"Health: {health['healthy']}")
print(f"Warnings: {health['warnings']}")
print(f"Errors: {health['errors']}")

🔗 Related Documentation

Topic	Document	Description
Overview	Agent Registration Overview	Registration architecture and process
Registration Flow	Registration Flow	Step-by-step registration process
Device Registry	Device Registry	Registry component implementation
Galaxy Devices Config	Galaxy Devices Configuration	YAML configuration reference
Device Info	Device Info Provider	Telemetry collection details
AIP Protocol	AIP Overview	Agent Interaction Protocol
Constellation System	Constellation Overview	Multi-device coordination
WebSocket Client	Client AIP Integration	Client-side implementation

💡 Best Practices

1. Meaningful Capabilities

# ✅ Good: Specific, actionable capabilities
capabilities = ["web_browsing", "office_excel", "file_management", "email_sending"]

# ❌ Bad: Vague capabilities
capabilities = ["desktop", "general"]

2. Rich Metadata

# ✅ Good: Comprehensive metadata for smart routing
metadata = {
    "location": "datacenter_us_west",
    "performance": "high",
    "description": "GPU workstation for ML training",
    "tags": ["production", "gpu", "ml"],
    "operation_engineer_email": "ml-team@example.com"
}

3. Monitor Heartbeats

# Regularly check heartbeat freshness
if profile.last_heartbeat:
    age = datetime.now(timezone.utc) - profile.last_heartbeat
    if age > timedelta(minutes=5):
        logger.warning(f"Device {profile.device_id} heartbeat stale")

4. Use System Info for Resource-Aware Routing

# Check if device has enough resources
system_info = profile.metadata.get("system_info", {})
if system_info.get("memory_total_gb", 0) >= 16:
    # Assign memory-intensive task
    pass

🚀 Next Steps

Learn Registration Process: Read Registration Flow
Configure Devices: See Galaxy Devices Configuration
Understand DeviceRegistry: Check Device Registry
Study Telemetry: Read Device Info Provider

📚 Source Code References

AgentProfile Definition: galaxy/client/components/types.py
DeviceRegistry: galaxy/client/components/device_registry.py
ConstellationDeviceManager: galaxy/client/device_manager.py
DeviceInfoProvider: ufo/client/device_info_provider.py

FilesExpand file tree

agent_profile.md

Latest commit

History