feat: 1.2.0 multi-agent coordination methods + usage() fix → 0.2.0 (#6)

Brings the Python SDK in line with @delega-dev/mcp 1.2.0. Agents using
this SDK can now delegate, inspect chains, pass shared context, and
dedup — the surfaces that turn Delega from a task tracker into an
actual coordination layer.

New methods (sync + async):
- tasks.assign(task_id, agent_id) — PUT /tasks/:id with
  assigned_to_agent_id (None to unassign). For multi-agent handoffs
  use delegate() instead — assign() doesn't record a chain.
- tasks.chain(task_id) — GET /tasks/:id/chain. Returns
  DelegationChain dataclass. Client normalizes hosted {root_id}
  vs self-hosted {root: Task} shape divergence so root_id is
  always populated.
- tasks.update_context(task_id, context) — PATCH /tasks/:id/context.
  Deep-merges keys (not replace). Normalizes hosted bare-dict vs
  self-hosted full-task response; always returns the merged dict.
- tasks.find_duplicates(content, threshold=None) — POST /tasks/dedup.
  Returns DedupResult with DuplicateMatch entries.

Existing method updates:
- tasks.delegate() gains project_id / labels / due_date /
  assigned_to_agent_id kwargs matching the delega-mcp tool signature.
  Existing callers unaffected (all additions are optional kwargs).
- Delega.usage() BUG FIX: was hitting /stats, now correctly /usage.
  Raises DelegaError before any HTTP call on self-hosted backends
  (which don't expose /usage — mirrors the delega-mcp client gate).
  Async usage() also gets the gate.

Model updates:
- Task now surfaces parent_task_id, root_task_id, delegation_depth,
  status, assigned_to_agent_id, created_by_agent_id,
  completed_by_agent_id, context. from_dict() auto-parses the
  JSON-encoded context string that hosted returns (D1/SQLite text
  column) vs the dict self-hosted returns (SQLAlchemy JSON).
- New: DelegationChain, DedupResult, DuplicateMatch dataclasses.
  All exposed from the package root.

Test updates:
- 10 new unit tests covering all new methods and both shape branches
  of chain() and update_context(). Existing test_usage split into
  _hosted + _self_hosted_raises_before_fetch (asserts mock never
  called).
- HTTPClient.path_prefix property added to both sync and async
  transports so the usage gate can inspect the namespace.

Version: 0.1.3 → 0.2.0 (minor — new methods, usage() endpoint
change is a semantic fix not a signature change).

63/63 pytest tests passing.

Out of scope (follow-up): async test coverage for the new methods.
The async paths are line-by-line mirrors of sync; a later PR can
add async-specific tests.

Co-authored-by: Delega Bot <hello@delega.dev>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "delega"
-version = "0.1.3"
+version = "0.2.0"
 description = "Official Python SDK for the Delega API"
 readme = "README.md"
 license = "MIT"

src/delega/__init__.py

@@ -9,19 +9,30 @@
     DelegaNotFoundError,
     DelegaRateLimitError,
 )
-from .models import Agent, Comment, Project, Task
+from .models import (
+    Agent,
+    Comment,
+    DedupResult,
+    DelegationChain,
+    DuplicateMatch,
+    Project,
+    Task,
+)
 from .webhooks import verify_webhook
 
 __all__ = [
     "Agent",
     "AsyncDelega",
     "Comment",
+    "DedupResult",
     "Delega",
     "DelegaAPIError",
     "DelegaAuthError",
     "DelegaError",
     "DelegaNotFoundError",
     "DelegaRateLimitError",
+    "DelegationChain",
+    "DuplicateMatch",
     "Project",
     "Task",
     "verify_webhook",

src/delega/_http.py

@@ -60,6 +60,11 @@ def __init__(self, base_url: str, api_key: str, timeout: int = _DEFAULT_TIMEOUT)
         self._api_key = api_key
         self._timeout = timeout
 
+    @property
+    def path_prefix(self) -> str:
+        """Return the API namespace path ("/v1" for hosted, "/api" for self-hosted)."""
+        return urllib.parse.urlparse(self._base_url).path or ""
+
     def _headers(self) -> dict[str, str]:
         return {
             "X-Agent-Key": self._api_key,

src/delega/_version.py

@@ -1,4 +1,4 @@
 """Package version metadata."""
 
-__version__ = "0.1.3"
+__version__ = "0.2.0"
 USER_AGENT = f"delega-python/{__version__}"

src/delega/async_client.py

@@ -13,7 +13,7 @@
     DelegaNotFoundError,
     DelegaRateLimitError,
 )
-from .models import Agent, Comment, Project, Task
+from .models import Agent, Comment, DedupResult, DelegationChain, Project, Task
 from ._version import USER_AGENT
 
 _DEFAULT_BASE_URL = "https://api.delega.dev"
@@ -49,6 +49,12 @@ def __init__(self, base_url: str, api_key: str, timeout: int = 30) -> None:
             timeout=timeout,
         )
 
+    @property
+    def path_prefix(self) -> str:
+        """Return the API namespace path ("/v1" for hosted, "/api" for self-hosted)."""
+        import urllib.parse
+        return urllib.parse.urlparse(self._base_url).path or ""
+
     async def request(
         self,
         method: str,
@@ -201,16 +207,74 @@ async def delegate(
         *,
         description: Optional[str] = None,
         priority: Optional[int] = None,
+        project_id: Optional[str] = None,
+        labels: Optional[list[str]] = None,
+        due_date: Optional[str] = None,
+        assigned_to_agent_id: Optional[str] = None,
     ) -> Task:
-        """Create a delegated sub-task under a parent task."""
+        """Create a delegated child task under a parent.
+
+        The parent's ``status`` flips to ``"delegated"``. Use this — not
+        ``assign()`` — for multi-agent handoffs so the parent/child
+        accountability chain is recorded.
+        """
         body: dict[str, Any] = {"content": content}
         if description is not None:
             body["description"] = description
         if priority is not None:
             body["priority"] = priority
+        if project_id is not None:
+            body["project_id"] = project_id
+        if labels is not None:
+            body["labels"] = labels
+        if due_date is not None:
+            body["due_date"] = due_date
+        if assigned_to_agent_id is not None:
+            body["assigned_to_agent_id"] = assigned_to_agent_id
         data = await self._http.post(f"/tasks/{parent_task_id}/delegate", body=body)
         return Task.from_dict(data)
 
+    async def assign(self, task_id: str, agent_id: Optional[str]) -> Task:
+        """Assign a task to an agent (or ``None`` to unassign)."""
+        data = await self._http.put(
+            f"/tasks/{task_id}", body={"assigned_to_agent_id": agent_id}
+        )
+        return Task.from_dict(data)
+
+    async def chain(self, task_id: str) -> DelegationChain:
+        """Get the full parent/child delegation chain for a task."""
+        data = await self._http.get(f"/tasks/{task_id}/chain")
+        return DelegationChain.from_dict(data)
+
+    async def update_context(
+        self, task_id: str, context: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Deep-merge keys into a task's persistent context blob.
+
+        Existing keys are preserved; supplied keys are added or overwritten.
+        """
+        data = await self._http.patch(f"/tasks/{task_id}/context", body=context)
+        if isinstance(data, dict) and "content" in data and "id" in data:
+            raw_ctx = data.get("context") or {}
+            if isinstance(raw_ctx, str):
+                import json as _json
+                try:
+                    raw_ctx = _json.loads(raw_ctx) if raw_ctx.strip() else {}
+                except Exception:
+                    raw_ctx = {}
+            return raw_ctx if isinstance(raw_ctx, dict) else {}
+        return data if isinstance(data, dict) else {}
+
+    async def find_duplicates(
+        self, content: str, *, threshold: Optional[float] = None
+    ) -> DedupResult:
+        """Check whether content is similar to existing open tasks."""
+        body: dict[str, Any] = {"content": content}
+        if threshold is not None:
+            body["threshold"] = threshold
+        data = await self._http.post("/tasks/dedup", body=body)
+        return DedupResult.from_dict(data)
+
     async def add_comment(self, task_id: str, content: str) -> Comment:
         """Add a comment to a task."""
         data = await self._http.post(f"/tasks/{task_id}/comments", body={"content": content})
@@ -368,7 +432,17 @@ async def me(self) -> dict[str, Any]:
         return await self._http.get("/agent/me")  # type: ignore[no-any-return]
 
     async def usage(self) -> dict[str, Any]:
-        """Get API usage information."""
+        """Get quota and rate-limit information for the current plan.
+
+        Hosted API only (``api.delega.dev``). Self-hosted deployments
+        will raise :class:`DelegaError` before making a request.
+        """
+        if self._http.path_prefix != "/v1":
+            raise DelegaError(
+                "usage() is only available on the hosted Delega API "
+                "(api.delega.dev). Self-hosted deployments do not expose "
+                "a usage endpoint."
+            )
         return await self._http.get("/usage")  # type: ignore[no-any-return]
 
     async def aclose(self) -> None:

src/delega/client.py

@@ -7,7 +7,14 @@
 
 from ._http import HTTPClient
 from .exceptions import DelegaError
-from .models import Agent, Comment, Project, Task
+from .models import (
+    Agent,
+    Comment,
+    DedupResult,
+    DelegationChain,
+    Project,
+    Task,
+)
 
 _DEFAULT_BASE_URL = "https://api.delega.dev"
 
@@ -153,23 +160,121 @@ def delegate(
         *,
         description: Optional[str] = None,
         priority: Optional[int] = None,
+        project_id: Optional[str] = None,
+        labels: Optional[list[str]] = None,
+        due_date: Optional[str] = None,
+        assigned_to_agent_id: Optional[str] = None,
     ) -> Task:
-        """Create a delegated sub-task under a parent task.
+        """Create a delegated child task under a parent.
+
+        The parent's ``status`` flips to ``"delegated"``. Use this — not
+        ``assign()`` — for multi-agent handoffs so the parent/child
+        accountability chain is recorded (inspectable via ``chain()``).
 
         Args:
             parent_task_id: The parent task identifier.
-            content: The sub-task content/title.
+            content: The child task content/title.
             description: Optional longer description.
-            priority: Optional priority level.
+            priority: Optional priority level (1-4).
+            project_id: Optional project to attach the child to.
+            labels: Optional labels for the child.
+            due_date: Optional due date (YYYY-MM-DD).
+            assigned_to_agent_id: Optional agent to assign the child to.
         """
         body: dict[str, Any] = {"content": content}
         if description is not None:
             body["description"] = description
         if priority is not None:
             body["priority"] = priority
+        if project_id is not None:
+            body["project_id"] = project_id
+        if labels is not None:
+            body["labels"] = labels
+        if due_date is not None:
+            body["due_date"] = due_date
+        if assigned_to_agent_id is not None:
+            body["assigned_to_agent_id"] = assigned_to_agent_id
         data = self._http.post(f"/tasks/{parent_task_id}/delegate", body=body)
         return Task.from_dict(data)
 
+    def assign(self, task_id: str, agent_id: Optional[str]) -> Task:
+        """Assign a task to an agent (or ``None`` to unassign).
+
+        For multi-agent handoffs where you want the parent/child chain
+        recorded, use ``delegate()`` instead — ``assign()`` only updates
+        the assignee on an existing task.
+
+        Args:
+            task_id: The task identifier.
+            agent_id: The agent identifier, or ``None`` to unassign.
+        """
+        data = self._http.put(
+            f"/tasks/{task_id}", body={"assigned_to_agent_id": agent_id}
+        )
+        return Task.from_dict(data)
+
+    def chain(self, task_id: str) -> DelegationChain:
+        """Get the full parent/child delegation chain for a task.
+
+        Normalizes hosted (``{root_id}``) vs self-hosted (``{root: Task}``)
+        response shapes so ``DelegationChain.root_id`` is always populated.
+
+        Args:
+            task_id: Any task identifier in the chain.
+        """
+        data = self._http.get(f"/tasks/{task_id}/chain")
+        return DelegationChain.from_dict(data)
+
+    def update_context(
+        self, task_id: str, context: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Deep-merge keys into a task's persistent context blob.
+
+        Existing keys are preserved; supplied keys are added or overwritten.
+        Use this to pass shared state between delegated agents instead of
+        re-describing context in task descriptions.
+
+        Args:
+            task_id: The task identifier.
+            context: Keys to merge into the existing context.
+
+        Returns:
+            The merged context dict.
+        """
+        # Hosted returns the bare merged context dict; self-hosted returns
+        # the full Task with a ``context`` field. Normalize to always
+        # return the merged context.
+        data = self._http.patch(f"/tasks/{task_id}/context", body=context)
+        if isinstance(data, dict) and "content" in data and "id" in data:
+            # Looks like a full Task.
+            raw_ctx = data.get("context") or {}
+            if isinstance(raw_ctx, str):
+                import json as _json
+                try:
+                    raw_ctx = _json.loads(raw_ctx) if raw_ctx.strip() else {}
+                except Exception:
+                    raw_ctx = {}
+            return raw_ctx if isinstance(raw_ctx, dict) else {}
+        return data if isinstance(data, dict) else {}
+
+    def find_duplicates(
+        self, content: str, *, threshold: Optional[float] = None
+    ) -> DedupResult:
+        """Check whether content is similar to existing open tasks.
+
+        Call before ``create()`` to avoid redundant work. Uses Jaccard
+        similarity against open tasks.
+
+        Args:
+            content: The proposed task content to check.
+            threshold: Similarity threshold 0-1 (default 0.6 server-side).
+        """
+        body: dict[str, Any] = {"content": content}
+        if threshold is not None:
+            body["threshold"] = threshold
+        data = self._http.post("/tasks/dedup", body=body)
+        return DedupResult.from_dict(data)
+
     def add_comment(self, task_id: str, content: str) -> Comment:
         """Add a comment to a task.
 
@@ -386,9 +491,21 @@ def me(self) -> dict[str, Any]:
         return self._http.get("/agent/me")  # type: ignore[no-any-return]
 
     def usage(self) -> dict[str, Any]:
-        """Get API usage information.
+        """Get quota and rate-limit information for the current plan.
+
+        Hosted API only (``api.delega.dev``). Self-hosted deployments
+        will raise :class:`DelegaError` before making a request.
 
         Returns:
-            Dictionary with usage statistics.
+            Dict with ``plan``, ``task_count_month``, ``task_limit``,
+            ``reset_date``, ``agent_count``, ``agent_limit``,
+            ``webhook_count``, ``webhook_limit``, ``project_count``,
+            ``project_limit``, ``rate_limit_rpm``, ``max_content_chars``.
         """
-        return self._http.get("/stats")  # type: ignore[no-any-return]
+        if self._http.path_prefix != "/v1":
+            raise DelegaError(
+                "usage() is only available on the hosted Delega API "
+                "(api.delega.dev). Self-hosted deployments do not expose "
+                "a usage endpoint."
+            )
+        return self._http.get("/usage")  # type: ignore[no-any-return]