Python SDK Reference

Complete API reference for the DataGen Python SDK

Installation

pip install datagen-python-sdk

Requires Python 3.10+.

Client Configuration

from datagen_sdk import DatagenClient

client = DatagenClient(
    api_key=None,           # Default: reads from DATAGEN_API_KEY env var
    base_url="https://api.datagen.dev",
    timeout=30,             # Request timeout in seconds
    retries=0,              # Retry attempts for failed requests
    backoff_seconds=0.5     # Initial backoff (exponential with 2^attempt multiplier)
)

Parameter	Type	Default	Description
api_key	Optional[str]	None	API key. Falls back to DATAGEN_API_KEY env var
base_url	str	"https://api.datagen.dev"	DataGen API base URL
timeout	int	30	Request timeout in seconds
retries	int	0	Number of retry attempts
backoff_seconds	float	0.5	Initial backoff time for retries

---

Tool Execution

execute_tool

Execute an MCP tool by its alias name.

result = client.execute_tool(
    tool_alias_name: str,
    parameters: Optional[Dict[str, Any]] = None
) -> Any

Parameter	Type	Required	Description
tool_alias_name	str	Yes	Tool alias (e.g., "mcp_Gmail_gmail_send_email")
parameters	Dict[str, Any]	No	Tool-specific parameters

Returns: Tool execution result (type varies by tool).

Raises: DatagenAuthError, DatagenToolError, DatagenHttpError

---

Custom Tool Deployment

deploy_custom_tool

Deploy a Python workflow as an API endpoint.

result = client.deploy_custom_tool(
    name: str,
    code: str,
    description: Optional[str] = None,
    input_schema: Optional[Dict[str, Any]] = None,
    output_vars: Optional[List[str]] = None,
    expected_tools: Optional[List[str]] = None,
    additional_imports: Optional[List[str]] = None,
    deployment_type: str = "private",
    default_input_vars: Optional[Dict[str, Any]] = None,
    mcp_server_names: Optional[List[str]] = None,
    required_secrets: Optional[List[str]] = None,
) -> Dict[str, Any]

Parameter	Type	Required	Default	Description
name	str	Yes	—	API name for the tool
code	str	Yes	—	Python code to deploy
description	str	No	None	Tool description
input_schema	Dict	No	None	OpenAPI/JSON Schema for inputs
output_vars	List[str]	No	None	Output variable names
expected_tools	List[str]	No	None	Required MCP tools
additional_imports	List[str]	No	None	Python packages
deployment_type	str	No	"private"	"private" or "public"
default_input_vars	Dict	No	None	Default input values
mcp_server_names	List[str]	No	None	MCP server names
required_secrets	List[str]	No	None	Required secret names

Returns: Dict with deployment_uuid and deployment details.

Raises: DatagenDeploymentError

update_custom_tool

Update an existing custom tool. Only provided fields are updated.

result = client.update_custom_tool(
    deployment_uuid: str,
    name: Optional[str] = None,
    description: Optional[str] = None,
    code: Optional[str] = None,
    input_schema: Optional[Dict[str, Any]] = None,
    output_vars: Optional[List[str]] = None,
    default_input_vars: Optional[Dict[str, Any]] = None,
    additional_imports: Optional[List[str]] = None,
    expected_tools: Optional[List[str]] = None,
    required_secrets: Optional[List[str]] = None,
) -> Dict[str, Any]

Parameter	Type	Required	Description
deployment_uuid	str	Yes	UUID of the tool to update
name	str	No	Updated name
description	str	No	Updated description
code	str	No	Updated Python code
input_schema	Dict	No	Updated input schema
output_vars	List[str]	No	Updated output variables
default_input_vars	Dict	No	Updated default inputs
additional_imports	List[str]	No	Updated imports
expected_tools	List[str]	No	Updated required tools
required_secrets	List[str]	No	Updated required secrets

Returns: Dict with updated deployment details.

Raises: DatagenDeploymentError

get_custom_tool

Get details of a custom tool.

result = client.get_custom_tool(deployment_uuid: str) -> Dict[str, Any]

list_custom_tools

List all custom tools.

result = client.list_custom_tools(
    sort_by: str = "created_at",
    order_by: str = "desc",
    skip: int = 0,
    limit: int = 100,
) -> List[Dict[str, Any]]

Parameter	Type	Default	Description
sort_by	str	"created_at"	Sort field
order_by	str	"desc"	"asc" or "desc"
skip	int	0	Pagination offset
limit	int	100	Max results

---

Custom Tool Execution

run_custom_tool

Run a custom tool synchronously (blocks until complete).

result = client.run_custom_tool(
    deployment_uuid: str,
    input_vars: Optional[Dict[str, Any]] = None,
    execution_timeout: int = 120,
) -> Dict[str, Any]

Parameter	Type	Required	Default	Description
deployment_uuid	str	Yes	—	Tool UUID
input_vars	Dict	No	None	Input variables
execution_timeout	int	No	120	Max execution time in seconds

Returns: Dict with run_uuid, status, and output_vars.

Raises: DatagenDeploymentError

run_custom_tool_async

Run a custom tool asynchronously (returns immediately).

result = client.run_custom_tool_async(
    deployment_uuid: str,
    input_vars: Optional[Dict[str, Any]] = None,
    execution_timeout: int = 120,
) -> Dict[str, Any]

Same parameters as run_custom_tool.

Returns: Dict with run_uuid and status: "pending".

Raises: DatagenDeploymentError

check_run_status

Check the status of a custom tool run.

result = client.check_run_status(run_uuid: str) -> Dict[str, Any]

Returns: Dict with run_uuid and status ("pending", "running", "completed", or "failed").

get_run

Get full details of a custom tool run including output.

result = client.get_run(run_uuid: str) -> Dict[str, Any]

Returns: Dict with complete run details including output_vars.

wait_for_run

Poll until an async run completes or fails.

result = client.wait_for_run(
    run_uuid: str,
    timeout: int = 300,
    poll_interval: float = 2.0,
) -> Dict[str, Any]

Parameter	Type	Default	Description
run_uuid	str	—	Run UUID
timeout	int	300	Max wait time in seconds
poll_interval	float	2.0	Seconds between status checks

Returns: Final run result with output_vars.

Raises: DatagenDeploymentError on failure or timeout.

---

Secret Management

list_secrets

List all secrets (values are never returned).

result = client.list_secrets() -> List[Dict[str, Any]]

Returns: List of secret metadata (name, masked_value, provider, etc.).

set_secret

Create or update a secret.

result = client.set_secret(
    name: str,
    value: str,
    display_name: Optional[str] = None,
    description: Optional[str] = None,
    category: str = "api_key",
    force: bool = False,
) -> Dict[str, Any]

Parameter	Type	Required	Default	Description
name	str	Yes	—	Secret name (alphanumeric, underscores, hyphens)
value	str	Yes	—	Secret value
display_name	str	No	None	Human-readable name
description	str	No	None	Description
category	str	No	"api_key"	Secret category
force	bool	No	False	If True, updates existing secret

Raises: DatagenSecretError if secret exists and force=False.

---

Error Types

All exceptions inherit from DatagenError.

Exception	Description
DatagenError	Base exception class
DatagenAuthError	Authentication failed (401/403) or missing API key
DatagenToolError	Tool execution failed
DatagenHttpError	HTTP-level errors (4xx/5xx)
DatagenDeploymentError	Custom tool deployment or run failed
DatagenSecretError	Secret management errors

from datagen_sdk import (
    DatagenClient,
    DatagenError,
    DatagenAuthError,
    DatagenToolError,
    DatagenHttpError,
    DatagenDeploymentError,
    DatagenSecretError,
)