Overview FastSkill provides execution services that enable AI agents to perform various tasks directly. Each skill exposes execution capabilities through service interfaces that agents can use to accomplish their objectives.
Execution services provide the functionality that AI agents need. Agents integrate with these services directly to access skill capabilities without tool calling intermediaries.
Execution Service Categories FastSkill execution services are organized into categories based on their functionality:
📄 Text Processing Services
Text Extractor Extract text from various document formats including PDF, DOCX, HTML, and plain text files.
Text Analyzer Analyze text content for sentiment, keywords, entities, and other insights.
Text Transformer Transform text through operations like translation, summarization, and formatting.
Content Generator Generate text content based on prompts, templates, and context.
📊 Data Analysis Services
Data Processor Process structured data from CSV, JSON, XML, and database sources.
Statistical Analyzer Perform statistical analysis including correlation, regression, and hypothesis testing.
Visualization Generator Create charts, graphs, and visualizations from data.
Pattern Detector Identify patterns, trends, and anomalies in datasets.
🌐 Web and Network Services
Web Scraper Extract data from websites and web APIs with configurable crawling rules.
API Client Make HTTP requests to REST APIs with authentication and rate limiting.
Content Fetcher Download and process web content including images, documents, and media.
Social Media Analyzer Analyze social media content and engagement metrics.
📁 File Management Services
File Organizer Organize files by type, size, date, or content with automatic categorization.
Format Converter Convert files between different formats (PDF, DOCX, HTML, etc.).
File Validator Validate file formats, integrity, and compliance with standards.
Backup Manager Create, manage, and restore file backups with compression and encryption.
Service Interface Each execution service follows a consistent interface pattern:
# Service interface structure { "name" : "service_name" , # Service identifier "description" : "What the service does" , # Human-readable description "skill_id" : "skill_identifier" , # Parent skill ID "version" : "1.0.0" , # Service version # Parameter specification "parameters" : { "param_name" : { "type" : "string|number|boolean|array|object" , "required" : true | false, "description" : "Parameter description" , "default" : "default_value" , # Optional default "enum" : [ "option1" , "option2" ] # Optional enum values } }, # Execution metadata "execution" : { "timeout" : 30 , # Timeout in seconds "memory_limit_mb" : 256 , # Memory limit "requires_network" : true | false, # Network requirement "sandbox_level" : "basic|strict" # Security level }, # Capabilities and features "capabilities" : [ "feature1" , "feature2" ], "supported_formats" : [ "format1" , "format2" ], "tags" : [ "tag1" , "tag2" ] } Service Catalog
PDF Text Extractor
# Service: pdf_text_extractor { "name" : "extract_text_from_pdf" , "description" : "Extract text content from PDF documents" , "parameters" : { "file_path" : { "type" : "string" , "required" : true, "description" : "Path to the PDF file" }, "page_range" : { "type" : "array" , "required" : false, "description" : "Specific pages to extract (e.g., [1, 2, 3])" , "default" : "all pages" }, "preserve_layout" : { "type" : "boolean" , "required" : false, "description" : "Maintain original text layout and formatting" , "default" : false } }, "execution" : { "timeout" : 60 , "memory_limit_mb" : 512 , "requires_network" : false } } # Usage example result = await service.tool_service.execute_tool( "pdf_text_extractor" , "extract_text_from_pdf" , { "file_path" : "/path/to/document.pdf" , "page_range" : [ 1 , 2 , 3 ], "preserve_layout" : True } ) print ( f "Extracted text: { result[ 'text' ] } " ) print ( f "Pages processed: { result[ 'pages_processed' ] } " )
DOCX Text Extractor
# Tool: docx_text_extractor { "name" : "extract_text_from_docx" , "description" : "Extract text and formatting from Microsoft Word documents" , "parameters" : { "file_path" : { "type" : "string" , "required" : true, "description" : "Path to the DOCX file" }, "include_formatting" : { "type" : "boolean" , "required" : false, "description" : "Include text formatting information" , "default" : false }, "extract_metadata" : { "type" : "boolean" , "required" : false, "description" : "Extract document metadata (author, creation date, etc.)" , "default" : false } } }
HTML Content Extractor
# Tool: html_content_extractor { "name" : "extract_text_from_html" , "description" : "Extract clean text content from HTML pages" , "parameters" : { "html_content" : { "type" : "string" , "required" : true, "description" : "HTML content as string or URL" }, "remove_scripts" : { "type" : "boolean" , "required" : false, "description" : "Remove JavaScript and CSS content" , "default" : true }, "preserve_links" : { "type" : "boolean" , "required" : false, "description" : "Convert links to reference format" , "default" : false } } } Data Analysis Services
CSV Data Analyzer
# Service: csv_data_analyzer { "name" : "analyze_csv_data" , "description" : "Perform comprehensive analysis on CSV datasets" , "parameters" : { "file_path" : { "type" : "string" , "required" : true, "description" : "Path to CSV file" }, "analysis_type" : { "type" : "string" , "required" : false, "description" : "Type of analysis to perform" , "enum" : [ "summary" , "correlation" , "trends" , "anomalies" , "complete" ], "default" : "summary" }, "columns" : { "type" : "array" , "required" : false, "description" : "Specific columns to analyze" , "default" : "all columns" } } }
Statistical Calculator
# Tool: statistical_calculator { "name" : "calculate_statistics" , "description" : "Calculate statistical measures for numerical data" , "parameters" : { "data" : { "type" : "array" , "required" : true, "description" : "Numerical data array" }, "statistics" : { "type" : "array" , "required" : false, "description" : "Statistical measures to calculate" , "enum" : [ "mean" , "median" , "mode" , "std" , "variance" , "min" , "max" , "quartiles" ], "default" : [ "mean" , "median" , "std" ] } } }
Chart Generator
# Tool: chart_generator { "name" : "generate_chart" , "description" : "Generate charts and visualizations from data" , "parameters" : { "data" : { "type" : "object" , "required" : true, "description" : "Data to visualize" }, "chart_type" : { "type" : "string" , "required" : true, "description" : "Type of chart to generate" , "enum" : [ "bar" , "line" , "pie" , "scatter" , "histogram" , "boxplot" ] }, "output_format" : { "type" : "string" , "required" : false, "description" : "Output format for the chart" , "enum" : [ "png" , "svg" , "pdf" , "html" ], "default" : "png" } } } Service Discovery Find services using various discovery methods:
By Capability async def find_services_by_capability (): service = FastSkillService() await service.initialize() # Find all text extraction services extraction_services = await service.execution_service.find_services_by_capability( "text_extraction" ) print ( f "📄 Found { len (extraction_services) } text extraction services" ) # Find all analysis services analysis_services = await service.execution_service.find_services_by_capability( "data_analysis" ) print ( f "📊 Found { len (analysis_services) } data analysis services" ) # Find all conversion services conversion_services = await service.execution_service.find_services_by_capability( "format_conversion" ) print ( f "🔄 Found { len (conversion_services) } format conversion services" ) await service.shutdown() By Tag async def find_services_by_tag (): service = FastSkillService() await service.initialize() # Find services by domain web_services = await service.execution_service.find_services_by_tag( "web" ) print ( f "🌐 Web services: { len (web_services) } " ) data_services = await service.execution_service.find_services_by_tag( "data" ) print ( f "📊 Data services: { len (data_services) } " ) text_services = await service.execution_service.find_services_by_tag( "text" ) print ( f "📝 Text services: { len (text_services) } " ) await service.shutdown() Advanced Service Search async def advanced_service_search (): service = FastSkillService() await service.initialize() # Search with multiple criteria search_criteria = { "capabilities" : [ "text_extraction" , "analysis" ], "tags" : [ "text" , "nlp" ], "max_execution_time" : 60 , "requires_network" : False } matching_services = await service.execution_service.search_services(search_criteria) print ( f "🎯 Found { len (matching_services) } services matching criteria" ) # Display service details for service in matching_services[: 5 ]: print ( f " \n Service: { service[ 'name' ] } " ) print ( f " Description: { service[ 'description' ] } " ) print ( f " Skill: { service[ 'skill_id' ] } " ) print ( f " Parameters: { list (service[ 'parameters' ].keys()) } " ) await service.shutdown() Service Execution Execute services with proper parameter handling:
Basic Execution # Basic service execution result = await service.execution_service.execute_service( skill_id = "text-extractor" , service_name = "extract_text_from_pdf" , parameters = { "file_path" : "/path/to/document.pdf" , "page_range" : [ 1 , 2 , 3 ] } ) print ( f "✅ Execution successful: { result[ 'success' ] } " ) print ( f "📄 Extracted text: { result[ 'data' ][: 200 ] } ..." ) print ( f "⏱️ Execution time: { result[ 'execution_time' ] } s" )
Batch Execution Execute multiple services in sequence:
async def batch_service_execution (): service = FastSkillService() await service.initialize() # Define batch operations operations = [ { "skill_id" : "text-extractor" , "service_name" : "extract_text_from_pdf" , "parameters" : { "file_path" : "doc1.pdf" } }, { "skill_id" : "text-analyzer" , "service_name" : "analyze_sentiment" , "parameters" : { "text" : "extracted_text_placeholder" } }, { "skill_id" : "data-analyzer" , "service_name" : "generate_summary" , "parameters" : { "analysis" : "sentiment_results_placeholder" } } ] # Execute batch with dependency chaining results = await service.execution_service.execute_batch(operations) for i, result in enumerate (results): print ( f "Step { i + 1 } : { operations[i][ 'service_name' ] } " ) print ( f " Success: { result[ 'success' ] } " ) print ( f " Result: { result[ 'data' ] } " ) await service.shutdown() Service Development Create custom services for your specific needs:
Define service specification
{ "name" : "custom_data_processor" , "description" : "Process custom data formats with specialized logic" , "skill_id" : "custom-processor" , "parameters" : { "input_data" : { "type" : "object" , "required" : true , "description" : "Input data to process" }, "processing_options" : { "type" : "object" , "required" : false , "description" : "Processing configuration options" , "default" : {} }, "output_format" : { "type" : "string" , "required" : false , "description" : "Desired output format" , "enum" : [ "json" , "csv" , "xml" ], "default" : "json" } }, "execution" : { "timeout" : 120 , "memory_limit_mb" : 1024 , "requires_network" : false , "sandbox_level" : "basic" }, "capabilities" : [ "custom_processing" , "data_transformation" ], "tags" : [ "custom" , "data" , "processing" ] }
Implement service logic
# custom_processor.py import asyncio from typing import Dict, Any import json class CustomDataProcessor : """Custom data processing implementation.""" async def process_data ( self , input_data : Dict[ str , Any], options : Dict[ str , Any] = None , output_format : str = "json" ) -> Dict[ str , Any]: """Process input data according to custom logic.""" # Custom processing logic here processed_data = { "original_size" : len ( str (input_data)), "processing_timestamp" : asyncio.get_event_loop().time(), "options_applied" : options or {}, "output_format" : output_format } # Format output if output_format == "csv" : output = self ._to_csv(processed_data) elif output_format == "xml" : output = self ._to_xml(processed_data) else : # json output = processed_data return { "success" : True , "data" : output, "metadata" : { "processing_time" : 0.1 , "output_size" : len ( str (output)) } } # FastSkill service interface async def process_custom_data ( input_data : Dict[ str , Any], processing_options : Dict[ str , Any] = None , output_format : str = "json" ) -> Dict[ str , Any]: """Process custom data with specialized logic. Args: input_data: Data to process processing_options: Processing configuration output_format: Output format (json, csv, xml) Returns: Processed data in the specified format """ processor = CustomDataProcessor() return await processor.process_data(input_data, processing_options, output_format)
Register and test
async def register_custom_service (): service = FastSkillService() await service.initialize() # Register the skill skill_id = await service.register_skill({ "id" : "custom-processor" , "name" : "Custom Data Processor" , "description" : "Process data with custom logic" , "version" : "1.0.0" , "tags" : [ "custom" , "data" , "processing" ], "capabilities" : [ "custom_processing" , "data_transformation" ], "skill_file" : "custom_processor.py" , "enabled" : True }) # Test the service test_data = { "field1" : "value1" , "field2" : "value2" } result = await service.execution_service.execute_service( "custom-processor" , "process_custom_data" , { "input_data" : test_data, "output_format" : "json" } ) print ( f "✅ Custom service executed: { result[ 'success' ] } " ) print ( f "📊 Result: { result[ 'data' ] } " ) await service.shutdown() Service Management Service Lifecycle async def manage_service_lifecycle (): service = FastSkillService() await service.initialize() # 1. List available services services = await service.execution_service.get_available_services() print ( f "🔧 Available services: { len (services) } " ) # 2. Get service details if services: service_details = await service.execution_service.get_service_details(services[ 0 ][ 'id' ]) print ( f "📋 Service: { service_details[ 'name' ] } " ) print ( f " Parameters: { list (service_details[ 'parameters' ].keys()) } " ) # 3. Update service configuration await service.execution_service.update_service( services[ 0 ][ 'id' ], { "execution_timeout" : 60 , "enabled" : True } ) # 4. Disable service await service.execution_service.disable_service(services[ 0 ][ 'id' ]) # 5. Remove service await service.execution_service.remove_service(services[ 0 ][ 'id' ]) await service.shutdown() Service Validation Ensure services meet quality standards:
async def validate_services (): service = FastSkillService() await service.initialize() # Get all services services = await service.execution_service.get_available_services() # Validate each service validation_results = [] for service_item in services: try : # Basic validation is_valid = await service.execution_service.validate_service(service_item[ 'id' ]) # Detailed validation details = await service.execution_service.get_service_details(service_item[ 'id' ]) validation_results.append({ "service_id" : service_item[ 'id' ], "valid" : is_valid, "parameters_count" : len (details.get( 'parameters' , {})), "has_description" : bool (details.get( 'description' )), "has_timeout" : bool (details.get( 'execution' , {}).get( 'timeout' )) }) except Exception as e: validation_results.append({ "service_id" : service_item[ 'id' ], "valid" : False , "error" : str (e) }) # Report validation results valid_services = [r for r in validation_results if r[ 'valid' ]] invalid_services = [r for r in validation_results if not r[ 'valid' ]] print ( f "✅ Valid services: { len (valid_services) } " ) print ( f "❌ Invalid services: { len (invalid_services) } " ) await service.shutdown() Best Practices
Design clear interfaces
Create services with clear, consistent parameter names and types that are easy for AI agents to understand.
Provide comprehensive documentation
Include detailed descriptions, parameter explanations, and usage examples for each service.
Handle errors gracefully
Implement proper error handling and provide meaningful error messages for debugging.
Optimize for performance
Consider execution time, memory usage, and resource constraints when designing services.
Test thoroughly
Test services with various input types, edge cases, and error conditions before deployment.
Always validate service inputs and handle edge cases. Services are executed in potentially untrusted environments and should be designed with security in mind.
