pip install danube

from danube import DanubeClient

# Initialize with API key (or set DANUBE_API_KEY environment variable)
with DanubeClient(api_key="YOUR_API_KEY") as client:
    # 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")

    # Execute a tool
    result = client.tools.execute(
        tool_name="Gmail - Send Email",
        parameters={
            "to": "user@example.com",
            "subject": "Hello from Danube!"
        }
    )

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

import asyncio
from danube import AsyncDanubeClient

async def main():
    async with AsyncDanubeClient(api_key="YOUR_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())

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

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

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

# Get full skill content
skill = client.skills.get(skill_id="skill-uuid")
# or by name
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)

# Share a skill folder with your organization (or pass a .zip, or zip bytes)
skill = client.skills.upload("./refund-policy", visibility="internal", org_id="org-uuid")
# Publish a new version of an existing skill
skill = client.skills.upload("./refund-policy", visibility="internal", replace=True)

# Your organization's internal skills, and a skill as a zip for .claude/skills/
org_skills = client.skills.list_org("org-uuid")
open("refund-policy.zip", "wb").write(client.skills.download(skill.id))

context = client.context.get()
print(context.markdown)
if context.organization:
    print(context.organization.name, context.organization.role)

# Only the text
text = client.context.get_markdown()

# Remember something in a personal document (created when missing)
client.context.remember("Prefers weekly summaries on Mondays.")
client.context.remember("# Role\nSupport lead", document="role.md", mode="replace")

client.context.set_timezone("Europe/Oslo")

# List public workflows
workflows = client.workflows.list(query="data pipeline", limit=10)

# Get a specific workflow
workflow = client.workflows.get("workflow-uuid")

# Execute a workflow with inputs
execution = client.workflows.execute(
    workflow_id="workflow-uuid",
    inputs={"query": "latest news"}
)
print(f"Status: {execution.status}")

for step in execution.step_results:
    print(f"Step {step.step_number}: {step.status}")

# Get a past execution result
result = client.workflows.get_execution("execution-uuid")
print(f"Completed in {result.execution_time_ms}ms")

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

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

for tx in client.wallet.get_transactions(limit=20, offset=0):
    print(tx)

# USDC spending limits for x402 payments, in atomic units (1 USDC = 1_000_000)
limits = client.wallet.get_spending_limits()
print(limits.max_per_call_atomic, limits.daily_limit_atomic)
client.wallet.update_spending_limits(max_per_call_atomic=2_000_000, daily_limit_atomic=20_000_000)

for key in client.api_keys.list():
    print(key)

created = client.api_keys.create(
    name="Staging agent",
    permissions={"allowed_services": ["service-uuid"], "max_spend_per_day_cents": 500},
)
print(created.key)  # shown once; store it now

rotated = client.api_keys.rotate("key-uuid")   # new secret, same permissions
client.api_keys.revoke("key-uuid")

hook = client.webhooks.create(
    url="https://example.com/hooks/danube",
    events=["tool.execution.failed", "workflow.failed"],
    description="Failure alerts",
)
print(hook.secret)  # signing secret, shown once

client.webhooks.update("webhook-uuid", is_active=False)
for delivery in client.webhooks.get_deliveries("webhook-uuid", limit=20):
    print(delivery)
client.webhooks.delete("webhook-uuid")

# Accounts created through POST /v1/agents (see Register Agent)
info = client.agents.get_info()
print(info.name, info.wallet)

# Add funds: a Stripe checkout URL, or a crypto deposit address
fund = client.agents.fund_wallet(method="card_checkout", amount_cents=2000)
print(fund.checkout_url)
fund = client.agents.fund_wallet(method="crypto")
print(fund.deposit_address, fund.network)

# Log a person in from a CLI or a headless agent without pasting a key
code = client.device_auth.request_code(client_name="My CLI")
print(f"Open {code.verification_url} and enter {code.user_code}")

# Poll every `code.interval` seconds; 428 means not yet, 410 means start over
token = client.device_auth.poll_token(code.device_code)

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

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

# Inspect or switch org context at runtime
print(client.org_id)
client.org_id = "<other-org-uuid>"