Delega Python SDK

Official Python SDK for the Delega API.

Installation

pip install delega

For async support:

pip install 'delega[async]'

Quick Start

from delega import Delega

client = Delega(api_key="dlg_...")

# List tasks
tasks = client.tasks.list()

# Create a task
task = client.tasks.create("Deploy to production", priority=1, labels=["ops"])

# Complete a task
client.tasks.complete(task.id)

Authentication

Pass your API key directly or set the DELEGA_API_KEY environment variable. DELEGA_AGENT_KEY is also accepted as a fallback for shells that are already configured for the Delega MCP; when both are set, DELEGA_API_KEY wins.

# Explicit
client = Delega(api_key="dlg_...")

# From environment
# export DELEGA_API_KEY=dlg_...
# or: export DELEGA_AGENT_KEY=dlg_...
client = Delega()

To target a custom endpoint (advanced), point base_url at the API namespace:

client = Delega(api_key="dlg_...", base_url="http://localhost:18890")
# or: Delega(api_key="dlg_...", base_url="https://delega.yourcompany.com/api")

Passing a bare localhost URL defaults to the /api namespace. Passing a bare HTTPS remote URL defaults to /v1; include /api explicitly for custom endpoints that expose the /api namespace. Plain HTTP is rejected unless the host is localhost.

Tasks

# List with filters
tasks = client.tasks.list(priority=1, completed=False)
tasks = client.tasks.list(labels=["urgent"], due_before="2026-12-31")

# Search
tasks = client.tasks.search("deploy")

# CRUD
task = client.tasks.create("Fix bug", description="Crash on login", priority=1)
task = client.tasks.get("task_id")
task = client.tasks.update("task_id", content="Updated title", priority=3)
client.tasks.delete("task_id")

# Completion
client.tasks.complete("task_id")
client.tasks.uncomplete("task_id")

# Delegation and assignment
subtask = client.tasks.delegate(
    "parent_task_id",
    "Research options",
    priority=2,
    assigned_to_agent_id="agent_id",
)
task = client.tasks.assign("task_id", "agent_id")  # or None to unassign
chain = client.tasks.chain("task_id")

# Duplicate detection
dedup = client.tasks.find_duplicates("Research options", threshold=0.7)

# Comments
client.tasks.add_comment("task_id", "Looks good, shipping it")
comments = client.tasks.list_comments("task_id")

Claiming (work queues)

Worker agents can pull tasks from a shared queue with claim(). Claims are atomic (no two workers get the same task) and ordered by priority, then creation time. A claim holds a lease — extend it with heartbeat() while you work, release it with release() if you can't finish, or complete() the task when done:

import time

while True:
    task = client.tasks.claim(labels=["worker"], lease_seconds=300)
    if task is None:
        time.sleep(10)  # queue empty — back off (or break)
        continue

    try:
        # ... do the work, periodically extending the lease:
        client.tasks.heartbeat(task.id, lease_seconds=300)
        # ...
        client.tasks.complete(task.id)
    except Exception:
        client.tasks.release(task.id)  # hand it back to the queue
        raise

claim() returns None when no claimable task is available. lease_seconds accepts 30-3600 (default 300); if the lease expires without a heartbeat, the task becomes claimable again. Claiming sets status to "claimed" but never touches assigned_to_agent_id. Filter claimed/unclaimed tasks with client.tasks.list(claimed=True) or claimed=False.

Session State

Report what a worker is doing on a task — without touching the claim lease — so orchestrators and dashboards can see working, waiting_input, or errored states:

client.tasks.set_state(task.id, "waiting_input", detail="Need repo credentials")

Task Context & Provenance

Each task carries a persistent context blob shared across sessions and agents. Writes merge (existing keys are preserved) and every write is recorded in an append-only provenance ledger:

# Read the current context and its version
snap = client.tasks.get_context(task.id, include_provenance=True)
print(snap.context, snap.version, snap.provenance)

# Merge keys, attributing the write and guarding against concurrent writers
client.tasks.update_context(
    task.id,
    {"decision": "use Postgres", "files": ["db.py"]},
    source="agent_observed",        # human_stated | agent_inferred | agent_observed | imported
    expected_version=snap.version,  # raises a 409 DelegaAPIError on conflict
)

# Audit who wrote what, when
history = client.tasks.context_history(task.id, key="decision")
for entry in history.entries:
    print(entry.version, entry.author_name, entry.source, entry.value)

# Mark a live entry as stale without changing the value
client.tasks.supersede_context(task.id, "decision")

Task Links

Attach repo activity or URLs to a task (the hosted GitHub integration creates these automatically from delega:#<task-id> mentions):

link = client.tasks.add_link(task.id, "pr", "42", repo="acme/webapp")
links = client.tasks.list_links(task.id)
client.tasks.delete_link(task.id, link.id)

Recurrences

Recurring task templates spawn normal task instances on a schedule. Completing a spawned task does not delete the recurrence.

recurrences = client.recurrences.list()
recurrence = client.recurrences.create(
    "Replace furnace filter",
    rule_type="monthly",
    timezone="America/Chicago",
    anchor_day=1,
    labels=["home-family"],
)
client.recurrences.update(recurrence.id, active=False)
client.recurrences.delete(recurrence.id)

Agents

agents = client.agents.list()
agent = client.agents.create("deploy-bot", display_name="Deploy Bot")
print(agent.api_key)  # Only available at creation time

# Role presets (admin key required): worker (own-task scope, default),
# coordinator (sees + can comment on all account tasks), admin
scrum = client.agents.create("scrum-bot", role="coordinator")
agent = client.agents.set_role(agent.id, "coordinator")
print(agent.role)

client.agents.update(agent.id, description="Handles deployments")
result = client.agents.rotate_key(agent.id)
print(result["api_key"])

client.agents.delete(agent.id)

Projects

projects = client.projects.list()
project = client.projects.create("Backend", emoji="⚙️", color="#3498db")

Webhooks

webhooks = client.webhooks.list()
webhook = client.webhooks.create(
    "https://example.com/webhook",
    events=["task.created", "task.completed"],
    secret="whsec_...",
)
client.webhooks.delete(webhook["id"])

Verify incoming webhook signatures with the raw payload bytes and the X-Delega-Signature / X-Delega-Timestamp header values:

from delega import verify_webhook

verify_webhook(payload, signature, timestamp, "whsec_...")

Account

me = client.me()       # Get authenticated agent info
usage = client.usage()  # Get API usage stats

usage() is only available on the hosted API (api.delega.dev) and raises DelegaError before making a request when the client is pointed at a custom /api namespace. Custom /api-namespace endpoints expose task, recurrence, agent, project, and webhook APIs; me() depends on whether that endpoint is implemented by the target API.

Async Client

from delega import AsyncDelega

async with AsyncDelega(api_key="dlg_...") as client:
    tasks = await client.tasks.list()
    task = await client.tasks.create("Async task")
    await client.tasks.complete(task.id)

The async client has the same interface as the sync client, but all methods are coroutines. Requires httpx (pip install 'delega[async]').

Error Handling

from delega import DelegaError, DelegaAPIError, DelegaAuthError, DelegaNotFoundError, DelegaRateLimitError

try:
    task = client.tasks.get("nonexistent")
except DelegaNotFoundError:
    print("Task not found")
except DelegaAuthError:
    print("Invalid API key")
except DelegaRateLimitError:
    print("Too many requests")
except DelegaAPIError as e:
    print(f"API error {e.status_code}: {e.error_message}")
except DelegaError as e:
    print(f"SDK error: {e}")

Models

All resource methods return typed dataclasses:

• Task - id, content, description, priority, labels, due_date, completed, project_id, parent_id, parent_task_id, root_task_id, delegation_depth, status, assigned_to_agent_id, created_by_agent_id, completed_by_agent_id, claimed_by_agent_id, claimed_at, lease_expires_at, session_state, session_state_detail, accountable_agent_id, context, context_version, created_at, updated_at

The claiming fields (claimed_by_agent_id, claimed_at, lease_expires_at) are set while a task is claimed via tasks.claim() and are None otherwise. A claimed task has status == "claimed".

• Recurrence - id, content, description, project_id, priority, labels, assigned_to_agent_id, rule_type, interval, timezone, anchor_day, anchor_month, anchor_weekday, next_due_at, last_spawned_at, active, skip_if_open, created_by_agent_id, created_at, updated_at
• DelegationChain - root_id, chain, depth, completed_count, total_count
• TaskLink - id, task_id, kind (branch/commit/pr/url), repo, ref, url, created_by_agent_id, created_at
• ContextSnapshot - context, version, provenance (from tasks.get_context())
• ContextEntry / ContextHistory - the provenance ledger (from tasks.context_history() / tasks.supersede_context())
• DedupResult / DuplicateMatch - duplicate-detection result and matches from tasks.find_duplicates()
• Comment - id, task_id, content, created_at
• Agent - id, name, display_name, description, role, api_key, created_at, updated_at

The api_key field is returned on agent creation and key rotation responses, but it is hidden from the default dataclass repr() to reduce accidental secret leakage in logs.

• Project - id, name, emoji, color, created_at, updated_at

Development

The CI workflow installs and tests the package with:

pip install -e ".[async]"
pip install pytest pytest-asyncio
pytest tests/ -v

License

MIT