API Reference Overview

API Overview

FastSkill provides multiple API interfaces to accommodate different integration needs and programming languages. All APIs provide access to the same core functionality with consistent behavior and data formats.

The FastSkill API follows REST principles and provides both synchronous and asynchronous interfaces across all supported languages.

API Types

1. Python SDK

The primary interface for Python applications, providing both synchronous and asynchronous methods with full type hints and comprehensive error handling.

import asyncio
from fastskill import FastSkillService, ServiceConfig

async def python_sdk_example():
    # Create service with configuration
    config = ServiceConfig(
        skill_storage_path="./skills",
        execution_timeout=30,
        enable_hot_reload=True
    )

    # Initialize service
    service = FastSkillService(config)
    await service.initialize()

    # Use the API
    skills = await service.list_skills()
    print(f"Found {len(skills)} skills")

    await service.shutdown()

asyncio.run(python_sdk_example())

2. CLI Tool

HTTP-based API for web applications, microservices, and cross-language integration.

# List skills
curl -X GET http://localhost:8080/api/skills \
  -H "Content-Type: application/json"

# Register skill
curl -X POST http://localhost:8080/api/skills \
  -H "Content-Type: application/json" \
  -d '{
    "id": "my-skill",
    "name": "My Skill",
    "description": "Example skill",
    "version": "1.0.0"
  }'

# Execute tool
curl -X POST http://localhost:8080/api/tools/execute \
  -H "Content-Type: application/json" \
  -d '{
    "skill_id": "text-extractor",
    "tool_name": "extract_text",
    "parameters": {
      "file_path": "document.pdf"
    }
  }'

3. WebSocket API

Real-time API for streaming data and live updates.

JavaScript

// WebSocket connection for real-time updates
const ws = new WebSocket('ws://localhost:8080/ws');

ws.onopen = function(event) {
  // Subscribe to skill events
  ws.send(JSON.stringify({
    type: 'subscribe',
    channels: ['skill.registered', 'skill.executed']
  }));
};

ws.onmessage = function(event) {
  const data = JSON.parse(event.data);
  console.log('Skill event:', data);
};

4. Rust SDK

High-performance interface for Rust applications.

use fastskill::{FastSkillService, ServiceConfig};
use std::path::PathBuf;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = ServiceConfig {
        skill_storage_path: PathBuf::from("./skills"),
        ..Default::default()
    };

    let service = FastSkillService::new(config).await?;
    service.initialize().await?;

    // Rust API usage
    let skills = service.skill_manager().list_skills(None).await?;
    println!("Found {} skills", skills.len());

    Ok(())
}

API Architecture

Request Flow

Response Format

All APIs return consistent response formats:

{
  "success": true,
  "data": {
    // Response data
  },
  "metadata": {
    "timestamp": "2024-01-15T10:30:00Z",
    "request_id": "req_123456",
    "processing_time_ms": 150
  }
}

Common Patterns

Error Handling

try:
    skills = await service.discover_skills("text processing")
    print(f"Found {len(skills)} skills")
except ValidationError as e:
    print(f"Validation error: {e}")
except ExecutionError as e:
    print(f"Execution error: {e}")
except ServiceError as e:
    print(f"Service error: {e}")

Pagination

Handle large result sets with pagination:

# Paginated requests
page = 1
page_size = 20

while True:
    result = await service.list_skills(
        page=page,
        page_size=page_size
    )

    skills = result['data']
    total_count = result['total_count']

    print(f"Page {page}: {len(skills)} skills")

    # Process skills
    for skill in skills:
        print(f"  - {skill['name']}")

    # Check if more pages
    if page * page_size >= total_count:
        break

    page += 1

Filtering and Sorting

Apply filters and sorting to API responses:

# Filtered and sorted requests
filters = {
    "enabled": True,
    "tags": ["text", "nlp"],
    "capabilities": ["text_extraction"]
}

sort_options = {
    "field": "name",
    "order": "asc"
}

skills = await service.list_skills(
    filters=filters,
    sort=sort_options
)

Rate Limiting

APIs are rate-limited to ensure fair usage:

{
  "X-RateLimit-Limit": "1000",
  "X-RateLimit-Remaining": "987",
  "X-RateLimit-Reset": "1640995200"
}

API Versions

FastSkill APIs are versioned to ensure backward compatibility:

Versioning Strategy

URL Versioning: /api/v1/skills, /api/v2/skills
Header Versioning: X-API-Version: v1
Accept Header: Accept: application/vnd.fastskill.v1+json

Migration Guide

Check current version

# Check API version
curl -X GET http://localhost:8080/api/version \
  -H "Content-Type: application/json"

Review changes

# Get changelog for version differences
curl -X GET http://localhost:8080/api/changelog?from=v1&to=v2 \
  -H "Content-Type: application/json"

Update client code

# Update to new API version
import fastskill

# Old way (deprecated)
service = fastskill.FastSkillService()

# New way (v2)
from fastskill import ServiceConfig
config = ServiceConfig(api_version="v2")
service = fastskill.FastSkillService(config)

Integration Examples

With AI Frameworks

from langchain.tools import FastSkillTool
from langchain.agents import initialize_agent

async def langchain_integration():
    # Create FastSkill service
    service = FastSkillService()
    await service.initialize()

    # Create LangChain tool
    fastskill_tool = FastSkillTool(
        service=service,
        description="Access to FastSkill ecosystem"
    )

    # Initialize agent
    agent = initialize_agent(
        tools=[fastskill_tool],
        llm=your_llm,
        agent_type="conversational-react-description"
    )

    # Use in conversation
    response = await agent.arun("extract text from this PDF")
    return response

    await service.shutdown()

Microservices Integration

# FastAPI integration
from fastapi import FastAPI, HTTPException
from fastskill import FastSkillService

app = FastAPI()
service = None

@app.on_event("startup")
async def startup_event():
    global service
    service = FastSkillService()
    await service.initialize()

@app.on_event("shutdown")
async def shutdown_event():
    if service:
        await service.shutdown()

@app.get("/api/skills")
async def list_skills():
    try:
        skills = await service.list_skills()
        return {"success": True, "data": skills}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/tools/execute")
async def execute_tool(request: ToolExecutionRequest):
    try:
        result = await service.tool_service.execute_tool(
            request.skill_id,
            request.tool_name,
            request.parameters
        )
        return {"success": True, "data": result}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

Testing APIs

Unit Testing

import pytest
import asyncio
from fastskill import FastSkillService

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

@pytest.mark.asyncio
async def test_list_skills(service):
    skills = await service.list_skills()
    assert isinstance(skills, list)

@pytest.mark.asyncio
async def test_discover_skills(service):
    skills = await service.discover_skills("text processing")
    assert len(skills) >= 0  # May be empty in test environment

@pytest.mark.asyncio
async def test_tool_execution(service):
    # Register test skill first
    skill_id = await service.register_skill(test_skill_definition)

    result = await service.tool_service.execute_tool(
        skill_id,
        "test_tool",
        {"test_param": "test_value"}
    )

    assert result["success"] == True

Integration Testing

# Test API endpoints
import requests

def test_rest_api():
    base_url = "http://localhost:8080"

    # Test health endpoint
    response = requests.get(f"{base_url}/health")
    assert response.status_code == 200

    # Test skills endpoint
    response = requests.get(f"{base_url}/api/skills")
    assert response.status_code == 200

    data = response.json()
    assert "success" in data
    assert "data" in data

Performance Considerations

API Optimization

Use pagination for large datasets

# Instead of getting all skills at once
all_skills = await service.list_skills()  # ❌ May be slow

# Use pagination
page = 1
all_skills = []
while True:
    result = await service.list_skills(page=page, page_size=50)
    skills = result['data']
    all_skills.extend(skills)

    if len(skills) < 50:  # Last page
        break
    page += 1

Cache frequently accessed data

# Cache skill metadata
cached_skills = None

async def get_cached_skills():
    global cached_skills
    if cached_skills is None:
        cached_skills = await service.list_skills()
    return cached_skills

Batch operations when possible

# Batch multiple operations
operations = [
    {"action": "register", "skill": skill1},
    {"action": "update", "skill": skill2},
    {"action": "enable", "skill_id": "skill3"}
]

results = await service.batch_execute(operations)

API Limits

Request Limits

Endpoint	Rate Limit	Burst Limit
`/api/skills`	1000/hour	100/minute
`/api/tools/execute`	500/hour	50/minute
`/api/discovery/*`	2000/hour	200/minute
`/api/routing/*`	1000/hour	100/minute

Data Limits

Resource	Size Limit	Description
Skill definition	1MB	JSON size limit
Tool parameters	10MB	Total parameter size
Execution timeout	300s	Maximum execution time
Response size	50MB	Maximum response payload

Exceeding API limits will result in rate limiting or request rejection. Monitor your usage and implement appropriate backoff strategies.

Troubleshooting

Connection Issues

Connection refused: Ensure the FastSkill service is running and accessible on the specified host and port.

# Check if service is running
curl http://localhost:8080/health

Timeout errors: Check network connectivity and service health. Consider increasing timeout values.

config = ServiceConfig(request_timeout=60)  # Increase timeout

Authentication Issues

API key errors: Verify your API key is correct and has the required permissions.

config = ServiceConfig(api_key="your-valid-api-key")

Permission denied: Check that your API key has access to the requested resources.

Rate Limiting

Implement backoff: Add exponential backoff when you hit rate limits.

import time

async def retry_with_backoff(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await func()
        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                await asyncio.sleep(wait_time)
            else:
                raise

The FastSkill API is designed to be intuitive and consistent across all interfaces. Start with the Python SDK for the best development experience, then expand to REST API for production deployments.

Getting Started

Configuration

Skill Management

Discovery & Routing

Tool Calling

Progressive Loading

API Reference

CLI Reference

Registry

Examples & Tutorials

Advanced Topics

Integration

Community

​API Overview

​API Types

​1. Python SDK

​2. CLI Tool

​3. WebSocket API

​4. Rust SDK

​API Architecture

​Request Flow

​Response Format

​Common Patterns

​Error Handling

​Pagination

​Filtering and Sorting

​Rate Limiting

​API Versions

​Versioning Strategy

​Migration Guide

​Integration Examples

​With AI Frameworks

​Microservices Integration

​Testing APIs

​Unit Testing

​Integration Testing

​Performance Considerations

​API Optimization

​API Limits

​Request Limits

​Data Limits

​Troubleshooting

API Overview

API Types

1. Python SDK

2. CLI Tool

3. WebSocket API

4. Rust SDK

API Architecture

Request Flow

Response Format

Common Patterns

Error Handling

Pagination

Filtering and Sorting

Rate Limiting

API Versions

Versioning Strategy

Migration Guide

Integration Examples

With AI Frameworks

Microservices Integration

Testing APIs

Unit Testing

Integration Testing

Performance Considerations

API Optimization

API Limits

Request Limits

Data Limits

Troubleshooting