name: agent-skills-orchestration
description: Design and orchestrate multiple AI agent skills that work together. Use when building systems with skill discovery, skill composition, registry patterns, error handling across skills, and multi-agent workflows. Essential for complex agent applications.

Agent Skills Orchestration & Multi-Agent Workflows

Quick Start

Skills are modular capabilities agents can discover and use. Organize them in a registry for efficient discovery and composition.

from dataclasses import dataclass
from typing import Callable, Dict, Any

@dataclass
class AgentSkill:
    name: str
    description: str  # For agent discovery
    parameters: Dict[str, Any]
    handler: Callable

class SkillsRegistry:
    def __init__(self):
        self.skills: Dict[str, AgentSkill] = {}

    def register(self, skill: AgentSkill):
        self.skills[skill.name] = skill

    def get_skill(self, name: str) -> AgentSkill:
        return self.skills.get(name)

    def list_skills(self):
        return [
            {
                "name": s.name,
                "description": s.description,
                "parameters": s.parameters
            }
            for s in self.skills.values()
        ]

Skill Definition Structure

Each skill needs clear metadata for agent discovery:

weather_skill = AgentSkill(
    name="get_weather",
    description="Get current weather for any city. Use when user asks about weather, temperature, or climate conditions in a specific location.",
    parameters={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "City name"
            },
            "units": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "description": "Temperature units"
            }
        },
        "required": ["city"]
    },
    handler=lambda city, units="celsius": get_weather_api(city, units)
)

Progressive Disclosure for Skills

Skills should load information in stages to manage context window:

Level 1: Metadata (Always Loaded)
├── name: "email-sender"
├── description: "Send emails with templates"
└── tags: ["communication", "notifications"]
↓
Level 2: Instructions (When Skill Triggered)
├── Full SKILL.md content
├── Step-by-step guidance
└── Usage examples
↓
Level 3: Resources (As Needed)
├── Template files
├── Helper scripts
└── Reference documentation

Implementation:

class SkillsRegistry:
    def discover_skills(self):
        """Level 1: Return only metadata for discovery"""
        return [
            {
                "name": skill.name,
                "description": skill.description,
                "tags": skill.tags
            }
            for skill in self.skills.values()
        ]

    def load_skill_details(self, skill_name: str):
        """Level 2: Load full instructions when triggered"""
        skill = self.skills[skill_name]
        return {
            "name": skill.name,
            "instructions": skill.instructions,
            "examples": skill.examples,
            "parameters": skill.parameters
        }

    def load_skill_resources(self, skill_name: str, resource_path: str):
        """Level 3: Load supporting files on demand"""
        # Read from filesystem only when needed
        return read_file(f"skills/{skill_name}/{resource_path}")

Skill Composition Patterns

Pattern 1: Sequential Skills

Agent uses multiple skills in sequence:

def sequential_workflow(agent, request: str):
    """
    Workflow: 
    1. Get user profile
    2. Query available products
    3. Send recommendation email
    """

    # Step 1: Get user data
    user_result = agent.use_skill(
        "fetch_user_profile",
        {"user_id": request.user_id}
    )

    if not user_result.success:
        return {"error": "Failed to fetch user"}

    # Step 2: Query products
    products = agent.use_skill(
        "search_products",
        {
            "category": user_result.data.preferred_category,
            "budget": user_result.data.budget
        }
    )

    # Step 3: Send email
    email_result = agent.use_skill(
        "send_email",
        {
            "to": user_result.data.email,
            "subject": "Personalized Recommendations",
            "products": products.data
        }
    )

    return email_result

Pattern 2: Conditional Skills

Agent chooses skills based on conditions:

def conditional_workflow(agent, request: str):
    """
    If urgent: Use fast_support skill
    Else: Use detailed_support skill
    """

    # First, analyze the request
    analysis = agent.use_skill(
        "analyze_request",
        {"text": request.message}
    )

    if analysis.data.severity == "high":
        # Use specialized urgent skill
        return agent.use_skill(
            "emergency_support",
            {"request_id": request.id}
        )
    else:
        # Use standard support skill
        return agent.use_skill(
            "standard_support",
            {"request_id": request.id}
        )

Pattern 3: Parallel Skills

Execute independent skills concurrently:

import asyncio

async def parallel_workflow(agent, user_id: str):
    """
    Execute in parallel:
    1. Fetch user profile
    2. Fetch order history
    3. Fetch recommendations
    """

    tasks = [
        agent.use_skill_async("fetch_user_profile", {"user_id": user_id}),
        agent.use_skill_async("fetch_order_history", {"user_id": user_id}),
        agent.use_skill_async("fetch_recommendations", {"user_id": user_id})
    ]

    profile, orders, recommendations = await asyncio.gather(*tasks)

    return {
        "profile": profile.data,
        "orders": orders.data,
        "recommendations": recommendations.data
    }

Pattern 4: Dependent Skills (Fanout-Fanin)

Use output of one skill as input to multiple others:

def fanout_fanin_workflow(agent, category: str):
    """
    1. Get products in category (fanout)
    2. For each product:
       - Get price
       - Get reviews
       - Get availability
    3. Combine results (fanin)
    """

    # Step 1: Get products
    products = agent.use_skill(
        "search_products",
        {"category": category}
    )

    # Step 2: Fanout - get details for each product
    details = []
    for product in products.data:
        pricing = agent.use_skill(
            "get_pricing",
            {"product_id": product.id}
        )
        reviews = agent.use_skill(
            "get_reviews",
            {"product_id": product.id}
        )
        availability = agent.use_skill(
            "check_availability",
            {"product_id": product.id}
        )

        details.append({
            "product": product,
            "pricing": pricing.data,
            "reviews": reviews.data,
            "availability": availability.data
        })

    # Step 3: Fanin - combine results
    return {
        "category": category,
        "products": details,
        "total_count": len(details)
    }

Error Handling Across Skills

class SkillResult:
    def __init__(self, success: bool, data=None, error=None):
        self.success = success
        self.data = data
        self.error = error

    def __bool__(self):
        return self.success

def use_skill_with_retry(agent, skill_name: str, args: dict, max_retries: int = 3):
    """Use a skill with automatic retry on failure"""

    for attempt in range(max_retries):
        try:
            result = agent.use_skill(skill_name, args)

            if result.success:
                return result

            # Log failure
            print(f"Attempt {attempt + 1} failed: {result.error}")

            # Exponential backoff
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)

        except Exception as e:
            print(f"Exception in skill {skill_name}: {e}")
            if attempt == max_retries - 1:
                return SkillResult(
                    success=False,
                    error=f"Failed after {max_retries} attempts: {str(e)}"
                )

    return SkillResult(
        success=False,
        error=f"Skill {skill_name} failed after {max_retries} attempts"
    )

Skill Dependencies & Validation

class SkillDependency:
    def __init__(self, skill_name: str, required: bool = True):
        self.skill_name = skill_name
        self.required = required

class ValidatedSkill(AgentSkill):
    def __init__(self, name, description, parameters, handler, dependencies=None):
        super().__init__(name, description, parameters, handler)
        self.dependencies = dependencies or []

    def validate_dependencies(self, registry: SkillsRegistry) -> tuple[bool, list[str]]:
        """Check if all required dependencies are available"""
        missing = []

        for dep in self.dependencies:
            if not registry.get_skill(dep.skill_name):
                if dep.required:
                    missing.append(dep.skill_name)

        return len(missing) == 0, missing

    def validate_parameters(self, input_params: dict) -> tuple[bool, list[str]]:
        """Validate input parameters against schema"""
        errors = []

        for required in self.parameters.get("required", []):
            if required not in input_params:
                errors.append(f"Missing required parameter: {required}")

        return len(errors) == 0, errors

# Usage
email_skill = ValidatedSkill(
    name="send_email",
    description="Send emails",
    parameters={
        "type": "object",
        "properties": {
            "to": {"type": "string"},
            "subject": {"type": "string"},
            "body": {"type": "string"}
        },
        "required": ["to", "subject"]
    },
    handler=send_email_handler,
    dependencies=[
        SkillDependency("validate_email", required=True),
        SkillDependency("log_email", required=False)
    ]
)

Skill Versioning

class VersionedSkill(AgentSkill):
    def __init__(self, name, version, description, parameters, handler):
        super().__init__(name, description, parameters, handler)
        self.version = version

    def is_compatible(self, required_version: str) -> bool:
        """Check version compatibility"""
        return self.version >= required_version

# Register multiple versions
skills_registry.register(
    VersionedSkill(
        "weather",
        version="1.0",
        description="Get weather (v1)",
        parameters={...},
        handler=get_weather_v1
    )
)

skills_registry.register(
    VersionedSkill(
        "weather",
        version="2.0",
        description="Get weather with forecasts (v2)",
        parameters={...},
        handler=get_weather_v2
    )
)

# Use specific version
agent.use_skill("weather", args, version="2.0")

Skill Monitoring & Analytics

class SkillMetrics:
    def __init__(self):
        self.calls: Dict[str, int] = {}
        self.failures: Dict[str, int] = {}
        self.latency: Dict[str, list[float]] = {}

    def record_call(self, skill_name: str):
        self.calls[skill_name] = self.calls.get(skill_name, 0) + 1

    def record_failure(self, skill_name: str):
        self.failures[skill_name] = self.failures.get(skill_name, 0) + 1

    def record_latency(self, skill_name: str, duration: float):
        if skill_name not in self.latency:
            self.latency[skill_name] = []
        self.latency[skill_name].append(duration)

    def get_success_rate(self, skill_name: str) -> float:
        calls = self.calls.get(skill_name, 0)
        failures = self.failures.get(skill_name, 0)

        if calls == 0:
            return 0.0

        return ((calls - failures) / calls) * 100

    def get_avg_latency(self, skill_name: str) -> float:
        latencies = self.latency.get(skill_name, [])

        if not latencies:
            return 0.0

        return sum(latencies) / len(latencies)

# Usage
metrics = SkillMetrics()

def use_skill_with_metrics(agent, skill_name: str, args: dict):
    import time

    metrics.record_call(skill_name)
    start = time.time()

    try:
        result = agent.use_skill(skill_name, args)

        if not result.success:
            metrics.record_failure(skill_name)

        return result
    finally:
        duration = time.time() - start
        metrics.record_latency(skill_name, duration)

Real-World Multi-Agent Workflow

class CustomerSupportAgent:
    def __init__(self, skills_registry: SkillsRegistry):
        self.registry = skills_registry

    def handle_customer_request(self, request: dict):
        """
        Multi-skill workflow:
        1. Analyze request priority
        2. Fetch customer data
        3. Check order history
        4. Get refund policy
        5. Generate response
        """

        # Step 1: Analyze
        analysis = self.use_skill(
            "analyze_request",
            {"text": request["message"]}
        )

        if not analysis:
            return {"error": "Failed to analyze request"}

        # Step 2: Get customer data
        customer = self.use_skill(
            "fetch_customer",
            {"customer_id": request["customer_id"]}
        )

        # Step 3: Check order history
        orders = self.use_skill(
            "fetch_orders",
            {"customer_id": request["customer_id"]}
        )

        # Step 4: Get policy
        policy = self.use_skill(
            "get_refund_policy",
            {"order_id": orders.data[0]["id"]}
        )

        # Step 5: Generate response using all data
        response = self.use_skill(
            "generate_response",
            {
                "request": request["message"],
                "priority": analysis.data["priority"],
                "customer": customer.data,
                "orders": orders.data,
                "policy": policy.data
            }
        )

        return response

    def use_skill(self, skill_name: str, args: dict):
        """Execute skill with error handling"""
        skill = self.registry.get_skill(skill_name)

        if not skill:
            return SkillResult(False, error=f"Skill {skill_name} not found")

        try:
            result = skill.handler(**args)
            return SkillResult(True, data=result)
        except Exception as e:
            return SkillResult(False, error=str(e))

Best Practices for Skills

✓ DO:
- Write clear descriptions (include "when to use")
- Keep skills focused on single responsibility
- Return consistent structured results
- Validate all inputs
- Handle errors gracefully
- Use timeouts for external calls
- Document dependencies
- Test skills independently

✗ DON'T:
- Mix multiple concerns in one skill
- Expect agents to understand vague descriptions
- Return raw/unstructured results
- Skip error handling
- Make unlimited blocking calls
- Create circular dependencies
- Hardcode configuration
- Forget to version skills

References

• Agent Skills Overview
• Skills Best Practices
• Equipping Agents