pip install danube

from danube import DanubeClient

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

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

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

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

# Search the agent-friendly site directory
sites = client.sites.search("payments", limit=10)

# Get a site by ID
site = client.sites.get("site-uuid")
print(f"{site.domain}: {site.status}")

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

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="dk_...",
    base_url="https://api.danubeai.com",
    timeout=60.0,
    max_retries=5,
)