Python SDK

A comprehensive Python SDK for building and interacting with intelligent agents, data processing pipelines, and real-time applications. Complete with built-in LLM provider management - no additional CLI tools required!

Features

🤖 Intelligent Agents

Create, deploy, and execute AI agents with automatic dependency injection and multi-provider LLM support (OpenAI, Anthropic, Google)

🔧 Built-in Provider Management

Add and manage LLM providers directly from Python - no CLI required

🗄️ Dynamic Data Models

Create and manage database schemas dynamically with automatic field definitions, validation rules, and type-safe operations

⚡ Real-time Communication

WebSocket-based streaming for live agent responses and event processing

🗃️ Dynamic Data Operations

Type-safe database operations with automatic query building and schema validation

🛡️ Full Type Safety

Complete Pydantic model integration with runtime validation, serialization, and IDE autocomplete

Installation

Local Installation (Recommended)

The Python SDK is included with your Fiberwise installation:

# Navigate to the Python SDK directory
cd fiber-sdk-python/fiberwise-sdk

# Install in development mode
pip install -e .

✅ That's it! The SDK includes everything you need:

  • ✅ LLM provider management (OpenAI, Anthropic, Google AI, local models)
  • ✅ Local database for provider storage
  • ✅ Agent execution and management
  • ✅ Real-time communication capabilities
  • ✅ Type-safe data operations

No additional CLI tools or setup required!

From Local Installation

# The Python SDK is included with Fiberwise
# Navigate to the SDK directory in your Fiberwise installation
cd fiber-sdk-python/fiberwise-sdk
pip install -e .

Quick Start

Setup: Add LLM Providers

Set up your LLM providers directly with the SDK:

from fiberwise_sdk import FiberApp

app = FiberApp()

# Add your LLM providers with custom names
await app.add_provider(
    name="GPT-4 Turbo",
    provider_type="openai",
    api_key="sk-your-openai-key-here",
    model="gpt-4-turbo-preview",
    set_default=True
)

await app.add_provider(
    name="Claude Sonnet", 
    provider_type="anthropic",
    api_key="sk-ant-your-anthropic-key",
    model="claude-3-sonnet-20240229"
)

# Verify your providers
providers = await app.list_providers()
for provider in providers:
    print(f"- {provider['name']} ({provider['provider_type']})")

Configuration Options

The SDK supports two distinct configuration modes:

Option 1: Local Client Mode (Recommended for Development)

Uses local provider database - perfect for development and testing:

import asyncio
from fiberwise_sdk import FiberApp

# Local client - uses local provider database
app = FiberApp({
    'app_id': 'F1BE8ACD-E0F1-4A2B-915E-C3D4E5F6A7B8',  # Default for local development
    'use_local': True,  # Uses local provider database
    
    # Required: Database provider
    'database_provider': 'sqlite',
    'database_url': 'sqlite:///./data/providers.db',
    
    # Required: Storage provider  
    'storage_provider': 'local',
    'storage_config': {'base_path': './storage'}
})

Option 2: Remote API Mode (Production)

Connects to remote Fiberwise instance:

# Remote API client - connects to Fiberwise instance
app = FiberApp({
    'api_key': 'your-api-key',
    'base_url': 'https://api.fiberwise.ai/api/v1',
    'app_id': 'your-app-id'
})

Access SDK Services

agents = app.agents
data = app.data
storage = app.storage

LLM Provider Management

The SDK now includes full provider management capabilities, allowing you to add, configure, and manage LLM providers directly from your Python code.

Provider Management Options

You have two approaches for managing LLM providers:

  1. SDK-based Management (shown below): Add providers programmatically
  2. CLI-based Management: Use the Fiberwise CLI to add providers to remote instances

📖 For CLI-based provider management

See the Instance Routing Documentation for details on adding providers to specific Fiberwise instances using the fiber CLI with --to-instance routing.

Adding Providers

from fiberwise_sdk import FiberApp

app = FiberApp()

# Add OpenAI provider
openai_config = await app.add_provider(
    name="My OpenAI",
    provider_type="openai", 
    api_key="sk-...",
    model="gpt-4",
    temperature=0.7,
    max_tokens=2000,
    set_default=True
)

# Add Anthropic provider
anthropic_config = await app.add_provider(
    name="Claude Assistant",
    provider_type="anthropic",
    api_key="sk-ant-...",
    model="claude-3-sonnet-20240229",
    temperature=0.3
)

# Add Google provider
google_config = await app.add_provider(
    name="Gemini Pro",
    provider_type="google",
    api_key="AIza...",
    model="gemini-1.5-pro",
    base_url="https://generativelanguage.googleapis.com"
)

# Add local Ollama provider
local_config = await app.add_provider(
    name="Local Llama",
    provider_type="local",
    api_key="none",  # Not needed for local
    model="llama2",
    base_url="http://localhost:11434/api"
)

Managing Providers

# List all providers
providers = await app.list_providers()
for provider in providers:
    print(f"- {provider['name']} ({provider['provider_type']})")

# Filter by provider type
openai_providers = await app.list_providers(provider_type="openai")

# Get specific provider
provider = await app.get_provider("My OpenAI")
if provider:
    print(f"Provider: {provider['name']}")
    print(f"Model: {provider['default_model']}")
    print(f"Created: {provider['created_at']}")

# Set default provider
await app.set_default_provider("Claude Assistant")

# Get current default
default = await app.get_default_provider()
if default:
    print(f"Default provider: {default['name']}")

# Delete a provider
deleted = await app.delete_provider("Old Provider")
if deleted:
    print("Provider deleted successfully")

Using Named Providers in Agent Execution

# Execute agent with specific provider
result = await app.agents.execute(
    "my-agent",
    input_data={"query": "Explain quantum computing"},
    metadata={
        "llm_provider": "Claude Assistant",  # Use named provider
        "temperature": 0.1,  # Override default temperature
        "max_tokens": 500
    }
)

# The agent will use the "Claude Assistant" provider configuration
print(result['output'])

Working with Agents

Basic Agent Operations

async def main():
    # List all agents
    agents_response = await app.agents.list(page=1, page_size=10)
    print(f"Found {agents_response.total} agents")
    
    # Get a specific agent
    agent = await app.agents.get("agent-id")
    print(f"Agent: {agent.name} ({agent.agent_type})")
    
    # Execute an agent
    from fiberwise_sdk import AgentExecutionRequest
    request = AgentExecutionRequest(
        input_data={"prompt": "Hello, how are you?"},
        metadata={
            "user_id": "user-123",
            "llm_provider": "OpenAI GPT-4",  # Use named LLM provider
            "temperature": 0.7,
            "max_tokens": 150
        }
    )
    response = await app.agents.execute("agent-id", request)
    print(f"Agent response: {response.activation.output_data}")
    
    # Execute with different provider
    claude_request = AgentExecutionRequest(
        input_data={"prompt": "Explain quantum computing"},
        metadata={
            "user_id": "user-123", 
            "llm_provider": "Anthropic Claude",  # Different named provider
            "temperature": 0.3
        }
    )
    claude_response = await app.agents.execute("agent-id", claude_request)
    print(f"Claude response: {claude_response.activation.output_data}")
    
    # Create a new agent
    new_agent = await app.agents.create(
        name="My AI Assistant",
        description="A helpful AI assistant",
        system_prompt="You are a helpful assistant.",
        model="gpt-4",
        temperature=0.7
    )
    print(f"Created agent: {new_agent.agent_id}")

asyncio.run(main())

Streaming Agent Responses

async def stream_agent_response():
    async def on_message(data):
        print(f"Received: {data}")
    
    async def on_complete(data):
        print(f"Complete: {data}")
    
    async def on_error(error):
        print(f"Error: {error}")
    
    # Stream agent activation
    async for chunk in app.agents.activate_streaming(
        "agent-id",
        input_data={"prompt": "Tell me a story"},
        on_message=on_message,
        on_complete=on_complete,
        on_error=on_error
    ):
        print(f"Stream chunk: {chunk}")

asyncio.run(stream_agent_response())

Data Operations

async def data_operations():
    # Create a new chat (using the Chat model from activation-chat app)
    result = await app.data.create_item("chats", {
        "title": "My First Chat",
        # chat_id (UUID), user_id, and created_at are handled automatically
        # as system fields with defaults (CURRENT_USER, CURRENT_TIMESTAMP, etc.)
    })
    print(f"Created chat: {result}")
    
    # List all chats (with pagination)
    chats_response = await app.data.list_items("chats", page=1, limit=10)
    chats = chats_response.get('items', [])
    print(f"Found {len(chats)} chats")
    
    # Get a specific chat
    chat = await app.data.get_item("chats", "chat-id")
    if chat:
        print(f"Chat: {chat['title']} (ID: {chat['chat_id']})")
        print(f"Created: {chat['created_at']}")
    
    # Update a chat title
    updated_chat = await app.data.update_item("chats", "chat-id", {
        "title": "Updated Chat Title"
    })
    print(f"Updated chat: {updated_chat}")
    
    # Delete a chat
    deleted = await app.data.delete_item("chats", "chat-id")
    if deleted:
        print("Chat deleted successfully")

asyncio.run(data_operations())

Data Models & Fields

The SDK provides dynamic data model management for creating and managing database schemas within your applications. Define custom data structures, field types, and relationships for your app's data storage needs.

📝 Note

Data models define the database table structure and field definitions for storing structured data within apps. For comprehensive app development including data models, see the App Development Guide.

Working with Data Models

async def work_with_models():
    # List available data models in an app
    models = await app.data.list_models("app_id")
    print(f"Available models: {[m['name'] for m in models]}")
    
    # Get model schema details
    chat_model = await app.data.get_model_schema("app_id", "chats")
    print(f"Chat model fields:")
    for field in chat_model['fields']:
        print(f"  • {field['name']}: {field['type']} ({field.get('description', 'No description')})")

asyncio.run(work_with_models())

Dynamic Field Operations

# Create items with dynamic fields
chat_data = {
    "title": "AI Strategy Discussion",
    "participant_count": 3,
    "tags": ["strategy", "ai", "planning"],
    "metadata": {
        "duration_minutes": 45,
        "follow_up_required": True
    }
}

new_chat = await app.data.create_item("chats", chat_data)
print(f"Created chat: {new_chat['chat_id']}")

# Query with field-specific filters
filtered_chats = await app.data.list_items("chats", 
    filters={
        "participant_count": {"gte": 2},
        "tags": {"contains": "ai"},
        "metadata.follow_up_required": True
    },
    sort_by="created_at",
    sort_order="desc"
)

# Update specific fields
await app.data.update_item("chats", chat_id, {
    "metadata.follow_up_required": False,
    "status": "completed"
})

Configuration Details

Local Client Mode (Development)

Uses local provider database - no remote API calls needed:

# Local client configuration
config = {
    'app_id': 'F1BE8ACD-E0F1-4A2B-915E-C3D4E5F6A7B8',  # Default for local development
    'use_local': True,  # Enable local client mode
    
    # Database provider - Required for local mode
    'database_url': 'sqlite:///./data/providers.db',  # Local database path
    'database_provider': 'sqlite',  # Database type: sqlite, postgresql
    
    # Storage provider - Required for local mode  
    'storage_provider': 'local',  # Storage type: local, s3, azure
    'storage_config': {
        'base_path': './storage'  # Local storage directory
    }
}

app = FiberApp(config)

Remote API Mode (Production)

Connects to remote Fiberwise instance:

# Remote API configuration
config = {
    # Authentication - Required for remote
    'api_key': 'your-api-key',  # Your Fiberwise instance API key
    
    # Instance Configuration - Required for remote
    'base_url': 'https://your-fiberwise-instance.com/api/v1',  # Your instance URL
    'app_id': 'your-app-id',  # Required - specific to your instance
    
    # Optional Settings
    'session_id': 'optional-session-id',
}

app = FiberApp(config)

💡 Note

Choose the right mode for your setup:

  • Local Client Mode (use_local=True): Uses local provider database and storage directly. Requires database_provider, database_url, storage_provider, and storage_config. No base_url or api_key needed.
  • Remote API Mode: Requires api_key and base_url to connect to a remote Fiberwise instance. Database and storage are handled by the remote instance.

The app_id is required in both modes. Use F1BE8ACD-E0F1-4A2B-915E-C3D4E5F6A7B8 for local development.

Error Handling

from fiberwise_sdk import (
    FiberWiseError, AuthenticationError, AgentNotFoundError,
    ValidationError, RateLimitError
)

async def robust_agent_call():
    try:
        result = await app.agents.execute("agent-id", request)
        return result
    except AuthenticationError:
        print("Invalid API key")
    except AgentNotFoundError as e:
        print(f"Agent {e.agent_id} not found")
    except ValidationError as e:
        print(f"Invalid request: {e.message}")
    except RateLimitError as e:
        print(f"Rate limited. Retry after {e.retry_after} seconds")
    except FiberWiseError as e:
        print(f"SDK error: {e.message}")

API Reference

Core Classes

FiberApp

Main SDK entry point

FiberApp(config: Dict[str, Any], client: Optional[BaseClient] = None)

Agents

Agent management and execution

  • list(page=1, page_size=10) → PaginatedResponse[Agent]
  • get(agent_id: str) → Agent
  • execute(agent_id: str, request: AgentExecutionRequest) → AgentExecutionResponse
  • create(name: str, **kwargs) → Agent

DynamicData

Database operations

  • create_item(model: str, data: Dict) → Dict
  • list_items(model: str, page: int, limit: int) → Dict
  • get_item(model: str, item_id: str) → Dict
  • update_item(model: str, item_id: str, data: Dict) → Dict
  • delete_item(model: str, item_id: str) → bool

StorageService

File storage operations

  • upload(file, filename: str, **kwargs) → str
  • retrieve(key: str) → Any
  • store(key: str, data: Any, **kwargs) → None
  • delete(key: str) → bool

Model Classes

  • Agent - Agent definition and metadata
  • AgentActivation - Agent execution result
  • AgentExecutionRequest - Execution parameters
  • AgentExecutionResponse - Execution response
  • PaginatedResponse[T] - Type-safe pagination

Exception Classes

  • FiberWiseError - Base exception class
  • AuthenticationError - Authentication failures
  • ValidationError - Input validation errors
  • AgentNotFoundError - Missing agent errors
  • RateLimitError - API rate limiting

API Reference

Complete reference for all Python SDK classes and methods:

FiberApp Class

Main application class for managing the Fiberwise platform integration.

Constructor

FiberApp(config: dict = None)

Parameters:

  • config (dict, optional): Configuration dictionary with app settings

Example:

app = FiberApp({
    'app_id': 'your-app-id',
    'use_local': True,
    'database_url': 'sqlite:///./data/providers.db'
})

LLM Provider Methods

async add_provider(name: str, provider_type: str, api_key: str, model: str, set_default: bool = False) -> dict

Add a new LLM provider to the app.

Parameters:

  • name (str): Custom name for the provider
  • provider_type (str): Provider type ("openai", "anthropic", "google")
  • api_key (str): API key for the provider
  • model (str): Model name to use
  • set_default (bool): Whether to set as default provider

Returns: Provider configuration dictionary

async list_providers() -> list

List all configured LLM providers.

Returns: List of provider dictionaries

async remove_provider(name: str) -> bool

Remove an LLM provider by name.

Parameters:

  • name (str): Name of provider to remove

Returns: True if removed successfully

Agent Methods

async activate_agent(agent_name: str, input_data: dict, **kwargs) -> dict

Activate an agent with input data.

Parameters:

  • agent_name (str): Name of agent to activate
  • input_data (dict): Input data for the agent
  • **kwargs: Additional activation parameters

Returns: Agent activation result

async list_agents() -> list

List all available agents in the app.

Returns: List of agent information dictionaries

Data Operations

app.data.create_item(model: str, data: dict) -> dict

Create a new data item in the specified model.

Parameters:

  • model (str): Data model name
  • data (dict): Item data to create

Returns: Created item with ID

app.data.get_item(model: str, item_id: str) -> dict

Retrieve a specific item by ID.

Parameters:

  • model (str): Data model name
  • item_id (str): ID of item to retrieve

Returns: Item data dictionary

app.data.list_items(model: str, filters: dict = None, limit: int = 100) -> list

List items from a data model with optional filtering.

Parameters:

  • model (str): Data model name
  • filters (dict, optional): Filter criteria
  • limit (int): Maximum items to return (default: 100)

Returns: List of item dictionaries

app.data.update_item(model: str, item_id: str, updates: dict) -> dict

Update an existing item with new data.

Parameters:

  • model (str): Data model name
  • item_id (str): ID of item to update
  • updates (dict): Fields to update

Returns: Updated item data

app.data.delete_item(model: str, item_id: str) -> bool

Delete an item by ID.

Parameters:

  • model (str): Data model name
  • item_id (str): ID of item to delete

Returns: True if deleted successfully

FiberAgent Class

Base class for creating custom Python agents with automatic dependency injection.

Abstract Methods

def run_agent(self, input_data: dict, **kwargs) -> dict

Main agent execution method (must be implemented by subclasses).

Parameters:

  • input_data (dict): Input data for processing
  • **kwargs: Automatically injected dependencies

Returns: Agent result dictionary

Optional Override Methods

def run_agent(self, input_data: dict, service: ServiceType) -> dict

Modern dependency injection via type hints and parameter names.

Parameters: Services automatically injected based on parameter names and type hints

Returns: Agent result dictionary

async def initialize(self)

Called once when agent is first loaded. Use for setup tasks.

async def cleanup(self)

Called when agent is being shut down. Use for cleanup tasks.

Error Classes

Exception classes for handling SDK errors.

FiberwiseSDKError

Base exception class for all SDK errors.

ConfigurationError

Raised when SDK configuration is invalid or incomplete.

AuthenticationError

Raised when API authentication fails.

ProviderError

Raised when LLM provider operations fail.

AgentNotFoundError

Raised when requested agent cannot be found.

RateLimitError

Raised when API rate limits are exceeded.

Support