Tasks¶

TasksClient¶

Accessed via client.tasks.

pyawe.client.TasksClient ¶

TasksClient(http: _HttpSession)

Methods for the /tasks resource.

list ¶

list(workflow_id: Optional[Union[UUID, str]] = None) -> List[Task]

List tasks, optionally filtered by workflow.

Parameters:

Name	Type	Description	Default
`workflow_id`	`Optional[Union[UUID, str]]`	When supplied, returns only tasks within this workflow.	`None`

Returns:

Type	Description
`List[Task]`	List of :class:`~pyawe.models.Task` objects.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

list_mine ¶

list_mine(role_ids: Optional[List[Union[UUID, str]]] = None) -> List[TaskWithContext]

Return tasks assigned to the authenticated user or their team roles.

Parameters:

Name	Type	Description	Default
`role_ids`	`Optional[List[Union[UUID, str]]]`	Team role UUIDs to include alongside direct assignments.	`None`

Returns:

Type	Description
`List[TaskWithContext]`	List of :class:`~pyawe.models.TaskWithContext` objects, each enriched
`List[TaskWithContext]`	with workflow and job names.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

create ¶

create(name: str, workflow_id: Union[UUID, str], *, is_template: Optional[bool] = None, description: Optional[str] = None, status: Optional[Union[Status, str]] = None, schedule_status: Optional[Union[ScheduleStatus, str]] = None, rework_task_id: Optional[Union[UUID, str]] = None, is_start: Optional[bool] = None, is_end: Optional[bool] = None, task_type: Optional[Union[TaskType, str]] = None, effort: Optional[int] = None) -> Task

Create a task within a workflow.

Parameters:

Name	Type	Description	Default
`name`	`str`	Task name.	required
`workflow_id`	`Union[UUID, str]`	UUID of the containing workflow.	required
`is_template`	`Optional[bool]`	Mark as a template task.	`None`
`description`	`Optional[str]`	Optional description.	`None`
`status`	`Optional[Union[Status, str]]`	Initial status (default `"Not Started"`).	`None`
`schedule_status`	`Optional[Union[ScheduleStatus, str]]`	Initial schedule status (default `"N/A"`).	`None`
`rework_task_id`	`Optional[Union[UUID, str]]`	UUID of the task this one reworks, if any.	`None`
`is_start`	`Optional[bool]`	Designate as the workflow start task.	`None`
`is_end`	`Optional[bool]`	Designate as the workflow end task.	`None`
`task_type`	`Optional[Union[TaskType, str]]`	`"standard"` (default), `"decision"`, `"automated"`, or `"loop_block"`.	`None`
`effort`	`Optional[int]`	Unitless effort estimate (e.g. story points).	`None`

Returns:

Type	Description
`Task`	The newly created :class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweAuthError`	If not authenticated.
`AweValidationError`	On a 400 response.

get ¶

get(task_id: Union[UUID, str]) -> Task

Fetch a single task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`Task`	class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

update ¶

update(task_id: Union[UUID, str], *, name: Optional[str] = None, is_template: Optional[bool] = None, description: Optional[str] = None, status: Optional[Union[Status, str]] = None, schedule_status: Optional[Union[ScheduleStatus, str]] = None, rework_task_id: Optional[Union[UUID, str]] = None, is_start: Optional[bool] = None, is_end: Optional[bool] = None, task_type: Optional[Union[TaskType, str]] = None, assigned_to: Any = _UNSET, decision_input_port: Any = _UNSET, is_locked: Optional[bool] = None, canvas_x: Optional[float] = None, canvas_y: Optional[float] = None, effort: Any = _UNSET) -> Task

Update a task; only supplied fields are changed.

assigned_to, decision_input_port, and effort are tri-state fields:

Omitted (default): field is left unchanged.
None: clears the current value.
A value: sets the field.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task to update.	required
`name`	`Optional[str]`	New name.	`None`
`is_template`	`Optional[bool]`	Template flag.	`None`
`description`	`Optional[str]`	New description.	`None`
`status`	`Optional[Union[Status, str]]`	New status. Completed/cancelled tasks cannot be transitioned. Setting to `"In Progress"` stamps `start_time`; setting to `"Complete"` stamps `end_time`.	`None`
`schedule_status`	`Optional[Union[ScheduleStatus, str]]`	New schedule status.	`None`
`rework_task_id`	`Optional[Union[UUID, str]]`	Rework back-link.	`None`
`is_start`	`Optional[bool]`	Start task flag.	`None`
`is_end`	`Optional[bool]`	End task flag.	`None`
`task_type`	`Optional[Union[TaskType, str]]`	Task type string.	`None`
`assigned_to`	`Any`	User ID string, `None` to clear, or omit to leave unchanged.	`_UNSET`
`decision_input_port`	`Any`	Input port name for auto-decide, `None` to clear, or omit to leave unchanged.	`_UNSET`
`is_locked`	`Optional[bool]`	Lock or unlock structural edits (requires owner/leader/admin).	`None`
`canvas_x`	`Optional[float]`	Canvas X position from the workflow editor.	`None`
`canvas_y`	`Optional[float]`	Canvas Y position from the workflow editor.	`None`
`effort`	`Any`	Unitless effort estimate, `None` to clear, or omit to leave unchanged.	`_UNSET`

Returns:

Type	Description
`Task`	The updated :class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweValidationError`	If the status transition is invalid.
`AweAuthError`	If not authenticated.

delete ¶

delete(task_id: Union[UUID, str]) -> None

Delete a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task to delete.	required

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

decide ¶

decide(task_id: Union[UUID, str], branch_label: str) -> Task

Resolve a decision task by choosing an outgoing branch.

Sets the task status to "Complete", records decision_outcome, activates the chosen successor, and cancels tasks on rejected branches.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of a decision task.	required
`branch_label`	`str`	Label of the branch to follow. Must match an outgoing link's `branch_label`.	required

Returns:

Type	Description
`Task`	The decided :class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweValidationError`	If the task is not a decision task, is already complete, or the branch label is not found.
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

clear_rework ¶

clear_rework(task_id: Union[UUID, str]) -> Task

Remove the rework back-link from a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`Task`	The updated :class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

TaskLinksClient¶

Accessed via client.task_links.

pyawe.client.TaskLinksClient ¶

TaskLinksClient(http: _HttpSession)

Methods for task link management (/task-links and /tasks/{id}/…).

create ¶

create(from_task_id: Union[UUID, str], to_task_id: Union[UUID, str], *, branch_label: Optional[str] = None) -> TaskLink

Create a directed link between two tasks.

Parameters:

Name	Type	Description	Default
`from_task_id`	`Union[UUID, str]`	UUID of the upstream task.	required
`to_task_id`	`Union[UUID, str]`	UUID of the downstream task.	required
`branch_label`	`Optional[str]`	Required when the upstream task is a decision task.	`None`

Returns:

Type	Description
`TaskLink`	The created :class:`~pyawe.models.TaskLink`.

Raises:

Type	Description
`AweAuthError`	If not authenticated.
`AweValidationError`	On a 400 response.

delete ¶

delete(from_task_id: Union[UUID, str], to_task_id: Union[UUID, str]) -> None

Delete the link between two tasks.

Parameters:

Name	Type	Description	Default
`from_task_id`	`Union[UUID, str]`	UUID of the upstream task.	required
`to_task_id`	`Union[UUID, str]`	UUID of the downstream task.	required

Raises:

Type	Description
`AweNotFoundError`	If the link does not exist.
`AweAuthError`	If not authenticated.

get_next ¶

get_next(task_id: Union[UUID, str]) -> List[Task]

Return the immediate successor tasks of task_id.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[Task]`	List of :class:`~pyawe.models.Task` objects.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

get_previous ¶

get_previous(task_id: Union[UUID, str]) -> List[Task]

Return the immediate predecessor tasks of task_id.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[Task]`	List of :class:`~pyawe.models.Task` objects.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

get_outgoing ¶

get_outgoing(task_id: Union[UUID, str]) -> List[TaskLink]

Return all outgoing links from task_id.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[TaskLink]`	List of :class:`~pyawe.models.TaskLink` objects.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

update_bindings ¶

update_bindings(from_task_id: Union[UUID, str], to_task_id: Union[UUID, str], bindings: List[DataBinding]) -> TaskLink

Replace the data bindings on a task link edge.

Bindings are applied by the propagator when the upstream task completes, copying output values to the downstream task's inputs.

Parameters:

Name	Type	Description	Default
`from_task_id`	`Union[UUID, str]`	UUID of the upstream task.	required
`to_task_id`	`Union[UUID, str]`	UUID of the downstream task.	required
`bindings`	`List[DataBinding]`	New list of :class:`~pyawe.models.DataBinding` objects.	required

Returns:

Type	Description
`TaskLink`	The updated :class:`~pyawe.models.TaskLink`.

Raises:

Type	Description
`AweNotFoundError`	If the link does not exist.
`AweAuthError`	If not authenticated.

TaskPortsClient¶

Accessed via client.task_ports.

pyawe.client.TaskPortsClient ¶

TaskPortsClient(http: _HttpSession)

Methods for task port spec and value management (/tasks/{id}/ports, etc.).

list_specs ¶

list_specs(task_id: Union[UUID, str]) -> List[TaskPortSpec]

Return all port specs (inputs and outputs) for a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[TaskPortSpec]`	List of :class:`~pyawe.models.TaskPortSpec` objects.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

create_spec ¶

create_spec(task_id: Union[UUID, str], direction: str, name: str, value_type: str, *, required: Optional[bool] = None, description: Optional[str] = None, default_value: Optional[Any] = None, sort_order: Optional[int] = None) -> TaskPortSpec

Add a port spec to a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`direction`	`str`	`"input"` or `"output"`.	required
`name`	`str`	Port name (unique within direction on the task).	required
`value_type`	`str`	`"string"`, `"number"`, `"boolean"`, `"json"`, `"file"`, or `"dam_asset"`.	required
`required`	`Optional[bool]`	Whether this input is required (default `False`; ignored for outputs).	`None`
`description`	`Optional[str]`	Human-readable description.	`None`
`default_value`	`Optional[Any]`	JSON default applied at instantiation for optional inputs.	`None`
`sort_order`	`Optional[int]`	Display order.	`None`

Returns:

Type	Description
`TaskPortSpec`	The created :class:`~pyawe.models.TaskPortSpec`.

Raises:

Type	Description
`AweAuthError`	If not authenticated.
`AweValidationError`	On a 400 response.

update_spec ¶

update_spec(task_id: Union[UUID, str], port_id: Union[UUID, str], *, name: Optional[str] = None, value_type: Optional[str] = None, required: Optional[bool] = None, description: Optional[str] = None, default_value: Optional[Any] = None, sort_order: Optional[int] = None) -> TaskPortSpec

Update a port spec; only supplied fields are changed.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`port_id`	`Union[UUID, str]`	UUID of the port spec.	required
`name`	`Optional[str]`	New port name.	`None`
`value_type`	`Optional[str]`	New value type.	`None`
`required`	`Optional[bool]`	Required flag.	`None`
`description`	`Optional[str]`	New description.	`None`
`default_value`	`Optional[Any]`	New default value.	`None`
`sort_order`	`Optional[int]`	New sort order.	`None`

Returns:

Type	Description
`TaskPortSpec`	The updated :class:`~pyawe.models.TaskPortSpec`.

Raises:

Type	Description
`AweNotFoundError`	If the port spec does not exist.
`AweAuthError`	If not authenticated.

delete_spec ¶

delete_spec(task_id: Union[UUID, str], port_id: Union[UUID, str]) -> None

Delete a port spec.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`port_id`	`Union[UUID, str]`	UUID of the port spec.	required

Raises:

Type	Description
`AweNotFoundError`	If the port spec does not exist.
`AweAuthError`	If not authenticated.

get_inputs ¶

get_inputs(task_id: Union[UUID, str]) -> TaskPortValues

Return input port specs and current values for a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`TaskPortValues`	class:`~pyawe.models.TaskPortValues` with `direction == "input"` specs.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

patch_inputs ¶

patch_inputs(task_id: Union[UUID, str], values: Dict[str, Any]) -> TaskPortValues

Partially update the input values of a task.

Only the keys present in values are written; others are left unchanged.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`values`	`Dict[str, Any]`	Mapping of `{port_name: value}` pairs to set.	required

Returns:

Name	Type	Description
`Updated`	`TaskPortValues`	class:`~pyawe.models.TaskPortValues`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

get_outputs ¶

get_outputs(task_id: Union[UUID, str]) -> TaskPortValues

Return output port specs and current values for a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`TaskPortValues`	class:`~pyawe.models.TaskPortValues` with `direction == "output"` specs.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

patch_outputs ¶

patch_outputs(task_id: Union[UUID, str], values: Dict[str, Any]) -> Task

Partially update the output values of a task (runner use).

Only the keys present in values are written; others are left unchanged. Typically called by the automated runner after a task completes.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`values`	`Dict[str, Any]`	Mapping of `{port_name: value}` pairs to set.	required

Returns:

Name	Type	Description
`Updated`	`Task`	class:`~pyawe.models.Task`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

TaskScriptsClient¶

Accessed via client.task_scripts.

pyawe.client.TaskScriptsClient ¶

TaskScriptsClient(http: _HttpSession)

Methods for automated task script management (/tasks/{id}/script).

get ¶

get(task_id: Union[UUID, str]) -> TaskScript

Fetch the script attached to a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`TaskScript`	class:`~pyawe.models.TaskScript`.

Raises:

Type	Description
`AweNotFoundError`	If no script is attached.
`AweAuthError`	If not authenticated.

upsert ¶

upsert(task_id: Union[UUID, str], *, script_type: Optional[str] = None, endpoint: Optional[str] = None, script_body: Optional[str] = None, timeout_secs: Optional[int] = None, retry_limit: Optional[int] = None, execution_profile_id: Optional[Union[UUID, str]] = None) -> TaskScript

Create or fully replace the script on a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`script_type`	`Optional[str]`	`"webhook"` (default), `"shell"`, `"python"`, or `"mcp_tool"`.	`None`
`endpoint`	`Optional[str]`	Webhook URL; required when `script_type == "webhook"`.	`None`
`script_body`	`Optional[str]`	Script source; required for `"shell"`, `"python"`, and `"mcp_tool"` types.	`None`
`timeout_secs`	`Optional[int]`	Execution timeout in seconds.	`None`
`retry_limit`	`Optional[int]`	Maximum number of retry attempts.	`None`
`execution_profile_id`	`Optional[Union[UUID, str]]`	Execution profile for Kubernetes dispatch (`"shell"` and `"python"` only).	`None`

Returns:

Type	Description
`TaskScript`	The upserted :class:`~pyawe.models.TaskScript`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

delete ¶

delete(task_id: Union[UUID, str]) -> None

Remove the script from a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Raises:

Type	Description
`AweNotFoundError`	If no script is attached.
`AweAuthError`	If not authenticated.

TaskRunsClient¶

Accessed via client.task_runs.

pyawe.client.TaskRunsClient ¶

TaskRunsClient(http: _HttpSession)

Methods for automated task execution history (/tasks/{id}/runs).

list ¶

list(task_id: Union[UUID, str]) -> List[TaskRun]

Return all execution run records for a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[TaskRun]`	List of :class:`~pyawe.models.TaskRun` objects, newest first.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

create ¶

create(task_id: Union[UUID, str], outcome: str, started_at: datetime, completed_at: datetime, *, run_id: Optional[Union[UUID, str]] = None, runner_id: Optional[str] = None, input_json: Optional[Any] = None, output_json: Optional[Any] = None, error_message: Optional[str] = None) -> TaskRun

Record a completed execution run.

This endpoint is called by the runner after executing an automated task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`outcome`	`str`	`"Success"`, `"Failure"`, or `"Timeout"`.	required
`started_at`	`datetime`	When execution began.	required
`completed_at`	`datetime`	When execution ended (or timed out).	required
`run_id`	`Optional[Union[UUID, str]]`	Optional client-supplied UUID for idempotent recording.	`None`
`runner_id`	`Optional[str]`	Identifier of the runner that executed the task.	`None`
`input_json`	`Optional[Any]`	Snapshot of the dispatch payload.	`None`
`output_json`	`Optional[Any]`	Structured output produced by the task.	`None`
`error_message`	`Optional[str]`	Error details when `outcome != "Success"`.	`None`

Returns:

Type	Description
`TaskRun`	The recorded :class:`~pyawe.models.TaskRun`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

get ¶

get(task_id: Union[UUID, str], run_id: Union[UUID, str]) -> TaskRun

Fetch a single execution run record.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`run_id`	`Union[UUID, str]`	UUID of the run.	required

Returns:

Type	Description
`TaskRun`	class:`~pyawe.models.TaskRun`.

Raises:

Type	Description
`AweNotFoundError`	If the run does not exist.
`AweAuthError`	If not authenticated.

TaskTeamRolesClient¶

Accessed via client.task_team_roles.

pyawe.client.TaskTeamRolesClient ¶

TaskTeamRolesClient(http: _HttpSession)

Methods for task team-role assignment (/tasks/{id}/team-roles).

list ¶

list(task_id: Union[UUID, str]) -> List[TaskTeamRole]

Return all team-role assignments for a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required

Returns:

Type	Description
`List[TaskTeamRole]`	List of :class:`~pyawe.models.TaskTeamRole` objects.

Raises:

Type	Description
`AweAuthError`	If not authenticated.

assign ¶

assign(task_id: Union[UUID, str], team_role_id: Union[UUID, str]) -> TaskTeamRole

Assign a team role to a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`team_role_id`	`Union[UUID, str]`	UUID of the team role.	required

Returns:

Type	Description
`TaskTeamRole`	The created :class:`~pyawe.models.TaskTeamRole`.

Raises:

Type	Description
`AweNotFoundError`	If the task does not exist.
`AweAuthError`	If not authenticated.

remove ¶

remove(task_id: Union[UUID, str], team_role_id: Union[UUID, str]) -> None

Remove a team-role assignment from a task.

Parameters:

Name	Type	Description	Default
`task_id`	`Union[UUID, str]`	UUID of the task.	required
`team_role_id`	`Union[UUID, str]`	UUID of the team role assignment to remove.	required

Raises:

Type	Description
`AweNotFoundError`	If the assignment does not exist.
`AweAuthError`	If not authenticated.