Agents

Agents are intelligent, autonomous components that form the core of the Fiberwise platform. They process input data, execute business logic, and integrate with external services using advanced dependency injection and activation patterns.

What are Agents?

Fiberwiseagents are Python classes that inherit from the FiberAgent base class, providing a powerful foundation for building AI-powered applications. They combine dependency injection, schema validation, and lifecycle management into a cohesive development experience.

Agent Architecture Flow

🔍

Detection

Activation processor scans agent files

→

⚡

Instantiation

Agent class created with config

→

💉

Injection

Services injected via signatures

→

🚀

Execution

Agent processes input data

Core Agent Capabilities:

Advanced Dependency Injection: Automatic service injection based on method signatures and type hints
Schema Validation: Pydantic models for input/output validation and documentation
Activation Processor: Sophisticated execution engine with pattern detection and lifecycle management
Multi-Pattern Support: Function-based, class-based, and async execution patterns
Service Integration: Built-in access to LLM providers, storage, OAuth, and platform services
Error Handling: Comprehensive error tracking and graceful degradation
Performance Monitoring: Real-time execution metrics and logging
Multi-language SDKs: Python and Node.js support with unified API

Agent Development Patterns

Fiberwisesupports multiple agent development patterns to accommodate different use cases and complexity levels.

1. Function-Based Agent (Simple)

Perfect for simple data processing tasks with minimal dependencies.

async def execute(input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
    """Simple function-based agent with automatic dependency injection"""
    fiber = kwargs.get('fiber')
    llm_service = kwargs.get('llm_service')

    # Process input data
    response = await llm_service.generate(
        prompt=input_data.get('message', ''),
        max_tokens=150
    )

    return {
        "status": "success",
        "response": response.text,
        "model_used": response.model
    }

2. Class-Based Agent (Recommended)

Full-featured agent with schema validation and comprehensive lifecycle management.

from fiberwise_sdk import FiberAgent
from pydantic import BaseModel, Field

class InputSchema(BaseModel):
    message: str = Field(..., description="User message to process")
    temperature: float = Field(0.7, description="LLM temperature")

class OutputSchema(BaseModel):
    response: str = Field(..., description="Agent response")
    model_used: str = Field(..., description="LLM model used")
    token_usage: dict = Field(default_factory=dict)

class ChatAgent(FiberAgent):
    """Advanced chat agent with schema validation"""
    
    name = "ChatAgent"
    description = "Processes chat messages using LLM providers"
    version = "1.2.0"
    
    def get_input_schema(self) -> type[BaseModel]:
        return InputSchema
    
    def get_output_schema(self) -> type[BaseModel]:
        return OutputSchema

    async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        """Execute the agent."""
        llm_service = kwargs.get('llm_service')

        response = await llm_service.generate(
            prompt=input_data.get('message'),
            temperature=input_data.get('temperature', 0.7)
        )

        return {
            "response": response.text,
            "model_used": response.model,
            "token_usage": response.usage
        }

3. Advanced Multi-Service Agent

Complex agent with multiple service dependencies and advanced features.

class ChatSummarizerAgent(FiberAgent):
    """Advanced agent with multiple services and complex logic"""

    async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        """Execute the agent with multiple service integrations."""
        fiber = kwargs.get('fiber')
        llm_service = kwargs.get('llm_service')
        storage_service = kwargs.get('storage_service')

        # Access chat history through FiberApp SDK
        activations = await fiber.get_agent_activations(limit=50)
        chat_history = []
        
        for activation in activations:
            if activation.get('result', {}).get('messages'):
                chat_history.extend(activation['result']['messages'])
        
        # Generate comprehensive summary using LLM
        summary_prompt = f"""
        Analyze this conversation history and provide:
        1. Key topics discussed
        2. Important decisions or conclusions
        3. Action items or follow-ups
        
        Chat History:
        {json.dumps(chat_history[-20:], indent=2)}
        """
        
        summary = await llm_service.generate(
            prompt=summary_prompt,
            max_tokens=500,
            temperature=0.3
        )
        
        # Store summary in storage service
        summary_doc = {
            "timestamp": datetime.now().isoformat(),
            "summary": summary.text,
            "message_count": len(chat_history)
        }
        
        await storage_service.store_document(
            key=f"chat_summary_{int(time.time())}",
            document=summary_doc
        )
        
        return {
            "status": "success",
            "summary": summary.text,
            "analysis": {
                "message_count": len(chat_history),
                "participants": self._identify_participants(chat_history)
            }
        }

Dependency Injection System

Fiberwisefeatures a sophisticated dependency injection system that automatically provides agents with required services based on method signatures and declared dependencies.

🔍 Automatic Detection

The activation processor scans agent methods and automatically injects services based on parameter names and type hints.

🏷️ Type Hint Resolution

Services are automatically injected based on type hints and parameter names, eliminating manual configuration.

🔌 Service Registry

Services are managed through a centralized registry with lifecycle management and configuration.

⚡ Performance Optimized

Service instances are cached and reused across agent activations for optimal performance.

Available Services

LLMProviderService Access to configured LLM providers (OpenAI, Anthropic, etc.)

StorageService Multi-cloud storage with unified API

OAuthService OAuth integration for external services

FiberApp Platform SDK for accessing activations, agents, and platform features

Dependency Injection Patterns

🎯 Pattern 1: kwargs-Based Service Access

Services are accessed through kwargs for clean dependency injection:

async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
    # Services accessed via kwargs.get():
    fiber = kwargs.get('fiber')
    llm_service = kwargs.get('llm_service')
    storage_service = kwargs.get('storage_service')
    pass

🏷️ Pattern 2: Type Hints with kwargs

Use type hints for better IDE support while accessing services via kwargs:

from fiberwise_sdk.services import LLMProviderService, StorageService
from typing import Dict, Any

class DocumentProcessor(FiberAgent):
    async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        llm_service: LLMProviderService = kwargs.get('llm_service')
        storage_service: StorageService = kwargs.get('storage_service')
        # Services accessed via kwargs with type hints for IDE support
        pass

🔌 Pattern 3: Class with execute Method

Use the execute method pattern for class-based agents:

class ContentAnalyzer(FiberAgent):
    async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        """Execute the agent."""
        llm_service = kwargs.get('llm_service')
        storage_service = kwargs.get('storage_service')

        analysis = await llm_service.analyze(input_data['text'])
        await storage_service.save_analysis(analysis)
        return {"analysis": analysis}

⚡ Pattern 4: Dynamic Service Resolution

Access any registered service dynamically at runtime:

class FlexibleAgent(FiberAgent):
    async def execute(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        """Execute the agent."""
        fiber = kwargs.get('fiber')

        # Get services dynamically from kwargs or fiber
        llm_service = kwargs.get('llm_service')

        if llm_service:
            result = await llm_service.generate(input_data['prompt'])
            return {"result": result}
        else:
            return {"error": "LLM service not available"}

Service Registry & Lifecycle

🏗️ Automatic Registration

Services are automatically registered when the platform starts:

LLMProviderService - Configured LLM providers (OpenAI, Anthropic, etc.)
StorageService - Cloud storage with unified API
OAuthService - OAuth integrations for external services
DatabaseService - Direct database access
FiberApp - Platform SDK for activations and app features

⚡ Performance & Caching

The service registry optimizes performance through:

Singleton Pattern - Services are instantiated once and reused
Lazy Loading - Services are created only when first accessed
Context Isolation - Each activation gets its own service context
Resource Cleanup - Automatic cleanup after activation completion

🔧 Configuration & Customization

Services can be configured and customized:

Environment Variables - Service-specific configuration
Runtime Configuration - Dynamic service parameters
Custom Services - Register your own services
Service Overrides - Replace default services with custom implementations

Activation Processor System

The Activation Processor is the sophisticated execution engine that manages agent lifecycle, dependency injection, and performance monitoring.

🎯 Pattern Detection

Automatically detects agent execution patterns:

Function-based agents (async def execute)
Class-based agents with execute method
FiberAgent subclasses with execute method

💉 Smart Injection

Advanced dependency injection capabilities:

Parameter name matching (llm_service → LLMProviderService)
Type hint resolution for complex dependencies
Service configuration and lifecycle management
Error handling for missing dependencies

📊 Execution Context

Rich execution environment:

Context isolation between activations
Performance metrics and timing
Detailed logging and error tracking
Resource usage monitoring

Activation Lifecycle

Queue & Detection

Activation is queued and processor detects agent execution pattern

Service Resolution

Dependencies are resolved and services are prepared for injection

Agent Execution

Agent is instantiated and executed with injected services and input data

Result Processing

Output is validated, stored, and made available via WebSocket notifications

Activation Methods

🖥️ CLI Activation

# Activate with simple input
fiber activate --input '{"text": "Hello world"}' my_agent.py

# Activate with file input
fiber activate --input-file input.json my_agent.py

# Activate with specific LLM provider
fiber activate --provider openai --input '{"message": "Hi"}' chat_agent.py

🌐 Web API

POST /api/v1/agents/my_agent/activate
{
    "input_data": {
        "text": "Hello world",
        "temperature": 0.7
    },
    "agent_config": {
        "max_retries": 3,
        "timeout": 300
    }
}

🖱️ Web Interface

Interactive agent activation through the Fiberwiseweb dashboard with:

Form-based input validation
Real-time execution monitoring
WebSocket result streaming
Execution history and analytics

Common Application Patterns

Learn proven patterns for building different types of applications with Fiberwiseagents.

🗨️ Chat Applications

Build conversational interfaces with context management and message persistence:

class ChatApp {
  async sendMessage(message, sessionId) {
    // Store user message
    await fiber.data.create('messages', {
      session_id: sessionId,
      role: 'user',
      content: message,
      timestamp: new Date()
    });

    // Get agent response
    const response = await fiber.agents.activate('chat-agent', {
      message,
      context: { session_id: sessionId }
    });

    // Store agent response  
    await fiber.data.create('messages', {
      session_id: sessionId,
      role: 'assistant', 
      content: response.text,
      timestamp: new Date()
    });

    return response;
  }
}

📊 Data Processing Applications

Process files and data with functions and store results:

class DataProcessor {
  async processFile(file, processingType) {
    // Upload file
    const fileRef = await fiber.files.upload(file);
    
    // Process with function
    const result = await fiber.functions.activate('file-processor', {
      file_ref: fileRef,
      processing_type: processingType
    });

    // Store results
    await fiber.data.create('processing_results', {
      file_ref: fileRef,
      result: result,
      processed_at: new Date()
    });

    return result;
  }
}

🔄 Workflow Applications

Create multi-step workflows with state management:

class WorkflowApp {
  async startWorkflow(workflowId, initialData) {
    // Create workflow instance
    const instance = await fiber.data.create('workflow_instances', {
      workflow_id: workflowId,
      status: 'running',
      data: initialData,
      started_at: new Date()
    });

    // Execute first step
    await this.executeWorkflowStep(instance.id, 0);
    
    return instance;
  }

  async executeWorkflowStep(instanceId, stepIndex) {
    // Get workflow definition and current data
    const instance = await fiber.data.get('workflow_instances', instanceId);
    
    // Execute step with agent or function
    const stepResult = await fiber.agents.activate('workflow-agent', {
      step_index: stepIndex,
      instance_data: instance.data
    });

    // Update instance with results
    await fiber.data.update('workflow_instances', instanceId, {
      data: { ...instance.data, ...stepResult.data },
      current_step: stepIndex + 1,
      updated_at: new Date()
    });

    return stepResult;
  }
}

Security & Authentication

Fiberwiseprovides comprehensive security features for enterprise applications:

🔑 API Key Management

Secure API access for both app-level and user-level credentials:

// App-level API access (configured in dashboard)
const response = await fiber.agents.activate('agent-id', data);

// User-level API access (OAuth or user-provided keys)
const userResponse = await fiber.agents.activate('agent-id', data, {
  user_credentials: 'user-specific-token'
});

🔐 OAuth Integration

Connect to third-party services securely:

// Get OAuth authorization URL
const authUrl = await fiber.auth.getAuthUrl('google', {
  scopes: ['profile', 'email'],
  redirect_uri: 'https://yourapp.com/callback'
});

// Handle OAuth callback
const tokens = await fiber.auth.handleCallback('google', {
  code: 'auth-code-from-callback',
  state: 'csrf-protection-state'
});

// Use OAuth tokens in agents
const emailData = await fiber.agents.activate('email-agent', {
  action: 'list_emails',
  oauth_provider: 'google'
});

Best Practices

🎯 Agent Design

Single Responsibility: Each agent should have one clear purpose. Create multiple agents rather than one complex agent.

Clear Documentation: Document your agent's purpose, inputs, outputs, and dependencies.

Error Handling: Always handle errors gracefully and return meaningful error messages.

⚡ Performance

Streaming: Use streaming for long-running agent tasks to improve user experience.

Error Handling: Implement proper error handling and retries for robust applications.

Caching: Cache frequently accessed data to reduce latency.

Real-time Updates: Use real-time updates only when necessary to avoid unnecessary overhead.

🔒 Security

API Keys: Never expose API keys in frontend code. Use environment variables or secure storage.

OAuth: Use OAuth for user-specific external service access rather than storing credentials.

Input Validation: Validate all user input before processing to prevent security vulnerabilities.

Access Controls: Implement proper access controls and user permissions.

👤 User Experience

Loading States: Provide loading states for async operations to keep users informed.

Optimistic Updates: Use optimistic updates where possible to make applications feel faster.

Offline Handling: Handle offline scenarios gracefully with appropriate messaging.

Error Messaging: Implement proper error messaging that helps users understand what went wrong.

🏗️ Code Organization

Web Components: Use Web Components for modularity and reusability.

Separation of Concerns: Separate concerns between UI, logic, and data layers.

Reusable Functions: Create reusable functions for common operations.

Documentation: Document your components and APIs thoroughly.

🧪 Testing

Unit Testing: Write tests for your agents using mock dependencies.

Integration Testing: Test agent interactions with the platform services.

Error Scenarios: Test error scenarios and edge cases.