​

marvin.tasks.task

Tasks for Marvin.

A Task is a container for a prompt and its associated state.

​

Constants

​

NOTSET

NOTSET = "'__NOTSET__'"

​

T

T = TypeVar('T')

​

Classes

​

Task

class Task(instructions: str | Sequence[UserContent], attachments: Sequence[UserContent] | None = None, result_type: ResultType[T] = NOTSET, name: str | None = None, prompt_template: str | Path = Path('task.jinja'), agents: list[Actor | Team] | None = None, context: dict[str, Any] | None = None, tools: list[Callable[..., Any]] | None = None, memories: list[Memory] | None = None, result_validator: Callable[..., Any] | None = None, parent: Task[Any] | None | Literal['__NOTSET__'] = NOTSET, depends_on: Sequence[Task[Any]] | None = None, allow_fail: bool = False, allow_skip: bool = False, verbose: bool = False, cli: bool = False, plan: bool = False)

A task is a container for a prompt and its associated state.

Methods:

• friendly_name

  def friendly_name(self, verbose: bool = True) -> str
  

  Get a friendly name for this task.

• get_actor

  def get_actor(self) -> Actor
  

  Retrieve the actor assigned to this task.

• get_end_turn_tools

  def get_end_turn_tools(self) -> list[type[EndTurn]]
  

  Get the result tool for this task.

• get_prompt

  def get_prompt(self) -> str | Sequence[UserContent]
  

  Get the rendered prompt for this task.

  Uses the task’s prompt_template (or default if None) and renders it with this task instance as the task variable.

• get_result_type

  def get_result_type(self) -> type[T]
  

  Get the effective result type for this task. For classification tasks, returns the type that should be used for validation (e.g., int or list[int]).

• get_result_type_str

  def get_result_type_str(self) -> str
  

  Get a string representation of the result type.

• get_tools

  def get_tools(self) -> list[Callable[..., Any]]
  

  Get the tools assigned to this task.

• is_classifier

  def is_classifier(self) -> bool
  

  Return True if this task is a classification task.

• is_complete

  def is_complete(self) -> bool
  

  Check if the task is complete.

• is_failed

  def is_failed(self) -> bool
  

  Check if the task is failed.

• is_incomplete

  def is_incomplete(self) -> bool
  

  Check if the task is incomplete.

• is_pending

  def is_pending(self) -> bool
  

  Check if the task is pending.

• is_ready

  def is_ready(self) -> bool
  

  Check if the task is ready to run.

  A task is ready if it is incomplete and all of its dependencies (including subtasks) are complete.

• is_running

  def is_running(self) -> bool
  

  Check if the task is running.

• is_skipped

  def is_skipped(self) -> bool
  

  Check if the task is skipped.

• is_successful

  def is_successful(self) -> bool
  

  Check if the task is successful.

• mark_failed

  def mark_failed(self, error: str, thread: Thread | None = None)
  

  Mark the task as failed with an error message.

• mark_failed_tool

  def mark_failed_tool(self) -> type[marvin.engine.end_turn.MarkTaskFailed]
  

• mark_running

  def mark_running(self, thread: Thread | None = None)
  

  Mark the task as running.

• mark_skipped

  def mark_skipped(self, thread: Thread | None = None)
  

  Mark the task as skipped.

• mark_skipped_tool

  def mark_skipped_tool(self) -> type[marvin.engine.end_turn.MarkTaskSkipped]
  

• mark_successful

  def mark_successful(self, result: T = None, validate_result: bool = True, thread: Thread | None = None)
  

  Mark the task as successful with an optional result.

• mark_successful_tool

  def mark_successful_tool(self) -> type[marvin.engine.end_turn.MarkTaskSuccessful]
  

• run

  def run(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> T
  

• run_async

  def run_async(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> T
  

• run_stream

  def run_stream(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> AsyncGenerator[Event, None]
  

• validate_result

  def validate_result(self, raw_result: Any) -> T
  

  Validate a result against the expected type and custom validator.

​

TaskState

class TaskState()

State of a task.

​

Functions

​

get_type_adapter

def get_type_adapter(result_type: type[T]) -> TypeAdapter[T]

Parent Module: tasks