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:
- SDK-based Management (shown below): Add providers programmatically
- 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. Requiresdatabase_provider
,database_url
,storage_provider
, andstorage_config
. Nobase_url
orapi_key
needed. - Remote API Mode: Requires
api_key
andbase_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 metadataAgentActivation
- Agent execution resultAgentExecutionRequest
- Execution parametersAgentExecutionResponse
- Execution responsePaginatedResponse[T]
- Type-safe pagination
Exception Classes
FiberWiseError
- Base exception classAuthenticationError
- Authentication failuresValidationError
- Input validation errorsAgentNotFoundError
- Missing agent errorsRateLimitError
- 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 providerprovider_type
(str): Provider type ("openai", "anthropic", "google")api_key
(str): API key for the providermodel
(str): Model name to useset_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 activateinput_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 namedata
(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 nameitem_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 namefilters
(dict, optional): Filter criterialimit
(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 nameitem_id
(str): ID of item to updateupdates
(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 nameitem_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
- Documentation: https://docs.fiberwise.ai
- GitHub Issues: https://github.com/fiberwise/fiber-sdk-python/issues
- Discord: Join our community
- Email: [email protected]