add back-office task models and create_back_office_task: POST /back-office-tasks

src/gradient_labs/__init__.py

@@ -7,3 +7,4 @@
 from .note import *
 from .traffic_group import *
 from .voice_call_context import *
+from .back_office_task import *

src/gradient_labs/_back_office_task_create.py

@@ -0,0 +1,45 @@
+from typing import Optional, Dict, Any, List
+from datetime import datetime
+from dataclasses import dataclass
+from dataclasses_json import dataclass_json
+
+from ._http_client import HttpClient
+from .back_office_task import BackOfficeTask
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class BackOfficeTaskCreateAttachment:
+    file_name: str
+    # Provide either url (remote file) or base64_contents (inline upload), not both.
+    url: Optional[str] = None
+    base64_contents: Optional[str] = None
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class BackOfficeTaskCreateParams:
+    id: str
+    agent_id: str
+    input: Dict[str, Any]
+    created: Optional[datetime] = None
+    metadata: Optional[Dict[str, str]] = None
+    attachments: Optional[List[BackOfficeTaskCreateAttachment]] = None
+
+
+def create_back_office_task(
+    *, client: HttpClient, params: BackOfficeTaskCreateParams
+) -> BackOfficeTask:
+    body: Dict[str, Any] = {
+        "id": params.id,
+        "agent_id": params.agent_id,
+        "input": params.input,
+    }
+    if params.created is not None:
+        body["created"] = HttpClient.localize(params.created)
+    if params.metadata is not None:
+        body["metadata"] = params.metadata
+    if params.attachments is not None:
+        body["attachments"] = [a.to_dict() for a in params.attachments]
+    rsp = client.post(path="back-office-tasks", body=body)
+    return BackOfficeTask.from_dict(rsp)

src/gradient_labs/back_office_task.py

@@ -0,0 +1,68 @@
+from typing import Optional, Any, List
+from enum import Enum
+from datetime import datetime
+from dataclasses import dataclass, field
+from dataclasses_json import dataclass_json, config
+from marshmallow import fields
+
+
+class BackOfficeTaskStatus(str, Enum):
+    PENDING = "pending"
+    IN_PROGRESS = "in-progress"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    HANDED_OFF = "handed-off"
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class BackOfficeTaskResult:
+    result_type: str
+    custom: Optional[Any] = None
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class BackOfficeTaskAttachment:
+    idempotency_key: str
+    file_name: str
+    external_url: Optional[str] = None
+
+
+# Reusable marshmallow config for optional datetime fields; avoids repeating allow_none/load_default.
+_optional_datetime_config = config(
+    encoder=lambda x: x.isoformat() if x is not None else None,
+    decoder=lambda x: datetime.fromisoformat(x) if x is not None else None,
+    mm_field=fields.DateTime(format="iso", allow_none=True, load_default=None),
+)
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class BackOfficeTask:
+    id: str
+    agent_id: str
+    status: BackOfficeTaskStatus
+    input: Any
+    created: datetime = field(
+        metadata=config(
+            encoder=datetime.isoformat,
+            decoder=datetime.fromisoformat,
+            mm_field=fields.DateTime(format="iso"),
+        )
+    )
+    metadata: Optional[Any] = None
+    attachments: Optional[List[BackOfficeTaskAttachment]] = None
+    updated: Optional[datetime] = field(
+        default=None, metadata=_optional_datetime_config
+    )
+    completed: Optional[datetime] = field(
+        default=None, metadata=_optional_datetime_config
+    )
+    failed: Optional[datetime] = field(default=None, metadata=_optional_datetime_config)
+    failure_reasons: Optional[List[str]] = None
+    handed_off: Optional[datetime] = field(
+        default=None, metadata=_optional_datetime_config
+    )
+    hand_off_reason: Optional[str] = None
+    result: Optional[BackOfficeTaskResult] = None

src/gradient_labs/client.py

@@ -108,6 +108,14 @@
 )
 from .voice_call_context import VoiceCallContext
 
+from .back_office_task import (
+    BackOfficeTask,
+)
+from ._back_office_task_create import (
+    create_back_office_task,
+    BackOfficeTaskCreateParams,
+)
+
 from ._http_client import HttpClient, API_BASE_URL
 from .tool import *
 from .note import Note
@@ -808,3 +816,15 @@ def read_voice_call_context(
             phone_number=phone_number,
             params=params,
         )
+
+    # ==================== Back-Office Task Operations ====================
+
+    def create_back_office_task(
+        self, *, params: BackOfficeTaskCreateParams
+    ) -> BackOfficeTask:
+        """create_back_office_task creates a new back-office task routed to a
+        configurable back-office agent."""
+        return create_back_office_task(
+            client=self.http_client,
+            params=params,
+        )