daita-client

Python SDK for calling hosted Daita agents and workflows from any Python project.

Installation

pip install daita-client

Authentication

Set your API key as an environment variable:

export DAITA_API_KEY=sk-...

Or pass it directly when creating a client:

from daita_client import DaitaClient

client = DaitaClient(api_key="sk-...")

Quick start

from daita_client import DaitaClient

with DaitaClient() as client:
    result = client.run_agent("my_agent", prompt="Summarise last month's sales", wait=True)
    print(result.output)

Usage

Run an agent and wait for the result

from daita_client import DaitaClient

with DaitaClient() as client:
    result = client.run_agent(
        "analyst",
        prompt="Which product had the highest margin in Q4?",
        wait=True,
        timeout=120,
    )
    print(result.output)
    print(f"Cost: ${result.cost:.6f} | Tokens: {result.total_tokens}")

Fire and poll manually

with DaitaClient() as client:
    # Returns immediately with an execution_id
    execution = client.run_agent("analyst", prompt="Run full report", wait=False)
    print(f"Started: {execution.execution_id}")

    # Do other work while the agent runs ...

    # Block until complete
    result = client.wait_for_execution(execution.execution_id, timeout=300)
    print(result.output)

Pass structured data

with DaitaClient() as client:
    result = client.run_agent(
        "data_processor",
        data={"records": [{"id": 1, "value": 42}, {"id": 2, "value": 17}]},
        wait=True,
    )

Run a workflow

with DaitaClient() as client:
    result = client.run_workflow(
        "etl_pipeline",
        data={"source": "s3://my-bucket/data.csv"},
        wait=True,
    )

List and inspect executions

with DaitaClient() as client:
    # List recent executions
    history = client.list_executions(limit=10, status="completed", target_type="agent")
    for ex in history:
        print(f"{ex.execution_id[:8]}  {ex.status:<12}  {ex.duration_seconds:.1f}s  {ex.target_name}")

    # Get a specific execution
    result = client.get_execution("exec_abc12345")

    # Get the latest execution for a specific agent
    latest = client.get_latest_execution(agent_name="analyst")

Cancel an execution

with DaitaClient() as client:
    execution = client.run_agent("long_task", wait=False)
    client.cancel_execution(execution.execution_id)

One-off convenience functions

For scripts that only need a single call, import run_agent or run_workflow directly — no client setup required:

from daita_client import run_agent, run_workflow

result = run_agent("analyst", prompt="Quick summary", wait=True)
print(result.output)

result = run_workflow("pipeline", data={"source": "s3"}, wait=True)

Async interface

All methods have an _async suffix counterpart. Use async with for the client:

import asyncio
from daita_client import DaitaClient

async def main():
    async with DaitaClient() as client:
        result = await client.run_agent_async(
            "analyst",
            prompt="Async analysis",
            wait=True,
        )
        print(result.output)

asyncio.run(main())

Module-level async convenience functions are also available:

from daita_client import run_agent_async, run_workflow_async

result = await run_agent_async("analyst", prompt="Quick summary")

ExecutionResult properties

Property	Type	Description
execution_id	str	Unique execution identifier
status	str	queued, running, completed, failed, cancelled
target_name	str	Name of the agent or workflow
output	Any	Primary output from the agent (text or structured data)
total_tokens	int | None	Total tokens used
prompt_tokens	int | None	Input tokens
completion_tokens	int | None	Output tokens
cost	float | None	Estimated cost in USD
duration_seconds	float | None	Total wall-clock time
processing_time_seconds	float | None	Time spent in LLM calls
iterations	int | None	Number of tool-call loops
tool_calls	list	List of tool calls made
is_complete	bool	True if status is terminal
is_success	bool	True if status is completed or success
is_running	bool	True if status is queued or running
dashboard_url	str | None	Link to execution in the Daita dashboard

Error handling

from daita_client import DaitaClient
from daita_client.exceptions import (
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ExecutionTimeoutError,
    ServerError,
)

try:
    with DaitaClient() as client:
        result = client.run_agent("my_agent", prompt="...", wait=True, timeout=60)
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Agent not found — is it deployed?")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ExecutionTimeoutError as e:
    print(f"Timed out after {e.timeout_seconds}s")
except ServerError:
    print("Daita server error")

All exceptions inherit from ExecutionError, so you can catch them all with a single except ExecutionError.

Client options

client = DaitaClient(
    api_key="sk-...",         # defaults to DAITA_API_KEY env var
    api_base="https://...",   # defaults to DAITA_API_ENDPOINT env var or https://api.daita-tech.io
    timeout=300,              # request timeout in seconds (default: 300)
    max_retries=3,            # retries on transient failures (default: 3)
    retry_delay=1.0,          # base retry delay in seconds (default: 1.0)
)

License

Apache 2.0 — see LICENSE.