Skip to content

antikythera.models.blueprints ¤

Classes¤

Blueprint ¤

Blueprint(
    id: str,
    name: str,
    version: str | None = "1.0",
    description: str | None = None,
    tasks: list[Task] = None,
    **kwargs
)

Represents a complete blueprint.

Attributes:

  • id (str) –

    Unique identifier for the blueprint.

  • name (str) –

    A human-readable name for the blueprint.

  • version (str) –

    The version of the blueprint schema.

  • description ((str, optional)) –

    A human-readable description of the blueprint.

  • tasks ((list[Task], optional)) –

    A list of tasks that make up the blueprint.

  • scopes (list[Scope]) –

    Scope definitions derived from scope_start/scope_end task pairs.

Returned by: Used by:

Methods:¤

check_dataflow ¤
check_dataflow() -> list[str]

Report condition expressions that read names no task in this blueprint produces.

Conditions are evaluated at runtime against session data, which is populated exclusively from declared task outputs. A condition reading a name that no task declares raises NameError at runtime, long after the blueprint was accepted.

Unlike :meth:validate this never raises; the caller decides how severe a problem is. POST /blueprints/upload rejects a blueprint that reports any, while loading one from file only logs them, so that blueprints already in storage stay loadable and repairable.

Note that an unresolved name is not provably wrong: agents may return outputs the blueprint never declared (see :meth:Task.set_output_value), and an inner blueprint additionally receives context from the composite task that invokes it. Both cases are reported here.

Returns:

  • list[str] –

    Human-readable problems, one per problematic expression.

validate ¤
validate() -> None

Validates the blueprint structure.

Raises:

BlueprintSession ¤

BlueprintSession(
    bsid: str,
    blueprint: Blueprint,
    inner_blueprints: dict[str, Blueprint] = None,
    state: BlueprintSessionState = PENDING,
    params: dict[str, str] = None,
    composite_to_inner_blueprint_map: dict[str, str] = None,
    blueprint_contexts: dict[str, Any] = None,
    scope_iterations: dict[str, int] = None,
    last_task_error: TaskError | None = None,
)

Represents a session of the execution of a blueprint.

Attributes:

  • bsid (str) –

    The ID of the blueprint session.

  • blueprint (Blueprint) –

    The blueprint.

  • inner_blueprints ((dict[str, Blueprint], optional)) –

    A dictionary of inner blueprints used in this session.

Returned by: Used by:

Methods:¤

get_context_for_blueprint ¤
get_context_for_blueprint(blueprint_id: str) -> dict | None

Returns the context for a given blueprint, if it exists.

BlueprintSessionState ¤

Enumeration of possible blueprint session states.

Used by:

Dependency ¤

Dependency(id: str, type: DependencyType = FS)

Represents a dependency of a task on another.

Attributes:

  • id (str) –

    The ID of the task this task depends on.

  • type (DependencyType) –

    The type of dependency, by default DependencyType.FS (Finish-to-Start).

Used by:

Scope ¤

Scope(
    id: str,
    label: str,
    task_ids: list[str] = None,
    end_task_id: str = "",
    policy_type: str = "skip",
    policy: dict = None,
)

A scope spanning a contiguous region of a blueprint's task DAG.

A scope is uniquely identified by the task ID of its opening (scope_start) task. An optional human-readable label can be provided via scope_start["name"].

Attributes:

  • id (str) –

    Identifier for this scope — the task ID of the scope_start task.

  • label (str) –

    Human-readable label for the scope.

  • task_ids (list[str]) –

    All task IDs that belong to this scope (including start and end).

  • end_task_id (str) –

    Task ID of the scope_end task that closes this scope.

  • policy_type (str) –

    One of "skip", "retry", or "while".

  • policy (dict) –

    Raw policy dict from task.scope_start.

Task ¤

Task(
    id: str,
    type: str,
    description: str | None = None,
    condition: str | None = None,
    inputs: list[TaskInput] = None,
    outputs: list[TaskOutput] = None,
    params: list[TaskParam] = None,
    depends_on: list[Dependency] = None,
    state: TaskState = PENDING,
    context: dict[str, Any] = None,
    scope_start: dict | None = None,
    scope_end: str | None = None,
    **kwargs
)

Represents a single task in a blueprint.

Attributes:

  • id (str) –

    Unique identifier for the task.

  • type (str) –

    The type of the task, which determines the agent that will execute it.

  • description ((str, optional)) –

    A human-readable description of the task.

  • inputs ((list[TaskInput], optional)) –

    A list of inputs for the task.

  • outputs ((list[TaskOutput], optional)) –

    A list of outputs for the task.

  • depends_on ((list[Dependency], optional)) –

    A list of dependencies on other tasks.

  • params ((list[TaskParam], optional)) –

    A list of task-specific parameters.

Returned by: Used by:

Methods:¤

from_dynamic_task classmethod ¤
from_dynamic_task(dynamic_task: Task, new_task_id: str, element_id: str) -> Task

Creates a new dynamically expanded task from a composite task.

Parameters:

  • dynamic_task (Task) –

    The original composite task to expand.

  • new_task_id (str) –

    The ID for the new expanded task.

  • element_id (str) –

    The element ID to associate with the new task.

Returns:

  • Task –

    The newly created expanded task.

then ¤
then(task: Task | list[Task], type: DependencyType = FS) -> Task | list[Task]

Adds a dependency from the given task(s) to this task.

Parameters:

  • task (Task or List[Task]) –

    The task(s) that will depend on this task.

  • type (DependencyType, default: FS ) –

    The type of dependency, by default DependencyType.FS.

Returns:

  • Task or List[Task] –

    The task(s) that was passed in, to allow chaining.

TaskIO ¤

TaskIO(
    name: str,
    value: Any = None,
    type_hint: str | None = None,
    description: str | None = None,
    type: str | None = None,
)

Base class for task inputs, outputs, and parameters.

Attributes¤

type property ¤
type: str | None

Deprecated alias for type_hint, kept for in-process readers.