# API Reference Source: https://docs.fibonacci.today/api-reference/overview Complete REST API reference for Fibonacci Cloud The Fibonacci REST API allows you to programmatically manage workflows, execute them, and interact with the Fibonacci platform. ## Base URL ``` https://api.fibonacci.dev/v1 ``` ## Authentication All API requests require authentication via API key: ```bash theme={null} curl https://api.fibonacci.dev/v1/workflows \ -H "Authorization: Bearer YOUR_API_KEY" ``` Get your API key from the [Fibonacci Dashboard](https://app.fibonacci.dev/settings/api-keys). ## Quick Reference ### Workflows | Method | Endpoint | Description | | -------- | ----------------- | -------------------- | | `GET` | `/workflows` | List all workflows | | `POST` | `/workflows` | Create a workflow | | `GET` | `/workflows/{id}` | Get workflow details | | `PUT` | `/workflows/{id}` | Update a workflow | | `DELETE` | `/workflows/{id}` | Delete a workflow | ### Execution | Method | Endpoint | Description | | ------ | ------------------------- | -------------------- | | `POST` | `/workflows/{id}/execute` | Execute a workflow | | `GET` | `/executions/{id}` | Get execution status | | `GET` | `/executions` | List executions | | `POST` | `/executions/{id}/cancel` | Cancel execution | ### Workflow Versions | Method | Endpoint | Description | | ------ | --------------------------------------------- | ---------------------------------- | | `GET` | `/workflows/{id}/versions` | List all versions | | `GET` | `/workflows/{id}/versions/{version}` | Inspect a specific version | | `POST` | `/workflows/{id}/versions/{version}/rollback` | Roll back (creates a new snapshot) | ### Workflow Registry | Method | Endpoint | Description | | ------ | ------------------------------------------ | ------------------------------------------------- | | `POST` | `/workflow-registry/{id}/approve-template` | Approve a workflow as an org template | | `POST` | `/workflow-registry/{id}/deprecate` | Mark a template as deprecated | | `POST` | `/workflow-registry/{id}/archive` | Archive a template | | `GET` | `/workflow-registry` | List registry entries (hides archived by default) | Pass `?include_archived=true` to `/workflow-registry` to include archived entries in the response. ### Organizations — Governance | Method | Endpoint | Description | | ------- | ------------------------------------------ | ------------------------------ | | `GET` | `/organizations/{id}/governance` | Get governance settings | | `PATCH` | `/organizations/{id}/governance` | Update governance settings | | `GET` | `/organizations/{id}/governance/report` | Governance summary dashboard | | `GET` | `/organizations/{id}/governance/audit` | Paginated governance audit log | | `GET` | `/organizations/{id}/governance/approvals` | Approval statistics | ### Organizations — Quotas | Method | Endpoint | Description | | ------- | --------------------------- | ----------------------------------------- | | `GET` | `/organizations/{id}/quota` | Get quota limits (plan + overrides) | | `PATCH` | `/organizations/{id}/quota` | Set quota overrides (owner role required) | ### Organizations — Tool Policies | Method | Endpoint | Description | | -------- | ------------------------------------------- | ---------------------------- | | `GET` | `/organizations/{id}/tool-policies` | List org-level tool policies | | `POST` | `/organizations/{id}/tool-policies` | Create a tool policy | | `PATCH` | `/organizations/{id}/tool-policies/{tool}` | Update a tool policy | | `DELETE` | `/organizations/{id}/tool-policies/{tool}` | Delete a tool policy | | `GET` | `/organizations/{id}/role-policy-overrides` | List role-level overrides | | `POST` | `/organizations/{id}/role-policy-overrides` | Create a role override | ### Memory | Method | Endpoint | Description | | -------- | --------------- | ------------------- | | `GET` | `/memory/{key}` | Get memory value | | `PUT` | `/memory/{key}` | Set memory value | | `DELETE` | `/memory/{key}` | Delete memory value | ### Tools | Method | Endpoint | Description | | ------ | ----------------------- | -------------------- | | `GET` | `/tools` | List available tools | | `GET` | `/tools/{name}` | Get tool details | | `POST` | `/tools/{name}/connect` | Connect a tool | ### Health | Method | Endpoint | Description | | ------ | ----------------- | ---------------------------------- | | `GET` | `/health` | Service health check | | `GET` | `/health/secrets` | Secrets backend connectivity check | ## Request Format All requests should include: ```bash theme={null} Content-Type: application/json Authorization: Bearer YOUR_API_KEY ``` ## Response Format Successful responses return JSON: ```json theme={null} { "data": { ... }, "meta": { "request_id": "req_abc123" } } ``` Error responses: ```json theme={null} { "error": { "code": "validation_error", "message": "Invalid input", "details": [...] } } ``` ## Rate Limits | Plan | Requests/minute | Requests/day | | ---------- | --------------- | ------------ | | Free | 60 | 1,000 | | Pro | 300 | 50,000 | | Enterprise | Custom | Custom | Rate limit headers are included in responses: ``` X-RateLimit-Limit: 60 X-RateLimit-Remaining: 45 X-RateLimit-Reset: 1706054400 ``` ## SDKs We recommend using our official SDKs: * **Python**: `pip install fibonacci-sdk` * **Node.js**: `npm install @fibonacci/sdk` (coming soon) ## Example: Execute a Workflow ```bash theme={null} curl -X POST https://api.fibonacci.dev/v1/workflows/wf-abc123/execute \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "inputs": { "text": "Analyze this text" } }' ``` Response: ```json theme={null} { "data": { "execution_id": "exec_abc123", "status": "completed", "outputs": { "analyzer": "The text sentiment is positive..." }, "duration_ms": 1523, "nodes_executed": 3, "nodes_failed": 0 } } ``` ## Example: Check Secrets Backend Health ```bash theme={null} curl https://api.fibonacci.dev/v1/health/secrets \ -H "Authorization: Bearer YOUR_API_KEY" ``` Response: ```json theme={null} { "status": "healthy", "backend": "aws_secrets_manager", "checks": { "list_secrets": "ok", "get_secret": "ok" } } ``` See the [Security guide](/guides/security) for IAM requirements and the [Governance guide](/guides/governance) for org-level policy endpoints. # Authentication Source: https://docs.fibonacci.today/getting-started/authentication Configure API keys and secure credential storage ## Getting Your API Key Create an account at [fibonacci.today](https://fibonacci.today/signup) Go to your [Dashboard](https://app.fibonacci.today) → Settings → API Keys Click "Create New Key" and give it a descriptive name Copy the key immediately - it won't be shown again! API keys start with `fib_live_` (production) or `fib_test_` (sandbox). Keep them secure and never share them publicly. ## Authentication Methods Fibonacci supports multiple ways to provide your API key, in order of security: | Method | Security Level | Best For | | -------------------- | -------------- | ----------------------- | | System Keychain | 🔐 Highest | Production environments | | Environment Variable | 🔒 High | CI/CD, containers | | `.env` File | ⚠️ Medium | Local development | | Direct Parameter | ❌ Low | Testing only | *** ## Method 1: System Keychain (Recommended) The most secure option. Your API key is encrypted by the operating system. ### Save to Keychain ```python theme={null} from fibonacci import save_api_key_secure # Save once - it's stored encrypted save_api_key_secure("fib_live_your_api_key_here") ``` ### Use in Your Code ```python theme={null} from fibonacci import Workflow, get_config_secure # Automatically loads from keychain config = get_config_secure() wf = Workflow(name="My Workflow", config=config) wf.deploy() # Uses keychain API key ``` ### Supported Platforms | Platform | Backend | | -------- | --------------------------------------- | | macOS | Keychain Services | | Windows | Credential Manager | | Linux | Secret Service (GNOME Keyring, KWallet) | ### Check Security Status ```python theme={null} from fibonacci import check_security_status status = check_security_status() print(f"Security Level: {status['security_level']}") print(f"API Key Location: {'Keychain' if status['api_key_in_keychain'] else 'Environment'}") ``` ### CLI: Manage Credentials ```bash theme={null} # Save API key interactively fibonacci security save # Check security status fibonacci security status # Migrate from .env to keychain fibonacci security migrate ``` *** ## Method 2: Environment Variable Good for CI/CD pipelines and containerized deployments. ### Set the Variable ```bash Linux/macOS theme={null} export FIBONACCI_API_KEY=fib_live_your_api_key_here ``` ```powershell Windows PowerShell theme={null} $env:FIBONACCI_API_KEY = "fib_live_your_api_key_here" ``` ```yaml GitHub Actions theme={null} env: FIBONACCI_API_KEY: ${{ secrets.FIBONACCI_API_KEY }} ``` ```yaml Docker theme={null} environment: - FIBONACCI_API_KEY=fib_live_your_api_key_here ``` ### Use in Your Code ```python theme={null} from fibonacci import Workflow from fibonacci.config import Config # Automatically loads from environment config = Config.from_env() wf = Workflow(name="My Workflow", config=config) ``` *** ## Method 3: `.env` File Convenient for local development. **Never commit to version control.** ### Create `.env` File ```bash .env theme={null} FIBONACCI_API_KEY=fib_live_your_api_key_here FIBONACCI_BASE_URL=https://api.fibonacci.ai FIBONACCI_TIMEOUT=300 FIBONACCI_DEBUG=false ``` ### Add to `.gitignore` ```bash .gitignore theme={null} # API Keys - Never commit! .env .env.local .env.*.local ``` ### Use in Your Code ```python theme={null} from dotenv import load_dotenv from fibonacci import Workflow from fibonacci.config import Config load_dotenv() # Load .env file config = Config.from_env() wf = Workflow(name="My Workflow", config=config) ``` *** ## Method 4: Direct Parameter Only use for quick testing. **Never use in production code.** ```python theme={null} # ⚠️ Warning: Only for testing! wf.deploy(api_key="fib_live_your_api_key_here") ``` *** ## Configuration Options All available configuration options: | Variable | Default | Description | | ------------------------- | -------------------------- | -------------------------- | | `FIBONACCI_API_KEY` | - | Your API key (required) | | `FIBONACCI_BASE_URL` | `https://api.fibonacci.ai` | API endpoint | | `FIBONACCI_TIMEOUT` | `300` | Request timeout in seconds | | `FIBONACCI_MAX_RETRIES` | `3` | Max retry attempts | | `FIBONACCI_VERIFY_SSL` | `true` | Verify SSL certificates | | `FIBONACCI_DEBUG` | `false` | Enable debug logging | | `FIBONACCI_LOG_REQUESTS` | `false` | Log API requests | | `FIBONACCI_LOG_RESPONSES` | `false` | Log API responses | ### Programmatic Configuration ```python theme={null} from fibonacci.config import Config config = Config( api_key="fib_live_xxx", base_url="https://api.fibonacci.ai", timeout=300, max_retries=3, verify_ssl=True, debug=False ) ``` *** ## Migrating to Secure Storage If you're currently using a `.env` file, migrate to keychain storage: ### Option 1: CLI ```bash theme={null} fibonacci security migrate ``` ### Option 2: Python ```python theme={null} from fibonacci import migrate_to_keychain if migrate_to_keychain(): print("✅ Migrated! You can now delete FIBONACCI_API_KEY from .env") else: print("❌ Migration failed") ``` *** ## Verifying Authentication Test that your API key works: ```python theme={null} from fibonacci import FibonacciClient from fibonacci.config import Config config = Config.from_env() async def test_auth(): async with FibonacciClient(config) as client: workflows = await client.list_workflows() print(f"✅ Authenticated! Found {len(workflows)} workflows") import asyncio asyncio.run(test_auth()) ``` Or use the CLI: ```bash theme={null} fibonacci list ``` If you see your workflows listed, authentication is working correctly! ## Next Steps Start building with Fibonacci Learn more about securing your workflows # Installation Source: https://docs.fibonacci.today/getting-started/installation Install and configure the Fibonacci SDK ## System Requirements Python 3.11 or higher is required macOS, Linux, Windows (64-bit) ## Installation Methods ### Using pip (Recommended) ```bash theme={null} pip install fibonacci-sdk ``` ### Using Poetry ```bash theme={null} poetry add fibonacci-sdk ``` ### Using uv ```bash theme={null} uv add fibonacci-sdk ``` ### From Source ```bash theme={null} git clone https://github.com/fibonacci-ai/fibonacci-sdk.git cd fibonacci-sdk pip install -e . ``` ## Optional Dependencies ### Security Features For secure keychain storage and enhanced security features: ```bash theme={null} pip install fibonacci-sdk[security] ``` This includes: * `keyring` - System keychain integration (macOS Keychain, Windows Credential Manager, Linux Secret Service) * Secure API key storage * Migration tools from plaintext to encrypted storage ### Development Dependencies For contributing to Fibonacci or running tests: ```bash theme={null} pip install fibonacci-sdk[dev] ``` This includes: * `pytest` - Testing framework * `pytest-asyncio` - Async test support * `black` - Code formatting * `ruff` - Linting * `mypy` - Type checking ### Full Installation Install everything: ```bash theme={null} pip install fibonacci-sdk[security,dev] ``` ## Verify Installation After installation, verify everything is working: ```python theme={null} import fibonacci # Check version print(f"Fibonacci SDK v{fibonacci.__version__}") # Check available classes from fibonacci import ( Workflow, LLMNode, ToolNode, CriticNode, ConditionalNode, Memory, FibonacciClient ) print("✅ All core classes imported successfully!") ``` You should see: ``` Fibonacci SDK v0.1.0 ✅ All core classes imported successfully! ``` ## CLI Installation The SDK includes a command-line interface. After installation, the `fibonacci` command is available: ```bash theme={null} fibonacci version ``` Output: ``` _____ _ _ _ ____ ____ _ __ | ___(_) |__ ___ _ __ __ _ ___ ___(_) / ___|| _ \| |/ / | |_ | | '_ \ / _ \| '_ \ / _` |/ __/ __| | \___ \| | | | ' / | _| | | |_) | (_) | | | | (_| | (_| (__| | ___) | |_| | . \ |_| |_|_.__/ \___/|_| |_|\__,_|\___\___|_| |____/|____/|_|\_\ AI Workflow Automation • Build. Deploy. Execute. v0.1.0 ``` ## Troubleshooting Make sure you've activated your virtual environment: ```bash theme={null} source venv/bin/activate # Linux/macOS .\venv\Scripts\activate # Windows ``` Then reinstall: ```bash theme={null} pip install fibonacci-sdk ``` Fibonacci requires Python 3.11+. Check your version: ```bash theme={null} python --version ``` If needed, install a newer Python version or use pyenv: ```bash theme={null} pyenv install 3.11.0 pyenv local 3.11.0 ``` On Linux, you may need to install the Secret Service backend: ```bash theme={null} # Ubuntu/Debian sudo apt-get install gnome-keyring # Fedora sudo dnf install gnome-keyring ``` If you're behind a corporate proxy: ```bash theme={null} pip install --trusted-host pypi.org --trusted-host files.pythonhosted.org fibonacci-sdk ``` ## Next Steps Configure your API key securely Build your first workflow # Quickstart Source: https://docs.fibonacci.today/getting-started/quickstart Get up and running with Fibonacci in under 5 minutes ## Prerequisites Before you begin, make sure you have: Fibonacci requires Python 3.11 or higher Sign up at [fibonacci.today](https://fibonacci.today) to get your API key ## Installation Install the Fibonacci SDK using pip: ```bash pip theme={null} pip install fibonacci-sdk ``` ```bash poetry theme={null} poetry add fibonacci-sdk ``` ```bash uv theme={null} uv add fibonacci-sdk ``` For secure API key storage, install with the security extras: ```bash theme={null} pip install fibonacci-sdk[security] ``` ## Set Up Authentication Create a `.env` file in your project root: ```bash .env theme={null} FIBONACCI_API_KEY=fib_live_your_api_key_here ``` Never commit your `.env` file to version control. Add it to your `.gitignore`. Or use secure keychain storage (recommended for production): ```python theme={null} from fibonacci import save_api_key_secure # Save to system keychain (macOS Keychain, Windows Credential Manager, etc.) save_api_key_secure("fib_live_your_api_key_here") ``` ## Your First Workflow Let's create a simple workflow that uses AI to analyze text: ```python theme={null} from fibonacci import Workflow, LLMNode from dotenv import load_dotenv import os # Load environment variables load_dotenv() # Create a workflow wf = Workflow( name="Text Analyzer", description="Analyzes text for sentiment and key points" ) # Add an LLM node analyze = LLMNode( id="analyze_text", name="Analyze Text", instruction=""" Analyze the following text and provide: 1. Overall sentiment (positive, negative, neutral) 2. Key points (bullet list) 3. Suggested actions Text: {{input.text}} """ ) wf.add_node(analyze) # Deploy to Fibonacci workflow_id = wf.deploy(api_key=os.getenv("FIBONACCI_API_KEY")) print(f"✅ Deployed! Workflow ID: {workflow_id}") # Execute the workflow result = wf.run(input_data={ "text": "Our Q4 sales exceeded expectations by 15%. The team did an amazing job!" }) print(f"Status: {result.status}") print(f"Output: {result.output_data['analyze_text']}") ``` ## Expected Output ``` ✅ Deployed! Workflow ID: wf_abc123xyz Status: completed Output: **Sentiment:** Positive **Key Points:** - Q4 sales exceeded expectations by 15% - Team performance was exceptional **Suggested Actions:** - Recognize and reward the team - Analyze what contributed to success - Set ambitious Q1 targets ``` ## What's Next? Create a complete workflow with multiple nodes and tool integrations Understand the different node types and when to use each Connect to Google Sheets, Slack, and 100+ other services Learn how to secure your API keys and audit your workflows **Congratulations!** You've just created and deployed your first Fibonacci workflow. # Your First Workflow Source: https://docs.fibonacci.today/getting-started/your-first-workflow Build a complete workflow with multiple nodes and tool integrations In this guide, you'll build a complete workflow that: 1. Reads data from a Google Sheet 2. Analyzes the data with AI 3. Generates a summary report ## Prerequisites Make sure you've completed the [Quickstart](/getting-started/quickstart) and have your API key configured. ## Step 1: Create the Workflow Start by creating a new Python file: ```python sales_report.py theme={null} from fibonacci import Workflow, LLMNode, ToolNode from dotenv import load_dotenv import os load_dotenv() # Create the workflow wf = Workflow( name="Sales Report Generator", description="Reads sales data and generates an analysis report", tags=["sales", "reporting", "weekly"] ) ``` ## Step 2: Add the Data Source Node Use a `ToolNode` to read data from Google Sheets: ```python theme={null} # Node 1: Read sales data from Google Sheets read_data = ToolNode( id="read_sales", name="Read Sales Data", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "{{input.range}}" } ) wf.add_node(read_data) ``` The `{{input.sheet_id}}` syntax is a **template variable**. It will be replaced with actual values when you run the workflow. ## Step 3: Add the Analysis Node Use an `LLMNode` to analyze the data: ```python theme={null} # Node 2: Analyze the sales data analyze = LLMNode( id="analyze_data", name="Analyze Sales Data", instruction=""" Analyze this sales data and provide: 1. **Total Revenue**: Sum of all sales 2. **Top Performers**: Top 3 salespeople by revenue 3. **Trends**: Week-over-week comparison 4. **Concerns**: Any red flags or areas needing attention Sales Data: {{read_sales}} Format your response with clear sections and bullet points. """, dependencies=["read_sales"], # This node depends on read_sales model="claude-sonnet-4-5-20250929", max_tokens=2000 ) wf.add_node(analyze) ``` ## Step 4: Add the Summary Node Add another LLM node to create an executive summary: ```python theme={null} # Node 3: Generate executive summary summary = LLMNode( id="executive_summary", name="Executive Summary", instruction=""" Based on this analysis, create a brief executive summary (3-4 sentences) suitable for a busy executive. Focus on the most important insights and any actions needed. Analysis: {{analyze_data}} """, dependencies=["analyze_data"] ) wf.add_node(summary) ``` ## Step 5: Complete Workflow Code Here's the complete workflow: ```python sales_report.py theme={null} from fibonacci import Workflow, LLMNode, ToolNode from dotenv import load_dotenv import os load_dotenv() # Create the workflow wf = Workflow( name="Sales Report Generator", description="Reads sales data and generates an analysis report", tags=["sales", "reporting", "weekly"] ) # Node 1: Read data read_data = ToolNode( id="read_sales", name="Read Sales Data", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "{{input.range}}" } ) # Node 2: Analyze analyze = LLMNode( id="analyze_data", name="Analyze Sales Data", instruction=""" Analyze this sales data and provide: 1. **Total Revenue**: Sum of all sales 2. **Top Performers**: Top 3 salespeople by revenue 3. **Trends**: Week-over-week comparison 4. **Concerns**: Any red flags or areas needing attention Sales Data: {{read_sales}} """, dependencies=["read_sales"] ) # Node 3: Summary summary = LLMNode( id="executive_summary", name="Executive Summary", instruction=""" Create a brief executive summary (3-4 sentences) based on: {{analyze_data}} """, dependencies=["analyze_data"] ) # Add all nodes wf.add_nodes([read_data, analyze, summary]) if __name__ == "__main__": # Validate the workflow wf.validate() print("✅ Workflow validated!") # Deploy api_key = os.getenv("FIBONACCI_API_KEY") workflow_id = wf.deploy(api_key=api_key) print(f"✅ Deployed! ID: {workflow_id}") # Execute result = wf.run(input_data={ "sheet_id": "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms", # Sample Sheet "range": "Sales!A1:E10" }) # Print results print(f"\n📊 Status: {result.status}") print(f"⏱️ Duration: {result.duration_seconds:.2f}s") print(f"💰 Cost: ${result.total_cost:.4f}") print("\n" + "="*60) print("📈 ANALYSIS") print("="*60) print(result.output_data["analyze_data"]) print("\n" + "="*60) print("📋 EXECUTIVE SUMMARY") print("="*60) print(result.output_data["executive_summary"]) ``` ## Step 6: Run Your Workflow Execute the workflow: ```bash theme={null} python sales_report.py ``` Expected output: ``` ✅ Workflow validated! ✅ Deployed! ID: wf_abc123xyz 📊 Status: completed ⏱️ Duration: 4.32s 💰 Cost: $0.0023 ============================================================ 📈 ANALYSIS ============================================================ ## Total Revenue $1,234,567 for the period ## Top Performers 1. Sarah Johnson - $245,000 2. Mike Chen - $198,000 3. Lisa Park - $187,000 ## Trends - 12% increase from last week - Strong performance in Enterprise segment ## Concerns - SMB segment down 8% - Two reps below quota ============================================================ 📋 EXECUTIVE SUMMARY ============================================================ Sales exceeded targets by 12% this week, driven by strong Enterprise performance from Sarah Johnson and Mike Chen. However, the SMB segment declined 8%, requiring immediate attention. Recommend a team meeting to address underperforming segments. ``` ## Understanding the Workflow Workflow Diagram ### How Nodes Connect 1. **read\_sales** runs first (no dependencies) 2. **analyze\_data** waits for read\_sales to complete 3. **executive\_summary** waits for analyze\_data to complete ### Template Variables * `{{input.sheet_id}}` - Replaced with input data at runtime * `{{read_sales}}` - Replaced with the output of the read\_sales node * `{{analyze_data}}` - Replaced with the output of the analyze\_data node ## Adding Conditional Logic Want to send an alert only if sales drop? Add a conditional node: ```python theme={null} from fibonacci import ConditionalNode # Check if we need an alert check_alert = ConditionalNode( id="check_alert", name="Check if Alert Needed", left_value="{{analyze_data}}", operator="contains", right_value="down", true_branch=["send_alert"], false_branch=[], dependencies=["analyze_data"] ) # Send Slack alert (only runs if condition is true) send_alert = ToolNode( id="send_alert", name="Send Slack Alert", tool="slack_send_message", params={ "channel": "#sales-alerts", "message": "⚠️ Sales Alert!\n\n{{executive_summary}}" }, dependencies=["check_alert", "executive_summary"] ) ``` ## Exporting to YAML You can also define workflows in YAML: ```python theme={null} # Export your workflow to YAML wf.to_yaml("sales_report.yaml") ``` This creates: ```yaml sales_report.yaml theme={null} name: Sales Report Generator description: Reads sales data and generates an analysis report version: 1 is_active: true tags: - sales - reporting - weekly nodes: - id: read_sales type: tool name: Read Sales Data tool_name: google_sheets_read tool_params: spreadsheet_id: "{{input.sheet_id}}" range: "{{input.range}}" - id: analyze_data type: llm name: Analyze Sales Data instruction: | Analyze this sales data and provide: 1. **Total Revenue**: Sum of all sales 2. **Top Performers**: Top 3 salespeople by revenue ... dependencies: - read_sales - id: executive_summary type: llm name: Executive Summary instruction: | Create a brief executive summary... dependencies: - analyze_data ``` ## Next Steps Learn about all available node types Connect to 100+ tools and services Build dynamic workflows with branching Persist data across workflow runs **Great job!** You've built a real workflow with multiple nodes, dependencies, and AI analysis. # Best Practices Source: https://docs.fibonacci.today/guides/best-practices Production patterns and recommendations for building reliable Fibonacci workflows This guide covers proven patterns and recommendations for building reliable, maintainable, and scalable Fibonacci workflows in production. ## Workflow Design ### Keep Workflows Focused Each workflow should do one thing well: ```python theme={null} # ✅ Good - focused workflows customer_classifier = Workflow(name="classify-customer-inquiry") response_generator = Workflow(name="generate-support-response") escalation_handler = Workflow(name="handle-escalation") # ❌ Avoid - monolithic workflow doing everything support_system = Workflow(name="complete-customer-support-system") ``` ### Use Meaningful Names ```python theme={null} # ✅ Good - descriptive names workflow = Workflow(name="quarterly-sales-report-generator") sentiment_node = LLMNode( id="analyze_customer_sentiment", ... ) # ❌ Avoid - vague names workflow = Workflow(name="processor") node = LLMNode(id="node1", ...) ``` ### Document Your Workflows ```python theme={null} workflow = Workflow( name="customer-churn-predictor", description=""" Analyzes customer behavior patterns to predict churn risk. Inputs: - customer_id: Customer identifier - timeframe: Analysis period (default: 90 days) Outputs: - risk_score: 0-100 churn probability - factors: Contributing factors - recommendations: Retention actions Owner: data-science@company.com Last updated: 2025-01-15 """ ) ``` ## Node Patterns ### Chain of Responsibility Break complex tasks into specialized nodes: ```python theme={null} from fibonacci import Workflow, LLMNode workflow = Workflow(name="content-pipeline") # Step 1: Generate outline outliner = LLMNode( id="create_outline", model="claude-sonnet-4-5-20250929", prompt="Create an outline for: {{input.topic}}" ) # Step 2: Write draft writer = LLMNode( id="write_draft", model="claude-sonnet-4-5-20250929", prompt="Write content following this outline: {{create_outline}}", dependencies=["create_outline"] ) # Step 3: Edit and polish editor = LLMNode( id="edit_content", model="claude-sonnet-4-5-20250929", prompt="Edit for clarity and style: {{write_draft}}", dependencies=["write_draft"] ) # Step 4: Final review reviewer = LLMNode( id="final_review", model="claude-opus-4-5-20251101", # Premium model for final check prompt="Final quality review: {{edit_content}}", dependencies=["edit_content"] ) ``` ### Parallel Processing Maximize throughput with parallel execution: ```python theme={null} workflow = Workflow(name="multi-analysis") # These run in parallel (no dependencies) sentiment = LLMNode( id="sentiment", model="claude-haiku-4-5-20251001", prompt="Sentiment: {{input.text}}" ) entities = LLMNode( id="entities", model="claude-haiku-4-5-20251001", prompt="Extract entities: {{input.text}}" ) summary = LLMNode( id="summary", model="claude-haiku-4-5-20251001", prompt="Summarize: {{input.text}}" ) # Combine results combine = LLMNode( id="combine", model="claude-sonnet-4-5-20250929", prompt=""" Combine these analyses: Sentiment: {{sentiment}} Entities: {{entities}} Summary: {{summary}} """, dependencies=["sentiment", "entities", "summary"] ) ``` ### Critic Pattern for Quality Use CriticNode for iterative improvement: ```python theme={null} from fibonacci import LLMNode, CriticNode writer = LLMNode( id="writer", model="claude-sonnet-4-5-20250929", prompt="Write a professional email: {{input.request}}" ) critic = CriticNode( id="quality_critic", model="claude-sonnet-4-5-20250929", target_node="writer", criteria=[ "Professional tone throughout", "Clear and concise language", "Proper grammar and spelling", "Appropriate length (not too long)", "Clear call to action" ], min_score=8, max_iterations=3, improvement_prompt=""" Improve the email based on this feedback: {{feedback}} Original email: {{writer}} """ ) ``` ## Prompt Engineering ### Use Clear Structure ```python theme={null} analyzer = LLMNode( id="analyzer", model="claude-sonnet-4-5-20250929", prompt="""You are analyzing customer feedback for a software company. Product: {{input.product_name}} Customer tier: {{input.customer_tier}} {{input.feedback_text}} 1. Identify the main issue or request 2. Assess the urgency (low/medium/high/critical) 3. Categorize: bug, feature_request, question, complaint, praise 4. Suggest an appropriate response approach Return JSON: { "main_issue": "...", "urgency": "...", "category": "...", "response_approach": "..." } """ ) ``` ### Include Examples (Few-Shot) ```python theme={null} classifier = LLMNode( id="intent_classifier", model="claude-sonnet-4-5-20250929", prompt="""Classify the customer intent. Input: "I can't log into my account" Output: {"intent": "technical_support", "urgency": "high"} Input: "Do you offer annual billing?" Output: {"intent": "billing_question", "urgency": "low"} Input: "This is unacceptable, I want a refund!" Output: {"intent": "complaint", "urgency": "high"} Input: "Love the new feature!" Output: {"intent": "feedback_positive", "urgency": "low"} {{input.message}} Classify this input following the same format as the examples.""" ) ``` ### Specify Output Format ```python theme={null} # Structured JSON output structured_node = LLMNode( id="structured", model="claude-sonnet-4-5-20250929", prompt="Analyze: {{input.text}}", output_format="json", output_schema={ "type": "object", "required": ["sentiment", "confidence", "keywords"], "properties": { "sentiment": {"enum": ["positive", "negative", "neutral"]}, "confidence": {"type": "number", "minimum": 0, "maximum": 1}, "keywords": {"type": "array", "items": {"type": "string"}} } } ) ``` ## Error Handling ### Always Set Timeouts ```python theme={null} # Node-level timeout node = LLMNode( id="analyzer", model="claude-opus-4-5-20251101", prompt="...", timeout=60 # 60 seconds ) # Workflow-level timeout result = workflow.execute( inputs=data, timeout=300 # 5 minutes for entire workflow ) ``` ### Implement Graceful Degradation ```python theme={null} from fibonacci import Workflow, LLMNode workflow = Workflow(name="resilient-analysis") # Premium analysis with expensive model premium = LLMNode( id="premium_analysis", model="claude-opus-4-5-20251101", prompt="Detailed analysis: {{input.text}}", timeout=60, on_error="standard_analysis" ) # Standard fallback standard = LLMNode( id="standard_analysis", model="claude-sonnet-4-5-20250929", prompt="Analysis: {{input.text}}", timeout=30, on_error="basic_analysis" ) # Basic fallback basic = LLMNode( id="basic_analysis", model="claude-haiku-4-5-20251001", prompt="Quick analysis: {{input.text}}", timeout=15 ) ``` ### Use Retry with Backoff ```python theme={null} from fibonacci import ToolNode, RetryConfig api_node = ToolNode( id="external_api", tool="http.request", inputs={"url": "https://api.example.com"}, retry=RetryConfig( max_attempts=5, delay=1.0, backoff="exponential", max_delay=30.0, jitter=True, retry_on=["timeout", "5xx", "rate_limit"] ) ) ``` ## Performance Optimization ### Choose the Right Model | Use Case | Recommended Model | | --------------------- | -------------------------- | | Simple classification | claude-haiku-4-5-20251001 | | General tasks | claude-sonnet-4-5-20250929 | | Complex reasoning | claude-opus-4-5-20251101 | | High volume, low cost | claude-haiku-4-5-20251001 | ```python theme={null} # Use Haiku for simple, high-volume tasks classifier = LLMNode( id="quick_classify", model="claude-haiku-4-5-20251001", # Fast and cost-effective prompt="Classify as spam/not_spam: {{input.text}}" ) # Use Opus for complex analysis deep_analyzer = LLMNode( id="deep_analysis", model="claude-opus-4-5-20251101", # Most capable prompt="Provide comprehensive analysis: {{input.text}}" ) ``` ### Optimize Token Usage ```python theme={null} # Be concise in prompts # ❌ Verbose verbose = LLMNode( id="verbose", prompt="""I would like you to please analyze the following text and provide me with a detailed sentiment analysis. The text is provided below for your analysis...""" ) # ✅ Concise concise = LLMNode( id="concise", prompt="Analyze sentiment: {{input.text}}\nReturn: positive/negative/neutral" ) ``` ### Cache Expensive Operations ```python theme={null} from fibonacci import Workflow from fibonacci.patterns import cached workflow = Workflow(name="cached-workflow") @cached(ttl=3600) # Cache for 1 hour def expensive_analysis(text: str): return workflow.execute(inputs={"text": text}) # Or use memory-based caching def get_or_compute(workflow, key: str, compute_fn): cached = workflow.memory.get(f"cache:{key}", scope="organization") if cached: return cached result = compute_fn() workflow.memory.set( f"cache:{key}", result, scope="organization", ttl=3600 ) return result ``` ## Testing ### Unit Test Nodes ```python theme={null} import pytest from fibonacci import LLMNode from fibonacci.testing import MockLLM def test_sentiment_classifier(): # Mock the LLM response mock = MockLLM(responses={ "positive": '{"sentiment": "positive", "confidence": 0.95}' }) node = LLMNode( id="sentiment", model=mock, prompt="Classify: {{input.text}}" ) result = node.execute({"text": "I love this product!"}) assert result["sentiment"] == "positive" assert result["confidence"] > 0.9 ``` ### Integration Test Workflows ```python theme={null} def test_support_workflow(): workflow = Workflow.from_yaml("workflows/support.yaml") # Test happy path result = workflow.execute( inputs={ "message": "How do I reset my password?", "customer_id": "test-123" } ) assert result["category"] == "technical_support" assert "password" in result["response"].lower() def test_error_handling(): workflow = Workflow.from_yaml("workflows/support.yaml") # Test with invalid input with pytest.raises(ValidationError): workflow.execute(inputs={"message": ""}) ``` ### Use Test Fixtures ```python theme={null} import pytest from fibonacci import Workflow @pytest.fixture def sample_workflow(): return Workflow.from_yaml("test_fixtures/sample.yaml") @pytest.fixture def test_inputs(): return { "text": "Sample text for testing", "user_id": "test-user" } def test_workflow_execution(sample_workflow, test_inputs): result = sample_workflow.execute(inputs=test_inputs) assert "output" in result ``` ## Deployment ### Environment Separation ```yaml theme={null} # fibonacci.yaml environments: development: model_default: claude-haiku-4-5-20251001 # Cheaper for dev timeout_default: 30 log_level: debug staging: model_default: claude-sonnet-4-5-20250929 timeout_default: 60 log_level: info production: model_default: claude-sonnet-4-5-20250929 timeout_default: 120 log_level: warning retry_default: 3 ``` ### Version Your Workflows ```yaml theme={null} name: customer-support version: "2.1.0" # Semantic versioning changelog: | 2.1.0 - Added escalation handling 2.0.0 - Breaking: Changed output format 1.1.0 - Added sentiment analysis 1.0.0 - Initial release ``` ### CI/CD Pipeline ```yaml theme={null} # .github/workflows/deploy.yaml name: Deploy Workflows on: push: branches: [main] pull_request: branches: [main] jobs: validate: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: '3.11' - run: pip install fibonacci-sdk - run: fibonacci validate workflows/ --strict test: needs: validate runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: '3.11' - run: pip install fibonacci-sdk pytest - run: pytest tests/ deploy-staging: needs: test if: github.ref == 'refs/heads/main' runs-on: ubuntu-latest environment: staging steps: - uses: actions/checkout@v4 - run: pip install fibonacci-sdk - run: fibonacci deploy --env staging env: FIBONACCI_API_KEY: ${{ secrets.FIBONACCI_API_KEY }} deploy-production: needs: deploy-staging if: github.ref == 'refs/heads/main' runs-on: ubuntu-latest environment: production steps: - uses: actions/checkout@v4 - run: pip install fibonacci-sdk - run: fibonacci deploy --env production env: FIBONACCI_API_KEY: ${{ secrets.FIBONACCI_API_KEY_PROD }} ``` ## Monitoring ### Track Key Metrics ```python theme={null} from fibonacci import Workflow from fibonacci.metrics import MetricsCollector metrics = MetricsCollector( backend="datadog", # or "prometheus", "cloudwatch" config={"api_key": "..."} ) workflow = Workflow( name="monitored-workflow", metrics=metrics ) # Tracked automatically: # - fibonacci.workflow.duration # - fibonacci.workflow.success_rate # - fibonacci.node.duration # - fibonacci.node.token_usage # - fibonacci.errors.count ``` ### Set Up Alerts ```yaml theme={null} # alerts.yaml alerts: - name: high-error-rate condition: fibonacci.errors.count > 10 window: 5m severity: critical notify: ["pagerduty", "slack:#alerts"] - name: slow-response condition: fibonacci.workflow.duration.p95 > 30s window: 15m severity: warning notify: ["slack:#monitoring"] - name: high-token-usage condition: fibonacci.node.token_usage > 100000 window: 1h severity: info notify: ["email:team@company.com"] ``` ## Checklist * [ ] Workflows are focused and single-purpose * [ ] Names are descriptive and consistent * [ ] Documentation is complete * [ ] Dependencies are clearly defined * [ ] Timeouts set on all nodes and workflows * [ ] Retry configuration with backoff * [ ] Fallback handlers for critical paths * [ ] Input validation schemas defined * [ ] Appropriate models selected for each task * [ ] Parallel execution where possible * [ ] Caching for expensive operations * [ ] Token usage optimized * [ ] Unit tests for individual nodes * [ ] Integration tests for workflows * [ ] Edge cases and error scenarios covered * [ ] Test data fixtures maintained * [ ] Environment separation configured * [ ] Semantic versioning applied * [ ] CI/CD pipeline set up * [ ] Rollback procedures documented * [ ] Key metrics tracked * [ ] Alerts configured * [ ] Logging enabled * [ ] Dashboards created ## Next Steps Secure your workflows Handle failures gracefully # Command Line Interface Source: https://docs.fibonacci.today/guides/cli Use the Fibonacci CLI for workflow management, execution, and development The Fibonacci CLI provides powerful commands for managing workflows, executing them locally or remotely, and streamlining development workflows. ## Installation The CLI is included with the Fibonacci SDK: ```bash theme={null} pip install fibonacci-sdk ``` Verify installation: ```bash theme={null} fibonacci --version ``` ## Quick Reference | Command | Description | | -------------------- | -------------------------- | | `fibonacci init` | Initialize a new project | | `fibonacci run` | Execute a workflow locally | | `fibonacci list` | List deployed workflows | | `fibonacci validate` | Validate workflow files | | `fibonacci deploy` | Deploy to Fibonacci Cloud | | `fibonacci logs` | View execution logs | | `fibonacci tools` | Manage tool integrations | ## Commands ### `fibonacci init` Initialize a new Fibonacci project: ```bash theme={null} # Interactive setup fibonacci init # With project name fibonacci init my-project # From template fibonacci init --template chatbot fibonacci init --template data-pipeline fibonacci init --template api-integration ``` This creates: ``` my-project/ ├── fibonacci.yaml # Project configuration ├── workflows/ # Workflow definitions │ └── example.yaml ├── .env.example # Environment template └── README.md ``` ### `fibonacci run` Execute workflows locally: ```bash theme={null} # Run a Python workflow fibonacci run workflow.py # Run a YAML workflow fibonacci run workflows/analyzer.yaml # With input data fibonacci run workflow.py --input '{"text": "Hello world"}' # From input file fibonacci run workflow.py --input-file data.json # With specific environment fibonacci run workflow.py --env production ``` **Options:** | Option | Description | | ------------------ | ------------------------------------ | | `--input, -i` | JSON input data | | `--input-file, -f` | Path to JSON input file | | `--env, -e` | Environment (dev/staging/production) | | `--timeout, -t` | Execution timeout in seconds | | `--verbose, -v` | Verbose output | | `--dry-run` | Validate without executing | To trigger an already-deployed workflow via the platform (with queue, risk assessment, and approval gates applied), use the API: `POST /api/v1/workflows/{id}/execute`. The `fibonacci run` command runs workflows locally without these controls. ### `fibonacci list` List workflows deployed to Fibonacci Cloud: ```bash theme={null} # List all workflows fibonacci list # Show only active workflows fibonacci list --active # Output as JSON fibonacci list --format json ``` **Example output:** ``` ID NAME ACTIVE wf-abc123 customer-support yes wf-def456 data-pipeline yes wf-ghi789 legacy-analyzer no ``` ### `fibonacci validate` Validate workflow syntax and configuration: ```bash theme={null} # Validate a single file fibonacci validate workflow.yaml # Validate all workflows in directory fibonacci validate workflows/ # Verbose validation with suggestions fibonacci validate workflow.yaml --verbose # Strict mode (treat warnings as errors) fibonacci validate workflow.yaml --strict ``` **Output example:** ``` ✓ workflow.yaml ✓ Schema validation passed ✓ Node dependencies valid ✓ Tool references valid ⚠ Warning: Node 'analyzer' has no description Validated 1 file with 0 errors, 1 warning ``` ### `fibonacci deploy` Deploy workflows to Fibonacci Cloud: ```bash theme={null} # Deploy all workflows fibonacci deploy # Deploy specific workflow fibonacci deploy workflows/production.yaml # Deploy to specific environment fibonacci deploy --env staging # Deploy with version tag fibonacci deploy --version v1.2.0 # Preview changes without deploying fibonacci deploy --dry-run ``` **Options:** | Option | Description | | --------------- | ------------------------- | | `--env, -e` | Target environment | | `--version, -v` | Version tag | | `--dry-run` | Preview without deploying | | `--force` | Skip confirmation prompts | | `--message, -m` | Deployment message | ### `fibonacci logs` View execution logs: ```bash theme={null} # View recent logs fibonacci logs # Follow logs in real-time fibonacci logs --follow # Filter by workflow fibonacci logs --workflow customer-support # Filter by time range fibonacci logs --since "1 hour ago" fibonacci logs --since "2025-01-15T10:00:00" # Filter by status fibonacci logs --status failed # View specific execution fibonacci logs --execution-id abc123 ``` **Options:** | Option | Description | | ---------------- | ----------------------------------------- | | `--follow, -f` | Stream logs in real-time | | `--workflow, -w` | Filter by workflow name | | `--since, -s` | Show logs since time | | `--until, -u` | Show logs until time | | `--status` | Filter by status (success/failed/running) | | `--limit, -n` | Number of log entries | | `--format` | Output format (text/json) | ### `fibonacci tools` Manage tool integrations: ```bash theme={null} # List available tools fibonacci tools list # Search for tools fibonacci tools search "google sheets" # Get tool details fibonacci tools info google_sheets.read_range # Connect a tool fibonacci tools connect google fibonacci tools connect slack fibonacci tools connect salesforce # List connected tools fibonacci tools connected # Disconnect a tool fibonacci tools disconnect google ``` ### `fibonacci auth` Manage authentication: ```bash theme={null} # Login to Fibonacci Cloud fibonacci auth login # Login with API key fibonacci auth login --api-key YOUR_KEY # Check authentication status fibonacci auth status # Logout fibonacci auth logout # Switch organization fibonacci auth org switch my-org ``` ### `fibonacci config` Manage configuration: ```bash theme={null} # View current config fibonacci config list # Set a config value fibonacci config set default_model claude-sonnet-4-6 # Get a config value fibonacci config get default_model # Reset to defaults fibonacci config reset # Edit config file fibonacci config edit ``` **Common configuration options:** | Key | Description | Default | | --------------- | --------------------- | ------------------- | | `default_model` | Default LLM model | `claude-sonnet-4-6` | | `default_env` | Default environment | `development` | | `log_level` | CLI log verbosity | `info` | | `output_format` | Default output format | `text` | ### `fibonacci secret` Manage secrets: ```bash theme={null} # Set a secret fibonacci secret set API_KEY # Set from value (not recommended for sensitive data) fibonacci secret set API_KEY --value "sk-..." # List secrets (names only) fibonacci secret list # Delete a secret fibonacci secret delete API_KEY # Sync secrets to environment fibonacci secret sync --env production ``` ## Project Configuration The `fibonacci.yaml` file configures your project: ```yaml theme={null} # fibonacci.yaml name: my-project version: "1.0.0" # Default settings defaults: model: claude-sonnet-4-6 timeout: 60 retry: max_attempts: 3 delay: 1.0 # Environments environments: development: api_url: http://localhost:8000 log_level: debug staging: api_url: https://staging-api.fibonacci.dev log_level: info production: api_url: https://api.fibonacci.dev log_level: warning # Workflow locations workflows: - workflows/*.yaml - src/**/*.py # Tool configurations tools: google: credentials_file: ./credentials/google.json slack: default_channel: "#notifications" ``` ## Environment Variables The CLI respects these environment variables: | Variable | Description | | --------------------- | --------------------------- | | `FIBONACCI_API_KEY` | API key for Fibonacci Cloud | | `FIBONACCI_ENV` | Default environment | | `FIBONACCI_CONFIG` | Path to config file | | `FIBONACCI_LOG_LEVEL` | Log verbosity | | `FIBONACCI_NO_COLOR` | Disable colored output | ## Shell Completion Enable tab completion for your shell: ```bash theme={null} # Bash fibonacci completion bash >> ~/.bashrc # Zsh fibonacci completion zsh >> ~/.zshrc # Fish fibonacci completion fish > ~/.config/fish/completions/fibonacci.fish ``` ## CI/CD Integration ### GitHub Actions ```yaml theme={null} # .github/workflows/deploy.yaml name: Deploy Workflows on: push: branches: [main] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Setup Python uses: actions/setup-python@v5 with: python-version: '3.11' - name: Install Fibonacci run: pip install fibonacci-sdk - name: Validate Workflows run: fibonacci validate workflows/ --strict - name: Deploy run: fibonacci deploy --env production env: FIBONACCI_API_KEY: ${{ secrets.FIBONACCI_API_KEY }} ``` ### GitLab CI ```yaml theme={null} # .gitlab-ci.yml stages: - validate - deploy validate: stage: validate image: python:3.11 script: - pip install fibonacci-sdk - fibonacci validate workflows/ --strict deploy: stage: deploy image: python:3.11 script: - pip install fibonacci-sdk - fibonacci deploy --env production only: - main environment: name: production ``` ## Troubleshooting ### Common Issues Ensure Fibonacci is installed and in your PATH: ```bash theme={null} pip install fibonacci-sdk echo $PATH which fibonacci ``` Re-authenticate with Fibonacci Cloud: ```bash theme={null} fibonacci auth logout fibonacci auth login ``` Run with verbose mode for detailed errors: ```bash theme={null} fibonacci validate workflow.yaml --verbose ``` This was a known bug in SDK versions before v0.2.4 where workflow objects were accessed as dicts. Upgrade to the latest version: ```bash theme={null} pip install --upgrade fibonacci-sdk ``` ### Debug Mode Enable debug output: ```bash theme={null} FIBONACCI_LOG_LEVEL=debug fibonacci run workflow.py ``` Or in the command: ```bash theme={null} fibonacci --debug run workflow.py ``` ## Next Steps Define workflows in YAML Production deployment patterns # Error Handling Source: https://docs.fibonacci.today/guides/error-handling Handle errors gracefully in Fibonacci workflows with retries, fallbacks, and monitoring Robust error handling is essential for production workflows. Fibonacci provides multiple mechanisms for handling failures gracefully, from automatic retries to custom error handlers. ## Error Types Fibonacci defines specific exception types for different error scenarios: ```python theme={null} from fibonacci.exceptions import ( FibonacciError, # Base exception WorkflowError, # Workflow-level errors NodeExecutionError, # Node execution failures ToolError, # Tool integration errors ValidationError, # Input/output validation AuthenticationError, # Auth failures RateLimitError, # API rate limits TimeoutError, # Execution timeouts MemoryError, # Memory operations ConfigurationError, # Invalid configuration ) ``` ## Basic Error Handling ### Try-Catch Pattern ```python theme={null} from fibonacci import Workflow from fibonacci.exceptions import ( NodeExecutionError, ToolError, TimeoutError ) workflow = Workflow.from_yaml("workflow.yaml") try: result = workflow.execute(inputs={"text": "Hello"}) print(result) except NodeExecutionError as e: print(f"Node '{e.node_id}' failed: {e.message}") print(f"Details: {e.details}") except ToolError as e: print(f"Tool error: {e.tool_name} - {e.message}") except TimeoutError as e: print(f"Execution timed out after {e.timeout}s") except FibonacciError as e: print(f"Workflow error: {e}") ``` ### Accessing Error Context ```python theme={null} try: result = workflow.execute(inputs=data) except NodeExecutionError as e: # Rich error context print(f"Node ID: {e.node_id}") print(f"Node Type: {e.node_type}") print(f"Input: {e.input_data}") print(f"Error: {e.message}") print(f"Traceback: {e.traceback}") # Partial results from successful nodes print(f"Partial results: {e.partial_results}") ``` ## Retry Configuration ### Node-Level Retries Configure retries for individual nodes: ```python theme={null} from fibonacci import LLMNode, ToolNode, RetryConfig # Simple retry config analyzer = LLMNode( id="analyzer", model="claude-sonnet-4-5-20250929", prompt="Analyze: {{input.text}}", retry=RetryConfig( max_attempts=3, delay=1.0 ) ) # Advanced retry config api_call = ToolNode( id="fetch_data", tool="http.request", inputs={"url": "https://api.example.com/data"}, retry=RetryConfig( max_attempts=5, delay=1.0, backoff="exponential", # or "linear", "constant" max_delay=30.0, jitter=True, # Add randomness to prevent thundering herd retry_on=[ "timeout", "rate_limit", "5xx", # Retry on 500-599 status codes ], no_retry_on=[ "4xx", # Don't retry client errors "validation", ] ) ) ``` ### YAML Retry Configuration ```yaml theme={null} nodes: - id: unreliable_api type: tool tool: http.request inputs: url: "https://api.example.com/data" retry: max_attempts: 5 delay: 2.0 backoff: exponential max_delay: 60.0 jitter: true retry_on: - timeout - rate_limit - connection_error ``` ### Workflow-Level Retry Defaults ```python theme={null} from fibonacci import Workflow, RetryConfig workflow = Workflow( name="resilient-workflow", default_retry=RetryConfig( max_attempts=3, delay=1.0, backoff="exponential" ) ) ``` ## Fallback Handlers ### Node Fallbacks Define fallback behavior when a node fails: ```python theme={null} from fibonacci import LLMNode, ToolNode # Fallback to simpler model primary = LLMNode( id="primary_analyzer", model="claude-opus-4-5-20251101", prompt="Complex analysis: {{input.text}}", fallback="fallback_analyzer" ) fallback = LLMNode( id="fallback_analyzer", model="claude-haiku-4-5-20251001", prompt="Basic analysis: {{input.text}}" ) workflow.add_node(primary) workflow.add_node(fallback) ``` ### Fallback Chains ```python theme={null} # Chain of fallbacks primary_api = ToolNode( id="primary_api", tool="http.request", inputs={"url": "https://primary.api.com/data"}, fallback="secondary_api" ) secondary_api = ToolNode( id="secondary_api", tool="http.request", inputs={"url": "https://secondary.api.com/data"}, fallback="cached_response" ) cached_response = ToolNode( id="cached_response", tool="cache.get", inputs={"key": "last_known_data"} ) ``` ### Conditional Fallbacks ```python theme={null} from fibonacci import LLMNode, ConditionalNode analyzer = LLMNode( id="analyzer", model="claude-sonnet-4-5-20250929", prompt="Analyze: {{input.text}}", on_error="error_handler" ) error_handler = ConditionalNode( id="error_handler", conditions=[ { "if": {"field": "{{error.type}}", "equals": "rate_limit"}, "then": "wait_and_retry" }, { "if": {"field": "{{error.type}}", "equals": "timeout"}, "then": "use_cache" } ], default="notify_admin" ) ``` ## Custom Error Handlers ### Error Handler Functions ```python theme={null} from fibonacci import Workflow, ErrorContext def my_error_handler(context: ErrorContext): """Custom error handler""" print(f"Error in {context.node_id}: {context.error}") # Log to monitoring system log_error( workflow=context.workflow_name, node=context.node_id, error=str(context.error), inputs=context.inputs ) # Return fallback value or re-raise if context.node_id == "analyzer": return {"result": "Analysis unavailable"} raise context.error workflow = Workflow( name="monitored-workflow", error_handler=my_error_handler ) ``` ### Async Error Handlers ```python theme={null} async def async_error_handler(context: ErrorContext): """Async error handler for async workflows""" await log_error_async(context) if context.retry_count < 3: await asyncio.sleep(context.retry_count * 2) return {"retry": True} return {"fallback": "default_value"} workflow = Workflow( name="async-workflow", error_handler=async_error_handler ) ``` ## Error Recovery Patterns ### Graceful Degradation ```python theme={null} from fibonacci import Workflow, LLMNode, ConditionalNode workflow = Workflow(name="graceful-degradation") # Primary: Full analysis with premium model full_analysis = LLMNode( id="full_analysis", model="claude-opus-4-5-20251101", prompt="Comprehensive analysis: {{input.text}}", timeout=60, on_error="degraded_analysis" ) # Degraded: Simpler analysis with faster model degraded_analysis = LLMNode( id="degraded_analysis", model="claude-haiku-4-5-20251001", prompt="Quick analysis: {{input.text}}", timeout=15, on_error="minimal_response" ) # Minimal: Static response minimal_response = LLMNode( id="minimal_response", model="claude-haiku-4-5-20251001", prompt="Return: 'Analysis temporarily unavailable. Please try again later.'" ) ``` ### Circuit Breaker ```python theme={null} from fibonacci import Workflow, ToolNode from fibonacci.patterns import CircuitBreaker # Circuit breaker for external API circuit_breaker = CircuitBreaker( failure_threshold=5, # Open after 5 failures recovery_timeout=60, # Try again after 60s half_open_requests=3 # Test with 3 requests ) api_node = ToolNode( id="external_api", tool="http.request", inputs={"url": "https://api.example.com"}, circuit_breaker=circuit_breaker ) ``` ### Partial Results ```python theme={null} from fibonacci import Workflow from fibonacci.exceptions import PartialExecutionError workflow = Workflow( name="partial-results", allow_partial_results=True # Don't fail entire workflow on node error ) try: result = workflow.execute(inputs=data) except PartialExecutionError as e: # Get results from successful nodes successful = e.partial_results failed = e.failed_nodes print(f"Completed: {list(successful.keys())}") print(f"Failed: {[n.id for n in failed]}") # Use partial results if "summarizer" in successful: return successful["summarizer"] ``` ## Timeout Handling ### Node Timeouts ```python theme={null} from fibonacci import LLMNode, ToolNode # LLM timeout slow_analyzer = LLMNode( id="deep_analysis", model="claude-opus-4-5-20251101", prompt="Deep analysis: {{input.text}}", timeout=120 # 2 minutes ) # Tool timeout api_call = ToolNode( id="api_call", tool="http.request", inputs={"url": "https://slow-api.example.com"}, timeout=30 ) ``` ### Workflow Timeout ```python theme={null} from fibonacci import Workflow from fibonacci.exceptions import TimeoutError workflow = Workflow( name="time-bounded", timeout=300 # 5 minute total timeout ) try: result = workflow.execute(inputs=data, timeout=60) # Override except TimeoutError as e: print(f"Workflow timed out: {e.elapsed}s") ``` ## Validation Errors ### Input Validation ```python theme={null} from fibonacci import Workflow, ValidationError workflow = Workflow( name="validated-workflow", input_schema={ "type": "object", "required": ["text", "language"], "properties": { "text": {"type": "string", "minLength": 1}, "language": {"type": "string", "enum": ["en", "es", "fr"]} } } ) try: result = workflow.execute(inputs={"text": ""}) except ValidationError as e: print(f"Invalid input: {e.errors}") # [{"field": "text", "error": "String too short", "min": 1}] ``` ### Output Validation ```python theme={null} from fibonacci import LLMNode # Validate LLM output analyzer = LLMNode( id="analyzer", model="claude-sonnet-4-5-20250929", prompt="Return JSON: {{input.text}}", output_format="json", output_schema={ "type": "object", "required": ["sentiment", "confidence"], "properties": { "sentiment": {"enum": ["positive", "negative", "neutral"]}, "confidence": {"type": "number", "minimum": 0, "maximum": 1} } }, retry_on_validation_error=True, # Retry if output doesn't match schema max_validation_retries=2 ) ``` ## Logging and Monitoring ### Structured Logging ```python theme={null} from fibonacci import Workflow import structlog logger = structlog.get_logger() workflow = Workflow( name="logged-workflow", logger=logger ) # Logs include: # - Workflow start/end # - Node execution timing # - Errors with full context # - Retry attempts ``` ### Error Hooks ```python theme={null} from fibonacci import Workflow def on_error(event): """Called on any error""" send_to_sentry(event.error) def on_retry(event): """Called on each retry""" metrics.increment("workflow.retries", tags={ "workflow": event.workflow, "node": event.node_id }) workflow = Workflow( name="monitored-workflow", hooks={ "on_error": on_error, "on_retry": on_retry, "on_timeout": lambda e: alert_ops_team(e), } ) ``` ### Metrics Export ```python theme={null} from fibonacci import Workflow from fibonacci.metrics import PrometheusExporter workflow = Workflow( name="metrics-workflow", metrics_exporter=PrometheusExporter(port=9090) ) # Exports: # fibonacci_workflow_duration_seconds # fibonacci_node_duration_seconds # fibonacci_errors_total # fibonacci_retries_total ``` ## Best Practices Prevent runaway executions with appropriate timeouts: ```python theme={null} node = LLMNode(..., timeout=60) workflow.execute(inputs=data, timeout=300) ``` For retries, exponential backoff prevents overwhelming failing services: ```python theme={null} retry=RetryConfig( backoff="exponential", max_delay=60.0, jitter=True ) ``` Include relevant context when logging errors: ```python theme={null} logger.error("Node failed", node_id=e.node_id, inputs=e.input_data, error=str(e) ) ``` Design workflows to handle partial results when appropriate: ```python theme={null} workflow = Workflow(allow_partial_results=True) ``` ## Next Steps Production workflow patterns Secure your workflows # Governance Source: https://docs.fibonacci.today/guides/governance Manage organization governance settings, quotas, tool policies, and workflow registry lifecycle Fibonacci's governance layer gives organization owners and admins control over approval workflows, resource quotas, tool access policies, and the workflow registry. All governance mutations require at minimum the **owner** role. ## Organization Governance Settings Governance settings control which workflow actions require human approval before execution. ### Get Current Settings ```bash theme={null} GET /organizations/{id}/governance ``` ```json theme={null} { "require_approval_for_external_comm": false, "require_approval_for_public_writes": true, "require_approval_for_permission_changes": true, "allow_self_approval": false, "approval_threshold": 1, "category_flags": { "financial": true, "pii": true } } ``` ### Update Settings ```bash theme={null} PATCH /organizations/{id}/governance ``` ```json theme={null} { "require_approval_for_external_comm": true, "allow_self_approval": false, "approval_threshold": 2 } ``` **Field reference:** | Field | Type | Description | | ----------------------------------------- | ------- | -------------------------------------------------------------------- | | `require_approval_for_external_comm` | boolean | Gate workflows that send email, webhooks, or external HTTP calls | | `require_approval_for_public_writes` | boolean | Gate workflows that write to public-facing storage or APIs | | `require_approval_for_permission_changes` | boolean | Gate any action that modifies user/role permissions | | `allow_self_approval` | boolean | Whether a workflow's creator can approve their own execution request | | `approval_threshold` | integer | Minimum number of approvals required | | `category_flags` | object | Per-category override flags (e.g. `"financial": true`) | The risk service enforces these flags at execution time — matching workflows are queued for approval rather than running immediately. ## Quota Management Quotas cap resource usage per organization. Plan defaults apply automatically; owners can override them up to platform-defined ceilings. ### Get Quota Limits ```bash theme={null} GET /organizations/{id}/quota ``` ```json theme={null} { "plan": "pro", "limits": { "workflow_runs_per_hour": 300, "concurrent_executions": 10, "memory_mb": 4096 }, "overrides": { "workflow_runs_per_hour": 500 }, "effective": { "workflow_runs_per_hour": 500, "concurrent_executions": 10, "memory_mb": 4096 } } ``` ### Set Quota Overrides Requires **owner** role. Overrides are validated against plan ceilings. ```bash theme={null} PATCH /organizations/{id}/quota ``` ```json theme={null} { "workflow_runs_per_hour": 500 } ``` Override values must be positive and cannot exceed 100× the plan default for any key. Invalid values are rejected with a `422` response. See [Security](/guides/security#quota-override-validation) for details. ## Tool Policy Administration Tool policies control which tools organization members can use, and optionally restrict usage by role. ### Organization-Level Tool Policies ```bash theme={null} # List all tool policies GET /organizations/{id}/tool-policies # Create a policy POST /organizations/{id}/tool-policies ``` ```json theme={null} { "tool_name": "http.request", "allowed": true, "restrictions": { "allowed_domains": ["api.example.com"], "max_response_size_kb": 1024 } } ``` The `tool_name` must exist in the tool registry. Unknown names return `422` with a hint to check `GET /tools`. ### Role-Level Policy Overrides Fine-tune access per role within the organization: ```bash theme={null} # List role overrides GET /organizations/{id}/role-policy-overrides # Create an override POST /organizations/{id}/role-policy-overrides ``` ```json theme={null} { "role": "developer", "tool_name": "shell.exec", "allowed": false } ``` Role overrides take precedence over org-level policies. Evaluation order: role override → org policy → platform default (deny). ## Workflow Version Management Every deployment of a workflow creates a version snapshot. Admins and owners can inspect and roll back versions. ### List Versions Accessible to any active org member (member or viewer). ```bash theme={null} GET /workflows/{id}/versions ``` ```json theme={null} { "versions": [ { "version": 3, "deployed_at": "2026-05-12T23:00:00Z", "deployed_by": "user-123", "message": "Add retry logic to API node" }, { "version": 2, "deployed_at": "2026-05-10T14:00:00Z", "deployed_by": "user-456", "message": "Initial production deploy" } ] } ``` ### Inspect a Version ```bash theme={null} GET /workflows/{id}/versions/{version} ``` Returns the full workflow definition snapshot for that version. ### Roll Back Rollback creates a **new version snapshot** from the target version's definition rather than overwriting history. ```bash theme={null} POST /workflows/{id}/versions/{version}/rollback ``` ```json theme={null} { "message": "Rolling back to v2 — v3 caused timeout errors" } ``` Response: ```json theme={null} { "new_version": 4, "rolled_back_from": 2, "deployed_at": "2026-05-13T10:00:00Z" } ``` ## Workflow Registry The workflow registry is an organization-level catalogue of approved workflow templates. Templates can be approved, deprecated, and archived through a structured lifecycle. ### Lifecycle States ``` active ──► deprecated ──► archived ▲ │ approve-template │ (draft workflow) ``` | State | Visible in registry | Usable as template | | ---------- | ------------------------------------ | ------------------ | | active | Yes | Yes | | deprecated | Yes | Yes (with warning) | | archived | No (unless `?include_archived=true`) | No | ### Approve as Template Requires **owner** role. Only workflows that belong to an organization can be approved (personal workflows are rejected with `400`). ```bash theme={null} POST /workflow-registry/{id}/approve-template ``` ```json theme={null} { "template_name": "customer-support-v2", "description": "Handles tier-1 support queries with escalation", "tags": ["support", "production-ready"] } ``` ### Deprecate a Template ```bash theme={null} POST /workflow-registry/{id}/deprecate ``` ```json theme={null} { "reason": "Superseded by customer-support-v3", "replacement_id": "wf-xyz789" } ``` ### Archive a Template Archived templates are hidden from the default registry listing and cannot be used as templates. This action is not reversible via the API. ```bash theme={null} POST /workflow-registry/{id}/archive ``` ### List Registry Entries ```bash theme={null} # Active and deprecated only (default) GET /workflow-registry # Include archived GET /workflow-registry?include_archived=true ``` ## Governance Reporting ### Summary Dashboard High-level governance health for the organization: ```bash theme={null} GET /organizations/{id}/governance/report ``` ```json theme={null} { "period": "last_30_days", "total_executions": 4820, "gated_executions": 312, "approved": 298, "rejected": 14, "approval_rate": 0.955, "avg_approval_time_minutes": 8.3, "top_approvers": [ {"user_id": "user-123", "approvals": 145} ] } ``` ### Paginated Audit Log Full governance event history with pagination: ```bash theme={null} GET /organizations/{id}/governance/audit?page=1&per_page=50 ``` Optional filters: | Parameter | Description | | ------------ | ---------------------------------------------------------------- | | `event_type` | Filter by event type (e.g. `approval.granted`, `quota.exceeded`) | | `user_id` | Filter by acting user | | `since` | ISO 8601 start date | | `until` | ISO 8601 end date | ### Approval Statistics Breakdown of approval activity by workflow, user, and category: ```bash theme={null} GET /organizations/{id}/governance/approvals ``` ```json theme={null} { "by_workflow": [ { "workflow_id": "wf-abc123", "name": "data-exporter", "total": 58, "approved": 55, "rejected": 3 } ], "by_category": { "external_comm": {"total": 120, "approved": 115}, "public_writes": {"total": 80, "approved": 78} } } ``` ## Access Requirements Summary | Operation | Minimum Role | | ------------------------------------------ | ---------------- | | Read governance settings | admin or owner | | Update governance settings | owner | | Read quota limits | admin or owner | | Set quota overrides | owner | | Read/write tool policies | owner | | List workflow versions | member or viewer | | Inspect/rollback workflow versions | admin or owner | | Approve/deprecate/archive registry entries | owner | | View governance reports | admin or owner | ## Next Steps Production hardening, secrets backends, and audit logging Production workflow patterns # Memory & State Source: https://docs.fibonacci.today/guides/memory Persist data across workflow executions with Fibonacci memory system Memory in Fibonacci allows workflows to store and retrieve data across executions, enabling stateful applications like chatbots, progressive data collection, and user preference tracking. ## Memory Scopes Fibonacci provides four memory scopes, each with different visibility and lifecycle: | Scope | Visibility | Use Case | | -------------- | ---------------------------- | --------------------------------------- | | `workflow` | Single workflow | Execution history, intermediate results | | `user` | Single user across workflows | Preferences, conversation history | | `organization` | All users in organization | Shared knowledge bases, company data | | `global` | All workflows and users | Application-wide settings | ## Basic Usage ### Storing Memory ```python theme={null} from fibonacci import Workflow, Memory workflow = Workflow(name="memory-demo") # Store a simple value workflow.memory.set( key="user_preference", value={"theme": "dark", "language": "en"}, scope="user" ) # Store with TTL (time-to-live in seconds) workflow.memory.set( key="session_cache", value={"last_query": "weather report"}, scope="workflow", ttl=3600 # Expires in 1 hour ) ``` ### Retrieving Memory ```python theme={null} # Get a value (returns None if not found) preference = workflow.memory.get( key="user_preference", scope="user" ) # Get with default value theme = workflow.memory.get( key="user_preference", scope="user", default={"theme": "light"} ) # Check if key exists if workflow.memory.exists("user_preference", scope="user"): print("Preference found!") ``` ### Deleting Memory ```python theme={null} # Delete a specific key workflow.memory.delete(key="session_cache", scope="workflow") # Clear all memory in a scope workflow.memory.clear(scope="workflow") ``` ## Memory in Nodes Nodes can read from and write to memory using template variables and the `memory` parameter: ### Reading Memory in Prompts ```python theme={null} from fibonacci import LLMNode chat_node = LLMNode( id="chatbot", model="claude-sonnet-4-5-20250929", prompt="""You are a helpful assistant. Previous conversation: {{memory.conversation_history}} User preferences: {{memory.user_preferences}} User message: {{input.message}}""", memory_read=["conversation_history", "user_preferences"] ) ``` ### Writing Memory from Nodes ```python theme={null} from fibonacci import LLMNode summarize_node = LLMNode( id="summarizer", model="claude-sonnet-4-5-20250929", prompt="Summarize: {{input.text}}", memory_write={ "key": "last_summary", "scope": "user" } ) ``` ## Building Stateful Applications ### Conversational Chatbot Build a chatbot that remembers conversation history: ```python theme={null} from fibonacci import Workflow, LLMNode def create_chatbot(): workflow = Workflow( name="stateful-chatbot", description="Chatbot with conversation memory" ) # Chat node that reads history and generates response chat_node = LLMNode( id="chat", model="claude-sonnet-4-5-20250929", prompt="""You are a friendly assistant with memory of past conversations. Conversation history: {{memory.history}} Current message: {{input.message}} Respond naturally, referencing past context when relevant.""", memory_read=["history"] ) workflow.add_node(chat_node) return workflow # Usage with history management workflow = create_chatbot() def chat(user_message: str, user_id: str): # Get existing history history = workflow.memory.get( key="history", scope="user", user_id=user_id, default=[] ) # Execute workflow result = workflow.execute( inputs={"message": user_message}, user_id=user_id ) # Update history history.append({"role": "user", "content": user_message}) history.append({"role": "assistant", "content": result["chat"]}) # Keep last 20 messages history = history[-20:] workflow.memory.set( key="history", value=history, scope="user", user_id=user_id ) return result["chat"] ``` ### User Preference Learning Track and apply user preferences over time: ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode workflow = Workflow(name="preference-learner") # Analyze user input for preferences analyzer = LLMNode( id="analyze_preferences", model="claude-sonnet-4-5-20250929", prompt="""Analyze this user input for implicit preferences: {{input.message}} Current known preferences: {{memory.preferences}} Return JSON with any new or updated preferences detected: { "communication_style": "formal|casual|technical", "detail_level": "brief|detailed|comprehensive", "topics_of_interest": ["topic1", "topic2"] } Only include fields where you detected a clear preference.""", memory_read=["preferences"], output_format="json" ) # Generate personalized response responder = LLMNode( id="respond", model="claude-sonnet-4-5-20250929", prompt="""Respond to: {{input.message}} Adapt your response style based on user preferences: {{memory.preferences}} Detected preference updates: {{analyze_preferences}}""", dependencies=["analyze_preferences"], memory_read=["preferences"] ) workflow.add_node(analyzer) workflow.add_node(responder) ``` ### Progress Tracking Track multi-step process progress: ```python theme={null} from fibonacci import Workflow, LLMNode, ConditionalNode from datetime import datetime workflow = Workflow(name="onboarding-tracker") def update_progress(workflow, user_id: str, step: str, data: dict): """Update user's onboarding progress""" progress = workflow.memory.get( key="onboarding_progress", scope="user", user_id=user_id, default={"completed_steps": [], "data": {}} ) progress["completed_steps"].append(step) progress["data"][step] = data progress["last_updated"] = datetime.now().isoformat() workflow.memory.set( key="onboarding_progress", value=progress, scope="user", user_id=user_id ) def get_next_step(workflow, user_id: str) -> str: """Determine next onboarding step""" progress = workflow.memory.get( key="onboarding_progress", scope="user", user_id=user_id, default={"completed_steps": []} ) all_steps = ["welcome", "profile", "preferences", "tutorial", "complete"] completed = progress["completed_steps"] for step in all_steps: if step not in completed: return step return "complete" ``` ## Memory Patterns ### Caching Expensive Operations ```python theme={null} from fibonacci import Workflow, ToolNode import hashlib workflow = Workflow(name="cached-operations") def get_cached_or_compute(workflow, operation: str, inputs: dict): """Cache expensive computations""" # Create cache key from inputs cache_key = hashlib.md5( f"{operation}:{str(inputs)}".encode() ).hexdigest() # Check cache cached = workflow.memory.get( key=f"cache_{cache_key}", scope="organization" ) if cached: return cached["result"] # Compute if not cached result = workflow.execute(inputs=inputs) # Cache for 24 hours workflow.memory.set( key=f"cache_{cache_key}", value={"result": result, "inputs": inputs}, scope="organization", ttl=86400 ) return result ``` ### Rate Limiting ```python theme={null} from datetime import datetime, timedelta def check_rate_limit(workflow, user_id: str, limit: int = 100) -> bool: """Check if user is within rate limit""" rate_data = workflow.memory.get( key="rate_limit", scope="user", user_id=user_id, default={"count": 0, "window_start": datetime.now().isoformat()} ) window_start = datetime.fromisoformat(rate_data["window_start"]) # Reset if window expired (1 hour) if datetime.now() - window_start > timedelta(hours=1): rate_data = {"count": 0, "window_start": datetime.now().isoformat()} if rate_data["count"] >= limit: return False rate_data["count"] += 1 workflow.memory.set( key="rate_limit", value=rate_data, scope="user", user_id=user_id, ttl=3600 ) return True ``` ### Session Management ```python theme={null} import uuid from datetime import datetime, timedelta def create_session(workflow, user_id: str) -> str: """Create a new session""" session_id = str(uuid.uuid4()) workflow.memory.set( key=f"session_{session_id}", value={ "user_id": user_id, "created_at": datetime.now().isoformat(), "data": {} }, scope="user", user_id=user_id, ttl=7200 # 2 hour session ) return session_id def get_session(workflow, session_id: str, user_id: str) -> dict | None: """Retrieve session data""" return workflow.memory.get( key=f"session_{session_id}", scope="user", user_id=user_id ) def update_session(workflow, session_id: str, user_id: str, data: dict): """Update session data""" session = get_session(workflow, session_id, user_id) if session: session["data"].update(data) session["last_activity"] = datetime.now().isoformat() workflow.memory.set( key=f"session_{session_id}", value=session, scope="user", user_id=user_id, ttl=7200 ) ``` ## Memory Configuration ### Default Memory Backend By default, Fibonacci uses an in-memory store. For production, configure a persistent backend: ```python theme={null} from fibonacci import Workflow, MemoryConfig # Redis backend workflow = Workflow( name="production-workflow", memory_config=MemoryConfig( backend="redis", connection_url="redis://localhost:6379/0", key_prefix="fibonacci:" ) ) # PostgreSQL backend workflow = Workflow( name="production-workflow", memory_config=MemoryConfig( backend="postgres", connection_url="postgresql://user:pass@localhost/fibonacci", table_name="workflow_memory" ) ) ``` ### Memory Encryption Enable encryption for sensitive data: ```python theme={null} from fibonacci import Workflow, MemoryConfig workflow = Workflow( name="secure-workflow", memory_config=MemoryConfig( backend="redis", connection_url="redis://localhost:6379/0", encryption_key="your-32-byte-encryption-key-here", encrypt_values=True ) ) ``` ## Best Practices Match memory scope to data sensitivity and sharing requirements: * `workflow`: Temporary execution data * `user`: Personal preferences, history * `organization`: Shared resources, team data * `global`: Application configuration Always set TTL for caches, sessions, and temporary data to prevent unbounded growth: ```python theme={null} workflow.memory.set(key="temp", value=data, ttl=3600) ``` Use consistent, hierarchical key naming: ```python theme={null} # Good "user:preferences:theme" "session:abc123:cart" "cache:api:weather:nyc" # Avoid "theme" "cart" "weather" ``` Always provide defaults when reading memory: ```python theme={null} prefs = workflow.memory.get("prefs", default={"theme": "light"}) ``` Cap array-based memory (like chat history) to prevent unbounded growth: ```python theme={null} history = history[-50:] # Keep last 50 items ``` ## Next Steps Learn workflow lifecycle management Add branching based on memory values # Nodes Source: https://docs.fibonacci.today/guides/nodes Understanding the different node types in Fibonacci Nodes are the building blocks of workflows. Each node performs a specific action and can depend on the outputs of other nodes. ## Node Types Overview Calls Claude or other LLMs to analyze, generate, or transform content Executes integrations like Google Sheets, Slack, APIs, etc. Evaluates and scores the output of other nodes Branches workflow based on conditions *** ## LLMNode The most common node type. Uses AI to process text. ### Basic Usage ```python theme={null} from fibonacci import LLMNode node = LLMNode( id="analyze", # Unique identifier (required) name="Analyze Text", # Human-readable name (required) instruction="Analyze: {{input.text}}" # Prompt (required) ) ``` ### Full Options ```python theme={null} analyze = LLMNode( id="analyze_sentiment", name="Sentiment Analysis", instruction=""" Analyze the sentiment of this text. Respond with: positive, negative, or neutral Text: {{input.text}} """, model="claude-sonnet-4-5-20250929", # Model to use max_tokens=500, # Max response tokens temperature=0.7, # Creativity (0-1) dependencies=[] # Nodes this depends on ) ``` ### Available Models | Model | Best For | Speed | Cost | | ---------------------------- | ----------------- | ------- | ------ | | `claude-sonnet-4-5-20250929` | Most tasks | Fast | \$ | | `claude-opus-4-5-20251101` | Complex reasoning | Slower | \$\$\$ | | `claude-haiku-4-5` | Simple tasks | Fastest | ¢ | ### Template Variables Reference inputs and other node outputs: ```python theme={null} LLMNode( id="summarize", instruction=""" Summarize this data: Input: {{input.raw_data}} Previous analysis: {{analyze_sentiment}} User preferences: {{input.preferences.format}} """ ) ``` *** ## ToolNode Executes integrations with external services. ### Basic Usage ```python theme={null} from fibonacci import ToolNode node = ToolNode( id="read_sheet", name="Read Google Sheet", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "A1:D100" } ) ``` ### Full Options ```python theme={null} send_email = ToolNode( id="send_report", name="Send Report Email", tool="gmail_send", params={ "to": "{{input.recipient}}", "subject": "Weekly Report", "body": "{{generate_report}}" }, timeout=30, # Timeout in seconds dependencies=["generate_report"] ) ``` ### Popular Tools * `google_sheets_read` - Read spreadsheet data * `google_sheets_write` - Write to spreadsheet * `google_docs_read` - Read document content * `google_docs_write` - Create/update documents * `gmail_send` - Send emails * `google_calendar_create` - Create events * `slack_send_message` - Send Slack messages * `slack_create_channel` - Create channels * `discord_send` - Send Discord messages * `twilio_sms` - Send SMS * `http_request` - Generic HTTP calls * `webhook_call` - Call webhooks * `database_query` - Query databases * `s3_upload` - Upload to S3 * `salesforce_create` - Create Salesforce records * `hubspot_contact` - Manage HubSpot contacts * `notion_page` - Create Notion pages * `airtable_record` - Manage Airtable records ### Discover Tools ```python theme={null} from fibonacci import list_tools, search_tools, print_tool_info # List all tools tools = list_tools() # Search for specific tools slack_tools = search_tools("slack") # Get detailed info print_tool_info("google_sheets_read") ``` *** ## CriticNode Evaluates the output of another node for quality. ### Basic Usage ```python theme={null} from fibonacci import CriticNode critic = CriticNode( id="evaluate_report", name="Evaluate Report Quality", target_node="generate_report", # Node to evaluate criteria=["clarity", "completeness", "accuracy"] ) ``` ### How It Works The critic node: 1. Receives the output of the target node 2. Evaluates it against the specified criteria 3. Returns a score and feedback ```python theme={null} result = wf.run(input_data={...}) evaluation = result.output_data["evaluate_report"] # { # "score": 8.5, # "feedback": { # "clarity": "Well organized with clear sections", # "completeness": "Missing Q4 projections", # "accuracy": "All figures verified" # } # } ``` ### Custom Criteria ```python theme={null} critic = CriticNode( id="check_code", name="Code Review", target_node="generate_code", criteria=[ "correctness", "efficiency", "readability", "error_handling", "security" ] ) ``` *** ## ConditionalNode Branches workflow execution based on conditions. ### Basic Usage ```python theme={null} from fibonacci import ConditionalNode condition = ConditionalNode( id="check_sentiment", name="Check Sentiment", left_value="{{analyze_sentiment}}", operator="contains", right_value="negative", true_branch=["send_alert"], # Nodes to run if true false_branch=["send_summary"] # Nodes to run if false ) ``` ### Operators | Operator | Description | Example | | -------------- | ------------------ | ------------------------------------- | | `equals` | Exact match | `"completed"` equals `"completed"` | | `not_equals` | Not equal | `"pending"` not\_equals `"completed"` | | `contains` | Substring match | `"error occurred"` contains `"error"` | | `not_contains` | No substring | `"success"` not\_contains `"error"` | | `greater_than` | Numeric comparison | `100` greater\_than `50` | | `less_than` | Numeric comparison | `25` less\_than `50` | | `starts_with` | String prefix | `"error: ..."` starts\_with `"error"` | | `ends_with` | String suffix | `"file.pdf"` ends\_with `".pdf"` | | `is_empty` | Empty check | `""` is\_empty | | `is_not_empty` | Not empty | `"text"` is\_not\_empty | ### Complex Branching ```python theme={null} # Check score and branch accordingly check_score = ConditionalNode( id="check_score", name="Check Quality Score", left_value="{{evaluate_report.score}}", operator="greater_than", right_value="7", true_branch=["publish_report", "notify_team"], false_branch=["request_revision"] ) ``` *** ## Common Node Properties All nodes share these properties: | Property | Type | Description | | -------------- | ----------- | ------------------------------------------ | | `id` | `str` | Unique identifier (lowercase, underscores) | | `name` | `str` | Human-readable name | | `type` | `str` | Node type (auto-set) | | `dependencies` | `list[str]` | Node IDs this depends on | | `condition` | `dict` | Optional execution condition | | `enable_retry` | `bool` | Enable retry on failure | | `max_retries` | `int` | Max retry attempts | | `retry_delay` | `float` | Delay between retries | ## Dependencies Control execution order by declaring dependencies: ```python theme={null} # Node B depends on Node A node_a = LLMNode(id="a", name="A", instruction="...") node_b = LLMNode(id="b", name="B", instruction="...", dependencies=["a"]) # Or use fluent API node_b = LLMNode(id="b", name="B", instruction="...").depends_on("a") ``` ### Execution Order ``` No dependencies → Runs immediately dependencies=["a"] → Waits for "a" to complete dependencies=["a", "b"] → Waits for both "a" AND "b" ``` ## Retry Configuration Handle transient failures: ```python theme={null} # Enable retries with fluent API node = ToolNode( id="api_call", tool="http_request", params={...} ).with_retry(max_retries=3, delay=2.0) # Or in constructor node = ToolNode( id="api_call", tool="http_request", params={...}, enable_retry=True, max_retries=3, retry_delay=2.0 # Seconds, with exponential backoff ) ``` ## Conditional Execution Run a node only if a condition is met: ```python theme={null} # Only run if input.send_email is "true" node = ToolNode( id="send", tool="gmail_send", params={...} ).with_condition( left_value="{{input.send_email}}", operator="equals", right_value="true" ) ``` ## Next Steps Explore all available tool integrations Build complex branching workflows # Security Source: https://docs.fibonacci.today/guides/security Secure your Fibonacci workflows with authentication, encryption, and access controls Security is critical for production workflows. This guide covers authentication, secrets management, encryption, access controls, and security best practices. ## Authentication ### API Key Management Store API keys securely using Fibonacci's keychain integration: ```python theme={null} from fibonacci import SecureConfig from fibonacci.security import KeychainStorage # Store API key in system keychain keychain = KeychainStorage() keychain.set("fibonacci_api_key", "your-api-key") # Use in workflow config = SecureConfig( api_key=keychain.get("fibonacci_api_key") ) ``` ### Supported Keychain Backends | Platform | Backend | | -------- | --------------------------------------- | | macOS | Keychain Services | | Windows | Windows Credential Manager | | Linux | Secret Service (GNOME Keyring, KWallet) | ### Environment-Based Authentication For CI/CD and containerized environments: ```python theme={null} from fibonacci import SecureConfig import os config = SecureConfig( api_key=os.environ.get("FIBONACCI_API_KEY"), require_api_key=True ) ``` ## Secrets Management ### Defining Secrets Never hardcode secrets in workflow files: ```yaml theme={null} # ❌ Bad - secrets in plain text nodes: - id: api_call type: tool tool: http.request inputs: headers: Authorization: "Bearer sk-1234567890" # ✅ Good - reference secrets nodes: - id: api_call type: tool tool: http.request inputs: headers: Authorization: "Bearer {{secrets.API_KEY}}" ``` ### Setting Secrets via CLI ```bash theme={null} # Set a secret (prompts for value) fibonacci secret set API_KEY # Set secret for specific environment fibonacci secret set API_KEY --env production # List secrets (names only, never values) fibonacci secret list ``` ### Secrets in Python ```python theme={null} from fibonacci import Workflow from fibonacci.security import SecretsManager # Default: database backend secrets = SecretsManager(backend="db") # AWS Secrets Manager backend # Requires: pip install boto3 # Env vars: SECRETS_BACKEND, AWS_REGION, AWS_SECRETS_PREFIX secrets = SecretsManager( backend="aws_secrets_manager", config={ "region": "us-east-1", "prefix": "fibonacci/prod" # optional } ) workflow = Workflow( name="secure-workflow", secrets_manager=secrets ) # Access secrets in code api_key = secrets.get("API_KEY") ``` ### Secret Rotation ```python theme={null} from fibonacci.security import SecretsManager, RotationPolicy secrets = SecretsManager( backend="vault", rotation_policy=RotationPolicy( enabled=True, interval_days=30, notify_before_days=7 ) ) # Check rotation status status = secrets.rotation_status("API_KEY") print(f"Days until rotation: {status.days_remaining}") ``` ## Production Environment Hardening ### Required Settings The server validates these settings at startup. Misconfigurations are **fatal errors** in `APP_ENV=production` and warnings in staging. | Variable | Production Requirement | Description | | ------------------------- | ---------------------- | ------------------------------------ | | `SESSION_SECRET_KEY` | Required (≥ 32 chars) | Signing key for sessions | | `TRUSTED_PROXY_HOSTS` | Recommended | Comma-separated trusted proxy IPs | | `ENABLE_DOCS` | Must be `false` | Disables the interactive API docs UI | | `ENABLE_DEVELOPER_ROUTES` | Must be `false` | Disables developer-only routes | ```bash theme={null} # Production .env APP_ENV=production SESSION_SECRET_KEY=your-random-secret-key-at-least-32-characters TRUSTED_PROXY_HOSTS=10.0.0.1,10.0.0.2 ENABLE_DOCS=false ENABLE_DEVELOPER_ROUTES=false ``` Setting `ENABLE_DOCS=true` or `ENABLE_DEVELOPER_ROUTES=true` with `APP_ENV=production` **prevents the server from starting**. In staging these are warnings only. ### Secrets Backend Configuration Configure where Fibonacci stores organization secrets: ```bash theme={null} # Default: database SECRETS_BACKEND=db # AWS Secrets Manager SECRETS_BACKEND=aws_secrets_manager AWS_REGION=us-east-1 AWS_SECRETS_PREFIX=fibonacci/prod # optional namespace prefix ``` When `SECRETS_BACKEND=aws_secrets_manager` is set, `boto3` must be installed — startup fails with a clear error if it is missing: ```bash theme={null} pip install boto3 ``` ### Secrets Health Endpoint Verify secrets backend connectivity without exposing secret values: ```bash theme={null} GET /health/secrets ``` ```json theme={null} { "status": "healthy", "backend": "aws_secrets_manager", "checks": { "list_secrets": "ok", "get_secret": "ok" } } ``` This endpoint confirms IAM connectivity only and never returns secret values. It requires `ListSecrets` and `GetSecretValue` IAM actions on the configured prefix. ### Quota Override Validation Quota overrides set by organization admins are validated to prevent abuse: * Negative or non-numeric values are silently discarded and replaced with the plan default. * Values exceeding 100× the plan default ceiling are rejected. This prevents both DoS via zero/negative limits and rate-limit bypass via excessively large overrides. See [Governance](/guides/governance) for how to set valid quota overrides. ## Encryption ### Data Encryption at Rest Encrypt sensitive data stored in memory: ```python theme={null} from fibonacci import Workflow, MemoryConfig from fibonacci.security import EncryptionConfig workflow = Workflow( name="encrypted-workflow", memory_config=MemoryConfig( backend="redis", encryption=EncryptionConfig( enabled=True, algorithm="AES-256-GCM", key_source="keychain" # or "env", "vault" ) ) ) # All memory operations are now encrypted workflow.memory.set("sensitive_data", {"ssn": "123-45-6789"}) ``` ### Encryption in Transit All Fibonacci Cloud communications use TLS 1.3: ```python theme={null} from fibonacci import FibonacciClient client = FibonacciClient( api_key="...", tls_config={ "verify_certificates": True, "min_version": "1.3", "client_cert": "/path/to/cert.pem", # mTLS "client_key": "/path/to/key.pem" } ) ``` ### Field-Level Encryption Encrypt specific fields in workflow data: ```python theme={null} from fibonacci import LLMNode from fibonacci.security import encrypt_field, decrypt_field # Encrypt sensitive input before processing encrypted_data = encrypt_field( data=user_pii, fields=["ssn", "credit_card"], key_id="pii-encryption-key" ) # Decrypt when needed decrypted = decrypt_field( data=encrypted_data, fields=["ssn"], key_id="pii-encryption-key" ) ``` ## Access Control ### Role-Based Access Control (RBAC) Define roles and permissions for workflows: ```python theme={null} from fibonacci import Workflow from fibonacci.security import AccessControl, Role # Define roles admin_role = Role( name="admin", permissions=["read", "write", "execute", "delete", "admin"] ) developer_role = Role( name="developer", permissions=["read", "write", "execute"] ) viewer_role = Role( name="viewer", permissions=["read"] ) workflow = Workflow( name="access-controlled", access_control=AccessControl( enabled=True, default_role="viewer", roles=[admin_role, developer_role, viewer_role] ) ) ``` ### Resource-Level Permissions ```yaml theme={null} # fibonacci.yaml access_control: enabled: true resources: workflows: "production/*": execute: - role: admin - role: production-deployer write: - role: admin "development/*": execute: - role: developer - role: admin write: - role: developer - role: admin secrets: "*": read: - role: admin write: - role: admin ``` ### API Key Scopes Create scoped API keys with limited permissions: ```bash theme={null} # Create read-only key fibonacci auth create-key --name "monitoring" --scope read # Create execution-only key fibonacci auth create-key --name "ci-cd" --scope execute # Create key with specific workflow access fibonacci auth create-key --name "production" \ --scope execute \ --workflows "production/*" ``` API key scopes are returned by the API as a `Dict[str, bool]` map (e.g. `{"runs:read": true, "runs:write": false}`). The SDK's `APIKeyInfo.scopes` field accepts both this dict form and the legacy list form. ## Audit Logging ### Enable Audit Logs ```python theme={null} from fibonacci import Workflow from fibonacci.security import AuditLogger audit = AuditLogger( enabled=True, destination="cloudwatch", # or "file", "splunk", "datadog" config={ "log_group": "/fibonacci/audit", "log_stream": "workflows" } ) workflow = Workflow( name="audited-workflow", audit_logger=audit ) ``` ### Audit Log Contents Audit logs capture: ```json theme={null} { "timestamp": "2025-01-23T10:30:00Z", "event_type": "workflow.execute", "workflow_id": "customer-support", "user_id": "user-123", "ip_address": "192.168.1.100", "action": "execute", "status": "success", "duration_ms": 1523, "metadata": { "input_hash": "abc123", "nodes_executed": ["classifier", "responder"], "api_key_id": "key-456" } } ``` Additional audit events include `DEVELOPER_SHELL_EXECUTED` (emitted when a superuser accesses the developer shell), quota enforcement events, and governance setting changes. ### Query Audit Logs ```bash theme={null} # View recent audit events fibonacci audit logs --since "1 hour ago" # Filter by workflow fibonacci audit logs --workflow production-pipeline # Filter by user fibonacci audit logs --user user-123 # Export for analysis fibonacci audit export --format json --output audit.json ``` For paginated API access to audit logs see the [Governance guide](/guides/governance). ## Input Validation and Sanitization ### Schema Validation ```python theme={null} from fibonacci import Workflow workflow = Workflow( name="validated-workflow", input_schema={ "type": "object", "required": ["user_input"], "properties": { "user_input": { "type": "string", "maxLength": 10000, "pattern": "^[a-zA-Z0-9\\s.,!?-]*$" }, "email": { "type": "string", "format": "email" } }, "additionalProperties": False } ) ``` ### Prompt Injection Prevention ```python theme={null} from fibonacci import LLMNode from fibonacci.security import sanitize_input # Sanitize user input before including in prompts sanitized = sanitize_input( user_input, rules=[ "escape_xml", "remove_system_prompts", "limit_length" ] ) node = LLMNode( id="analyzer", model="claude-sonnet-4-6", prompt=f"""Analyze this text: {sanitized} Provide sentiment analysis.""", prompt_injection_detection=True ) ``` ### Output Filtering ```python theme={null} from fibonacci import Workflow from fibonacci.security import OutputFilter workflow = Workflow( name="filtered-workflow", output_filter=OutputFilter( rules=[ {"type": "pii", "action": "redact"}, {"type": "secrets", "action": "block"}, {"type": "code", "action": "sandbox"} ] ) ) ``` ## Network Security ### IP Allowlisting ```yaml theme={null} # fibonacci.yaml security: network: ip_allowlist: - 10.0.0.0/8 - 192.168.1.0/24 - 203.0.113.50 ip_blocklist: - 0.0.0.0/0 # Block all except allowlist ``` ### VPC Configuration ```python theme={null} from fibonacci import FibonacciClient client = FibonacciClient( api_key="...", vpc_config={ "vpc_id": "vpc-12345", "subnet_ids": ["subnet-a", "subnet-b"], "security_group_ids": ["sg-12345"] } ) ``` ## Security Checklist * [ ] API keys stored in keychain or secrets manager * [ ] No hardcoded credentials in code or YAML * [ ] API key rotation policy in place * [ ] Scoped API keys for different environments * [ ] `SESSION_SECRET_KEY` set (≥ 32 characters) * [ ] `ENABLE_DOCS=false` in production * [ ] `ENABLE_DEVELOPER_ROUTES=false` in production * [ ] `TRUSTED_PROXY_HOSTS` configured * [ ] Secrets backend chosen and verified via `GET /health/secrets` * [ ] `boto3` installed if using `aws_secrets_manager` backend * [ ] Encryption at rest enabled for sensitive data * [ ] TLS 1.3 for all communications * [ ] PII handling compliant with regulations * [ ] Data retention policies defined * [ ] RBAC configured with least privilege * [ ] Resource-level permissions defined * [ ] Regular access reviews scheduled * [ ] Service accounts properly scoped * [ ] Audit logging enabled * [ ] Security alerts configured * [ ] Anomaly detection active * [ ] Regular security assessments * [ ] Input validation schemas defined * [ ] Prompt injection protection enabled * [ ] Output filtering configured * [ ] Rate limiting implemented ## Compliance ### GDPR Compliance ```python theme={null} from fibonacci import Workflow from fibonacci.compliance import GDPRConfig workflow = Workflow( name="gdpr-compliant", compliance=GDPRConfig( data_retention_days=90, enable_right_to_erasure=True, enable_data_portability=True, consent_tracking=True ) ) ``` ### SOC 2 Compliance Fibonacci Cloud is SOC 2 Type II certified. Enable additional controls: ```yaml theme={null} # fibonacci.yaml compliance: soc2: enabled: true controls: - access_logging - encryption_at_rest - encryption_in_transit - vulnerability_scanning - incident_response ``` ## Next Steps Organization governance, quotas, and policy administration Production workflow patterns # Tool Integrations Source: https://docs.fibonacci.today/guides/tool-nodes Connect workflows to external services with ToolNodes ToolNodes let you interact with external services like Google Sheets, Slack, Notion, databases, and HTTP APIs. Use `list_tools()` to see every tool available on your platform. ## Using ToolNodes ### Basic Structure ```python theme={null} from fibonacci import ToolNode node = ToolNode( id="unique_id", # Required: Unique identifier name="Human Name", # Required: Display name tool="tool_name", # Required: Tool to execute params={ # Required: Tool parameters "param1": "value", "param2": "{{input.field}}" } ) ``` ### With Dependencies ```python theme={null} # Read data, then write to another sheet read = ToolNode( id="read_source", name="Read Source", tool="google_sheets_read", params={"spreadsheet_id": "...", "range": "A1:Z100"} ) write = ToolNode( id="write_dest", name="Write Destination", tool="google_sheets_write", params={ "spreadsheet_id": "...", "range": "A1", "values": "{{read_source}}" }, dependencies=["read_source"] ) ``` *** ## Tool Discovery ### List All Tools ```python theme={null} from fibonacci import list_tools tools = list_tools() for tool in tools: print(f"{tool['name']}: {tool['description']}") ``` ### Search Tools ```python theme={null} from fibonacci import search_tools # Find Google-related tools google_tools = search_tools("google") # Find email tools email_tools = search_tools("email") ``` ### Get Tool Schema ```python theme={null} from fibonacci import print_tool_info # See full details including parameters and examples print_tool_info("google_sheets_read") ``` Output: ``` ====================================================================== 🔧 google_sheets_read ====================================================================== Category: google_workspace Description: Read data from a Google Sheets spreadsheet Cost per use: $0.0001 📥 Input Parameters: • spreadsheet_id (string) - required The ID of the spreadsheet • range (string) - required The A1 notation range to read (e.g., "Sheet1!A1:D10") 📖 Examples: 1. Read sales data Params: {"spreadsheet_id": "abc123", "range": "Sales!A1:E100"} ====================================================================== ``` ### Find Tool for Task ```python theme={null} from fibonacci import find_tool_for_task # Describe what you want to do tool = find_tool_for_task("send a message to slack") print(tool) # "slack_send_message" tool = find_tool_for_task("read data from google sheets") print(tool) # "google_sheets_read" ``` *** ## Google Workspace ### Google Sheets ```python Read theme={null} read = ToolNode( id="read_data", name="Read Sheet Data", tool="google_sheets_read", params={ "spreadsheet_id": "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms", "range": "Sheet1!A1:D100" } ) ``` ```python Write theme={null} write = ToolNode( id="write_data", name="Write to Sheet", tool="google_sheets_write", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Results!A1", "values": [ ["Name", "Score", "Status"], ["Alice", "95", "Pass"], ["Bob", "82", "Pass"] ] } ) ``` ```python Append theme={null} append = ToolNode( id="append_row", name="Append Row", tool="google_sheets_append", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Log!A:D", "values": [["{{timestamp}}", "{{analyze_data}}"]] } ) ``` ### Google Docs ```python Read Document theme={null} read_doc = ToolNode( id="read_doc", name="Read Document", tool="google_docs_read", params={ "document_id": "{{input.doc_id}}" } ) ``` ```python Create Document theme={null} create_doc = ToolNode( id="create_doc", name="Create Document", tool="google_docs_create", params={ "title": "{{input.title}}", "content": "{{generate_report}}" } ) ``` ### Gmail ```python theme={null} send_email = ToolNode( id="send_email", name="Send Email", tool="gmail_send", params={ "to": "recipient@example.com", "subject": "Your Report is Ready", "body": """ Hi {{input.recipient_name}}, Your weekly report is ready. Summary: {{executive_summary}} Best, Fibonacci Bot """ } ) ``` *** ## Communication Tools ### Slack ```python Send Message theme={null} slack_msg = ToolNode( id="notify_team", name="Notify Team", tool="slack_send_message", params={ "channel": "#team-updates", "message": "🎉 Report generated!\n\n{{executive_summary}}" } ) ``` ```python Send to User theme={null} slack_dm = ToolNode( id="dm_user", name="DM User", tool="slack_send_message", params={ "channel": "@john.doe", "message": "Hey! Your task is complete." } ) ``` ```python Rich Message theme={null} slack_rich = ToolNode( id="rich_msg", name="Rich Message", tool="slack_send_message", params={ "channel": "#alerts", "blocks": [ { "type": "header", "text": {"type": "plain_text", "text": "📊 Weekly Report"} }, { "type": "section", "text": {"type": "mrkdwn", "text": "{{summary}}"} } ] } ) ``` *** ## HTTP & APIs ### Generic HTTP Request ```python theme={null} api_call = ToolNode( id="call_api", name="Call External API", tool="http_request", params={ "method": "POST", "url": "https://api.example.com/data", "headers": { "Authorization": "Bearer {{input.api_token}}", "Content-Type": "application/json" }, "body": { "query": "{{input.search_term}}", "limit": 10 } } ) ``` ### Webhook ```python theme={null} webhook = ToolNode( id="trigger_webhook", name="Trigger Webhook", tool="webhook_call", params={ "url": "https://hooks.zapier.com/hooks/catch/123/abc", "payload": { "event": "report_generated", "data": "{{report_data}}" } } ) ``` *** ## Database Tools ### SQL Query ```python theme={null} query = ToolNode( id="query_db", name="Query Database", tool="database_query", params={ "connection_string": "{{secrets.DATABASE_URL}}", "query": """ SELECT customer_id, SUM(amount) as total FROM orders WHERE date >= '{{input.start_date}}' GROUP BY customer_id ORDER BY total DESC LIMIT 10 """ } ) ``` Never hardcode database credentials. Use secrets management. *** ## Cloud Storage ### AWS S3 ```python Upload theme={null} s3_upload = ToolNode( id="upload_s3", name="Upload to S3", tool="s3_upload", params={ "bucket": "my-reports", "key": "reports/{{input.report_id}}.pdf", "content": "{{report_pdf}}", "content_type": "application/pdf" } ) ``` ```python Download theme={null} s3_download = ToolNode( id="download_s3", name="Download from S3", tool="s3_download", params={ "bucket": "my-data", "key": "data/{{input.file_name}}" } ) ``` *** ## CRM & Business Tools ### Salesforce ```python theme={null} create_lead = ToolNode( id="create_lead", name="Create Salesforce Lead", tool="salesforce_create", params={ "object": "Lead", "data": { "FirstName": "{{input.first_name}}", "LastName": "{{input.last_name}}", "Email": "{{input.email}}", "Company": "{{input.company}}", "Description": "{{qualify_lead}}" } } ) ``` ### HubSpot ```python theme={null} hubspot_contact = ToolNode( id="update_contact", name="Update HubSpot Contact", tool="hubspot_update_contact", params={ "email": "{{input.email}}", "properties": { "lead_score": "{{score_lead}}", "last_activity": "{{timestamp}}" } } ) ``` ### Notion ```python theme={null} notion_page = ToolNode( id="create_notion", name="Create Notion Page", tool="notion_create_page", params={ "database_id": "{{input.database_id}}", "properties": { "Name": "{{input.title}}", "Status": "Ready", "Priority": "High" }, "content": "{{generate_notes}}" } ) ``` *** ## Tool Categories ```python theme={null} from fibonacci import list_tool_categories categories = list_tool_categories() for cat in categories: print(f"{cat['name']}: {cat['count']} tools") print(f" Examples: {', '.join(cat['tools'][:3])}") ``` Output: ``` google_workspace: 7 tools Examples: google_sheets_read, google_docs_read, gmail_send communication: 3 tools Examples: slack_send_message, slack_upload_file, slack_create_channel databases: 3 tools Examples: database_query, mongodb_find, redis_get cloud_storage: 2 tools Examples: s3_upload, s3_download ... ``` *** ## Error Handling ### Retry on Failure ```python theme={null} api_call = ToolNode( id="api_call", name="API Call", tool="http_request", params={...} ).with_retry(max_retries=3, delay=2.0) ``` ### Timeout Configuration ```python theme={null} slow_api = ToolNode( id="slow_api", name="Slow API", tool="http_request", params={...}, timeout=60 # override the 30s default ) ``` *** ## Next Steps Branch workflows based on tool outputs Store and retrieve data across runs # Workflows Source: https://docs.fibonacci.today/guides/workflows Understanding workflows and how to build them A **Workflow** is the core building block of Fibonacci. It's a directed graph of nodes that execute in order based on their dependencies. ## Anatomy of a Workflow ```python theme={null} from fibonacci import Workflow wf = Workflow( name="My Workflow", # Required: Human-readable name description="What it does", # Optional: Description version=1, # Optional: Version number is_active=True, # Optional: Whether workflow is active tags=["tag1", "tag2"] # Optional: Tags for organization ) ``` ## Creating a Workflow ### From Python Code ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode wf = Workflow(name="Data Pipeline") # Add nodes node1 = ToolNode(id="fetch", name="Fetch Data", tool="api_call", params={}) node2 = LLMNode(id="process", name="Process", instruction="...", dependencies=["fetch"]) wf.add_nodes([node1, node2]) ``` ### From YAML ```python theme={null} from fibonacci import Workflow wf = Workflow.from_yaml("workflow.yaml") ``` YAML format: ```yaml workflow.yaml theme={null} name: Data Pipeline description: Fetches and processes data version: 1 tags: - data - etl nodes: - id: fetch type: tool name: Fetch Data tool_name: api_call tool_params: {} - id: process type: llm name: Process instruction: "Process this data: {{fetch}}" dependencies: - fetch ``` ## Workflow Lifecycle Define your workflow and nodes in Python or YAML Check for errors like circular dependencies or missing nodes ```python theme={null} wf.validate() ``` Upload to Fibonacci platform ```python theme={null} workflow_id = wf.deploy(api_key="...") ``` Run with input data ```python theme={null} result = wf.run(input_data={"key": "value"}) ``` Check status and retrieve results ```python theme={null} status = wf.get_status(run_id) stats = wf.get_stats() ``` ## Workflow Properties | Property | Type | Description | | ------------- | ------------ | ------------------------------------- | | `name` | `str` | Human-readable workflow name | | `description` | `str` | Description of what the workflow does | | `version` | `int` | Version number (default: 1) | | `is_active` | `bool` | Whether workflow can be executed | | `tags` | `list[str]` | Tags for organization and filtering | | `workflow_id` | `str` | Assigned after deployment | | `nodes` | `list[Node]` | List of nodes in the workflow | | `node_count` | `int` | Number of nodes | ## Managing Workflows ### List All Workflows ```python theme={null} from fibonacci import FibonacciClient from fibonacci.config import Config config = Config.from_env() async with FibonacciClient(config) as client: workflows = await client.list_workflows() for wf in workflows: print(f"{wf['name']} ({wf['id']})") ``` Or use the CLI: ```bash theme={null} fibonacci list ``` ### Update a Workflow ```python theme={null} # Update name and description wf.update( name="New Name", description="New description" ) # Deactivate wf.deactivate() # Reactivate wf.activate() ``` ### Delete a Workflow ```python theme={null} wf.delete() # Or by ID wf.delete(workflow_id="wf_abc123") ``` ### Get Workflow Statistics ```python theme={null} stats = wf.get_stats() print(f"Total runs: {stats.total_runs}") print(f"Success rate: {stats.success_rate}%") print(f"Average duration: {stats.avg_duration_seconds}s") print(f"Total cost: ${stats.total_cost:.2f}") ``` ## Execution ### Synchronous Execution Wait for the workflow to complete: ```python theme={null} result = wf.run( input_data={"text": "Hello world"}, wait=True, # Wait for completion (default) timeout=300.0 # Timeout in seconds ) print(result.status) # "completed" or "failed" print(result.output_data) # Node outputs print(result.duration_seconds) print(result.total_cost) ``` ### Asynchronous Execution Start execution and check later: ```python theme={null} # Start without waiting result = wf.run( input_data={"text": "Hello"}, wait=False ) run_id = result.id print(f"Started run: {run_id}") # Check status later status = wf.get_status(run_id) print(f"Status: {status.status}") # Wait for completion final = await client.wait_for_completion(run_id) ``` ### Async/Await Pattern For high-performance applications: ```python theme={null} import asyncio from fibonacci import Workflow async def main(): wf = Workflow(name="Async Example") # ... add nodes ... # Deploy async workflow_id = await wf.deploy_async(api_key="...") # Run async result = await wf.run_async(input_data={"key": "value"}) print(result.output_data) asyncio.run(main()) ``` ## Input Data Pass data to your workflow at runtime: ```python theme={null} result = wf.run(input_data={ "customer_id": "cust_123", "report_type": "monthly", "include_charts": True, "filters": { "start_date": "2024-01-01", "end_date": "2024-01-31" } }) ``` Access in nodes with `{{input.field_name}}`: ```python theme={null} LLMNode( id="generate", instruction=""" Generate a {{input.report_type}} report for customer {{input.customer_id}}. Date range: {{input.filters.start_date}} to {{input.filters.end_date}} """ ) ``` ## Output Data Each node's output is stored by its ID: ```python theme={null} result = wf.run(input_data={...}) # Access individual node outputs raw_data = result.output_data["fetch_data"] analysis = result.output_data["analyze"] summary = result.output_data["summarize"] ``` ## Best Practices ```python theme={null} # ❌ Bad wf = Workflow(name="wf1") # ✅ Good wf = Workflow(name="Weekly Sales Report Generator") ``` ```python theme={null} wf = Workflow( name="Customer Onboarding", tags=["customers", "onboarding", "production", "v2"] ) ``` ```python theme={null} # In code wf = Workflow(name="Pipeline", version=2) # Or use YAML files with git version control wf = Workflow.from_yaml("pipelines/v2/main.yaml") ``` ```python theme={null} try: wf.validate() wf.deploy(api_key=api_key) except ValidationError as e: print(f"Fix these errors: {e.errors}") ``` ## Next Steps Learn about the different types of nodes Define workflows declaratively # YAML Workflows Source: https://docs.fibonacci.today/guides/yaml-workflows Define workflows declaratively using YAML configuration files YAML workflows let you define Fibonacci workflows in a declarative format, separating workflow structure from application code. This approach is ideal for version control, code review, and non-developer workflow authoring. ## Basic Structure A YAML workflow file defines the complete workflow configuration: ```yaml theme={null} name: text-analyzer description: Analyze text sentiment and extract key themes version: "1.0" nodes: - id: sentiment type: llm model: claude-sonnet-4-5-20250929 prompt: | Analyze the sentiment of this text: {{input.text}} Return: positive, negative, or neutral - id: themes type: llm model: claude-sonnet-4-5-20250929 prompt: | Extract key themes from this text: {{input.text}} Return as a comma-separated list. dependencies: - sentiment outputs: sentiment: "{{sentiment}}" themes: "{{themes}}" ``` ## Loading YAML Workflows ### From File ```python theme={null} from fibonacci import Workflow # Load from file workflow = Workflow.from_yaml("workflows/text-analyzer.yaml") # Execute result = workflow.execute(inputs={"text": "I love this product!"}) print(result) ``` ### From String ```python theme={null} from fibonacci import Workflow yaml_content = """ name: quick-summary nodes: - id: summarize type: llm model: claude-sonnet-4-5-20250929 prompt: "Summarize: {{input.text}}" """ workflow = Workflow.from_yaml_string(yaml_content) ``` ### From URL ```python theme={null} from fibonacci import Workflow # Load from remote URL workflow = Workflow.from_yaml_url( "https://example.com/workflows/analyzer.yaml" ) ``` ## Node Types ### LLM Nodes ```yaml theme={null} nodes: - id: writer type: llm model: claude-sonnet-4-5-20250929 prompt: | Write a {{input.style}} paragraph about {{input.topic}}. temperature: 0.7 max_tokens: 500 - id: structured_output type: llm model: claude-sonnet-4-5-20250929 prompt: "Extract entities from: {{input.text}}" output_format: json schema: type: object properties: people: type: array items: type: string places: type: array items: type: string ``` ### Tool Nodes ```yaml theme={null} nodes: - id: fetch_data type: tool tool: google_sheets.read_range inputs: spreadsheet_id: "{{input.sheet_id}}" range: "A1:D100" - id: send_notification type: tool tool: slack.post_message inputs: channel: "#alerts" message: "Analysis complete: {{analyzer}}" dependencies: - analyzer ``` ### Conditional Nodes ```yaml theme={null} nodes: - id: route_by_sentiment type: conditional conditions: - if: field: "{{sentiment}}" operator: equals value: "positive" then: positive_handler - if: field: "{{sentiment}}" operator: equals value: "negative" then: negative_handler default: neutral_handler dependencies: - sentiment ``` ### Critic Nodes ```yaml theme={null} nodes: - id: quality_check type: critic model: claude-sonnet-4-5-20250929 target_node: writer criteria: - "Content is factually accurate" - "Tone matches requested style" - "Length is appropriate" min_score: 7 max_iterations: 3 ``` ## Advanced Configuration ### Variables and Secrets Use environment variables and secrets in YAML: ```yaml theme={null} name: secure-workflow version: "1.0" # Reference environment variables env: API_ENDPOINT: ${API_ENDPOINT} DEFAULT_MODEL: ${FIBONACCI_MODEL:-claude-sonnet-4-5-20250929} # Reference secrets (from secure storage) secrets: - OPENAI_API_KEY - DATABASE_URL nodes: - id: fetch type: tool tool: http.request inputs: url: "{{env.API_ENDPOINT}}/data" headers: Authorization: "Bearer {{secrets.API_KEY}}" ``` ### Retry Configuration ```yaml theme={null} nodes: - id: unreliable_api type: tool tool: http.request inputs: url: "https://api.example.com/data" retry: max_attempts: 3 delay: 1.0 backoff: exponential max_delay: 30.0 retry_on: - timeout - 5xx ``` ### Memory Configuration ```yaml theme={null} name: stateful-workflow memory: backend: redis connection: ${REDIS_URL} key_prefix: "myapp:" nodes: - id: chatbot type: llm model: claude-sonnet-4-5-20250929 prompt: | Conversation history: {{memory.history}} User: {{input.message}} memory_read: - history memory_write: key: last_response scope: user ``` ### Timeouts ```yaml theme={null} nodes: - id: long_analysis type: llm model: claude-opus-4-5-20251101 prompt: "Deep analysis of: {{input.data}}" timeout: 120 # seconds - id: quick_check type: tool tool: http.request inputs: url: "https://api.example.com/health" timeout: 5 ``` ## Workflow Composition ### Including Other Workflows ```yaml theme={null} name: master-pipeline version: "1.0" includes: - path: ./preprocessing.yaml as: preprocess - path: ./analysis.yaml as: analyze - path: ./reporting.yaml as: report nodes: - id: run_preprocess type: workflow workflow: preprocess inputs: data: "{{input.raw_data}}" - id: run_analysis type: workflow workflow: analyze inputs: clean_data: "{{run_preprocess.output}}" dependencies: - run_preprocess - id: run_report type: workflow workflow: report inputs: results: "{{run_analysis}}" dependencies: - run_analysis ``` ### Parallel Execution ```yaml theme={null} name: parallel-analysis version: "1.0" nodes: # These run in parallel (no dependencies between them) - id: sentiment_analysis type: llm model: claude-sonnet-4-5-20250929 prompt: "Analyze sentiment: {{input.text}}" - id: entity_extraction type: llm model: claude-sonnet-4-5-20250929 prompt: "Extract entities: {{input.text}}" - id: topic_classification type: llm model: claude-sonnet-4-5-20250929 prompt: "Classify topics: {{input.text}}" # This waits for all parallel nodes - id: combine_results type: llm model: claude-sonnet-4-5-20250929 prompt: | Combine these analysis results: Sentiment: {{sentiment_analysis}} Entities: {{entity_extraction}} Topics: {{topic_classification}} dependencies: - sentiment_analysis - entity_extraction - topic_classification ``` ## Validation ### Schema Validation Fibonacci validates YAML workflows against a JSON Schema: ```python theme={null} from fibonacci import Workflow, WorkflowValidationError try: workflow = Workflow.from_yaml("workflow.yaml") except WorkflowValidationError as e: print(f"Validation errors: {e.errors}") ``` ### CLI Validation ```bash theme={null} # Validate a workflow file fibonacci validate workflow.yaml # Validate with verbose output fibonacci validate workflow.yaml --verbose # Validate all workflows in a directory fibonacci validate workflows/ ``` ### Custom Validation Rules ```yaml theme={null} name: validated-workflow version: "1.0" validation: require_descriptions: true max_nodes: 20 allowed_models: - claude-sonnet-4-5-20250929 - claude-haiku-4-5-20251001 forbidden_tools: - http.request # Require specific API tools instead nodes: - id: analyzer type: llm model: claude-sonnet-4-5-20250929 description: "Analyzes input text for sentiment" # Required prompt: "Analyze: {{input.text}}" ``` ## Complete Example Here's a full production workflow example: ```yaml theme={null} name: customer-support-router description: Routes customer inquiries to appropriate handlers version: "2.1" author: support-team@company.com env: SLACK_CHANNEL: ${SUPPORT_SLACK_CHANNEL:-#support} ESCALATION_THRESHOLD: ${ESCALATION_THRESHOLD:-0.8} memory: backend: redis connection: ${REDIS_URL} nodes: # Classify the inquiry - id: classify type: llm model: claude-sonnet-4-5-20250929 prompt: | Classify this customer inquiry: {{input.message}} Categories: billing, technical, general, complaint Urgency: low, medium, high, critical Return JSON: {"category": "...", "urgency": "..."} output_format: json timeout: 30 # Check for escalation triggers - id: check_escalation type: conditional conditions: - if: field: "{{classify.urgency}}" operator: equals value: "critical" then: escalate_immediately - if: field: "{{classify.category}}" operator: equals value: "complaint" then: complaint_handler default: standard_response dependencies: - classify # Generate standard response - id: standard_response type: llm model: claude-sonnet-4-5-20250929 prompt: | Previous interactions: {{memory.customer_history}} Customer inquiry ({{classify.category}}): {{input.message}} Provide a helpful response. memory_read: - customer_history dependencies: - classify # Handle complaints specially - id: complaint_handler type: llm model: claude-sonnet-4-5-20250929 prompt: | Handle this complaint with empathy: {{input.message}} Previous issues: {{memory.customer_history}} Acknowledge concerns and offer resolution. memory_read: - customer_history temperature: 0.3 # More consistent tone dependencies: - classify # Immediate escalation - id: escalate_immediately type: tool tool: slack.post_message inputs: channel: "{{env.SLACK_CHANNEL}}" message: | 🚨 CRITICAL ESCALATION Customer: {{input.customer_id}} Category: {{classify.category}} Message: {{input.message}} dependencies: - classify # Log interaction - id: log_interaction type: tool tool: database.insert inputs: table: support_interactions data: customer_id: "{{input.customer_id}}" category: "{{classify.category}}" urgency: "{{classify.urgency}}" timestamp: "{{now}}" dependencies: - classify outputs: response: "{{standard_response || complaint_handler}}" category: "{{classify.category}}" urgency: "{{classify.urgency}}" escalated: "{{escalate_immediately != null}}" ``` ## Best Practices Store YAML workflows in Git for change tracking, code review, and rollback capabilities. Use environment variables for configuration that differs between dev/staging/production: ```yaml theme={null} env: API_URL: ${API_URL} MODEL: ${MODEL:-claude-sonnet-4-5-20250929} ``` Add descriptions to workflows and nodes: ```yaml theme={null} name: my-workflow description: | This workflow processes customer feedback. Author: team@company.com Last updated: 2025-01-15 ``` Always run `fibonacci validate` in CI/CD pipelines before deploying workflow changes. ## Next Steps Learn the Fibonacci command-line interface Production workflow patterns # Fibonacci SDK Source: https://docs.fibonacci.today/index Build and deploy AI-powered workflows programmatically with enterprise-grade security Fibonacci SDK Hero Fibonacci SDK Hero ## Welcome to Fibonacci Fibonacci is the **Cursor for SaaS tools** - a powerful Python SDK that lets you build, deploy, and execute AI-powered workflows that orchestrate your favorite tools and services. Get up and running with Fibonacci in under 5 minutes Build your first AI workflow step-by-step Complete reference for all SDK classes and methods REST API documentation for direct integrations ## Why Fibonacci? Built from the ground up for LLM-powered automation. Our LLM nodes seamlessly integrate with Claude and other models to analyze, generate, and transform data. Connect to Google Sheets, Slack, Notion, Salesforce, and dozens more. Execute real actions across your entire tool stack. Secure keychain storage, comprehensive audit logging, and API key redaction. Built for teams that take security seriously. Define workflows programmatically in Python or declaratively in YAML. Version control your automation like any other code. ## Quick Example Here's a simple workflow that reads data from Google Sheets and generates an analysis: ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode # Create a workflow wf = Workflow( name="Sales Analysis", description="Analyze sales data from Google Sheets" ) # Read data from Google Sheets read_data = ToolNode( id="read_sales", name="Read Sales Data", tool="google_sheets_read", params={"spreadsheet_id": "{{input.sheet_id}}"} ) # Analyze with AI analyze = LLMNode( id="analyze", name="Generate Analysis", instruction=""" Analyze this sales data and provide: 1. Key trends 2. Top performers 3. Areas for improvement Data: {{read_sales}} """, dependencies=["read_sales"] ) # Add nodes and deploy wf.add_nodes([read_data, analyze]) workflow_id = wf.deploy(api_key="fib_live_xxx") # Execute result = wf.run(input_data={"sheet_id": "abc123"}) print(result.output_data["analyze"]) ``` ## What Can You Build? ETL workflows that extract, transform, and load data across services Autonomous agents that research, analyze, and take action Automated reports that pull data and generate insights AI-powered content workflows with human-in-the-loop Intelligent ticket routing and response generation Lead scoring, outreach, and CRM automation ## Getting Help Join our community for help and discussions Report bugs or request features # FibonacciClient Source: https://docs.fibonacci.today/sdk-reference/client Client for interacting with Fibonacci Cloud API The `FibonacciClient` class provides methods for interacting with Fibonacci Cloud, including workflow deployment, execution, and management. ## Constructor ```python theme={null} from fibonacci import FibonacciClient client = FibonacciClient( api_key="your-api-key", base_url="https://api.fibonacci.dev", timeout=60 ) ``` ### Parameters | Parameter | Type | Default | Description | | ----------------- | ----- | ----------------------------- | ----------------------- | | `api_key` | `str` | Required | Fibonacci API key | | `base_url` | `str` | `"https://api.fibonacci.dev"` | API base URL | | `timeout` | `int` | `60` | Default request timeout | | `max_retries` | `int` | `3` | Max retry attempts | | `organization_id` | `str` | `None` | Organization identifier | ## Authentication ### From Environment ```python theme={null} import os from fibonacci import FibonacciClient # Uses FIBONACCI_API_KEY environment variable client = FibonacciClient.from_env() ``` ### From Keychain ```python theme={null} from fibonacci import FibonacciClient # Uses system keychain client = FibonacciClient.from_keychain() ``` ## Workflow Methods ### deploy Deploy a workflow to Fibonacci Cloud. ```python theme={null} client.deploy( workflow: Workflow | str, environment: str = "production", version: str | None = None ) -> DeploymentResult ``` **Parameters:** * `workflow`: Workflow instance or path to YAML file * `environment`: Target environment (development/staging/production) * `version`: Version tag **Returns:** `DeploymentResult` **Example:** ```python theme={null} result = client.deploy( workflow="workflows/analyzer.yaml", environment="production", version="1.2.0" ) print(f"Deployed: {result.deployment_id}") ``` *** ### execute Execute a deployed workflow. ```python theme={null} client.execute( workflow_name: str, inputs: dict, environment: str = "production", timeout: int | None = None, async_execution: bool = False ) -> ExecutionResult | ExecutionHandle ``` **Parameters:** * `workflow_name`: Name of deployed workflow * `inputs`: Input data * `environment`: Target environment * `timeout`: Execution timeout * `async_execution`: Return immediately with handle **Returns:** `ExecutionResult` or `ExecutionHandle` (if async) **Example:** ```python theme={null} # Synchronous result = client.execute( workflow_name="customer-analyzer", inputs={"text": "Hello world"} ) print(result.outputs) # Asynchronous handle = client.execute( workflow_name="long-running", inputs={"data": large_data}, async_execution=True ) # Check status later status = client.get_execution(handle.execution_id) ``` *** ### get\_execution Get execution status and results. ```python theme={null} client.get_execution(execution_id: str) -> ExecutionResult ``` **Parameters:** * `execution_id`: Execution identifier **Returns:** `ExecutionResult` *** ### list\_executions List workflow executions. ```python theme={null} client.list_executions( workflow_name: str | None = None, environment: str | None = None, status: str | None = None, limit: int = 100, offset: int = 0 ) -> list[ExecutionSummary] ``` **Parameters:** * `workflow_name`: Filter by workflow * `environment`: Filter by environment * `status`: Filter by status (running/success/failed) * `limit`: Max results * `offset`: Pagination offset **Returns:** List of `ExecutionSummary` *** ### cancel\_execution Cancel a running execution. ```python theme={null} client.cancel_execution(execution_id: str) -> bool ``` **Parameters:** * `execution_id`: Execution identifier **Returns:** `True` if cancelled ## Workflow Management ### list\_workflows List deployed workflows. ```python theme={null} client.list_workflows( environment: str | None = None ) -> list[WorkflowSummary] ``` *** ### get\_workflow Get workflow details. ```python theme={null} client.get_workflow( workflow_name: str, environment: str = "production" ) -> WorkflowDetails ``` *** ### delete\_workflow Delete a deployed workflow. ```python theme={null} client.delete_workflow( workflow_name: str, environment: str = "production" ) -> bool ``` ## Tool Methods ### list\_tools List available tools. ```python theme={null} client.list_tools( category: str | None = None ) -> list[ToolInfo] ``` **Example:** ```python theme={null} tools = client.list_tools(category="google") for tool in tools: print(f"{tool.name}: {tool.description}") ``` *** ### get\_tool Get tool details. ```python theme={null} client.get_tool(tool_name: str) -> ToolDetails ``` *** ### connect\_tool Connect an external tool. ```python theme={null} client.connect_tool( tool_name: str, credentials: dict | None = None ) -> ConnectionResult ``` **Example:** ```python theme={null} # OAuth-based connection (opens browser) result = client.connect_tool("google") # Credential-based connection result = client.connect_tool("slack", credentials={ "bot_token": "xoxb-..." }) ``` ## Memory Methods ### get\_memory Retrieve a memory value. ```python theme={null} client.get_memory( key: str, scope: str = "workflow", user_id: str | None = None ) -> Any ``` *** ### set\_memory Store a memory value. ```python theme={null} client.set_memory( key: str, value: Any, scope: str = "workflow", user_id: str | None = None, ttl: int | None = None ) -> bool ``` *** ### delete\_memory Delete a memory value. ```python theme={null} client.delete_memory( key: str, scope: str = "workflow", user_id: str | None = None ) -> bool ``` ## Async Client For async applications, use `AsyncFibonacciClient`: ```python theme={null} from fibonacci import AsyncFibonacciClient import asyncio async def main(): client = AsyncFibonacciClient(api_key="...") result = await client.execute( workflow_name="analyzer", inputs={"text": "Hello"} ) print(result.outputs) asyncio.run(main()) ``` ## Response Types ### DeploymentResult ```python theme={null} @dataclass class DeploymentResult: deployment_id: str workflow_name: str version: str environment: str status: str url: str created_at: datetime ``` ### ExecutionResult ```python theme={null} @dataclass class ExecutionResult: execution_id: str workflow_name: str status: str # running, success, failed outputs: dict | None error: str | None duration_ms: int started_at: datetime completed_at: datetime | None ``` ### WorkflowSummary ```python theme={null} @dataclass class WorkflowSummary: name: str version: str environment: str status: str last_deployed: datetime execution_count: int ``` ## Error Handling ```python theme={null} from fibonacci import FibonacciClient from fibonacci.exceptions import ( AuthenticationError, RateLimitError, WorkflowNotFoundError ) client = FibonacciClient(api_key="...") try: result = client.execute( workflow_name="my-workflow", inputs={"text": "Hello"} ) except AuthenticationError: print("Invalid API key") except RateLimitError as e: print(f"Rate limited. Retry after {e.retry_after}s") except WorkflowNotFoundError: print("Workflow not deployed") ``` ## Complete Example ```python theme={null} from fibonacci import FibonacciClient, Workflow # Initialize client client = FibonacciClient.from_env() # Deploy workflow workflow = Workflow.from_yaml("workflows/analyzer.yaml") deployment = client.deploy( workflow=workflow, environment="production", version="1.0.0" ) print(f"Deployed: {deployment.url}") # Execute workflow result = client.execute( workflow_name="analyzer", inputs={"text": "Analyze this text"}, timeout=60 ) if result.status == "success": print(f"Result: {result.outputs}") else: print(f"Error: {result.error}") # List recent executions executions = client.list_executions( workflow_name="analyzer", status="failed", limit=10 ) for ex in executions: print(f"{ex.execution_id}: {ex.status}") ``` # ConditionalNode Source: https://docs.fibonacci.today/sdk-reference/conditional-node Node for branching workflow execution based on a condition The `ConditionalNode` class branches workflow execution based on a single condition, routing to one set of nodes when the condition is true and another when it is false. ## Constructor ```python theme={null} from fibonacci import ConditionalNode node = ConditionalNode( id="check_sentiment", name="Check Sentiment", left_value="{{analyze_sentiment}}", operator="contains", right_value="positive", true_branch=["celebrate"], false_branch=["investigate"], dependencies=["analyze_sentiment"] ) ``` ### Parameters | Parameter | Type | Default | Description | | -------------- | ----------- | -------- | ------------------------------------------------------------------------------------------------ | | `id` | `str` | Required | Unique node identifier (lowercase letters, numbers, underscores, hyphens) | | `name` | `str` | Required | Human-readable node name | | `left_value` | `str` | Required | Left side of the condition — supports template variables like `{{node_id}}` or `{{input.field}}` | | `operator` | `str` | Required | Comparison operator (see table below) | | `right_value` | `str` | `""` | Right side of the condition — leave empty for `is_empty` / `is_not_empty` operators | | `true_branch` | `list[str]` | `[]` | Node IDs to execute when the condition is **true** | | `false_branch` | `list[str]` | `[]` | Node IDs to execute when the condition is **false** | | `dependencies` | `list[str]` | `[]` | Node IDs this node waits for before running | | `enable_retry` | `bool` | `False` | Retry on failure | | `max_retries` | `int` | `3` | Maximum retry attempts (used when `enable_retry=True`) | | `retry_delay` | `float` | `1.0` | Initial delay in seconds between retries | ## Supported Operators The `operator` field accepts exactly these 10 values: | Operator | Description | Example | | -------------- | ----------------------------------------------------------- | -------------------------------------- | | `equals` | Exact string or numeric match | `"completed"` equals `"completed"` | | `not_equals` | Not equal | `"pending"` not\_equals `"completed"` | | `contains` | Left value contains right value as a substring | `"error on line 5"` contains `"error"` | | `not_contains` | Left value does **not** contain right value | `"success"` not\_contains `"error"` | | `greater_than` | Numeric greater-than | `100` greater\_than `50` | | `less_than` | Numeric less-than | `25` less\_than `50` | | `starts_with` | Left value starts with right value | `"error: msg"` starts\_with `"error"` | | `ends_with` | Left value ends with right value | `"report.pdf"` ends\_with `".pdf"` | | `is_empty` | Left value is empty or null (`right_value` is ignored) | `""` is\_empty | | `is_not_empty` | Left value has a non-empty value (`right_value` is ignored) | `"some text"` is\_not\_empty | Only these 10 operators are supported. Using any other string will raise a `ValidationError` at workflow build time. ## Basic Examples ### Simple True/False Branch ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode, ConditionalNode wf = Workflow(name="feedback-router") # Step 1: Classify the feedback classify = LLMNode( id="classify", name="Classify Feedback", instruction="Classify this feedback as positive or negative. Reply with one word only.\n\n{{input.feedback}}" ) # Step 2: Route based on result router = ConditionalNode( id="router", name="Route Feedback", left_value="{{classify}}", operator="contains", right_value="negative", true_branch=["escalate"], false_branch=["log_ok"], dependencies=["classify"] ) # True branch escalate = ToolNode( id="escalate", name="Escalate to Support", tool="slack_send_message", params={ "channel": "#support", "message": "Negative feedback: {{input.feedback}}" }, dependencies=["router"] ) # False branch log_ok = ToolNode( id="log_ok", name="Log Positive Feedback", tool="google_sheets_append", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Log!A:B", "values": [["{{input.feedback}}", "positive"]] }, dependencies=["router"] ) wf.add_nodes([classify, router, escalate, log_ok]) ``` ### Numeric Threshold ```python theme={null} check_score = ConditionalNode( id="check_score", name="Score Threshold", left_value="{{evaluate}}", operator="greater_than", right_value="7", true_branch=["publish"], false_branch=["revise"], dependencies=["evaluate"] ) ``` ### String Prefix/Suffix Check ```python theme={null} check_type = ConditionalNode( id="check_type", name="Check File Type", left_value="{{input.filename}}", operator="ends_with", right_value=".pdf", true_branch=["process_pdf"], false_branch=["unsupported_format"] ) ``` ### Empty Value Guard ```python theme={null} check_data = ConditionalNode( id="check_data", name="Data Present?", left_value="{{fetch_data}}", operator="is_not_empty", right_value="", # ignored for is_empty / is_not_empty true_branch=["process_data"], false_branch=["send_no_data_alert"] ) ``` ## Chaining Conditions (AND / OR Logic) The `ConditionalNode` evaluates a single condition. Chain multiple nodes together to build AND / OR logic. ### AND Logic — chain nodes ```python theme={null} # First check: sentiment negative? check_sentiment = ConditionalNode( id="check_sentiment", name="Negative Sentiment?", left_value="{{analyze}}", operator="contains", right_value="negative", true_branch=["check_urgency"], # only continue AND chain when true false_branch=["standard_response"], dependencies=["analyze"] ) # Second check: urgency high? (only reached when sentiment IS negative) check_urgency = ConditionalNode( id="check_urgency", name="High Urgency?", left_value="{{analyze_urgency}}", operator="equals", right_value="high", true_branch=["escalate"], # negative AND high urgency false_branch=["soft_response"], dependencies=["check_sentiment"] ) ``` ### OR Logic — multiple conditions pointing to the same branch ```python theme={null} # Escalate if sentiment is negative check_neg = ConditionalNode( id="check_neg", name="Negative?", left_value="{{analyze.sentiment}}", operator="equals", right_value="negative", true_branch=["escalate"], false_branch=["check_critical"], dependencies=["analyze"] ) # OR escalate if category is complaint check_critical = ConditionalNode( id="check_critical", name="Complaint?", left_value="{{analyze.category}}", operator="equals", right_value="complaint", true_branch=["escalate"], # same escalate node false_branch=["normal_flow"], dependencies=["check_neg"] ) ``` ## Per-Node Condition (Conditional Execution) Every node type supports `.with_condition()` to skip execution unless a condition is met: ```python theme={null} # Only send email if the user opted in send_email = ToolNode( id="send_email", name="Send Email", tool="gmail_send", params={...} ).with_condition( left_value="{{input.email_opt_in}}", operator="equals", right_value="true" ) # Only run expensive analysis on large payloads deep_analysis = LLMNode( id="deep_analysis", name="Deep Analysis", instruction="..." ).with_condition( left_value="{{input.data_size}}", operator="greater_than", right_value="1000" ) ``` ## Retry Configuration Use `.with_retry()` to enable retries on the conditional node itself: ```python theme={null} router = ConditionalNode( id="router", name="Route Request", left_value="{{classify}}", operator="equals", right_value="urgent", true_branch=["urgent_handler"], false_branch=["standard_handler"], dependencies=["classify"] ).with_retry(max_retries=3, delay=1.0) ``` ## YAML Configuration ```yaml theme={null} nodes: - id: router name: Route by Sentiment type: condition left_value: "{{analyze}}" operator: contains right_value: "negative" true_branch: - escalate false_branch: - log_ok dependencies: - analyze ``` ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode, ConditionalNode wf = Workflow(name="support-ticket-router") # Classify ticket classify = LLMNode( id="classify", name="Classify Ticket", instruction="""Classify this support ticket. Respond with exactly one word: urgent, billing, technical, or general. Ticket: {{input.message}}""" ) # Route urgent tickets immediately check_urgent = ConditionalNode( id="check_urgent", name="Urgent?", left_value="{{classify}}", operator="equals", right_value="urgent", true_branch=["page_on_call"], false_branch=["check_billing"], dependencies=["classify"] ) # Route billing tickets check_billing = ConditionalNode( id="check_billing", name="Billing Issue?", left_value="{{classify}}", operator="equals", right_value="billing", true_branch=["billing_team"], false_branch=["tech_team"], dependencies=["check_urgent"] ) # Urgent: page on-call engineer page_on_call = ToolNode( id="page_on_call", name="Page On-Call", tool="slack_send_message", params={ "channel": "#on-call", "message": "🚨 Urgent ticket: {{input.message}}" }, dependencies=["check_urgent"] ) # Billing team billing_team = ToolNode( id="billing_team", name="Notify Billing", tool="slack_send_message", params={ "channel": "#billing", "message": "Billing ticket: {{input.message}}" }, dependencies=["check_billing"] ) # Tech support (default) tech_team = LLMNode( id="tech_team", name="Tech Response", instruction="Provide a helpful technical response to: {{input.message}}", dependencies=["check_billing"] ) wf.add_nodes([classify, check_urgent, check_billing, page_on_call, billing_team, tech_team]) result = wf.run(input_data={"message": "My payment failed!"}) print(result.output_data) ``` # CriticNode Source: https://docs.fibonacci.today/sdk-reference/critic-node Node for evaluating output from another node against criteria The `CriticNode` class evaluates the output of a target node against a list of criteria. It is automatically added as a dependent of the target node so it always runs after the node it evaluates. ## Constructor ```python theme={null} from fibonacci import CriticNode node = CriticNode( id="evaluate_report", name="Evaluate Report Quality", target_node="generate_report", criteria=["clarity", "completeness", "accuracy"] ) ``` ### Parameters | Parameter | Type | Default | Description | | -------------- | ----------- | -------------------------------------------- | ------------------------------------------------------------------------- | | `id` | `str` | Required | Unique node identifier (lowercase letters, numbers, underscores, hyphens) | | `name` | `str` | Required | Human-readable node name | | `target_node` | `str` | Required | ID of the node whose output will be evaluated | | `criteria` | `list[str]` | `["quality", "correctness", "completeness"]` | Evaluation criteria labels | | `dependencies` | `list[str]` | auto | `target_node` is automatically added; add extras here if needed | | `enable_retry` | `bool` | `False` | Retry the critic node on failure | | `max_retries` | `int` | `3` | Maximum retry attempts (used when `enable_retry=True`) | | `retry_delay` | `float` | `1.0` | Initial delay in seconds between retries | The `target_node` is automatically added to `dependencies`. You do not need to list it manually. ## Basic Usage ```python theme={null} from fibonacci import Workflow, LLMNode, CriticNode wf = Workflow(name="quality-assured") # Content generator writer = LLMNode( id="writer", name="Write Email", instruction="Write a professional email about: {{input.topic}}" ) # Quality evaluator critic = CriticNode( id="critic", name="Evaluate Email Quality", target_node="writer", criteria=[ "professional tone throughout", "clear and concise language", "proper grammar and spelling", "appropriate greeting and closing" ] ) wf.add_nodes([writer, critic]) ``` ## Default Criteria When you omit `criteria`, the node uses three built-in labels: ```python theme={null} # These two are equivalent critic_a = CriticNode( id="critic_a", name="Default Critic", target_node="writer" ) critic_b = CriticNode( id="critic_b", name="Explicit Critic", target_node="writer", criteria=["quality", "correctness", "completeness"] ) ``` ## Custom Criteria Provide specific, measurable criteria for your use case: ### Content Writing ```python theme={null} content_critic = CriticNode( id="content_critic", name="Content Quality Check", target_node="blog_writer", criteria=[ "engaging opening hook", "clear value proposition", "actionable takeaways", "appropriate length for the audience" ] ) ``` ### Code Review ```python theme={null} code_critic = CriticNode( id="code_critic", name="Code Review", target_node="code_generator", criteria=[ "syntactically correct", "follows best practices", "includes error handling", "is well-documented" ] ) ``` ### Data Extraction ```python theme={null} extraction_critic = CriticNode( id="extraction_critic", name="Extraction Accuracy Check", target_node="extractor", criteria=[ "all required fields extracted", "values are in the correct format", "no hallucinated information" ] ) ``` ## Writing Good Criteria ```python theme={null} # ✅ Good — specific, observable criteria=[ "response is between 100 and 200 words", "includes at least one concrete example", "no grammatical errors" ] # ❌ Vague — hard to evaluate objectively criteria=[ "good quality", "sounds nice" ] ``` ```python theme={null} # For customer-facing content criteria=[ "friendly and empathetic tone", "directly addresses the user's question", "offers a clear next step" ] # For internal analysis criteria=[ "cites data from the source", "identifies at least 3 trends", "recommends concrete actions" ] ``` ## Retry Configuration ```python theme={null} critic = CriticNode( id="critic", name="Evaluator", target_node="writer" ).with_retry(max_retries=3, delay=1.0) ``` ## YAML Configuration ```yaml theme={null} nodes: - id: quality_check name: Quality Check type: critic target_node: writer evaluation_criteria: - professional tone - clear language - grammar and spelling dependencies: - writer ``` ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode, CriticNode wf = Workflow(name="blog-writer") # Generate blog post writer = LLMNode( id="writer", name="Write Blog Post", instruction=""" Write a blog post about {{input.topic}}. Target audience: {{input.audience}} Desired length: approximately {{input.word_count}} words Tone: {{input.tone}} """, model="claude-sonnet-4-6" ) # Evaluate quality critic = CriticNode( id="quality_check", name="Editorial Quality Check", target_node="writer", criteria=[ "engaging from the first paragraph", "well-researched and accurate", "writing style matches the target audience", "close to the requested word count", "tone is consistent throughout", "includes actionable takeaways", "compelling conclusion" ] ) wf.add_nodes([writer, critic]) result = wf.run(input_data={ "topic": "Introduction to Machine Learning", "audience": "business professionals", "word_count": 800, "tone": "informative yet approachable" }) print(result.output_data["quality_check"]) ``` # Exceptions Source: https://docs.fibonacci.today/sdk-reference/exceptions Exception classes for error handling in Fibonacci Fibonacci provides a hierarchy of exception classes for precise error handling. All exceptions inherit from `FibonacciError`. ## Exception Hierarchy ``` FibonacciError ├── WorkflowError │ ├── WorkflowNotFoundError │ ├── WorkflowValidationError │ └── WorkflowExecutionError ├── NodeExecutionError ├── ToolError │ ├── ToolNotFoundError │ ├── ToolConnectionError │ └── ToolExecutionError ├── ValidationError │ ├── InputValidationError │ └── OutputValidationError ├── AuthenticationError ├── RateLimitError ├── TimeoutError ├── MemoryError └── ConfigurationError ``` ## FibonacciError Base exception for all Fibonacci errors. ```python theme={null} from fibonacci.exceptions import FibonacciError try: result = workflow.execute(inputs=data) except FibonacciError as e: print(f"Fibonacci error: {e.message}") print(f"Error code: {e.code}") ``` ### Attributes | Attribute | Type | Description | | --------- | ------ | ------------------------------------ | | `message` | `str` | Human-readable error message | | `code` | `str` | Error code for programmatic handling | | `details` | `dict` | Additional error context | *** ## WorkflowError Errors related to workflow operations. ```python theme={null} from fibonacci.exceptions import WorkflowError try: workflow.validate() except WorkflowError as e: print(f"Workflow error: {e.message}") print(f"Workflow: {e.workflow_name}") ``` ### Attributes | Attribute | Type | Description | | --------------- | ----- | ----------------------------- | | `workflow_name` | `str` | Name of the affected workflow | ### Subclasses #### WorkflowNotFoundError ```python theme={null} from fibonacci.exceptions import WorkflowNotFoundError try: workflow = Workflow.from_yaml("missing.yaml") except WorkflowNotFoundError as e: print(f"Workflow not found: {e.path}") ``` #### WorkflowValidationError ```python theme={null} from fibonacci.exceptions import WorkflowValidationError try: workflow.validate() except WorkflowValidationError as e: print(f"Validation errors: {e.errors}") for error in e.errors: print(f" - {error['field']}: {error['message']}") ``` *** ## NodeExecutionError Errors during node execution. ```python theme={null} from fibonacci.exceptions import NodeExecutionError try: result = workflow.execute(inputs=data) except NodeExecutionError as e: print(f"Node '{e.node_id}' failed: {e.message}") print(f"Node type: {e.node_type}") print(f"Input data: {e.input_data}") print(f"Partial results: {e.partial_results}") ``` ### Attributes | Attribute | Type | Description | | ----------------- | ------ | ------------------------------ | | `node_id` | `str` | ID of the failed node | | `node_type` | `str` | Type of node (llm, tool, etc.) | | `input_data` | `dict` | Input provided to the node | | `partial_results` | `dict` | Results from successful nodes | | `traceback` | `str` | Full error traceback | ### Example: Handling with Partial Results ```python theme={null} try: result = workflow.execute(inputs=data) except NodeExecutionError as e: # Use partial results from successful nodes if "analyzer" in e.partial_results: analysis = e.partial_results["analyzer"] # Continue with degraded functionality return fallback_handler(analysis) raise ``` *** ## ToolError Errors from tool integrations. ```python theme={null} from fibonacci.exceptions import ToolError try: result = workflow.execute(inputs=data) except ToolError as e: print(f"Tool error: {e.tool_name}") print(f"Operation: {e.operation}") print(f"Details: {e.details}") ``` ### Attributes | Attribute | Type | Description | | ----------- | ----- | --------------------- | | `tool_name` | `str` | Name of the tool | | `operation` | `str` | Operation that failed | ### Subclasses #### ToolNotFoundError ```python theme={null} from fibonacci.exceptions import ToolNotFoundError try: node = ToolNode(id="test", tool="nonexistent.tool", inputs={}) except ToolNotFoundError as e: print(f"Tool not found: {e.tool_name}") print(f"Available tools: {e.available_tools}") ``` #### ToolConnectionError ```python theme={null} from fibonacci.exceptions import ToolConnectionError try: result = workflow.execute(inputs=data) except ToolConnectionError as e: print(f"Failed to connect to {e.tool_name}") print(f"Retry in {e.retry_after} seconds") ``` #### ToolExecutionError ```python theme={null} from fibonacci.exceptions import ToolExecutionError try: result = workflow.execute(inputs=data) except ToolExecutionError as e: print(f"Tool execution failed: {e.tool_name}") print(f"Response: {e.response}") ``` *** ## ValidationError Input or output validation failures. ```python theme={null} from fibonacci.exceptions import ValidationError try: result = workflow.execute(inputs={"text": ""}) except ValidationError as e: print(f"Validation failed: {e.message}") for error in e.errors: print(f" - {error['field']}: {error['error']}") ``` ### Attributes | Attribute | Type | Description | | --------- | ------------ | ------------------------- | | `errors` | `list[dict]` | List of validation errors | ### Error Format ```python theme={null} { "field": "text", "error": "String too short", "expected": {"minLength": 1}, "actual": "" } ``` ### Subclasses #### InputValidationError ```python theme={null} from fibonacci.exceptions import InputValidationError try: result = workflow.execute(inputs=invalid_data) except InputValidationError as e: print(f"Invalid input: {e.errors}") ``` #### OutputValidationError ```python theme={null} from fibonacci.exceptions import OutputValidationError try: result = workflow.execute(inputs=data) except OutputValidationError as e: print(f"Output didn't match schema: {e.errors}") print(f"Node: {e.node_id}") print(f"Output: {e.output}") ``` *** ## AuthenticationError Authentication and authorization failures. ```python theme={null} from fibonacci.exceptions import AuthenticationError try: client = FibonacciClient(api_key="invalid") client.list_workflows() except AuthenticationError as e: print(f"Auth failed: {e.message}") print(f"Reason: {e.reason}") # invalid_key, expired, insufficient_scope ``` ### Attributes | Attribute | Type | Description | | --------- | ----- | ---------------------------- | | `reason` | `str` | Specific auth failure reason | *** ## RateLimitError API rate limit exceeded. ```python theme={null} from fibonacci.exceptions import RateLimitError import time try: result = workflow.execute(inputs=data) except RateLimitError as e: print(f"Rate limited: {e.message}") print(f"Retry after: {e.retry_after} seconds") print(f"Limit: {e.limit}") print(f"Remaining: {e.remaining}") # Wait and retry time.sleep(e.retry_after) result = workflow.execute(inputs=data) ``` ### Attributes | Attribute | Type | Description | | ------------- | ----- | ---------------------------- | | `retry_after` | `int` | Seconds to wait before retry | | `limit` | `int` | Rate limit ceiling | | `remaining` | `int` | Remaining requests | *** ## TimeoutError Execution timeout exceeded. ```python theme={null} from fibonacci.exceptions import TimeoutError try: result = workflow.execute(inputs=data, timeout=30) except TimeoutError as e: print(f"Execution timed out after {e.timeout}s") print(f"Elapsed: {e.elapsed}s") print(f"Node: {e.node_id}") # Node that was running when timeout hit ``` ### Attributes | Attribute | Type | Description | | --------- | ------- | ----------------------- | | `timeout` | `int` | Configured timeout | | `elapsed` | `float` | Actual elapsed time | | `node_id` | `str` | Node running at timeout | *** ## MemoryError Memory operation failures. ```python theme={null} from fibonacci.exceptions import MemoryError try: workflow.memory.set("key", large_data) except MemoryError as e: print(f"Memory error: {e.message}") print(f"Operation: {e.operation}") # get, set, delete print(f"Key: {e.key}") ``` ### Attributes | Attribute | Type | Description | | ----------- | ----- | --------------------- | | `operation` | `str` | Memory operation type | | `key` | `str` | Memory key involved | *** ## ConfigurationError Invalid configuration. ```python theme={null} from fibonacci.exceptions import ConfigurationError try: workflow = Workflow( name="test", memory_config=MemoryConfig(backend="invalid") ) except ConfigurationError as e: print(f"Config error: {e.message}") print(f"Field: {e.field}") print(f"Value: {e.value}") ``` ### Attributes | Attribute | Type | Description | | --------- | ----- | ------------------- | | `field` | `str` | Configuration field | | `value` | `Any` | Invalid value | *** ## Comprehensive Error Handling ```python theme={null} from fibonacci import Workflow from fibonacci.exceptions import ( FibonacciError, NodeExecutionError, ToolError, ValidationError, AuthenticationError, RateLimitError, TimeoutError, ) import time import logging logger = logging.getLogger(__name__) def execute_with_handling(workflow, inputs, max_retries=3): """Execute workflow with comprehensive error handling.""" for attempt in range(max_retries): try: return workflow.execute(inputs=inputs) except ValidationError as e: # Input/output validation - don't retry, fix the data logger.error(f"Validation error: {e.errors}") raise except AuthenticationError as e: # Auth error - don't retry, fix credentials logger.error(f"Authentication failed: {e.reason}") raise except RateLimitError as e: # Rate limited - wait and retry logger.warning(f"Rate limited, waiting {e.retry_after}s") time.sleep(e.retry_after) continue except TimeoutError as e: # Timeout - may retry with longer timeout logger.warning(f"Timeout after {e.elapsed}s at node {e.node_id}") if attempt < max_retries - 1: continue raise except NodeExecutionError as e: # Node failed - log details, maybe use partial results logger.error(f"Node {e.node_id} failed: {e.message}") if e.partial_results: logger.info(f"Partial results available: {list(e.partial_results.keys())}") raise except ToolError as e: # Tool error - may retry for transient issues logger.error(f"Tool {e.tool_name} failed: {e.message}") if attempt < max_retries - 1: time.sleep(2 ** attempt) # Exponential backoff continue raise except FibonacciError as e: # Catch-all for other Fibonacci errors logger.error(f"Fibonacci error: {e.message}") raise raise RuntimeError(f"Failed after {max_retries} attempts") ``` # LLMNode Source: https://docs.fibonacci.today/sdk-reference/llm-node Node for calling Claude AI models with an instruction The `LLMNode` class sends an instruction to a Claude model and captures the response. It supports template variables so you can inject dynamic data from workflow inputs or previous nodes. ## Constructor ```python theme={null} from fibonacci import LLMNode node = LLMNode( id="analyzer", name="Analyze Text", instruction="Analyze the following text and summarize key points: {{input.text}}" ) ``` ### Parameters | Parameter | Type | Default | Description | | -------------- | ----------- | -------------------- | ------------------------------------------------------------------------- | | `id` | `str` | Required | Unique node identifier (lowercase letters, numbers, underscores, hyphens) | | `name` | `str` | Required | Human-readable node name | | `instruction` | `str` | Required | Instruction sent to Claude — supports `{{variable}}` template syntax | | `model` | `str` | `"claude-haiku-4-5"` | Claude model to use (see table below) | | `max_tokens` | `int` | `2000` | Maximum tokens in the response | | `temperature` | `float` | `1.0` | Sampling temperature — `0.0` (deterministic) to `1.0` (creative) | | `dependencies` | `list[str]` | `[]` | Node IDs this node waits for before running | | `enable_retry` | `bool` | `False` | Retry the node on failure | | `max_retries` | `int` | `3` | Maximum retry attempts (used when `enable_retry=True`) | | `retry_delay` | `float` | `1.0` | Initial delay in seconds between retries | ## Supported Models | Model | Identifier | Best For | | ----------------- | ------------------- | --------------------------------------------- | | Claude Haiku 4.5 | `claude-haiku-4-5` | Fast, cost-effective tasks — **SDK default** | | Claude Sonnet 4.6 | `claude-sonnet-4-6` | Balanced quality and speed — general purpose | | Claude Opus 4.6 | `claude-opus-4-6` | Most capable — complex reasoning and analysis | ## Template Variables Use `{{variable}}` syntax to inject dynamic values into the instruction: ```python theme={null} node = LLMNode( id="writer", name="Write Email", instruction=""" Write a {{input.tone}} email about {{input.topic}}. Previous analysis: {{analyzer}} """ ) ``` ### Available Variable Sources | Syntax | Where it comes from | | ------------------- | ---------------------------------------- | | `{{input.field}}` | Workflow input data | | `{{node_id}}` | Full output of another node | | `{{node_id.field}}` | Specific field from a node's JSON output | ## Selecting a Model ```python theme={null} # Default — fast and cheap quick = LLMNode( id="quick", name="Quick Summary", instruction="Summarize in one sentence: {{input.text}}" # model defaults to claude-haiku-4-5 ) # Balanced — good for most tasks balanced = LLMNode( id="balanced", name="Balanced Analysis", instruction="Analyze and explain: {{input.text}}", model="claude-sonnet-4-6" ) # Most capable — use for complex tasks complex_task = LLMNode( id="complex_task", name="Deep Reasoning", instruction="Perform a detailed multi-step analysis of: {{input.data}}", model="claude-opus-4-6" ) ``` ## Temperature Controls randomness. Range is `0.0` to `1.0`: ```python theme={null} # Deterministic — good for extraction, classification, structured output extract = LLMNode( id="extract", name="Extract Entities", instruction="Extract all names from: {{input.text}}", temperature=0.0 ) # Default (1.0) — good for general tasks general = LLMNode( id="general", name="General Response", instruction="Respond helpfully to: {{input.message}}" # temperature defaults to 1.0 ) # More creative — good for writing, brainstorming creative = LLMNode( id="creative", name="Creative Writing", instruction="Write a short story about: {{input.topic}}", temperature=0.9 ) ``` ## Max Tokens Limit how long the response can be: ```python theme={null} # Short response classify = LLMNode( id="classify", name="Classify", instruction="Classify as positive, negative, or neutral: {{input.text}}", max_tokens=10 # single word response ) # Long response report = LLMNode( id="report", name="Generate Report", instruction="Write a detailed report on: {{input.topic}}", max_tokens=2000 # default ) ``` ## Dependencies Control execution order by declaring which nodes must finish before this one runs: ```python theme={null} from fibonacci import Workflow, LLMNode wf = Workflow(name="pipeline") # First node — no dependencies analyzer = LLMNode( id="analyzer", name="Analyze", instruction="Analyze: {{input.text}}" ) # Second node — waits for analyzer to complete summarizer = LLMNode( id="summarizer", name="Summarize", instruction="Summarize this analysis: {{analyzer}}", dependencies=["analyzer"] ) # Third node — waits for both reporter = LLMNode( id="reporter", name="Create Report", instruction="""Create a report: Analysis: {{analyzer}} Summary: {{summarizer}}""", dependencies=["analyzer", "summarizer"] ) wf.add_nodes([analyzer, summarizer, reporter]) ``` You can also use the fluent `.depends_on()` method: ```python theme={null} reporter = LLMNode( id="reporter", name="Create Report", instruction="Report: {{analyzer}} / {{summarizer}}" ).depends_on("analyzer", "summarizer") ``` ## Retry Configuration Enable retries using `enable_retry=True` or the fluent `.with_retry()` method: ```python theme={null} # Via constructor node = LLMNode( id="analyzer", name="Analyze", instruction="Analyze: {{input.text}}", enable_retry=True, max_retries=5, retry_delay=2.0 ) # Via fluent method (equivalent) node = LLMNode( id="analyzer", name="Analyze", instruction="Analyze: {{input.text}}" ).with_retry(max_retries=5, delay=2.0) ``` ## Conditional Execution Run this node only when a condition is met using `.with_condition()`: ```python theme={null} # Only run expensive model when analysis level is detailed deep = LLMNode( id="deep", name="Deep Analysis", instruction="Perform deep analysis of: {{input.text}}", model="claude-opus-4-6" ).with_condition( left_value="{{input.analysis_level}}", operator="equals", right_value="detailed" ) ``` ## YAML Configuration ```yaml theme={null} nodes: - id: summarizer name: Summarize Text type: llm instruction: "Summarize the following text: {{input.text}}" config: model: claude-haiku-4-5 # optional, this is the default max_tokens: 500 temperature: 0.5 dependencies: - fetch_data ``` ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode wf = Workflow(name="blog-pipeline") # Outline outliner = LLMNode( id="outliner", name="Create Outline", instruction="Create a structured outline for a blog post about: {{input.topic}}", temperature=0.7 ) # Draft writer = LLMNode( id="writer", name="Write Draft", instruction=""" Write a blog post following this outline: {{outliner}} Style: {{input.style}} """, model="claude-sonnet-4-6", max_tokens=2000, dependencies=["outliner"] ) # Edit editor = LLMNode( id="editor", name="Edit Draft", instruction="Edit for clarity and engagement: {{writer}}", model="claude-opus-4-6", temperature=0.3, dependencies=["writer"] ) wf.add_nodes([outliner, writer, editor]) result = wf.run(input_data={ "topic": "AI in Healthcare", "style": "informative yet approachable" }) print(result.output_data["editor"]) ``` # Memory Source: https://docs.fibonacci.today/sdk-reference/memory Memory configuration and management for persistent workflow state The memory system enables workflows to persist and retrieve data across executions. ## MemoryConfig Configure the memory backend for a workflow. ```python theme={null} from fibonacci import Workflow, MemoryConfig workflow = Workflow( name="stateful-workflow", memory_config=MemoryConfig( backend="redis", connection_url="redis://localhost:6379/0" ) ) ``` ### Parameters | Parameter | Type | Default | Description | | ---------------- | ------ | -------------- | ------------------------------------- | | `backend` | `str` | `"memory"` | Storage backend | | `connection_url` | `str` | `None` | Connection string for remote backends | | `key_prefix` | `str` | `"fibonacci:"` | Prefix for all keys | | `default_ttl` | `int` | `None` | Default TTL in seconds | | `encryption_key` | `str` | `None` | Key for value encryption | | `encrypt_values` | `bool` | `False` | Enable encryption | ### Supported Backends | Backend | Connection URL Format | | ---------- | -------------------------------- | | `memory` | N/A (in-process) | | `redis` | `redis://host:port/db` | | `postgres` | `postgresql://user:pass@host/db` | | `dynamodb` | `dynamodb://region` | ## MemoryManager Access via `workflow.memory`: ```python theme={null} workflow = Workflow(name="my-workflow") memory = workflow.memory ``` ### Methods #### set Store a value in memory. ```python theme={null} memory.set( key: str, value: Any, scope: str = "workflow", user_id: str | None = None, ttl: int | None = None ) -> bool ``` **Parameters:** * `key`: Storage key * `value`: Value to store (must be JSON-serializable) * `scope`: Memory scope (workflow/user/organization/global) * `user_id`: User identifier (required for user scope) * `ttl`: Time-to-live in seconds **Example:** ```python theme={null} # Store workflow-scoped data memory.set("last_run", {"timestamp": "2025-01-23"}) # Store user-scoped data memory.set( "preferences", {"theme": "dark"}, scope="user", user_id="user-123" ) # Store with expiration memory.set("cache", data, ttl=3600) ``` *** #### get Retrieve a value from memory. ```python theme={null} memory.get( key: str, scope: str = "workflow", user_id: str | None = None, default: Any = None ) -> Any ``` **Parameters:** * `key`: Storage key * `scope`: Memory scope * `user_id`: User identifier * `default`: Default value if key not found **Example:** ```python theme={null} # Get with default prefs = memory.get("preferences", default={}) # Get user-scoped data history = memory.get( "chat_history", scope="user", user_id="user-123", default=[] ) ``` *** #### delete Delete a value from memory. ```python theme={null} memory.delete( key: str, scope: str = "workflow", user_id: str | None = None ) -> bool ``` *** #### exists Check if a key exists. ```python theme={null} memory.exists( key: str, scope: str = "workflow", user_id: str | None = None ) -> bool ``` *** #### clear Clear all memory in a scope. ```python theme={null} memory.clear( scope: str = "workflow", user_id: str | None = None ) -> bool ``` *** #### list\_keys List all keys in a scope. ```python theme={null} memory.list_keys( scope: str = "workflow", user_id: str | None = None, pattern: str | None = None ) -> list[str] ``` **Example:** ```python theme={null} # List all keys keys = memory.list_keys() # List with pattern cache_keys = memory.list_keys(pattern="cache:*") ``` ## Memory Scopes | Scope | Visibility | Use Case | | -------------- | --------------- | -------------------------- | | `workflow` | Single workflow | Execution state, temp data | | `user` | Single user | Preferences, history | | `organization` | All org users | Shared knowledge | | `global` | Everything | App-wide config | ### Scope Examples ```python theme={null} # Workflow scope - isolated per workflow memory.set("step_results", data, scope="workflow") # User scope - isolated per user across workflows memory.set("preferences", prefs, scope="user", user_id="u123") # Organization scope - shared across org memory.set("knowledge_base", kb, scope="organization") # Global scope - shared everywhere memory.set("app_config", config, scope="global") ``` ## Memory in Nodes ### Reading Memory in Prompts ```python theme={null} node = LLMNode( id="chat", model="claude-sonnet-4-5-20250929", prompt="""Previous context: {{memory.history}} User: {{input.message}}""", memory_read=["history"] # Specify keys to read ) ``` ### Writing Memory from Nodes ```python theme={null} node = LLMNode( id="summarizer", model="claude-sonnet-4-5-20250929", prompt="Summarize: {{input.text}}", memory_write={ "key": "last_summary", "scope": "user" } ) ``` ## Encryption Enable encryption for sensitive data: ```python theme={null} from fibonacci import Workflow, MemoryConfig workflow = Workflow( name="secure-workflow", memory_config=MemoryConfig( backend="redis", connection_url="redis://localhost:6379", encrypt_values=True, encryption_key="your-32-byte-key-here..." ) ) # Data is encrypted at rest workflow.memory.set("sensitive", {"ssn": "123-45-6789"}) ``` ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode, MemoryConfig # Configure with Redis backend workflow = Workflow( name="chatbot", memory_config=MemoryConfig( backend="redis", connection_url="redis://localhost:6379/0", key_prefix="chatbot:" ) ) # Chat node reads history chat = LLMNode( id="chat", model="claude-sonnet-4-5-20250929", prompt="""Conversation history: {{memory.history}} User: {{input.message}} Respond naturally.""", memory_read=["history"] ) workflow.add_node(chat) def chat_with_user(user_id: str, message: str): # Get history history = workflow.memory.get( "history", scope="user", user_id=user_id, default=[] ) # Execute result = workflow.execute( inputs={"message": message}, user_id=user_id ) # Update history history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": result["chat"]}) history = history[-20:] # Keep last 20 workflow.memory.set( "history", history, scope="user", user_id=user_id ) return result["chat"] ``` # SDK Reference Source: https://docs.fibonacci.today/sdk-reference/overview Complete API reference for the Fibonacci Python SDK The Fibonacci SDK provides a comprehensive Python API for building, executing, and managing AI workflows. This reference documents all public classes, methods, and configuration options. ## Installation ```bash theme={null} pip install fibonacci-sdk ``` ## Quick Links Core workflow class for building and executing pipelines Client for interacting with Fibonacci Cloud Node for AI model interactions Node for external tool integrations Node for quality evaluation and iteration Node for conditional branching logic ## Core Classes | Class | Description | | ---------------------------------------------------- | ----------------------------------------------- | | [`Workflow`](/sdk-reference/workflow) | Main class for defining and executing workflows | | [`FibonacciClient`](/sdk-reference/client) | Client for Fibonacci Cloud API | | [`LLMNode`](/sdk-reference/llm-node) | Node for LLM model calls | | [`ToolNode`](/sdk-reference/tool-node) | Node for tool integrations | | [`CriticNode`](/sdk-reference/critic-node) | Node for output evaluation | | [`ConditionalNode`](/sdk-reference/conditional-node) | Node for branching logic | ## Configuration Classes | Class | Description | | ---------------------------------------------------- | ---------------------------- | | [`MemoryConfig`](/sdk-reference/memory) | Memory backend configuration | | [`RetryConfig`](/sdk-reference/config#retryconfig) | Retry behavior settings | | [`SecureConfig`](/sdk-reference/secure-config) | Security configuration | | [`KeychainStorage`](/sdk-reference/keychain-storage) | Secure credential storage | ## Exceptions | Exception | Description | | -------------------------------------------------------------------- | ------------------------------ | | [`FibonacciError`](/sdk-reference/exceptions#fibonaccierror) | Base exception class | | [`WorkflowError`](/sdk-reference/exceptions#workflowerror) | Workflow-level errors | | [`NodeExecutionError`](/sdk-reference/exceptions#nodeexecutionerror) | Node execution failures | | [`ValidationError`](/sdk-reference/exceptions#validationerror) | Input/output validation errors | | [`TimeoutError`](/sdk-reference/exceptions#timeouterror) | Execution timeout errors | ## Basic Example ```python theme={null} from fibonacci import Workflow, LLMNode # Create workflow workflow = Workflow(name="example") # Add node analyzer = LLMNode( id="analyzer", model="claude-sonnet-4-5-20250929", prompt="Analyze: {{input.text}}" ) workflow.add_node(analyzer) # Execute result = workflow.execute(inputs={"text": "Hello world"}) print(result["analyzer"]) ``` ## Async Support All workflow operations support async/await: ```python theme={null} import asyncio from fibonacci import Workflow async def main(): workflow = Workflow.from_yaml("workflow.yaml") result = await workflow.execute_async(inputs={"text": "Hello"}) print(result) asyncio.run(main()) ``` ## Type Hints The SDK is fully typed for IDE support: ```python theme={null} from fibonacci import Workflow, LLMNode from fibonacci.types import WorkflowResult, NodeOutput workflow: Workflow = Workflow(name="typed") result: WorkflowResult = workflow.execute(inputs={}) ``` ## Versioning The SDK follows [semantic versioning](https://semver.org/): * **Major**: Breaking changes to public API * **Minor**: New features, backwards compatible * **Patch**: Bug fixes, backwards compatible Check your installed version: ```python theme={null} import fibonacci print(fibonacci.__version__) ``` # ToolNode Source: https://docs.fibonacci.today/sdk-reference/tool-node Node for executing external tool and API integrations The `ToolNode` class runs a Fibonacci platform tool — connecting workflows to services like Google Sheets, Slack, databases, and HTTP APIs. ## Constructor ```python theme={null} from fibonacci import ToolNode node = ToolNode( id="fetch_data", name="Fetch Sheet Data", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "A1:D100" } ) ``` ### Parameters | Parameter | Type | Default | Description | | -------------- | ----------- | -------- | ------------------------------------------------------------------------- | | `id` | `str` | Required | Unique node identifier (lowercase letters, numbers, underscores, hyphens) | | `name` | `str` | Required | Human-readable node name | | `tool` | `str` | Required | Tool name (e.g., `"google_sheets_read"`, `"slack_send_message"`) | | `params` | `dict` | Required | Input parameters for the tool — supports `{{variable}}` template syntax | | `timeout` | `int` | `30` | Execution timeout in seconds | | `dependencies` | `list[str]` | `[]` | Node IDs this node waits for before running | | `enable_retry` | `bool` | `False` | Retry on failure | | `max_retries` | `int` | `3` | Maximum retry attempts (used when `enable_retry=True`) | | `retry_delay` | `float` | `1.0` | Initial delay in seconds between retries | Tool names use **underscore format** (e.g., `google_sheets_read`, `slack_send_message`). Use `list_tools()` or `print_tool_info()` to see available tools and their exact names. ## Template Variables in Params Any value in `params` can reference dynamic data: ```python theme={null} node = ToolNode( id="notify", name="Send Notification", tool="slack_send_message", params={ "channel": "#{{input.team}}", "message": "Analysis complete: {{analyze_data}}" } ) ``` | Syntax | Source | | ------------------- | ---------------------------------------- | | `{{input.field}}` | Workflow input data | | `{{node_id}}` | Full output of another node | | `{{node_id.field}}` | Specific field from a node's JSON output | ## Tool Categories ### Google Workspace ```python theme={null} # Read from Google Sheets read_sheet = ToolNode( id="read_sheet", name="Read Sheet", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Sheet1!A1:D100" } ) # Write to Google Sheets write_sheet = ToolNode( id="write_sheet", name="Write Results", tool="google_sheets_write", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Results!A1", "values": [["Name", "Score"], ["Alice", "95"]] } ) # Append rows append_row = ToolNode( id="append_row", name="Append Log Row", tool="google_sheets_append", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Log!A:D", "values": [["{{input.id}}", "{{analyze_data}}"]] } ) # Read a Google Doc read_doc = ToolNode( id="read_doc", name="Read Document", tool="google_docs_read", params={ "document_id": "{{input.doc_id}}" } ) # Create a Google Doc create_doc = ToolNode( id="create_doc", name="Create Document", tool="google_docs_create", params={ "title": "{{input.title}}", "content": "{{generate_report}}" } ) # Send email via Gmail send_email = ToolNode( id="send_email", name="Send Email", tool="gmail_send", params={ "to": "recipient@example.com", "subject": "Your Report is Ready", "body": "{{executive_summary}}" } ) # Create a Google Calendar event create_event = ToolNode( id="create_event", name="Schedule Meeting", tool="google_calendar_create_event", params={ "summary": "Follow-up Meeting", "start": "2025-06-01T10:00:00", "end": "2025-06-01T11:00:00", "attendees": ["team@company.com"] } ) ``` ### Communication ```python theme={null} # Slack — send a message slack_msg = ToolNode( id="notify_team", name="Notify Team", tool="slack_send_message", params={ "channel": "#team-updates", "message": "Report ready: {{executive_summary}}" } ) # Slack — DM a user slack_dm = ToolNode( id="dm_user", name="DM User", tool="slack_send_message", params={ "channel": "@john.doe", "message": "Your task is complete." } ) ``` ### HTTP & APIs ```python theme={null} # Generic HTTP request api_call = ToolNode( id="call_api", name="Call External API", tool="http_request", params={ "method": "POST", "url": "https://api.example.com/data", "headers": { "Authorization": "Bearer {{input.api_token}}", "Content-Type": "application/json" }, "body": { "query": "{{input.search_term}}" } } ) # Webhook webhook = ToolNode( id="trigger_webhook", name="Trigger Webhook", tool="webhook_call", params={ "url": "https://hooks.zapier.com/hooks/catch/123/abc", "payload": { "event": "report_generated", "data": "{{report_data}}" } } ) ``` ### Databases ```python theme={null} query = ToolNode( id="query_db", name="Query Database", tool="database_query", params={ "connection_string": "{{secrets.DATABASE_URL}}", "query": """ SELECT customer_id, SUM(amount) AS total FROM orders WHERE date >= '{{input.start_date}}' GROUP BY customer_id ORDER BY total DESC LIMIT 10 """ } ) ``` Never hardcode database credentials. Use the `secrets` parameter in `wf.run()` or the Fibonacci secrets manager. ### Cloud Storage ```python theme={null} # Upload to S3 s3_upload = ToolNode( id="upload_s3", name="Upload to S3", tool="s3_upload", params={ "bucket": "my-reports", "key": "reports/{{input.report_id}}.pdf", "content": "{{report_pdf}}", "content_type": "application/pdf" } ) # Download from S3 s3_download = ToolNode( id="download_s3", name="Download from S3", tool="s3_download", params={ "bucket": "my-data", "key": "data/{{input.file_name}}" } ) ``` ### CRM & Business ```python theme={null} # Create a Salesforce lead create_lead = ToolNode( id="create_lead", name="Create Salesforce Lead", tool="salesforce_create", params={ "object": "Lead", "data": { "FirstName": "{{input.first_name}}", "LastName": "{{input.last_name}}", "Email": "{{input.email}}", "Company": "{{input.company}}" } } ) # Update a HubSpot contact hubspot_contact = ToolNode( id="update_contact", name="Update HubSpot Contact", tool="hubspot_update_contact", params={ "email": "{{input.email}}", "properties": { "lead_score": "{{score_lead}}" } } ) # Create a Notion page notion_page = ToolNode( id="create_notion", name="Create Notion Page", tool="notion_create_page", params={ "database_id": "{{input.database_id}}", "properties": { "Name": "{{input.title}}", "Status": "Ready" }, "content": "{{generate_notes}}" } ) ``` ## Tool Discovery ### List Available Tools ```python theme={null} from fibonacci import list_tools, list_tool_categories # All tools tools = list_tools() for tool in tools: print(f"{tool['name']}: {tool['description']}") # By category categories = list_tool_categories() for cat in categories: print(f"{cat['name']}: {cat['count']} tools") ``` ### Search Tools ```python theme={null} from fibonacci import search_tools results = search_tools("slack") for tool in results: print(f"{tool['name']}: {tool['description']}") ``` ### Inspect a Tool ```python theme={null} from fibonacci import print_tool_info print_tool_info("google_sheets_read") ``` Output: ``` ====================================================================== 🔧 google_sheets_read ====================================================================== Category: google_workspace Description: Read data from a Google Sheets spreadsheet 📥 Input Parameters: • spreadsheet_id (string) - required • range (string) - required A1 notation range (e.g., "Sheet1!A1:D10") 📖 Examples: 1. Read sales data Params: {"spreadsheet_id": "abc123", "range": "Sales!A1:E100"} ====================================================================== ``` ### Find the Right Tool ```python theme={null} from fibonacci import find_tool_for_task tool = find_tool_for_task("send a message to slack") print(tool) # "slack_send_message" tool = find_tool_for_task("read data from google sheets") print(tool) # "google_sheets_read" ``` ## Dependencies ```python theme={null} # Read data read = ToolNode( id="read_source", name="Read Source Sheet", tool="google_sheets_read", params={"spreadsheet_id": "...", "range": "A1:Z100"} ) # Write results (waits for read to finish) write = ToolNode( id="write_dest", name="Write Destination Sheet", tool="google_sheets_write", params={ "spreadsheet_id": "...", "range": "A1", "values": "{{read_source}}" }, dependencies=["read_source"] ) ``` ## Timeout Configuration ```python theme={null} slow_api = ToolNode( id="slow_api", name="Long-Running API", tool="http_request", params={"url": "https://api.example.com/slow-endpoint"}, timeout=120 # override the 30s default ) ``` ## Retry Configuration ```python theme={null} # Via constructor api_call = ToolNode( id="api_call", name="Reliable API Call", tool="http_request", params={"url": "https://api.example.com"}, enable_retry=True, max_retries=5, retry_delay=2.0 ) # Via fluent method api_call = ToolNode( id="api_call", name="Reliable API Call", tool="http_request", params={"url": "https://api.example.com"} ).with_retry(max_retries=5, delay=2.0) ``` ## Conditional Execution ```python theme={null} notify_urgent = ToolNode( id="notify_urgent", name="Notify Urgent", tool="slack_send_message", params={ "channel": "#urgent", "message": "Urgent issue: {{analyzer}}" } ).with_condition( left_value="{{analyzer_urgency}}", operator="equals", right_value="high" ) ``` ## YAML Configuration ```yaml theme={null} nodes: - id: notify name: Notify Team type: tool tool_name: slack_send_message tool_params: channel: "#analytics" message: "Analysis complete: {{analyze}}" dependencies: - analyze ``` ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode wf = Workflow(name="data-pipeline") # Fetch data fetch = ToolNode( id="fetch_data", name="Read Sheet", tool="google_sheets_read", params={ "spreadsheet_id": "{{input.sheet_id}}", "range": "Data!A1:Z1000" } ) # Analyze with LLM analyze = LLMNode( id="analyze", name="Analyze Data", instruction="Analyze this data and find key insights: {{fetch_data}}", model="claude-sonnet-4-6", dependencies=["fetch_data"] ) # Notify via Slack notify = ToolNode( id="notify", name="Send Report", tool="slack_send_message", params={ "channel": "#analytics", "message": "Analysis complete for {{input.sheet_id}}:\n{{analyze}}" }, dependencies=["analyze"] ) wf.add_nodes([fetch, analyze, notify]) result = wf.run(input_data={"sheet_id": "abc123"}) print(result.output_data) ``` # Workflow Source: https://docs.fibonacci.today/sdk-reference/workflow Core class for building, deploying, and running AI workflows The `Workflow` class is the foundation of Fibonacci. It orchestrates node execution, manages dependencies, and handles the full workflow lifecycle from building to deployment and execution. ## Constructor ```python theme={null} from fibonacci import Workflow wf = Workflow( name="customer-support-bot", description="Routes and responds to customer support tickets", version=1 ) ``` ### Parameters | Parameter | Type | Default | Description | | ------------- | -------- | -------- | --------------------------------------------------------- | | `name` | `str` | Required | Workflow name | | `description` | `str` | `""` | Human-readable description | | `version` | `int` | `1` | Workflow version number | | `is_active` | `bool` | `True` | Whether the deployed workflow is active | | `is_template` | `bool` | `False` | Save as a reusable template | | `config` | `Config` | `None` | SDK configuration object (uses default config if omitted) | ## Class Methods ### from\_yaml Load a workflow from a YAML file. ```python theme={null} wf = Workflow.from_yaml(path: str) -> Workflow ``` **Parameters:** * `path` (`str`): Path to the YAML workflow file **Returns:** `Workflow` instance **Example:** ```python theme={null} wf = Workflow.from_yaml("workflows/analyzer.yaml") ``` ## Instance Methods ### add\_node Add a single node to the workflow. ```python theme={null} wf.add_node(node: Node) -> Workflow ``` Returns `self` for chaining: ```python theme={null} wf.add_node(node_a).add_node(node_b).add_node(node_c) ``` *** ### add\_nodes Add multiple nodes at once. ```python theme={null} wf.add_nodes(nodes: list[Node]) -> Workflow ``` **Example:** ```python theme={null} wf.add_nodes([classifier, router, handler_a, handler_b]) ``` *** ### get\_node Look up a node by ID. ```python theme={null} wf.get_node(node_id: str) -> Node | None ``` *** ### remove\_node Remove a node from the workflow. ```python theme={null} wf.remove_node(node_id: str) -> Workflow ``` *** ### validate Validate the workflow DAG (checks for cycles, missing dependencies, etc.). ```python theme={null} wf.validate() -> bool ``` **Returns:** `True` if valid, raises `ValidationError` otherwise. ```python theme={null} if wf.validate(): wf.deploy() ``` *** ### deploy Deploy the workflow to the Fibonacci platform. ```python theme={null} wf.deploy( api_key: str | None = None, validate: bool = True ) -> str ``` **Parameters:** * `api_key` (`str`): API key to use for deployment. Falls back to the configured key if omitted. * `validate` (`bool`): Validate the workflow before deploying (default `True`). **Returns:** `workflow_id` string — save this to run the workflow later. **Raises:** `AuthenticationError`, `ValidationError`, `DeploymentError` **Example:** ```python theme={null} workflow_id = wf.deploy(api_key="fib_live_abc123") print(f"Deployed: {workflow_id}") ``` *** ### deploy\_async Async version of `deploy()`. ```python theme={null} workflow_id = await wf.deploy_async(api_key="fib_live_abc123") ``` *** ### run Execute the workflow synchronously. ```python theme={null} wf.run( input_data: dict, workflow_id: str | None = None, wait: bool = True, timeout: float = 300.0, secrets: dict | None = None ) -> WorkflowRunStatus ``` **Parameters:** * `input_data` (`dict`): Input data for the workflow. * `workflow_id` (`str`): Workflow ID to run. Uses the last deployed workflow if omitted. * `wait` (`bool`): Wait for the run to complete before returning (default `True`). * `timeout` (`float`): Maximum seconds to wait for completion (default `300.0`). * `secrets` (`dict`): Runtime secrets injected into the workflow (e.g. `{"SLACK_TOKEN": "xoxb-..."}`). **Returns:** `WorkflowRunStatus` with fields: * `.status` — `"completed"`, `"failed"`, `"running"`, etc. * `.output_data` — dict of node outputs keyed by node ID * `.error_message` — error details if status is `"failed"` * `.total_cost` — total cost of the run in USD * `.nodes_executed` — number of nodes that ran **Raises:** `AuthenticationError`, `ExecutionError`, `RateLimitError` **Example:** ```python theme={null} result = wf.run( input_data={"text": "Hello world"}, secrets={"GOOGLE_TOKEN": "my-token"} ) print(result.output_data["analyzer"]) ``` *** ### run\_async Async version of `run()`. ```python theme={null} import asyncio async def main(): result = await wf.run_async( input_data={"text": "Hello world"} ) print(result.output_data) asyncio.run(main()) ``` *** ### get\_status Get the status of a specific run. ```python theme={null} wf.get_status(run_id: str | None = None) -> WorkflowRunStatus ``` *** ### update Update workflow metadata on the platform. ```python theme={null} wf.update( name: str | None = None, description: str | None = None, is_active: bool | None = None ) -> WorkflowResponse ``` *** ### delete Delete the deployed workflow. ```python theme={null} wf.delete() -> bool ``` *** ### activate / deactivate Toggle a deployed workflow on or off without deleting it. ```python theme={null} wf.activate() # -> WorkflowResponse wf.deactivate() # -> WorkflowResponse ``` *** ### get\_stats Get aggregated statistics for the deployed workflow. ```python theme={null} wf.get_stats() -> WorkflowStats ``` **Returns:** `WorkflowStats` with `.total_runs`, `.success_rate`, `.avg_duration_seconds`, `.avg_cost`. *** ### to\_yaml Export the workflow to a YAML string (and optionally write to a file). ```python theme={null} yaml_str = wf.to_yaml() # return string wf.to_yaml("exported.yaml") # write to file ``` *** ### to\_yaml\_string Export the workflow to a YAML string (always returns a string, never writes). ```python theme={null} yaml_str = wf.to_yaml_string() ``` *** ### to\_dict Export the workflow as a Python dictionary. ```python theme={null} data = wf.to_dict() ``` ## Properties ### workflow\_id The ID assigned after deployment. `None` if the workflow has not been deployed yet. ```python theme={null} print(wf.workflow_id) # e.g. "wf_abc123" ``` *** ### nodes List of nodes currently in the workflow. ```python theme={null} for node in wf.nodes: print(node.id) ``` *** ### node\_count Number of nodes in the workflow. ```python theme={null} print(wf.node_count) # e.g. 4 ``` ## YAML Workflows Load a workflow from a YAML definition file: ```python theme={null} wf = Workflow.from_yaml("workflow.yaml") wf.deploy() result = wf.run(input_data={"text": "Hello!"}) ``` Export an existing workflow to YAML: ```python theme={null} wf.to_yaml("exported.yaml") ``` See the [YAML Workflows guide](/guides/yaml-workflows) for the full YAML schema. ## Complete Example ```python theme={null} from fibonacci import Workflow, LLMNode, ToolNode, ConditionalNode, Config # Optional: explicit config config = Config(api_key="fib_live_abc123") wf = Workflow( name="feedback-pipeline", description="Classifies and routes customer feedback", version=1, config=config ) # Classify classify = LLMNode( id="classify", name="Classify Feedback", instruction="Classify as positive, negative, or neutral. One word only.\n\n{{input.feedback}}" ) # Route router = ConditionalNode( id="router", name="Route by Sentiment", left_value="{{classify}}", operator="contains", right_value="negative", true_branch=["escalate"], false_branch=["log_ok"], dependencies=["classify"] ) # Escalate negative escalate = ToolNode( id="escalate", name="Alert Support Team", tool="slack_send_message", params={ "channel": "#support", "message": "Negative feedback: {{input.feedback}}" }, dependencies=["router"] ) # Log positive/neutral log_ok = ToolNode( id="log_ok", name="Log Feedback", tool="google_sheets_append", params={ "spreadsheet_id": "{{input.log_sheet}}", "range": "Log!A:C", "values": [["{{input.feedback}}", "{{classify}}"]] }, dependencies=["router"] ) wf.add_nodes([classify, router, escalate, log_ok]) # Validate before deploying wf.validate() # Deploy workflow_id = wf.deploy() print(f"Deployed: {workflow_id}") # Run result = wf.run( input_data={"feedback": "Great product!", "log_sheet": "abc123"} ) print(result.output_data) print(f"Cost: ${result.total_cost:.4f}") # Lifecycle management stats = wf.get_stats() print(f"Total runs: {stats.total_runs}, success rate: {stats.success_rate:.0%}") # Deactivate when not needed wf.deactivate() ```