> ## Documentation Index > Fetch the complete documentation index at: https://docs.danubeai.com/llms.txt > Use this file to discover all available pages before exploring further. # Python SDK > Official Python SDK for Danube AI # Python SDK The official Python SDK for Danube AI provides a clean, Pythonic interface for accessing services, tools, skills, and user identity through the Danube platform. ## Installation ```bash theme={null} pip install danube ``` ## Quick Start ```python theme={null} 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}") ``` ## Authentication The SDK uses your Danube API key for authentication. You can provide it in two ways: ```python Environment Variable theme={null} import os os.environ["DANUBE_API_KEY"] = "dk_your_api_key" from danube import DanubeClient client = DanubeClient() # Uses DANUBE_API_KEY automatically ``` ```python Constructor Parameter theme={null} from danube import DanubeClient client = DanubeClient(api_key="dk_your_api_key") ``` Get your API key from the [Danube Dashboard](https://danubeai.com/dashboard/api-keys). ## Async Support For better performance in async applications, use `AsyncDanubeClient`: ```python theme={null} 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 ```python theme={null} # 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 ```python theme={null} # 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 ```python theme={null} # 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) ``` ### Identity ```python theme={null} # Get user identity identity = client.identity.get() print(f"Name: {identity.name}") print(f"Email: {identity.email}") ``` ### Workflows ```python theme={null} # 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") ``` ### Sites ```python theme={null} # 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}") ``` ## Error Handling The SDK provides specific exception types for different error conditions: ```python theme={null} 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") ``` ## Configuration Options | Parameter | Environment Variable | Default | Description | | ------------- | -------------------- | -------------------------- | ------------------------- | | `api_key` | `DANUBE_API_KEY` | (required) | Your Danube API key | | `base_url` | `DANUBE_API_URL` | `https://api.danubeai.com` | API base URL | | `timeout` | `DANUBE_TIMEOUT` | `30` | Request timeout (seconds) | | `max_retries` | `DANUBE_MAX_RETRIES` | `3` | Max retry attempts | ```python theme={null} client = DanubeClient( api_key="dk_...", base_url="https://api.danubeai.com", timeout=60.0, max_retries=5, ) ``` ## 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 | ### Workflow | Field | Type | Description | | ------------------ | ---------- | --------------------- | | `id` | str | Workflow UUID | | `name` | str | Workflow name | | `description` | str | Workflow description | | `step_count` | int | Number of steps | | `visibility` | str | "public" or "private" | | `tags` | list\[str] | Workflow tags | | `total_executions` | int | Total execution count | ### WorkflowExecution | Field | Type | Description | | ------------------- | ---- | -------------------------------------------- | | `id` | str | Execution UUID | | `workflow_id` | str | Parent workflow ID | | `status` | str | "pending", "running", "success", or "failed" | | `inputs` | dict | Input values provided | | `step_results` | list | Per-step results | | `error` | str | Error message (if failed) | | `execution_time_ms` | int | Total execution time | ### AgentSite | Field | Type | Description | | ------------ | -------------- | --------------------------------------------------- | | `id` | str | Site UUID | | `domain` | str | Site domain | | `url` | str | Full URL | | `status` | str | "pending", "crawling", "analyzed", or "live" | | `components` | SiteComponents | Structured site data (contact, pricing, docs, etc.) | | `category` | str | Site category | | `tags` | list\[str] | Site tags | ## Source Code The SDK is open source and available on GitHub: * [GitHub Repository](https://github.com/danubeai/danube-python) * [PyPI Package](https://pypi.org/project/danube/)