Multi-Agent Systems & Custom Tools

nano ~/agent-zero/.env

# Multi-Agent Configuration
AGENT_MAX_DEPTH=5              # Maximum hierarchy depth
AGENT_MAX_SUBORDINATES=10      # Max subordinates per agent
AGENT_TIMEOUT=300              # Subordinate task timeout (seconds)
AGENT_MEMORY_SHARED=false      # Share memory between agents

cat ~/agent-zero/prompts/default/agent.system.md

from python.helpers.instrument import Instrument, Response

class ExampleInstrument(Instrument):
    """
    Brief description of what this instrument does.
    The agent sees this docstring when deciding whether to use the instrument.
    """
    
    # Define parameters the agent must provide
    async def execute(self, parameter1: str, parameter2: int = 10) -> Response:
        """
        Execute the instrument.
        
        Args:
            parameter1: Description of first parameter
            parameter2: Description of second parameter (default: 10)
            
        Returns:
            Response object with result or error
        """
        try:
            # Your implementation here
            result = f"Processed {parameter1} with value {parameter2}"
            return Response(success=True, message=result)
        except Exception as e:
            return Response(success=False, message=str(e))

nano ~/agent-zero/python/instruments/server_health.py

from python.helpers.instrument import Instrument, Response
import psutil
import platform
from datetime import datetime

class ServerHealthInstrument(Instrument):
    """
    Check server health metrics including CPU, memory, disk usage, and uptime.
    Use this instrument when asked about server status, resource usage, or system health.
    """
    
    async def execute(self, detailed: bool = False) -> Response:
        """
        Get server health metrics.
        
        Args:
            detailed: If True, include per-CPU and per-disk breakdown
            
        Returns:
            Response with health metrics
        """
        try:
            # Basic metrics
            cpu_percent = psutil.cpu_percent(interval=1)
            memory = psutil.virtual_memory()
            disk = psutil.disk_usage('/')
            boot_time = datetime.fromtimestamp(psutil.boot_time())
            uptime = datetime.now() - boot_time
            
            health = {
                "timestamp": datetime.now().isoformat(),
                "hostname": platform.node(),
                "cpu_percent": cpu_percent,
                "memory_percent": memory.percent,
                "memory_available_gb": round(memory.available / (1024**3), 2),
                "disk_percent": disk.percent,
                "disk_free_gb": round(disk.free / (1024**3), 2),
                "uptime_days": uptime.days,
                "uptime_hours": uptime.seconds // 3600
            }
            
            # Generate summary
            status = "healthy"
            warnings = []
            
            if cpu_percent > 80:
                warnings.append(f"High CPU usage: {cpu_percent}%")
                status = "warning"
            if memory.percent > 85:
                warnings.append(f"High memory usage: {memory.percent}%")
                status = "warning"
            if disk.percent > 90:
                warnings.append(f"Low disk space: {disk.percent}% used")
                status = "critical"
            
            health["status"] = status
            health["warnings"] = warnings
            
            return Response(
                success=True, 
                message=f"Server status: {status}",
                data=health
            )
            
        except Exception as e:
            return Response(success=False, message=f"Health check failed: {str(e)}")

cd ~/agent-zero
source venv/bin/activate
pip install psutil

sudo systemctl restart agent-zero

ls ~/agent-zero/prompts/default/

prompts/default/
├── agent.system.md      # Main system prompt
├── agent.tools.md       # Tool usage instructions
├── agent.formats.md     # Response formatting rules
├── memory.save.md       # Memory storage instructions
├── memory.recall.md     # Memory retrieval instructions
└── ...

cp -r ~/agent-zero/prompts/default ~/agent-zero/prompts/devops

nano ~/agent-zero/prompts/devops/agent.system.md

# Agent Identity

You are a DevOps specialist AI assistant focused on infrastructure automation, 
system administration, and deployment pipelines. You have deep expertise in:

- Linux system administration (Ubuntu, RHEL, Alpine)
- Container technologies (Docker, Podman, Kubernetes)
- CI/CD pipelines (GitHub Actions, GitLab CI, Jenkins)
- Infrastructure as Code (Terraform, Ansible, Pulumi)
- Cloud platforms (AWS, GCP, Azure)
- Monitoring and observability (Prometheus, Grafana, ELK)

## Communication Style

- Be direct and technical; avoid unnecessary explanations of basic concepts
- Provide working commands and configurations, not just descriptions
- Include error handling and edge cases in scripts
- Explain security implications of recommended actions
- Default to infrastructure-as-code approaches over manual configuration

## Safety Rules

- Never store credentials in code or logs
- Always use the principle of least privilege
- Verify backup existence before destructive operations
- Test in staging-equivalent environments when possible
- Alert the user to potential security issues proactively

PROMPT_PROFILE=devops

sudo systemctl restart agent-zero

# MCP Configuration
MCP_ENABLED=true
MCP_SERVERS=filesystem,github,slack

# Install MCP filesystem server
npm install -g @anthropic/mcp-server-filesystem

# MCP Filesystem Server
MCP_FILESYSTEM_ENABLED=true
MCP_FILESYSTEM_PATHS=/home/agentzero/projects,/var/log
MCP_FILESYSTEM_READ_ONLY=true

npm install -g @anthropic/mcp-server-github

# MCP GitHub Server
MCP_GITHUB_ENABLED=true
MCP_GITHUB_TOKEN=ghp_your_personal_access_token
MCP_GITHUB_REPOS=myorg/myrepo,myorg/another-repo

from mcp import Server, Tool, Resource
import httpx
import os

server = Server("internal-api")

@server.tool("get_user")
async def get_user(user_id: str) -> dict:
    """Fetch user details from internal API."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://internal.example.com/api/users/{user_id}",
            headers={"Authorization": f"Bearer {os.environ['INTERNAL_API_KEY']}"}
        )
        return response.json()

@server.tool("create_ticket")
async def create_ticket(title: str, description: str, priority: str = "medium") -> dict:
    """Create a support ticket in the internal system."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://internal.example.com/api/tickets",
            json={"title": title, "description": description, "priority": priority},
            headers={"Authorization": f"Bearer {os.environ['INTERNAL_API_KEY']}"}
        )
        return response.json()

if __name__ == "__main__":
    server.run()

MCP_CUSTOM_SERVERS=/home/agentzero/agent-zero/mcp_servers/internal_api.py