Python SDK

The official Python SDK for integrating Bleu AI workflows into Python applications.

Installation

pip install bleuai

Quick Start

from bleuai import BleuAI
import os

# Initialize client with your API key
client = BleuAI(api_key=os.environ.get("BLEU_API_KEY"))

# Run a workflow
result = await client.run_workflow('workflow-id', {
    'Input Name': 'value'  # Your workflow inputs
})

print(result.outputs)

# Clean up
await client.close()

Authentication

Getting Your API Key

1. Visit buildbleu.com/api-keys
2. Click "Create Key" and name it
3. Copy immediately - it won't be shown again

Using Environment Variables

# .env file
BLEU_API_KEY=bleu_xxxxxxxxxxxxxxxxxxxxx

# In your code
from dotenv import load_dotenv
import os

load_dotenv()
client = BleuAI(api_key=os.environ.get("BLEU_API_KEY"))

Security Warning

Never commit API keys to version control! If a key is leaked, revoke it immediately at buildbleu.com/api-keys

Synchronous Usage

For non-async environments:

import asyncio

def run_workflow_sync():
    async def run():
        client = BleuAI(api_key="BLEU_API_KEY")
        result = await client.run_workflow('workflow-id', inputs)
        await client.close()
        return result
    
    return asyncio.run(run())

result = run_workflow_sync()

Error Handling

Status	Error Type	Description
401	Authentication Error	Invalid or missing API key
404	Not Found	Workflow ID doesn't exist
429	Rate Limit	Too many requests
500	Server Error	Internal server error

Example error handling:

try:
    result = await client.run_workflow('workflow-id', inputs)
except Exception as e:
    if hasattr(e, 'status'):
        print(f"Error {e.status}: {str(e)}")

API Reference

Class: BleuAI

Main client for interacting with the Bleu AI API.

Constructor

BleuAI(api_key: str, headers: Optional[Dict[str, str]] = None)

Creates a new Bleu AI client instance.

Parameters:

• api_key (str, required): Your Bleu AI API key
• headers (Dict[str, str], optional): Additional HTTP headers to include in all requests

Returns:

• BleuAI client instance

Example:

client = BleuAI(api_key='bleu_xxxxxxxxxxxxxxxxxxxxx')

# With custom headers
client = BleuAI(
    api_key='bleu_xxxxxxxxxxxxxxxxxxxxx',
    headers={'X-Custom-Header': 'value'}
)

Methods

async run_workflow(workflow_id: str, inputs: Optional[Dict[str, Any]] = None) -> WorkflowResult

Executes a workflow with provided inputs.

Parameters:

• workflow_id (str, required): The UUID of the workflow to execute
• inputs (Dict[str, Any], optional): Input parameters required by the workflow. Keys must match the workflow's input field names.

Returns:

• WorkflowResult: Object containing execution results

Result Object Structure:

WorkflowResult:
    outputs: Any          # Workflow outputs (structure depends on workflow)
    status: str           # Execution status ('success', 'error')
    execution_time: int   # Execution time in milliseconds

Example:

result = await client.run_workflow('abc-123', {
    'Text Input': 'Hello world',
    'Count': 5
})
print(result.outputs)

Raises:

• Exception with status attribute for API errors
• Network-related exceptions for connectivity issues

async close() -> None

Closes the client and releases resources.

Parameters: None

Returns: None

Example:

await client.close()

Note: Always call this method when done with the client to properly clean up resources.

Type Hints

The SDK provides comprehensive type hints for IDE support:

from typing import Dict, Any, Optional
from bleuai import BleuAI, WorkflowResult

async def typed_example(
    workflow_id: str,
    inputs: Optional[Dict[str, Any]] = None
) -> WorkflowResult:
    client: BleuAI = BleuAI(api_key=os.environ.get("BLEU_API_KEY"))
    result: WorkflowResult = await client.run_workflow(workflow_id, inputs or {})
    await client.close()
    return result

Best Practices

• Store API keys in environment variables
• Always call close() when done with the client
• Use asyncio.gather() for concurrent workflows
• Implement retry logic for transient errors
• Validate inputs before API calls
• Use try-except blocks for all API operations