agently-dynamic-task

name	agently-dynamic-task
description	Use when the user needs Agently Dynamic Task, model-generated or app-submitted DAG planning, TaskDAG validation, DynamicTaskResolver handlers, or TaskDAGExecutor execution through Agently.create_dynamic_task. Dynamic Task is a first-class Agently API that uses TriggerFlow as an execution substrate.

Agently Dynamic Task

Use this skill when the problem is a dynamic task graph: a model or application submits DAG data that must be planned, validated, pruned, resolved to handlers, and executed.

Dynamic Task is an Agently first-class API. It is not a TriggerFlow sub-API. TriggerFlow is the execution substrate under TaskDAGExecutor, while Agently.create_dynamic_task(...) and agent.create_dynamic_task(...) are the ordinary app-facing entrypoints.

Native-First Rules

• use Agently.create_dynamic_task(...) or agent.create_dynamic_task(...) for ordinary app code
• split into AgentlyTaskDAGPlanner, TaskDAGValidator, DynamicTaskResolver, and TaskDAGExecutor only when staged control is required
• use plan=<TaskDAG> when the caller already owns the DAG and should skip model planning
• use TaskDAG.from_yaml(...) or TaskDAG.from_json(...) when the submitted DAG is a reviewed config artifact
• use submitted DAG inputs placeholders for small runtime wiring: ${INIT} / ${INIT.foo} for initial graph input, ${DEPS.task_id.path} for dependency results, ${STATE...} for execution-local state, and ${TRIGGER...} for the raw TriggerFlow trigger payload. Slot names are case-insensitive, but examples should use uppercase. Whole-string placeholders preserve the original value type; embedded placeholders stringify into the surrounding text. Missing runtime paths fail closed during execution
• when a submitted DAG runs through agent.create_execution(), let ${INIT...} read use_dynamic_task(graph_input=...) when explicit, otherwise the frozen execution prompt input slot, then {"target": task_target}; do not add a second inputs.input or task-local prompt mapping surface
• use planner=<agent-or-provider-settings> when the model must generate the DAG
• use model=<agent-or-provider-settings> for model task execution resources
• use handlers={"risk_check_handler": handler} for local/custom tasks; handler names should be explicit and usually end in _handler
• pass actions=... or skills=... only when the planner should be allowed to use those task kinds; do not expose them by default
• use Agently output schemas for model-owned semantic outputs instead of parsing model text with example-local JSON/code-block parsers
• keep backend business-system calls mockable in examples, but keep model-owned planning, classification, drafting, evaluation, revising, and final response generation model-driven
• require each model task that matters to have a clear output schema, either through facade-level output_schema or node-level inputs.output_schema
• set model-task inputs.output_format by result type when the plan is generated or submitted: json for compact machine-control data, action arguments, routing flags, booleans, numbers, strict extraction, model judges, and dense nested arrays/objects; flat_markdown for flat string long text/code/HTML/SVG/Markdown/SQL/templates; explicit hybrid for long prose plus structured lists, tables, citations, metadata, or nested evidence when retry latency is acceptable; auto only when structural schema-driven selection and retry latency are acceptable
• let TaskDAGValidator reject duplicate ids, missing dependencies, cycles, unsupported required task kinds, schema-version mismatches, unsafe side-effect policy, and unknown required handlers before execution
• allow unknown optional handlers to be pruned only when they do not affect required semantic outputs, required downstream nodes, approvals, or side-effect policy
• keep generated plan data in execution input or execution state; do not use shared TriggerFlow flow data for per-run DAG state
• use ${INIT...} / ${DEPS...} placeholders for small declarative input wiring; for larger transformations, keep the logic in a handler or model task
• do not teach planners to use internal resolver entries such as validate or emit as ordinary task kinds
• do not mark ordinary read-only model tasks as network side effects just because they call an LLM provider
• do not add approval gates to analysis, synthesis, drafting, or final response model tasks unless the user explicitly asks for approval

API Shape

Submitted DAG:

from agently import Agently

async def risk_check_handler(context):
    return {
        "task_id": context.task.id,
        "deps": dict(context.dependency_results),
    }

task = Agently.create_dynamic_task(
    target="review policy",
    plan={
        "graph_id": "review",
        "task_schema_version": "task_dag/v1",
        "tasks": [
            {"id": "extract", "kind": "local", "binding": "risk_check_handler"},
            {
                "id": "final",
                "kind": "local",
                "binding": "risk_check_handler",
                "depends_on": ["extract"],
                "inputs": {
                    "source": "${INIT}",
                    "extract_result": "${DEPS.extract}",
                },
            },
        ],
        "semantic_outputs": {"final": "final"},
    },
    handlers={"risk_check_handler": risk_check_handler},
)

snapshot = await task.async_start(timeout=10)
result = snapshot["semantic_outputs"]["final"]

Config-backed submitted DAG:

from agently import Agently
from agently.core import TaskDAG

graph = TaskDAG.from_yaml("plans/review.yaml")
task = Agently.create_dynamic_task(
    target="review policy",
    plan=graph,
    handlers={"risk_check_handler": risk_check_handler},
)
snapshot = await task.async_run(graph_input={"doc": "policy"}, timeout=10)

Use TaskDAG.from_json(...) for JSON/JSON5 files or raw content. Both loaders support task_dag_key_path="plans.review" for selecting one DAG from a larger config. Use graph.get_yaml(path) or graph.get_json(path) to export a normalized graph.

Auto-planned DAG:

from agently import Agently

task = Agently.create_dynamic_task(
    target=user_goal,
    planner=planner_agent,
    model=worker_agent,
    output_schema={
        "answer": (str, "final user-facing answer", True),
    },
)

graph = await task.async_plan(max_retries=3)
task.validate(graph, strict_schema_version=True)
snapshot = await task.async_run(graph_input={"goal": user_goal}, timeout=30)

Lower-level integration:

from agently.builtins.plugins import AgentlyTaskDAGPlanner
from agently.core import DynamicTaskResolver, TaskDAGExecutor, TaskDAGValidator

resolver = DynamicTaskResolver({"risk_check_handler": risk_check_handler})
validator = TaskDAGValidator(resolver)
planner = AgentlyTaskDAGPlanner(validator=validator)

graph = await planner.async_plan(planner_agent, {"target": "review policy"})
validation = validator.validate(graph, strict_schema_version=True)
snapshot = await TaskDAGExecutor(resolver, validator=validator).async_run(graph)

Planner Constraints

Planner plugins and planner prompts must declare:

• task schema version, normally task_dag/v1
• allowed task kinds and available binding names
• required semantic output leaves
• maximum task count when the scenario has a practical limit
• validation checks that become retryable model-output failures
• forbidden behavior: unstable task ids, dependency results embedded in static task inputs, implicit side effects, hidden action/skill use, and internal resolver task kinds

The facade planner should wire .output(planner.output_schema(), format="json"), required planner.ensure_keys(), and .validate(planner.validate_output) so structural problems become model-output validation failures before execution. Compile/run still revalidates the graph.

Layer ownership:

• TaskDAG / TaskDAGNode data contracts and YAML/JSON graph config belong to agently.types.data
• TaskDAGExecutor, TaskDAGValidator, and DynamicTaskResolver belong to core
• AgentlyTaskDAGPlanner owns planner output schema, ensure keys, and planner instructions as a plugin concern
• Agently.create_dynamic_task(...) and agent.create_dynamic_task(...) are the app-facing facade entrypoints

Anti-Patterns

• do not put Dynamic Task guidance under TriggerFlow as if it were TriggerFlow syntax
• do not hand-roll a scheduler for submitted DAG data
• do not parse model text with json.loads(...), regex marker splits, or code-block extraction when Agently output schemas can own the shape
• do not expose actions or skills to the planner by default
• do not hide model-owned classification, drafting, quality checks, or final response generation behind deterministic local substitutes in recommended examples
• do not call low-level TaskDAGExecutor from ordinary app code when Agently.create_dynamic_task(...) is enough
• do not use vague handler names such as risk_check; use risk_check_handler or another name that describes the callable role
• do not duplicate TriggerFlow docs here; explain only that Dynamic Task compiles to TriggerFlow internally

Read Next

• references/overview.md
• examples/minimal.py