Python SDK Reference

pip install fastskill

# Advanced configuration with all options
from fastskill import (
    ServiceConfig, ExecutionConfig, CacheConfig,
    RoutingConfig, SearchConfig
)

config = ServiceConfig(
    # Storage
    skill_storage_path="./skills",
    backup_storage_path="./backups",

    # Execution
    execution=ExecutionConfig(
        default_timeout=60,
        max_memory_mb=1024,
        max_cpu_percent=75,
        enable_networking=False,
        sandbox_level="strict",
        allowed_paths=["./data", "./temp"]
    ),

    # Caching
    cache=CacheConfig(
        metadata_cache_size=2000,
        content_cache_size=200,
        cache_ttl_seconds=3600,
        enable_persistence=True,
        cache_directory="./cache"
    ),

    # Service settings
    enable_hot_reload=True,
    watch_paths=["./skills", "./custom-skills"],
    enable_audit_logging=True,
    log_level="INFO",
    max_concurrent_executions=20,

    # Performance
    enable_metrics=True,
    metrics_retention_hours=48,
    enable_health_checks=True,

    # Search and routing
    search=SearchConfig(
        max_results=50,
        min_relevance_score=0.3,
        enable_fuzzy_matching=True
    ),

    routing=RoutingConfig(
        enable_learning=True,
        learning_rate=0.1,
        feedback_weight=0.2
    )
)

async def monitor_service_health():
    """Monitor service health and performance."""

    # Get service metrics
    metrics = await service.get_metrics()

    print("📊 Service Health:")
    print(f"   Uptime: {metrics['uptime_seconds']}s")
    print(f"   Total requests: {metrics['total_requests']}")
    print(f"   Success rate: {metrics['success_rate']:.1%}")
    print(f"   Average response time: {metrics['avg_response_time_ms']:.2f}ms")

    # Check component health
    health = await service.health_check()

    print("\n🔍 Component Health:")
    for component, status in health.items():
        print(f"   {component}: {'✅' if status['healthy'] else '❌'}")

    # Cache performance
    cache_stats = await service.loading_service.get_cache_stats()
    print("
💾 Cache Performance:"    print(f"   Hit rate: {cache_stats['hit_rate']:.1%}")
    print(f"   Memory usage: {cache_stats['memory_usage_mb']:.1f}MB")

async def event_driven_example():
    """Example of event-driven integration."""

    # Define event handlers
    async def on_skill_registered(event):
        print(f"🆕 Skill registered: {event['skill']['name']}")
        # Trigger downstream processes
        await update_skill_index(event['skill'])
        await notify_subscribers(event['skill'])

    async def on_skill_executed(event):
        print(f"⚡ Skill executed: {event['skill_id']}")
        # Log execution metrics
        await log_execution_metrics(event)

    async def on_discovery_performed(event):
        print(f"🔍 Discovery performed: '{event['query']}'")
        # Update search analytics
        await update_search_analytics(event)

    # Subscribe to events
    service.subscribe("skill.registered", on_skill_registered)
    service.subscribe("skill.executed", on_skill_executed)
    service.subscribe("discovery.performed", on_discovery_performed)

    # Events are now handled automatically
    await service.initialize()

    # Simulate some operations that will trigger events
    await service.register_skill(sample_skill)
    await service.discover_skills("text processing")

    # Cleanup
    await service.shutdown()

# For high-throughput applications
from fastskill import ServiceConfig

config = ServiceConfig(
    # Increase concurrent execution limits
    max_concurrent_executions=50,

    # Optimize caching
    cache=CacheConfig(
        metadata_cache_size=5000,
        content_cache_size=1000,
        enable_persistence=True
    ),

    # Enable performance metrics
    enable_metrics=True
)

# Reuse service instances
service = FastSkillService(config)
await service.initialize()

# Reuse for multiple operations
for query in queries:
    skills = await service.discover_skills(query)

# Efficient async patterns
async def efficient_skill_discovery():
    """Efficiently discover skills for multiple queries."""

    # Good: Batch similar operations
    queries = ["text processing", "data analysis", "file conversion"]

    # Execute discovery operations concurrently
    tasks = [service.discover_skills(query) for query in queries]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Process results
    for query, result in zip(queries, results):
        if isinstance(result, Exception):
            print(f"❌ Failed for '{query}': {result}")
        else:
            print(f"✅ Found {len(result)} skills for '{query}'")

# Good: Use connection pooling
class SkillDiscoveryService:
    """Service wrapper with connection pooling."""

    def __init__(self):
        self._service = None

    async def __aenter__(self):
        if self._service is None:
            self._service = FastSkillService()
            await self._service.initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._service:
            await self._service.shutdown()

    async def discover(self, query):
        return await self._service.discover_skills(query)

# Usage
async with SkillDiscoveryService() as discovery:
    skills = await discovery.discover("text processing")

import pytest
import asyncio
from unittest.mock import AsyncMock, patch
from fastskill import FastSkillService, ValidationError

@pytest.fixture
async def service():
    """Create a test service instance."""
    config = ServiceConfig(skill_storage_path="./test-skills")
    service = FastSkillService(config)
    await service.initialize()
    yield service
    await service.shutdown()

@pytest.mark.asyncio
async def test_skill_registration(service):
    """Test skill registration functionality."""

    skill_definition = {
        "id": "test-skill",
        "name": "Test Skill",
        "description": "A skill for testing",
        "version": "1.0.0",
        "tags": ["test"],
        "capabilities": ["testing"]
    }

    # Test successful registration
    skill_id = await service.register_skill(skill_definition)
    assert skill_id == "test-skill"

    # Test duplicate registration fails
    with pytest.raises(ValidationError):
        await service.register_skill(skill_definition)

@pytest.mark.asyncio
async def test_skill_discovery(service):
    """Test skill discovery functionality."""

    # Register test skills
    await service.register_skill({
        "id": "text-processor",
        "name": "Text Processor",
        "description": "Process text documents",
        "capabilities": ["text_processing"]
    })

    # Test discovery
    skills = await service.discover_skills("process text")
    assert len(skills) > 0
    assert skills[0]["id"] == "text-processor"

@pytest.mark.asyncio
async def test_tool_execution(service):
    """Test tool execution functionality."""

    # Register skill with tool
    await service.register_skill({
        "id": "calculator",
        "name": "Calculator",
        "description": "Basic calculator",
        "capabilities": ["math"],
        "tools": [{
            "name": "add",
            "description": "Add two numbers",
            "parameters": {
                "a": {"type": "number"},
                "b": {"type": "number"}
            }
        }]
    })

    # Test tool execution
    result = await service.tool_service.execute_tool(
        "calculator",
        "add",
        {"a": 5, "b": 3}
    )

    assert result["success"] == True
    assert result["data"] == 8

import pytest
import asyncio
from fastskill import FastSkillService
from fastskill.testing import TestEnvironment

@pytest.mark.asyncio
async def test_full_integration():
    """Test full integration with realistic scenarios."""

    # Setup test environment
    async with TestEnvironment() as env:
        service = env.service

        # Test complete workflow
        skill_id = await service.register_skill(sample_skill)

        # Discovery workflow
        skills = await service.discover_skills("process data")
        assert len(skills) > 0

        # Execution workflow
        result = await service.tool_service.execute_tool(
            skill_id, "process", {"data": test_data}
        )
        assert result["success"] == True

        # Cleanup
        await service.delete_skill(skill_id)