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="YOUR_DANUBE_API_KEY")

# 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": "user@example.com",
        "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="YOUR_DANUBE_API_KEY") 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="YOUR_DANUBE_API_KEY") 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": "user@example.com"}
)

# 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)

Agent frameworks

Three tools (search, describe, execute) for the framework you already use:

from danube import DanubeClient
from danube.integrations.openai_agents import danube_tools   # pip install "danube[openai-agents]"
# from danube.integrations.langchain import danube_tools     # pip install "danube[langchain]"
# from danube.integrations.crewai import danube_tools        # pip install "danube[crewai]"

tools = danube_tools(DanubeClient())

Search keeps the model's context small (the catalog has thousands of tools and changes daily); describe returns the schema, tips and readiness for the one tool the agent is about to call. See the docs: OpenAI Agents SDK, LangChain, CrewAI.

Configuration

Environment Variables

Programmatic Configuration

from danube import DanubeClient

client = DanubeClient(
    api_key="YOUR_DANUBE_API_KEY",
    base_url="https://api.danubeai.com",
    org_id="00000000-0000-0000-0000-000000000000",  # optional org context
    timeout=60.0,
    max_retries=5,
)

Organization Context

When org_id is set (constructor kwarg or DANUBE_ORG_ID), every request carries the danube-org-id header, so calls are attributed to — and governed by — that organization. Membership is validated server-side. You can inspect or switch it at runtime:

client = DanubeClient(api_key="YOUR_DANUBE_API_KEY", org_id="<org-uuid>")
print(client.org_id)          # current org context
client.org_id = "<other-org>"  # switch (applies to subsequent requests)

Device-code auth issues a per-user API key; after the user authorizes, set client.org_id to the org they selected.

Self-Hosted / In-VPC Deployments

Point the SDK at a customer-hosted control plane with base_url or DANUBE_BASE_URL:

client = DanubeClient(api_key="YOUR_DANUBE_API_KEY", base_url="https://danube.internal.example.com")

Note: on self-hosted deployments, storing plaintext credential values centrally is disabled — pass credential references (env://VAR_NAME or vault://path/to/secret) to client.credentials.store(...) instead, and the secret is resolved in your environment at execution time.

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

Tool

ToolResult

Skill / SkillContent

Identity / Contact

WorkflowDetail / WorkflowStep

WorkflowExecution / WorkflowStepResult

AgentSite / AgentSiteListItem

SiteComponents

WalletBalance / WalletTransaction

ToolRating / RatingAggregate

CredentialStoreResult

DeviceCode / 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.