danube 0.3.0

Official Python SDK for Danube

Danube Python SDK

Official Python SDK for Danube AI - access services, tools, skills, and user identity through a clean, Pythonic interface.

Installation

pip install danube

Or install from source:

pip install -e .

Quick Start

from danube import DanubeClient

# Initialize with API key (or set DANUBE_API_KEY environment variable)
client = DanubeClient(api_key="dk_...")

# List available services
services = client.services.list(limit=5)
for service in services:
    print(f"{service.name}: {service.tool_count} tools")

# Search for tools
tools = client.tools.search("send email")
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# Execute a tool
result = client.tools.execute(
    tool_name="Gmail - Send Email",
    parameters={
        "to": "[e-mail]",
        "subject": "Hello from Danube!",
        "body": "This email was sent using the Danube SDK."
    }
)

if result.success:
    print(f"Success: {result.content}")
else:
    print(f"Error: {result.error}")

# Clean up
client.close()

Usage

Context Manager (Recommended)

from danube import DanubeClient

with DanubeClient(api_key="dk_...") as client:
    services = client.services.list()
    # Client automatically closes when exiting the context

Async Usage

For better performance in async applications:

import asyncio
from danube import AsyncDanubeClient

async def main():
    async with AsyncDanubeClient(api_key="dk_...") as client:
        # Parallel requests
        services, tools = await asyncio.gather(
            client.services.list(limit=10),
            client.tools.search("weather"),
        )

        # Execute a tool
        result = await client.tools.execute(
            tool_name="Weather - Get Current",
            parameters={"city": "San Francisco"}
        )
        print(result.content)

asyncio.run(main())

API Reference

Services

# List/search services
services = client.services.list(query="github", limit=10)

# Get a specific service
service = client.services.get("service-uuid")

# Get tools for a service
result = client.services.get_tools("service-uuid")
if result.needs_configuration:
    print(f"Configure at: {result.configuration_url}")
else:
    for tool in result.tools:
        print(tool.name)

Tools

# Search for tools
tools = client.tools.search("send email", service_id="optional-filter")

# Get a specific tool
tool = client.tools.get("tool-uuid")

# Execute by ID (faster)
result = client.tools.execute(tool_id="tool-uuid", parameters={"key": "value"})

# Execute by name (searches first)
result = client.tools.execute(tool_name="Gmail - Send Email", parameters={...})

# Check result
if result.success:
    print(result.content)
    print(f"Took {result.duration_ms}ms")
else:
    print(f"Error: {result.error}")

# Batch execute (up to 10 calls)
results = client.tools.batch_execute([
    {"tool_id": "tool-uuid-1", "parameters": {"city": "San Francisco"}},
    {"tool_id": "tool-uuid-2", "parameters": {"city": "New York"}},
])
for r in results:
    print(f"{r.tool_id}: {'ok' if r.success else r.error}")

Skills

# Search for skills
skills = client.skills.search("pdf processing")

# Get full skill content
skill = client.skills.get(skill_id="skill-uuid")
# or
skill = client.skills.get(skill_name="pdf-processing")

print(f"Instructions:\n{skill.skill_md}")

for script in skill.scripts:
    print(f"Script: {script.name}")
    print(script.content)

Identity

# Get user identity
identity = client.identity.get()
print(f"Name: {identity.name}")
print(f"Email: {identity.email}")

Workflows

# List public workflows
workflows = client.workflows.list(query="email automation", limit=10)

# Get workflow details
workflow = client.workflows.get("workflow-uuid")
for step in workflow.steps:
    print(f"Step {step.step_number}: {step.tool_name}")

# Execute a workflow
execution = client.workflows.execute(
    "workflow-uuid",
    inputs={"query": "test", "recipient": "[e-mail]"}
)

# Check execution result
result = client.workflows.get_execution(execution.id)
print(f"Status: {result.status}")
for step in result.step_results:
    print(f"  Step {step.step_number}: {step.status}")

# Create a workflow
workflow = client.workflows.create(
    name="My Pipeline",
    steps=[
        {
            "step_number": 1,
            "tool_id": "tool-uuid",
            "tool_name": "Weather - Get Forecast",
            "description": "Fetch forecast",
            "input_mapping": {"city": "{{inputs.city}}"},
        }
    ],
    visibility="private",
    tags=["weather"],
)

# Update a workflow
updated = client.workflows.update("workflow-uuid", name="Renamed Pipeline")

# Delete a workflow
client.workflows.delete("workflow-uuid")

Sites

# Search agent-friendly site directory
sites = client.sites.search("stripe", category="finance", limit=10)

# Get site by ID
site = client.sites.get("site-uuid")
print(f"Domain: {site.domain}")
print(f"Components: {list(site.components.__dict__.keys())}")

# Get site by domain
site = client.sites.get_by_domain("stripe.com")
if site.components.pricing:
    print(f"Pricing: {site.components.pricing}")

Wallet

# Check balance
balance = client.wallet.get_balance()
print(f"Balance: ${balance.balance_dollars:.2f}")

# Get transaction history
transactions = client.wallet.get_transactions()
for tx in transactions:
    print(f"{tx.type}: {tx.amount_cents}c ({tx.created_at})")

Ratings

# Submit a rating
client.ratings.submit(tool_id="tool-uuid", rating=5, comment="Works great!")

# Get your rating for a tool
my_rating = client.ratings.get_mine("tool-uuid")

# Get aggregate ratings
aggregate = client.ratings.get_tool_ratings("tool-uuid")
print(f"Average: {aggregate.average_rating} ({aggregate.total_ratings} ratings)")

Credentials

# Store a credential for a service
result = client.credentials.store(
    service_id="service-uuid",
    credential_type="bearer",
    credential_value="sk-..."
)
print(f"Stored for: {result.service_name}")

Device Auth

# Initiate device auth flow (for CLI onboarding)
code = client.device_auth.request_code()
print(f"Go to {code.verification_url} and enter: {code.user_code}")

# Poll for token
import time
while True:
    token = client.device_auth.poll_token(code.device_code)
    if token:
        print(f"API key: {token.api_key}")
        break
    time.sleep(code.interval)

Configuration

Environment Variables

Variable	Description	Default
DANUBE_API_KEY	Your Danube API key	(required)
DANUBE_API_URL	API base URL	https://api.danubeai.com
DANUBE_TIMEOUT	Request timeout (seconds)	30
DANUBE_MAX_RETRIES	Max retry attempts	3

Programmatic Configuration

from danube import DanubeClient

client = DanubeClient(
    api_key="dk_...",
    base_url="https://api.danubeai.com",
    timeout=60.0,
    max_retries=5,
)

Error Handling

from danube import DanubeClient
from danube.exceptions import (
    NotFoundError,
    ExecutionError,
    ConfigurationRequiredError,
    RateLimitError,
    AuthenticationError,
)

with DanubeClient() as client:
    try:
        result = client.tools.execute(tool_id="invalid-id")
    except AuthenticationError:
        print("Invalid API key")
    except NotFoundError as e:
        print(f"Tool not found: {e}")
    except ExecutionError as e:
        print(f"Execution failed: {e}")
    except ConfigurationRequiredError as e:
        print(f"Configure credentials at: {e.configuration_url}")
    except RateLimitError as e:
        if e.retry_after:
            print(f"Rate limited. Retry in {e.retry_after}s")

Models

Service

Field	Type	Description
id	str	Service UUID
name	str	Service name
description	str	Service description
service_type	str	"mcp_server", "api", "internal", or "website"
tool_count	int	Number of available tools
is_connected	bool	Whether MCP service is connected

Tool

Field	Type	Description
id	str	Tool UUID
name	str	Tool name
description	str	Tool description
service_id	str	Parent service ID
parameters	dict	Parameter definitions

ToolResult

Field	Type	Description
success	bool	Whether execution succeeded
result	Any	Execution result
error	str	Error message (if failed)
content	str	Result as text (property)
duration_ms	float	Execution time

Skill / SkillContent

Field	Type	Description
id	str	Skill UUID
name	str	Skill name
description	str	Skill description
skill_md	str	Main SKILL.md content
scripts	list	Script files
references	list	Reference files
assets	list	Asset files

Identity / Contact

Field	Type	Description
profile	dict	User profile data
key_people	list	Important contacts
contacts	list	General contacts
name	str	Contact name
email	str	Contact email

WorkflowDetail / WorkflowStep

Field	Type	Description
id	str	Workflow UUID
name	str	Workflow name
description	str	Workflow description
steps	list[WorkflowStep]	Ordered execution steps
visibility	str	"private" or "public"
tags	list[str]	Workflow tags
step_number	int	Step order (on WorkflowStep)
tool_id	str	Tool UUID (on WorkflowStep)
tool_name	str	Tool name (on WorkflowStep)
input_mapping	dict	Template mapping (on WorkflowStep)

WorkflowExecution / WorkflowStepResult

Field	Type	Description
id	str	Execution UUID
status	str	"pending", "running", "success", or "failed"
inputs	dict	Input values used
step_results	list[WorkflowStepResult]	Per-step results
error	str?	Error message if failed
execution_time_ms	int?	Total execution time

AgentSite / AgentSiteListItem

Field	Type	Description
id	str	Site UUID
domain	str	Site domain
url	str	Full URL
status	str	"pending", "crawling", "analyzing", "review", "live", "error"
components	SiteComponents	Extracted structured components
discovered_tools	list[dict]	Auto-generated tool schemas
category	str?	Site category
tags	list[str]	Tags

SiteComponents

Field	Type	Description
contact	dict?	Emails, phones, forms
about	dict?	Company info
services	list[dict]?	Services offered
docs	dict?	Documentation links
pricing	dict?	Pricing plans
faq	list[dict]?	FAQ items
legal	dict?	Privacy policy, ToS
navigation	list[dict]?	Site navigation

WalletBalance / WalletTransaction

Field	Type	Description
balance_cents	int	Balance in cents
balance_dollars	float	Balance in dollars
type	str	Transaction type (on WalletTransaction)
amount_cents	int	Transaction amount (on WalletTransaction)

ToolRating / RatingAggregate

Field	Type	Description
tool_id	str	Tool UUID
rating	int	Star rating (1-5)
comment	str?	Optional comment
average_rating	float	Average stars (on RatingAggregate)
total_ratings	int	Total rating count (on RatingAggregate)

CredentialStoreResult

Field	Type	Description
success	bool	Whether storage succeeded
service_id	str	Service UUID
service_name	str	Service name
credential_type	str	"bearer" or "api_key"

DeviceCode / DeviceToken

Field	Type	Description
device_code	str	Device code for polling
user_code	str	User-facing code (XXXX-XXXX)
verification_url	str	URL to enter user_code
expires_in	int	Seconds until expiry
api_key	str	Returned API key (on DeviceToken)

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy src/danube

# Format code
black src/danube tests
ruff check src/danube tests

License

MIT License - see LICENSE for details.