danube 0.2.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}")

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}")

# Search contacts
contacts = client.identity.search_contacts("john", limit=5)
for contact in contacts:
    print(f"{contact.name} ({contact.email})")

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", or "internal"
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

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.