Skip to main content

Installation

Install the SDK using pip:
pip install bundleup-sdk
Using Poetry:
poetry add bundleup-sdk
Using pipenv:
pipenv install bundleup-sdk

Requirements

  • Python: 3.8 or higher
  • requests: >=2.25.0 (automatically installed)
  • typing-extensions: >=4.0.0 (automatically installed)

Python Compatibility

The BundleUp SDK is tested and supported on:
  • Python 3.8
  • Python 3.9
  • Python 3.10
  • Python 3.11
  • Python 3.12

Features

  • 🚀 Pythonic API - Follows Python best practices and PEP 8
  • 📦 Easy Integration - Simple, intuitive API design
  • Async Support - Built on requests with async capabilities
  • 🔌 100+ Integrations - Connect to Slack, GitHub, Jira, Linear, and many more
  • 🎯 Unified API - Consistent interface across all integrations via Unify API
  • 🔑 Proxy API - Direct access to underlying integration APIs
  • 🪶 Lightweight - Minimal dependencies
  • 🛡️ Error Handling - Comprehensive error messages and validation
  • 📚 Well Documented - Extensive documentation and examples
  • 🔍 Type Hints - Full type annotations for better IDE support
  • 🧪 Tested - Comprehensive test suite with pytest

Quick Start

Get started with BundleUp in just a few lines of code:
from bundleup import BundleUp
import os

# Initialize the client
client = BundleUp(os.environ['BUNDLEUP_API_KEY'])

# List all active connections
connections = client.connection.list()
print(f"You have {len(connections)} active connections")

# Use the Proxy API to make requests to integrated services
proxy = client.proxy('conn_123')
response = proxy.get('/api/users')
users = response.json()
print(f"Users: {users}")

# Use the Unify API for standardized data across integrations
unify = client.unify('conn_456')
channels = unify.chat.channels({'limit': 10})
print(f"Chat channels: {channels['data']}")

Authentication

The BundleUp SDK uses API keys for authentication. You can obtain your API key from the BundleUp Dashboard.

Getting Your API Key

  1. Sign in to your BundleUp Dashboard
  2. Navigate to API Keys
  3. Click Create API Key
  4. Copy your API key and store it securely

Initializing the SDK

from bundleup import BundleUp

# Initialize with API key
client = BundleUp('your_api_key_here')

# Or use environment variable (recommended)
import os
client = BundleUp(os.environ['BUNDLEUP_API_KEY'])

# Or use python-dotenv
from dotenv import load_dotenv
load_dotenv()
client = BundleUp(os.getenv('BUNDLEUP_API_KEY'))

Security Best Practices

  • DO store API keys in environment variables
  • DO use a secrets management service in production
  • DO rotate API keys regularly
  • DON’T commit API keys to version control
  • DON’T hardcode API keys in your source code
  • DON’T share API keys in public channels
Example .env file:
BUNDLEUP_API_KEY=bu_live_1234567890abcdefghijklmnopqrstuvwxyz
Loading environment variables: Install python-dotenv:
pip install python-dotenv
Then in your application:
from dotenv import load_dotenv
import os

load_dotenv()

from bundleup import BundleUp
client = BundleUp(os.getenv('BUNDLEUP_API_KEY'))
For Django applications:
# settings.py
import os
from pathlib import Path

BUNDLEUP_API_KEY = os.environ.get('BUNDLEUP_API_KEY')

# views.py or services
from django.conf import settings
from bundleup import BundleUp

client = BundleUp(settings.BUNDLEUP_API_KEY)
For Flask applications:
# config.py
import os

class Config:
    BUNDLEUP_API_KEY = os.environ.get('BUNDLEUP_API_KEY')

# app.py
from flask import Flask
from bundleup import BundleUp

app = Flask(__name__)
app.config.from_object('config.Config')
client = BundleUp(app.config['BUNDLEUP_API_KEY'])

Core Concepts

Platform API

The Platform API provides access to core BundleUp features like managing connections and integrations. Use this API to list, retrieve, and delete connections, as well as discover available integrations.

Proxy API

The Proxy API allows you to make direct HTTP requests to the underlying integration’s API through BundleUp. This is useful when you need access to integration-specific features not covered by the Unify API.

Unify API

The Unify API provides a standardized, normalized interface across different integrations. For example, you can fetch chat channels from Slack, Discord, or Microsoft Teams using the same API call.

API Reference

Connections

Manage your integration connections.

List Connections

Retrieve a list of all connections in your account.
connections = client.connection.list()
With query parameters:
connections = client.connection.list({
    'integration_id': 'int_slack',
    'limit': 50,
    'offset': 0,
    'external_id': 'user_123'
})
Query Parameters:
  • integration_id (str): Filter by integration ID
  • integration_identifier (str): Filter by integration identifier (e.g., ‘slack’, ‘github’)
  • external_id (str): Filter by external user/account ID
  • limit (int): Maximum number of results (default: 50, max: 100)
  • offset (int): Number of results to skip for pagination
Response:
[
    {
        'id': 'conn_123abc',
        'external_id': 'user_456',
        'integration_id': 'int_slack',
        'is_valid': True,
        'created_at': '2024-01-15T10:30:00Z',
        'updated_at': '2024-01-20T14:22:00Z',
        'refreshed_at': '2024-01-20T14:22:00Z',
        'expires_at': '2024-04-20T14:22:00Z'
    },
    # ... more connections
]

Retrieve a Connection

Get details of a specific connection by ID.
connection = client.connection.retrieve('conn_123abc')
Response:
{
    'id': 'conn_123abc',
    'external_id': 'user_456',
    'integration_id': 'int_slack',
    'is_valid': True,
    'created_at': '2024-01-15T10:30:00Z',
    'updated_at': '2024-01-20T14:22:00Z',
    'refreshed_at': '2024-01-20T14:22:00Z',
    'expires_at': '2024-04-20T14:22:00Z'
}

Delete a Connection

Remove a connection from your account.
client.connection.delete('conn_123abc')
Note: Deleting a connection will revoke access to the integration and cannot be undone.

Integrations

Discover and work with available integrations.

List Integrations

Get a list of all available integrations.
integrations = client.integration.list()
With query parameters:
integrations = client.integration.list({
    'status': 'active',
    'limit': 100,
    'offset': 0
})
Query Parameters:
  • status (str): Filter by status (‘active’, ‘inactive’, ‘beta’)
  • limit (int): Maximum number of results
  • offset (int): Number of results to skip for pagination
Response:
[
    {
        'id': 'int_slack',
        'identifier': 'slack',
        'name': 'Slack',
        'category': 'chat',
        'created_at': '2023-01-01T00:00:00Z',
        'updated_at': '2024-01-15T10:00:00Z'
    },
    # ... more integrations
]

Retrieve an Integration

Get details of a specific integration.
integration = client.integration.retrieve('int_slack')
Response:
{
    'id': 'int_slack',
    'identifier': 'slack',
    'name': 'Slack',
    'category': 'chat',
    'created_at': '2023-01-01T00:00:00Z',
    'updated_at': '2024-01-15T10:00:00Z'
}

Webhooks

Manage webhook subscriptions for real-time event notifications.

List Webhooks

Get all registered webhooks.
webhooks = client.webhook.list()
With pagination:
webhooks = client.webhook.list({
    'limit': 50,
    'offset': 0
})
Response:
[
    {
        'id': 'webhook_123',
        'name': 'My Webhook',
        'url': 'https://example.com/webhook',
        'events': {
            'connection.created': True,
            'connection.deleted': True
        },
        'created_at': '2024-01-15T10:30:00Z',
        'updated_at': '2024-01-20T14:22:00Z',
        'last_triggered_at': '2024-01-20T14:22:00Z'
    }
]

Create a Webhook

Register a new webhook endpoint.
webhook = client.webhook.create({
    'name': 'Connection Events Webhook',
    'url': 'https://example.com/webhook',
    'events': {
        'connection.created': True,
        'connection.deleted': True,
        'connection.updated': True
    }
})
Webhook Events:
  • connection.created - Triggered when a new connection is established
  • connection.deleted - Triggered when a connection is removed
  • connection.updated - Triggered when a connection is modified
Request Body:
  • name (str): Friendly name for the webhook
  • url (str): Your webhook endpoint URL
  • events (dict): Events to subscribe to
Response:
{
    'id': 'webhook_123',
    'name': 'Connection Events Webhook',
    'url': 'https://example.com/webhook',
    'events': {
        'connection.created': True,
        'connection.deleted': True,
        'connection.updated': True
    },
    'created_at': '2024-01-15T10:30:00Z',
    'updated_at': '2024-01-15T10:30:00Z'
}

Retrieve a Webhook

Get details of a specific webhook.
webhook = client.webhook.retrieve('webhook_123')

Update a Webhook

Modify an existing webhook.
updated = client.webhook.update('webhook_123', {
    'name': 'Updated Webhook Name',
    'url': 'https://example.com/new-webhook',
    'events': {
        'connection.created': True,
        'connection.deleted': False
    }
})

Delete a Webhook

Remove a webhook subscription.
client.webhook.delete('webhook_123')

Webhook Payload Example

When an event occurs, BundleUp sends a POST request to your webhook URL with the following payload:
{
  "id": "evt_1234567890",
  "type": "connection.created",
  "created_at": "2024-01-15T10:30:00Z",
  "data": {
    "id": "conn_123abc",
    "external_id": "user_456",
    "integration_id": "int_slack",
    "is_valid": true,
    "created_at": "2024-01-15T10:30:00Z"
  }
}

Webhook Security (Flask Example)

To verify webhook signatures in a Flask application:
from flask import Flask, request, jsonify
import hmac
import hashlib
import os

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('BundleUp-Signature')
    payload = request.get_data()

    if not verify_signature(payload, signature):
        return jsonify({'error': 'Invalid signature'}), 401

    event = request.get_json()
    process_webhook_event(event)

    return '', 200

def verify_signature(payload: bytes, signature: str) -> bool:
    secret = os.environ['BUNDLEUP_WEBHOOK_SECRET']
    computed = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

def process_webhook_event(event: dict):
    event_type = event['type']

    if event_type == 'connection.created':
        handle_connection_created(event['data'])
    elif event_type == 'connection.deleted':
        handle_connection_deleted(event['data'])
    elif event_type == 'connection.updated':
        handle_connection_updated(event['data'])
    elif event_type == 'connection.expired':
        handle_connection_expired(event['data'])
Django Example:
# views.py
from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings
import hmac
import hashlib
import json

@csrf_exempt
def bundleup_webhook(request):
    if request.method != 'POST':
        return HttpResponseForbidden()

    signature = request.META.get('HTTP_BUNDLEUP_SIGNATURE')
    payload = request.body

    if not verify_signature(payload, signature):
        return HttpResponseForbidden('Invalid signature')

    event = json.loads(payload)
    process_webhook_event(event)

    return HttpResponse(status=200)

def verify_signature(payload: bytes, signature: str) -> bool:
    secret = settings.BUNDLEUP_WEBHOOK_SECRET
    computed = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

Proxy API

Make direct HTTP requests to integration APIs through BundleUp.

Creating a Proxy Instance

proxy = client.proxy('conn_123abc')

GET Request

response = proxy.get('/api/users')
data = response.json()
print(data)
With query parameters:
response = proxy.get('/api/users', params={'limit': 10})
With custom headers:
response = proxy.get('/api/users', headers={
    'X-Custom-Header': 'value',
    'Accept': 'application/json'
})

POST Request

response = proxy.post('/api/users', body={
    'name': 'John Doe',
    'email': 'john@example.com',
    'role': 'developer'
})

new_user = response.json()
print(f"Created user: {new_user}")
With custom headers:
response = proxy.post(
    '/api/users',
    body={'name': 'John Doe'},
    headers={
        'Content-Type': 'application/json',
        'X-API-Version': '2.0'
    }
)

PUT Request

response = proxy.put('/api/users/123', body={
    'name': 'Jane Doe',
    'email': 'jane@example.com'
})

updated_user = response.json()

PATCH Request

response = proxy.patch('/api/users/123', body={
    'email': 'newemail@example.com'
})

partially_updated = response.json()

DELETE Request

response = proxy.delete('/api/users/123')

if response.ok:
    print('User deleted successfully')

Working with Response Objects

The Proxy API returns requests Response objects:
response = proxy.get('/api/users')

# Access response body as JSON
data = response.json()

# Check status code
print(response.status_code)  # 200

# Check if successful
print(response.ok)  # True

# Access headers
print(response.headers['content-type'])

# Access raw content
content = response.content

# Access text
text = response.text

# Handle errors
try:
    response = proxy.get('/api/invalid')
    response.raise_for_status()
except requests.HTTPError as e:
    print(f"Request failed: {e}")

Unify API

Access unified, normalized data across different integrations with a consistent interface.

Creating a Unify Instance

unify = client.unify('conn_123abc')

Chat API

The Chat API provides a unified interface for chat platforms like Slack, Discord, and Microsoft Teams.
List Channels
Retrieve a list of channels from the connected chat platform.
result = unify.chat.channels({
    'limit': 100,
    'after': None,
    'include_raw': False
})

print(f"Channels: {result['data']}")
print(f"Next cursor: {result['metadata']['next']}")
Parameters:
  • limit (int, optional): Maximum number of channels to return (default: 100, max: 1000)
  • after (str, optional): Pagination cursor from previous response
  • include_raw (bool, optional): Include raw API response from the integration (default: False)
Response:
{
    'data': [
        {
            'id': 'C1234567890',
            'name': 'general'
        },
        {
            'id': 'C0987654321',
            'name': 'engineering'
        }
    ],
    'metadata': {
        'next': 'cursor_abc123'  # Use this for pagination
    },
    '_raw': {  # Only present if include_raw=True
        # Original response from the integration API
    }
}
Pagination example:
all_channels = []
cursor = None

while True:
    result = unify.chat.channels({
        'limit': 100,
        'after': cursor
    })

    all_channels.extend(result['data'])
    cursor = result['metadata']['next']

    if cursor is None:
        break

print(f"Fetched {len(all_channels)} total channels")

Git API

The Git API provides a unified interface for version control platforms like GitHub, GitLab, and Bitbucket.
List Repositories
result = unify.git.repos({
    'limit': 50,
    'after': None,
    'include_raw': False
})

print(f"Repositories: {result['data']}")
Response:
{
    'data': [
        {
            'id': '123456',
            'name': 'my-awesome-project',
            'full_name': 'organization/my-awesome-project',
            'description': 'An awesome project',
            'url': 'https://github.com/organization/my-awesome-project',
            'created_at': '2023-01-15T10:30:00Z',
            'updated_at': '2024-01-20T14:22:00Z',
            'pushed_at': '2024-01-20T14:22:00Z'
        }
    ],
    'metadata': {
        'next': 'cursor_xyz789'
    }
}
List Pull Requests
result = unify.git.pulls('organization/repo-name', {
    'limit': 20,
    'after': None,
    'include_raw': False
})

print(f"Pull Requests: {result['data']}")
Parameters:
  • repo_name (str, required): Repository name in the format ‘owner/repo’
  • limit (int, optional): Maximum number of PRs to return
  • after (str, optional): Pagination cursor
  • include_raw (bool, optional): Include raw API response
Response:
{
    'data': [
        {
            'id': '12345',
            'number': 42,
            'title': 'Add new feature',
            'description': 'This PR adds an awesome new feature',
            'draft': False,
            'state': 'open',
            'url': 'https://github.com/org/repo/pull/42',
            'user': 'john-doe',
            'created_at': '2024-01-15T10:30:00Z',
            'updated_at': '2024-01-20T14:22:00Z',
            'merged_at': None
        }
    ],
    'metadata': {
        'next': None
    }
}
List Tags
result = unify.git.tags('organization/repo-name', {'limit': 50})

print(f"Tags: {result['data']}")
Response:
{
    'data': [
        {
            'name': 'v1.0.0',
            'commit_sha': 'abc123def456'
        },
        {
            'name': 'v0.9.0',
            'commit_sha': 'def456ghi789'
        }
    ],
    'metadata': {
        'next': None
    }
}
List Releases
result = unify.git.releases('organization/repo-name', {'limit': 10})

print(f"Releases: {result['data']}")
Response:
{
    'data': [
        {
            'id': '54321',
            'name': 'Version 1.0.0',
            'tag_name': 'v1.0.0',
            'description': 'Initial release with all the features',
            'prerelease': False,
            'url': 'https://github.com/org/repo/releases/tag/v1.0.0',
            'created_at': '2024-01-15T10:30:00Z',
            'released_at': '2024-01-15T10:30:00Z'
        }
    ],
    'metadata': {
        'next': None
    }
}

Project Management API

The PM API provides a unified interface for project management platforms like Jira, Linear, and Asana.
List Issues
result = unify.pm.issues({
    'limit': 100,
    'after': None,
    'include_raw': False
})

print(f"Issues: {result['data']}")
Response:
{
    'data': [
        {
            'id': 'PROJ-123',
            'url': 'https://jira.example.com/browse/PROJ-123',
            'title': 'Fix login bug',
            'status': 'in_progress',
            'description': 'Users are unable to log in',
            'created_at': '2024-01-15T10:30:00Z',
            'updated_at': '2024-01-20T14:22:00Z'
        }
    ],
    'metadata': {
        'next': 'cursor_def456'
    }
}
Filtering and sorting:
open_issues = [issue for issue in result['data'] if issue['status'] == 'open']
sorted_by_date = sorted(
    result['data'],
    key=lambda x: x['created_at'],
    reverse=True
)

Error Handling

The SDK raises exceptions for errors. Always wrap SDK calls in try-except blocks for proper error handling.
try:
    connections = client.connection.list()
except Exception as e:
    print(f"Failed to fetch connections: {e}")