AI Agent Skills Complete Guide 2026: Building Reusable Agent Capabilities

MCP-Only Approach:
  Load all 40 tool schemas (8,000 tokens) → Agent decides which to use
  Context: 8K tokens for tools + remaining for data and reasoning

Skills with MCP:
  Detect intent → Load matching skill → Skill lazily loads 3-5 relevant MCP tools
  Context: ~600 tokens for tools + majority for data and reasoning

# Example: MCP server exposing tools + skill instructing their use
# MCP Server (tools/ticket_system.py):
@mcp.tool()
async def search_tickets(query: str, status: str = None) -> List[Ticket]:
    """Search Jira tickets by query string and optional status filter."""
    pass

@mcp.tool()
async def create_ticket(summary: str, description: str, priority: str) -> Ticket:
    """Create a new ticket in the project management system."""
    pass

# Skill (sprint_planning.skill.md):
# Skill instructs the agent on how to compose these MCP tools
# into a sprint planning workflow: gather backlog → estimate →
# assign → create sprint → notify team

{
  "skill_id": "financial_analysis",
  "version": "1.2.0",
  "name": "Financial Analysis",
  "description": "Comprehensive financial analysis capabilities including ratio analysis, trend analysis, and forecasting",
  "capabilities": [
    "ratio_calculation",
    "trend_analysis", 
    "forecasting",
    "benchmarking"
  ],
  "required_tools": [
    "calculator",
    "data_query",
    "visualization_generator"
  ],
  "parameters": {
    "required": ["financial_data"],
    "optional": ["benchmark_data", "forecast_period"]
  },
  "constraints": {
    "max_execution_time": 300,
    "requires_approval_for": ["large_transactions"]
  },
  "dependencies": ["data_processing"]
}

class FinancialAnalysisSkill:
    """Financial analysis skill with built-in domain logic."""
    
    def __init__(self, tools: ToolRegistry):
        self.tools = tools
        self.analysis_pipeline = self._build_pipeline()
    
    def _build_pipeline(self):
        return [
            DataValidationStep(),
            RatioCalculationStep(),
            TrendAnalysisStep(),
            BenchmarkingStep(),
            ReportGenerationStep()
        ]
    
    async def execute(self, context: ExecutionContext) -> SkillResult:
        """Execute the complete financial analysis workflow."""
        
        # Validate input data
        validated_data = await self._validate(context.input_data)
        
        # Run analysis pipeline
        results = {}
        for step in self.analysis_pipeline:
            step_result = await step.execute(
                validated_data, 
                context,
                results  # Pass previous results
            )
            results[step.name] = step_result
            
            # Check for critical issues
            if step_result.has_critical_issues:
                return SkillResult(
                    status="failed",
                    error=f"Critical issue in {step.name}: {step_result.issue}"
                )
        
        # Generate final report
        report = await self._generate_report(results, context)
        
        return SkillResult(
            status="success",
            data=results,
            report=report,
            visualizations=results.get("visualizations", [])
        )

skill_prompts:
  intent_detection:
    template: |
      You are analyzing a user request for financial analysis.
      Based on the request, determine:
      1. The type of analysis needed (ratio, trend, forecasting, benchmarking)
      2. The data required
      3. Any constraints or preferences
      
      Request: {user_request}
      
      Available data: {available_data}
      
      Output a structured analysis plan.
  
  result_interpretation:
    template: |
      You are interpreting financial analysis results for a non-expert.
      
      Analysis performed: {analysis_type}
      Results: {results}
      
      Explain the findings in clear language:
      - What do the numbers mean?
      - What actions are recommended?
      - What are the key risks?
      
      Use analogies where helpful.
  
  error_handling:
    template: |
      A financial analysis encountered an issue:
      - Error: {error}
      - Partial results: {partial_results}
      
      Determine:
      1. Can we proceed with partial results?
      2. What alternative approaches exist?
      3. What information would resolve the issue?

from pydantic import BaseModel, Field
from typing import Optional, List

class SkillConfiguration(BaseModel):
    """Base configuration for skill behavior."""
    
    # Execution settings
    timeout_seconds: int = Field(default=300, ge=1, le=3600)
    max_retries: int = Field(default=3, ge=0, le=10)
    
    # Output settings
    include_raw_data: bool = True
    generate_visualizations: bool = True
    visualization_format: str = "png"  # png, svg, html
    
    # Safety settings
    require_human_approval: bool = False
    approval_threshold: float = 10000.0
    blocked_operations: List[str] = Field(default_factory=list)
    
    # Data handling
    cache_results: bool = True
    cache_ttl_seconds: int = 3600
    max_data_size_mb: int = 100

class FinancialAnalysisConfig(SkillConfiguration):
    """Configuration specific to financial analysis skill."""
    
    # Analysis-specific settings
    default_benchmark_source: str = "industry_avg"
    currency: str = "USD"
    
    # Ratio calculations
    include_liquidity_ratios: bool = True
    include_profitability_ratios: bool = True
    include_leverage_ratios: bool = True
    include_efficiency_ratios: bool = True
    
    # Forecasting
    default_forecast_periods: int = 4
    forecasting_model: str = "linear"  # linear, exponential, arima
    
    # Reporting
    report_format: str = "detailed"  # summary, detailed, executive
    include_recommendations: bool = True
    risk_threshold: float = 0.3

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import asyncio

class SkillStatus(Enum):
    """Skill execution status."""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class SkillInput:
    """Input to a skill execution."""
    parameters: Dict[str, Any]
    context: Dict[str, Any] = field(default_factory=dict)
    attachments: List[bytes] = field(default_factory=list)

@dataclass
class SkillOutput:
    """Output from a skill execution."""
    status: SkillStatus
    result: Optional[Any] = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    artifacts: Dict[str, Any] = field(default_factory=dict)

class BaseSkill(ABC):
    """Base class for all skills."""
    
    def __init__(self, skill_id: str, version: str):
        self.skill_id = skill_id
        self.version = version
        self.tools = {}
        self.configuration = {}
    
    @abstractmethod
    async def initialize(self, config: Dict[str, Any]) -> None:
        """Initialize the skill with configuration."""
        pass
    
    @abstractmethod
    async def execute(self, input_data: SkillInput) -> SkillOutput:
        """Execute the skill's primary function."""
        pass
    
    @abstractmethod
    async def validate_input(self, input_data: SkillInput) -> bool:
        """Validate input data before execution."""
        pass
    
    async def cleanup(self) -> None:
        """Cleanup resources after execution."""
        pass

class DocumentGenerationSkill(BaseSkill):
    """Skill for generating various document types."""
    
    def __init__(self):
        super().__init__("document_generation", "1.0.0")
        self.supported_formats = ["pdf", "docx", "html", "markdown"]
    
    async def initialize(self, config: Dict[str, Any]) -> None:
        """Initialize with templates and formatting settings."""
        self.templates = config.get("templates", {})
        self.default_format = config.get("default_format", "markdown")
        self.branding = config.get("branding", {})
        
        # Initialize required tools
        self.tools = {
            "formatter": await self._get_tool("document_formatter"),
            "template_engine": await self._get_tool("template_engine"),
            "image_processor": await self._get_tool("image_processor")
        }
    
    async def validate_input(self, input_data: SkillInput) -> bool:
        """Validate document generation request."""
        required_fields = ["content", "template"]
        
        for field in required_fields:
            if field not in input_data.parameters:
                return False
        
        # Validate format
        output_format = input_data.parameters.get("format", self.default_format)
        if output_format not in self.supported_formats:
            return False
        
        return True
    
    async def execute(self, input_data: SkillInput) -> SkillOutput:
        """Execute document generation."""
        try:
            # Parse parameters
            content = input_data.parameters["content"]
            template_name = input_data.parameters["template"]
            output_format = input_data.parameters.get("format", self.default_format)
            
            # Load template
            template = self.templates.get(template_name)
            if not template:
                return SkillOutput(
                    status=SkillStatus.FAILED,
                    error=f"Template '{template_name}' not found"
                )
            
            # Apply template
            rendered = await self._render_template(
                template, 
                content,
                input_data.context
            )
            
            # Format output
            formatted = await self._format_document(
                rendered, 
                output_format
            )
            
            return SkillOutput(
                status=SkillStatus.SUCCESS,
                result=formatted,
                metadata={
                    "format": output_format,
                    "template": template_name,
                    "pages": self._estimate_pages(formatted)
                }
            )
        
        except Exception as e:
            return SkillOutput(
                status=SkillStatus.FAILED,
                error=str(e)
            )
    
    async def _render_template(self, template: str, content: Dict, context: Dict) -> str:
        """Render template with content."""
        # Template rendering logic
        pass
    
    async def _format_document(self, content: str, format: str) -> bytes:
        """Format document to target format."""
        pass
    
    async def _get_tool(self, tool_name: str):
        """Retrieve tool from registry."""
        pass
    
    def _estimate_pages(self, content: str) -> int:
        """Estimate page count."""
        return max(1, len(content) // 3000)

class SkillComposer:
    """Composes multiple skills into cohesive workflows."""
    
    def __init__(self, skill_registry: "SkillRegistry"):
        self.skill_registry = skill_registry
        self.workflows = {}
    
    def create_workflow(
        self,
        workflow_id: str,
        steps: List[Dict[str, Any]]
    ) -> "Workflow":
        """Create a composed workflow from multiple skills."""
        
        workflow_steps = []
        for step_config in steps:
            skill = self.skill_registry.get(step_config["skill_id"])
            
            step = WorkflowStep(
                skill=skill,
                input_mapping=step_config.get("input_mapping", {}),
                output_mapping=step_config.get("output_mapping", {}),
                condition=step_config.get("condition"),
                on_error=step_config.get("on_error", "stop")
            )
            workflow_steps.append(step)
        
        workflow = Workflow(
            workflow_id=workflow_id,
            steps=workflow_steps
        )
        
        self.workflows[workflow_id] = workflow
        return workflow
    
    async def execute_workflow(
        self,
        workflow_id: str,
        initial_input: SkillInput
    ) -> List[SkillOutput]:
        """Execute a composed workflow."""
        
        workflow = self.workflows.get(workflow_id)
        if not workflow:
            raise ValueError(f"Workflow '{workflow_id}' not found")
        
        return await workflow.execute(initial_input)

class Workflow:
    """Represents a multi-skill workflow."""
    
    def __init__(self, workflow_id: str, steps: List["WorkflowStep"]):
        self.workflow_id = workflow_id
        self.steps = steps
    
    async def execute(self, initial_input: SkillInput) -> List[SkillOutput]:
        """Execute all steps in sequence."""
        
        outputs = []
        current_input = initial_input
        
        for step in self.steps:
            # Map input from previous step if needed
            if outputs:
                current_input = step.map_input(outputs[-1])
            
            # Check condition
            if step.condition and not step.condition.evaluate(current_input):
                continue
            
            # Execute step
            try:
                output = await step.skill.execute(current_input)
                outputs.append(output)
                
                # Handle errors
                if output.status == SkillStatus.FAILED:
                    if step.on_error == "stop":
                        break
                    elif step.on_error == "continue":
                        continue
                    
            except Exception as e:
                if step.on_error == "stop":
                    break
                outputs.append(SkillOutput(
                    status=SkillStatus.FAILED,
                    error=str(e)
                ))
        
        return outputs

# Example: Creating a report generation workflow
composer = SkillComposer(skill_registry)

report_workflow = composer.create_workflow(
    "business_report_generation",
    steps=[
        {
            "skill_id": "data_collection",
            "input_mapping": {"query": "input.query"},
            "output_mapping": {"data": "step.output"}
        },
        {
            "skill_id": "data_analysis", 
            "input_mapping": {"data": "previous.data"},
            "output_mapping": {"analysis": "step.output"}
        },
        {
            "skill_id": "document_generation",
            "input_mapping": {
                "content.analysis": "previous.analysis",
                "content.summary": "previous.summary"
            },
            "output_mapping": {"document": "step.output"}
        },
        {
            "skill_id": "document_formatter",
            "input_mapping": {"document": "previous.document"}
        }
    ]
)

class SkillRegistry:
    """Central registry for managing skills."""
    
    def __init__(self):
        self._skills: Dict[str, BaseSkill] = {}
        self._metadata: Dict[str, Dict] = {}
        self._versions: Dict[str, List[str]] = {}
    
    async def register(
        self,
        skill: BaseSkill,
        metadata: Dict[str, Any]
    ) -> None:
        """Register a new skill."""
        
        skill_id = skill.skill_id
        
        if skill_id in self._skills:
            # Handle version conflict
            existing_version = self._metadata[skill_id]["version"]
            new_version = metadata.get("version", "1.0.0")
            
            if self._compare_versions(new_version, existing_version) <= 0:
                raise ValueError(
                    f"Version {new_version} is not newer than existing {existing_version}"
                )
        
        # Initialize skill
        await skill.initialize(metadata.get("config", {}))
        
        # Store skill
        self._skills[skill_id] = skill
        self._metadata[skill_id] = metadata
        
        # Track versions
        if skill_id not in self._versions:
            self._versions[skill_id] = []
        self._versions[skill_id].append(metadata.get("version", "1.0.0"))
    
    def get(self, skill_id: str, version: str = None) -> BaseSkill:
        """Retrieve a skill by ID and optional version."""
        
        if skill_id not in self._skills:
            raise KeyError(f"Skill '{skill_id}' not found")
        
        if version:
            # Load specific version (would typically use versioned instances)
            pass
        
        return self._skills[skill_id]
    
    def discover(
        self,
        capability: str = None,
        tags: List[str] = None
    ) -> List[Dict[str, Any]]:
        """Discover skills by capability or tags."""
        
        results = []
        
        for skill_id, metadata in self._metadata.items():
            # Filter by capability
            if capability and capability not in metadata.get("capabilities", []):
                continue
            
            # Filter by tags
            if tags:
                skill_tags = set(metadata.get("tags", []))
                if not skill_tags.intersection(tags):
                    continue
            
            results.append({
                "skill_id": skill_id,
                **metadata
            })
        
        return results
    
    def _compare_versions(self, v1: str, v2: str) -> int:
        """Compare semantic versions. Returns -1, 0, or 1."""
        # Simplified version comparison
        parts1 = [int(x) for x in v1.split(".")]
        parts2 = [int(x) for x in v2.split(".")]
        
        for p1, p2 in zip(parts1, parts2):
            if p1 < p2:
                return -1
            elif p1 > p2:
                return 1
        
        return 0

class DynamicSkillLoader:
    """Handles on-demand skill loading."""
    
    def __init__(self, skill_registry: SkillRegistry):
        self.registry = skill_registry
        self.loaded_skills: Dict[str, BaseSkill] = {}
        self.loading_tasks: Dict[str, asyncio.Task] = {}
    
    async def load_skill(
        self,
        skill_id: str,
        version: str = None,
        priority: int = 0
    ) -> BaseSkill:
        """Load a skill with priority-based resource management."""
        
        # Check if already loaded
        cache_key = f"{skill_id}:{version or 'latest'}"
        if cache_key in self.loaded_skills:
            return self.loaded_skills[cache_key]
        
        # Check if loading in progress
        if cache_key in self.loading_tasks:
            return await self.loading_tasks[cache_key]
        
        # Start loading
        load_task = asyncio.create_task(
            self._load_skill_async(skill_id, version)
        )
        self.loading_tasks[cache_key] = load_task
        
        try:
            skill = await load_task
            self.loaded_skills[cache_key] = skill
            return skill
        finally:
            del self.loading_tasks[cache_key]
    
    async def _load_skill_async(self, skill_id: str, version: str) -> BaseSkill:
        """Async skill loading with initialization."""
        
        # Check resource limits
        await self._check_resource_limits()
        
        # Get skill from registry
        skill = self.registry.get(skill_id, version)
        
        # Initialize if needed
        if not skill.initialized:
            await skill.initialize({})
        
        return skill
    
    async def unload_skill(self, skill_id: str, version: str = None) -> None:
        """Unload a skill to free resources."""
        
        cache_key = f"{skill_id}:{version or 'latest'}"
        
        if cache_key in self.loaded_skills:
            skill = self.loaded_skills[cache_key]
            await skill.cleanup()
            del self.loaded_skills[cache_key]
    
    async def _check_resource_limits(self) -> None:
        """Check and enforce resource limits."""
        
        max_loaded = 10  # Maximum concurrent skills
        max_memory_mb = 2048  # Maximum memory usage
        
        if len(self.loaded_skills) >= max_loaded:
            # Evict lowest priority skill
            await self._evict_lowest_priority()

class SkillRouter:
    """Routes requests to appropriate skills."""
    
    def __init__(self, skill_registry: SkillRegistry):
        self.registry = skill_registry
        self.routing_rules: List[RoutingRule] = []
    
    def add_rule(self, rule: RoutingRule) -> None:
        """Add a routing rule."""
        self.routing_rules.append(rule)
    
    async def route(self, request: "AgentRequest") -> List[BaseSkill]:
        """Route request to appropriate skills."""
        
        # Match routing rules
        matched_skills = []
        
        for rule in self.routing_rules:
            if rule.matches(request):
                skill = self.registry.get(rule.skill_id)
                matched_skills.append(skill)
        
        # If no explicit rules matched, use capability matching
        if not matched_skills:
            matched_skills = await self._capability_match(request)
        
        return matched_skills
    
    async def _capability_match(self, request: "AgentRequest") -> List[BaseSkill]:
        """Match skills based on required capabilities."""
        
        required = request.required_capabilities
        discovered = self.registry.discover(capability=required[0])
        
        skills = []
        for meta in discovered:
            # Check if skill has all required capabilities
            if all(cap in meta.get("capabilities", []) for cap in required):
                skill = self.registry.get(meta["skill_id"])
                skills.append(skill)
        
        return skills

class RoutingRule:
    """Defines conditions for skill routing."""
    
    def __init__(
        self,
        skill_id: str,
        condition: "RoutingCondition"
    ):
        self.skill_id = skill_id
        self.condition = condition
    
    def matches(self, request: "AgentRequest") -> bool:
        return self.condition.evaluate(request)

class RoutingCondition:
    """Base class for routing conditions."""
    
    def evaluate(self, request: "AgentRequest") -> bool:
        raise NotImplementedError

class IntentRoutingCondition(RoutingCondition):
    """Route based on detected user intent."""
    
    def __init__(self, intents: List[str]):
        self.intents = intents
    
    def evaluate(self, request: "AgentRequest") -> bool:
        return request.detected_intent in self.intents

class ContextRoutingCondition(RoutingCondition):
    """Route based on request context."""
    
    def __init__(self, context_key: str, expected_value: Any):
        self.context_key = context_key
        self.expected_value = expected_value
    
    def evaluate(self, request: "AgentRequest") -> bool:
        return request.context.get(self.context_key) == self.expected_value

Orchestrator ──A2A──→ Specialist Agent (Data Analysis Skill)
                    └── Uses MCP to query databases
                    └── Uses local skill for statistical modeling
                    └── Returns result via A2A

Orchestrator ──A2A──→ Specialist Agent (Report Generation Skill)
                    └── Uses MCP to access templates
                    └── Uses local skill for formatting
                    └── Returns final document via A2A

class ProgressiveSkillLoader:
    """Loads skill capabilities progressively to minimize context usage."""

    def __init__(self):
        self.skill_index: Dict[str, SkillManifest] = {}
        self.active_skills: Dict[str, "BaseSkill"] = {}

    async def register_skills(self, skills: List["BaseSkill"]) -> None:
        """Register skills with their manifests only (no tool schemas yet)."""
        for skill in skills:
            self.skill_index[skill.skill_id] = SkillManifest(
                skill_id=skill.skill_id,
                capabilities=skill.get_capabilities(),
                estimated_tools=skill.tool_count(),
                description=skill.get_description(),
                version=skill.version
            )

    async def resolve_for_intent(self, intent: str) -> List[str]:
        """Resolve which skills match an intent using lightweight manifests."""
        matched = []
        for skill_id, manifest in self.skill_index.items():
            if any(cap in intent.lower() for cap in manifest.capabilities):
                matched.append(skill_id)
        return matched

    async def activate_skill(self, skill_id: str) -> "BaseSkill":
        """Fully load and activate a skill (pulls in tool schemas)."""
        if skill_id not in self.active_skills:
            skill = await self._load_skill(skill_id)
            self.active_skills[skill_id] = skill
        return self.active_skills[skill_id]

class SkillSecurityManager:
    """Manages security for skill execution."""
    
    def __init__(self):
        self.policies: Dict[str, SecurityPolicy] = {}
        self.access_control = AccessControlList()
    
    def register_policy(self, skill_id: str, policy: SecurityPolicy) -> None:
        """Register security policy for a skill."""
        self.policies[skill_id] = policy
    
    async def validate_execution(
        self,
        skill_id: str,
        user: "User",
        input_data: SkillInput
    ) -> ValidationResult:
        """Validate skill execution request."""
        
        policy = self.policies.get(skill_id)
        if not policy:
            # Default deny if no policy
            return ValidationResult(allowed=False, reason="No policy defined")
        
        # Check permissions
        if not self.access_control.can_access(user, skill_id):
            return ValidationResult(
                allowed=False, 
                reason="User not authorized for this skill"
            )
        
        # Check input validation
        if not policy.validate_input(input_data):
            return ValidationResult(
                allowed=False,
                reason="Input validation failed"
            )
        
        # Check data access
        required_resources = policy.get_required_resources(input_data)
        for resource in required_resources:
            if not self.access_control.can_access_data(user, resource):
                return ValidationResult(
                    allowed=False,
                    reason=f"Access denied to required resource: {resource}"
                )
        
        return ValidationResult(allowed=True)

class SecurityPolicy:
    """Defines security requirements for a skill."""
    
    def __init__(
        self,
        skill_id: str,
        required_permissions: List[str] = None,
        data_classification: str = "internal",
        requires_audit: bool = True,
        input_validation_rules: List[ValidationRule] = None
    ):
        self.skill_id = skill_id
        self.required_permissions = required_permissions or []
        self.data_classification = data_classification
        self.requires_audit = requires_audit
        self.input_validation_rules = input_validation_rules or []
    
    def validate_input(self, input_data: SkillInput) -> bool:
        """Validate input data against rules."""
        for rule in self.input_validation_rules:
            if not rule.validate(input_data):
                return False
        return True
    
    def get_required_resources(self, input_data: SkillInput) -> List[str]:
        """Get list of data resources required by this skill."""
        # Implementation would extract resource references from input
        return []

class SkillTestSuite:
    """Comprehensive testing for skills."""
    
    def __init__(self, skill: BaseSkill):
        self.skill = skill
    
    async def run_tests(self) -> TestResults:
        """Run complete test suite."""
        
        results = TestResults()
        
        # Unit tests
        results.add(await self._test_input_validation())
        results.add(await self._test_output_format())
        results.add(await self._test_error_handling())
        
        # Integration tests
        results.add(await self._test_tool_integration())
        results.add(await self._test_workflow_integration())
        
        # Performance tests
        results.add(await self._test_performance())
        
        return results
    
    async def _test_input_validation(self) -> TestResult:
        """Test input validation logic."""
        
        test_cases = [
            # Valid inputs
            (SkillInput(parameters={"valid": "data"}), True),
            
            # Invalid inputs
            (SkillInput(parameters={}), False),
            (SkillInput(parameters={"missing": "required"}), False),
        ]
        
        passed = 0
        failed = 0
        
        for input_data, should_pass in test_cases:
            try:
                result = await self.skill.validate_input(input_data)
                if result == should_pass:
                    passed += 1
                else:
                    failed += 1
            except Exception:
                failed += 1
        
        return TestResult(
            name="input_validation",
            passed=passed,
            failed=failed
        )

class SkillMetrics:
    """Metrics collection for skills."""
    
    def __init__(self, skill_id: str):
        self.skill_id = skill_id
        self.execution_count = 0
        self.success_count = 0
        self.failure_count = 0
        self.total_duration = 0.0
        self.error_types: Dict[str, int] = {}
    
    def record_execution(
        self,
        status: SkillStatus,
        duration: float,
        error: str = None
    ) -> None:
        """Record execution metrics."""
        
        self.execution_count += 1
        self.total_duration += duration
        
        if status == SkillStatus.SUCCESS:
            self.success_count += 1
        else:
            self.failure_count += 1
            if error:
                self.error_types[error] = self.error_types.get(error, 0) + 1
    
    def get_metrics(self) -> Dict:
        """Get current metrics."""
        
        success_rate = (
            self.success_count / self.execution_count 
            if self.execution_count > 0 else 0
        )
        
        avg_duration = (
            self.total_duration / self.execution_count
            if self.execution_count > 0 else 0
        )
        
        return {
            "skill_id": self.skill_id,
            "execution_count": self.execution_count,
            "success_rate": success_rate,
            "avg_duration_seconds": avg_duration,
            "error_distribution": self.error_types
        }

---
skill_id: web-research
name: Web Research Skill
version: 1.2.0
capabilities:
  - web_search
  - content_extraction
  - source_verification
  - summarization
requires:
  - mcp-server-web-search
  - mcp-server-content-extraction
install:
  - npx skills add org/web-research
---

# Web Research Skill

## Description
Conducts multi-source web research with source verification
and structured output generation.

## Capabilities
- **web_search**: Queries multiple search engines via MCP
- **content_extraction**: Fetches and parses page content
- **source_verification**: Cross-references claims across sources
- **summarization**: Generates structured research summaries

## Usage
This skill is invoked when the user asks to research a topic.
The agent should:
1. Decompose the research question into sub-queries
2. Execute searches via the web-search MCP server
3. Extract content from top results
4. Cross-reference findings for consistency
5. Generate a structured summary with citations

## MCP Dependencies
- `mcp-server-web-search`: Provides web search tool access
- `mcp-server-content-extraction`: Provides page fetching

# Install a skill (works on Claude Code, Cursor, Copilot, Codex, Gemini CLI)
npx skills add anthropic/web-research
npx skills add anthropic/skill-creator
npx skills add github/agentic-eval