script gen - support skyvern.loop & cleaner interfaces for generated code (no need to pass context.parameters, implicit template rendering) (#3542)

skyvern/__init__.py

@@ -34,9 +34,9 @@ from skyvern.services.script_service import (  # noqa: E402
     download,  # noqa: E402
     extract,  # noqa: E402
     http_request,  # noqa: E402
-    generate_text,  # noqa: E402
     goto,  # noqa: E402
     login,  # noqa: E402
+    loop,  # noqa: E402
     parse_file,  # noqa: E402
     prompt,  # noqa: E402
     render_list,  # noqa: E402
@@ -59,9 +59,9 @@ __all__ = [
     "download",
     "extract",
     "http_request",
-    "generate_text",
     "goto",
     "login",
+    "loop",
     "parse_file",
     "prompt",
     "render_list",

skyvern/core/script_generations/generate_script.py

@@ -168,57 +168,6 @@ def _render_value(
         return _value(prompt_text)
 
 
-def _generate_text_call(text_value: str, intention: str, parameter_key: str) -> cst.BaseExpression:
-    """Create a generate_text function call CST expression."""
-    return cst.Await(
-        expression=cst.Call(
-            func=cst.Attribute(value=cst.Name("skyvern"), attr=cst.Name("generate_text")),
-            whitespace_before_args=cst.ParenthesizedWhitespace(
-                indent=True,
-                last_line=cst.SimpleWhitespace(DOUBLE_INDENT),
-            ),
-            args=[
-                # First positional argument: context.parameters['parameter_key']
-                cst.Arg(
-                    value=cst.Subscript(
-                        value=cst.Attribute(
-                            value=cst.Name("context"),
-                            attr=cst.Name("parameters"),
-                        ),
-                        slice=[cst.SubscriptElement(slice=cst.Index(value=_value(parameter_key)))],
-                    ),
-                    whitespace_after_arg=cst.ParenthesizedWhitespace(
-                        indent=True,
-                        last_line=cst.SimpleWhitespace(DOUBLE_INDENT),
-                    ),
-                ),
-                # intention keyword argument
-                cst.Arg(
-                    keyword=cst.Name("intention"),
-                    value=_value(intention),
-                    whitespace_after_arg=cst.ParenthesizedWhitespace(
-                        indent=True,
-                        last_line=cst.SimpleWhitespace(DOUBLE_INDENT),
-                    ),
-                ),
-                # data keyword argument
-                cst.Arg(
-                    keyword=cst.Name("data"),
-                    value=cst.Attribute(
-                        value=cst.Name("context"),
-                        attr=cst.Name("parameters"),
-                    ),
-                    whitespace_after_arg=cst.ParenthesizedWhitespace(
-                        indent=True,
-                        last_line=cst.SimpleWhitespace(INDENT),
-                    ),
-                    comma=cst.Comma(),
-                ),
-            ],
-        )
-    )
-
-
 # --------------------------------------------------------------------- #
 # 2. utility builders                                                   #
 # --------------------------------------------------------------------- #
@@ -434,7 +383,7 @@ def _action_to_stmt(act: dict[str, Any], task: dict[str, Any], assign_to_output:
         args.append(
             cst.Arg(
                 keyword=cst.Name("prompt"),
-                value=_render_value(act["data_extraction_goal"]),
+                value=_value(act["data_extraction_goal"]),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(
                     indent=True,
                     last_line=cst.SimpleWhitespace(INDENT),
@@ -459,14 +408,6 @@ def _action_to_stmt(act: dict[str, Any], task: dict[str, Any], assign_to_output:
             cst.Arg(
                 keyword=cst.Name("intention"),
                 value=_value(act.get("intention") or act.get("reasoning") or ""),
-                whitespace_after_arg=cst.ParenthesizedWhitespace(
-                    indent=True,
-                    last_line=cst.SimpleWhitespace(INDENT),
-                ),
-            ),
-            cst.Arg(
-                keyword=cst.Name("data"),
-                value=cst.Attribute(value=cst.Name("context"), attr=cst.Name("parameters")),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(indent=True),
                 comma=cst.Comma(),
             ),
@@ -646,7 +587,7 @@ def _build_download_statement(
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(block.get("navigation_goal") or "", data_variable_name=data_variable_name),
+            value=_value(block.get("navigation_goal") or ""),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -657,7 +598,7 @@ def _build_download_statement(
         args.append(
             cst.Arg(
                 keyword=cst.Name("download_suffix"),
-                value=_render_value(block.get("download_suffix"), data_variable_name=data_variable_name),
+                value=_value(block.get("download_suffix")),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(
                     indent=True,
                     last_line=cst.SimpleWhitespace(INDENT),
@@ -694,7 +635,7 @@ def _build_action_statement(
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(block.get("navigation_goal", ""), data_variable_name=data_variable_name),
+            value=_value(block.get("navigation_goal", "")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -746,7 +687,7 @@ def _build_extract_statement(
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(block.get("data_extraction_goal", ""), data_variable_name=data_variable_name),
+            value=_value(block.get("data_extraction_goal", "")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -870,7 +811,7 @@ def _build_validate_statement(block: dict[str, Any]) -> cst.SimpleStatementLine:
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(block.get("navigation_goal", "")),
+            value=_value(block.get("navigation_goal", "")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
             ),
@@ -896,6 +837,14 @@ def _build_wait_statement(block: dict[str, Any]) -> cst.SimpleStatementLine:
         cst.Arg(
             keyword=cst.Name("seconds"),
             value=_value(block.get("wait_sec", 1)),
+            whitespace_after_arg=cst.ParenthesizedWhitespace(
+                indent=True,
+                last_line=cst.SimpleWhitespace(INDENT),
+            ),
+        ),
+        cst.Arg(
+            keyword=cst.Name("label"),
+            value=_value(block.get("label")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
             ),
@@ -920,7 +869,7 @@ def _build_goto_statement(block: dict[str, Any], data_variable_name: str | None
     args = [
         cst.Arg(
             keyword=cst.Name("url"),
-            value=_render_value(block.get("url", ""), data_variable_name=data_variable_name),
+            value=_value(block.get("url", "")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -1212,7 +1161,7 @@ def _build_prompt_statement(block: dict[str, Any]) -> cst.SimpleStatementLine:
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(block.get("prompt", "")),
+            value=_value(block.get("prompt", "")),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -1275,7 +1224,7 @@ def _build_for_loop_statement(block_title: str, block: dict[str, Any]) -> cst.Fo
 
     An example of a for loop statement:
     ```
-    for current_value in context.parameters["urls"]:
+    async for current_value in skyvern.loop(context.parameters["urls"]):
         await skyvern.goto(
             url=current_value,
             label="block_4",
@@ -1309,28 +1258,28 @@ def _build_for_loop_statement(block_title: str, block: dict[str, Any]) -> cst.Fo
     body_statements = []
 
     # Add loop_data assignment as the first statement
-    loop_data_variable_name = "loop_data"
-    loop_data_assignment = cst.SimpleStatementLine(
-        [
-            cst.Assign(
-                targets=[cst.AssignTarget(target=cst.Name(loop_data_variable_name))],
-                value=cst.Dict(
-                    [cst.DictElement(key=cst.SimpleString('"current_value"'), value=cst.Name("current_value"))]
-                ),
-            )
-        ]
-    )
-    body_statements.append(loop_data_assignment)
-
     for loop_block in loop_blocks:
-        stmt = _build_block_statement(loop_block, data_variable_name=loop_data_variable_name)
+        stmt = _build_block_statement(loop_block)
         body_statements.append(stmt)
 
-    # Create the for loop
+    # create skyvern.loop(loop_over_parameter_key, label=block_title)
+    loop_call_args = [cst.Arg(keyword=cst.Name("values"), value=_value(loop_over_parameter_key))]
+    if block.get("complete_if_empty"):
+        loop_call_args.append(
+            cst.Arg(keyword=cst.Name("complete_if_empty"), value=_value(block.get("complete_if_empty")))
+        )
+    loop_call_args.append(cst.Arg(keyword=cst.Name("label"), value=_value(block_title)))
+    loop_call = cst.Call(
+        func=cst.Attribute(value=cst.Name("skyvern"), attr=cst.Name("loop")),
+        args=loop_call_args,
+    )
+
+    # Create the async for loop
     for_loop = cst.For(
         target=target,
-        iter=_render_value(loop_over_parameter_key, render_func_name="render_list"),
+        iter=loop_call,
         body=cst.IndentedBlock(body=body_statements),
+        asynchronous=cst.Asynchronous(),
         whitespace_after_for=cst.SimpleWhitespace(" "),
         whitespace_before_in=cst.SimpleWhitespace(" "),
         whitespace_after_in=cst.SimpleWhitespace(" "),
@@ -1405,7 +1354,7 @@ def __build_base_task_statement(
     args = [
         cst.Arg(
             keyword=cst.Name("prompt"),
-            value=_render_value(prompt, data_variable_name=data_variable_name),
+            value=_value(prompt),
             whitespace_after_arg=cst.ParenthesizedWhitespace(
                 indent=True,
                 last_line=cst.SimpleWhitespace(INDENT),
@@ -1416,7 +1365,7 @@ def __build_base_task_statement(
         args.append(
             cst.Arg(
                 keyword=cst.Name("url"),
-                value=_render_value(block.get("url", "")),
+                value=_value(block.get("url", "")),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(
                     indent=True,
                     last_line=cst.SimpleWhitespace(INDENT),
@@ -1439,7 +1388,7 @@ def __build_base_task_statement(
         args.append(
             cst.Arg(
                 keyword=cst.Name("totp_identifier"),
-                value=_render_value(block.get("totp_identifier", "")),
+                value=_value(block.get("totp_identifier", "")),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(
                     indent=True,
                     last_line=cst.SimpleWhitespace(INDENT),
@@ -1450,7 +1399,7 @@ def __build_base_task_statement(
         args.append(
             cst.Arg(
                 keyword=cst.Name("totp_url"),
-                value=_render_value(block.get("totp_verification_url", "")),
+                value=_value(block.get("totp_verification_url", "")),
                 whitespace_after_arg=cst.ParenthesizedWhitespace(
                     indent=True,
                     last_line=cst.SimpleWhitespace(INDENT),

skyvern/core/script_generations/run_initializer.py

@@ -26,6 +26,7 @@ async def setup(
                 parameter = parameters_in_workflow_context[key]
                 if parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
                     parameters[key] = workflow_run_context.values[key]
+        context.script_run_parameters.update(parameters)
     skyvern_page = await SkyvernPage.create(browser_session_id=browser_session_id)
     run_context = RunContext(
         parameters=parameters,

skyvern/core/script_generations/skyvern_page.py

@@ -64,13 +64,29 @@ async def _get_element_id_by_xpath(xpath: str, page: Page) -> str | None:
     return element_id
 
 
+def _get_context_data(data: str | dict[str, Any] | None = None) -> dict[str, Any] | str | None:
+    context = skyvern_context.current()
+    global_context_data = context.script_run_parameters if context else None
+    if not data:
+        return global_context_data
+    result: dict[str, Any] | str | None
+    if isinstance(data, dict):
+        result = {k: v for k, v in data.items() if v}
+        if global_context_data:
+            result.update(global_context_data)
+    else:
+        global_context_data_str = json.dumps(global_context_data) if global_context_data else ""
+        result = f"{data}\n{global_context_data_str}"
+    return result
+
+
 def render_template(template: str, data: dict[str, Any] | None = None) -> str:
     """
     Refer to  Block.format_block_parameter_template_from_workflow_run_context
 
     TODO: complete this function so that block code shares the same template rendering logic
     """
-    template_data = data or {}
+    template_data = data.copy() if data else {}
     jinja_template = jinja_sandbox_env.from_string(template)
     context = skyvern_context.current()
     if context and context.workflow_run_id:
@@ -355,7 +371,7 @@ class SkyvernPage:
             try:
                 # Build the element tree of the current page for the prompt
                 context = skyvern_context.ensure_context()
-                payload_str = json.dumps(data) if isinstance(data, (dict, list)) else (data or "")
+                payload_str = _get_context_data(data)
                 refreshed_page = await self.scraped_page.generate_scraped_page_without_screenshots()
                 element_tree = refreshed_page.build_element_tree()
                 single_click_prompt = prompt_engine.load_prompt(
@@ -463,9 +479,7 @@ class SkyvernPage:
         if ai_infer and intention:
             try:
                 prompt = context.prompt if context else None
-                # Build the element tree of the current page for the prompt
+                data = _get_context_data(data)
-                # clean up empty data values
-                data = {k: v for k, v in data.items() if v} if isinstance(data, dict) else (data or "")
                 if (totp_identifier or totp_url) and context and organization_id and task_id:
                     verification_code = await poll_verification_code(
                         organization_id=organization_id,
@@ -488,11 +502,10 @@ class SkyvernPage:
                 self.scraped_page = refreshed_page
                 # get the element_id by the xpath
                 element_id = await _get_element_id_by_xpath(xpath, self.page)
-                payload_str = json.dumps(data) if isinstance(data, (dict, list)) else (data or "")
                 script_generation_input_text_prompt = prompt_engine.load_prompt(
                     template="script-generation-input-text-generatiion",
                     intention=intention,
-                    data=payload_str,
+                    data=data,
                     goal=prompt,
                 )
                 json_response = await app.SINGLE_INPUT_AGENT_LLM_API_HANDLER(
@@ -539,12 +552,11 @@ class SkyvernPage:
             try:
                 context = skyvern_context.current()
                 prompt = context.prompt if context else None
-                data = {k: v for k, v in data.items() if v} if isinstance(data, dict) else (data or "")
+                data = _get_context_data(data)
-                payload_str = json.dumps(data) if isinstance(data, (dict, list)) else (data or "")
                 script_generation_file_url_prompt = prompt_engine.load_prompt(
                     template="script-generation-file-url-generation",
                     intention=intention,
-                    data=payload_str,
+                    data=data,
                     goal=prompt,
                 )
                 json_response = await app.SINGLE_INPUT_AGENT_LLM_API_HANDLER(
@@ -578,15 +590,14 @@ class SkyvernPage:
             if ai_infer and intention and task and step:
                 try:
                     prompt = context.prompt if context else None
-                    data = {k: v for k, v in data.items() if v} if isinstance(data, dict) else (data or "")
+                    data = _get_context_data(data)
-                    payload_str = json.dumps(data) if isinstance(data, (dict, list)) else (data or "")
                     refreshed_page = await self.scraped_page.generate_scraped_page_without_screenshots()
                     self.scraped_page = refreshed_page
                     element_tree = refreshed_page.build_element_tree()
                     merged_goal = SELECT_OPTION_GOAL.format(intention=intention, prompt=prompt)
                     single_select_prompt = prompt_engine.load_prompt(
                         template="single-select-action",
-                        navigation_payload_str=payload_str,
+                        navigation_payload_str=data,
                         navigation_goal=merged_goal,
                         current_url=self.page.url,
                         elements=element_tree,

skyvern/forge/sdk/core/skyvern_context.py

@@ -1,5 +1,6 @@
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+from typing import Any
 from zoneinfo import ZoneInfo
 
 from playwright.async_api import Frame
@@ -28,13 +29,25 @@ class SkyvernContext:
     frame_index_map: dict[Frame, int] = field(default_factory=dict)
     dropped_css_svg_element_map: dict[str, bool] = field(default_factory=dict)
     max_screenshot_scrolls: int | None = None
+
+    # feature flags
+    enable_parse_select_in_extract: bool = False
+    use_prompt_caching: bool = False
+    cached_static_prompt: str | None = None
+
+    # script run context
     script_id: str | None = None
     script_revision_id: str | None = None
     action_order: int = 0
     prompt: str | None = None
-    enable_parse_select_in_extract: bool = False
+    parent_workflow_run_block_id: str | None = None
-    use_prompt_caching: bool = False
+    loop_metadata: dict[str, Any] | None = None
-    cached_static_prompt: str | None = None
+    loop_output_values: list[dict[str, Any]] | None = None
+    script_run_parameters: dict[str, Any] = field(default_factory=dict)
+    """
+    Example output value:
+    {"loop_value": "str", "output_parameter": "the key of the parameter", "output_value": Any}
+    """
 
     def __repr__(self) -> str:
         return f"SkyvernContext(request_id={self.request_id}, organization_id={self.organization_id}, task_id={self.task_id}, step_id={self.step_id}, workflow_id={self.workflow_id}, workflow_run_id={self.workflow_run_id}, task_v2_id={self.task_v2_id}, max_steps_override={self.max_steps_override}, run_id={self.run_id})"

skyvern/forge/sdk/workflow/models/block.py

@@ -918,14 +918,14 @@ class ForLoopBlock(Block):
 
         return context_parameters
 
-    async def get_loop_over_parameter_values(
+    async def get_values_from_loop_variable_reference(
         self,
         workflow_run_context: WorkflowRunContext,
         workflow_run_id: str,
         workflow_run_block_id: str,
         organization_id: str | None = None,
     ) -> list[Any]:
-        # parse the value from self.loop_variable_reference and then from self.loop_over
+        parameter_value = None
         if self.loop_variable_reference:
             LOG.debug("Processing loop variable reference", loop_variable_reference=self.loop_variable_reference)
 
@@ -1029,6 +1029,26 @@ class ForLoopBlock(Block):
                     raise FailedToFormatJinjaStyleParameter(value_template, str(e))
                 parameter_value = json.loads(value_json)
 
+        if isinstance(parameter_value, list):
+            return parameter_value
+        else:
+            return [parameter_value]
+
+    async def get_loop_over_parameter_values(
+        self,
+        workflow_run_context: WorkflowRunContext,
+        workflow_run_id: str,
+        workflow_run_block_id: str,
+        organization_id: str | None = None,
+    ) -> list[Any]:
+        # parse the value from self.loop_variable_reference and then from self.loop_over
+        if self.loop_variable_reference:
+            return await self.get_values_from_loop_variable_reference(
+                workflow_run_context,
+                workflow_run_id,
+                workflow_run_block_id,
+                organization_id,
+            )
         elif self.loop_over is not None:
             if isinstance(self.loop_over, WorkflowParameter):
                 parameter_value = workflow_run_context.get_value(self.loop_over.key)
@@ -1165,6 +1185,7 @@ class ForLoopBlock(Block):
 
         for loop_idx, loop_over_value in enumerate(loop_over_values):
             LOG.info("Starting loop iteration", loop_idx=loop_idx, loop_over_value=loop_over_value)
+            # context parameter has been deprecated. However, it's still used by task v2 - we should migrate away from it.
             context_parameters_with_value = self.get_loop_block_context_parameters(workflow_run_id, loop_over_value)
             for context_parameter in context_parameters_with_value:
                 workflow_run_context.set_value(context_parameter.key, context_parameter.value)

skyvern/services/script_service.py

@@ -2,12 +2,11 @@ import asyncio
 import base64
 import hashlib
 import importlib.util
-import json
 import os
 import uuid
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Callable, cast
+from typing import Any, AsyncGenerator, Callable, Sequence, cast
 
 import libcst as cst
 import structlog
@@ -21,7 +20,6 @@ from skyvern.core.script_generations.generate_script import _build_block_fn, cre
 from skyvern.core.script_generations.skyvern_page import script_run_context_manager
 from skyvern.exceptions import ScriptNotFound, WorkflowRunNotFound
 from skyvern.forge import app
-from skyvern.forge.prompts import prompt_engine
 from skyvern.forge.sdk.artifact.models import ArtifactType
 from skyvern.forge.sdk.core import skyvern_context
 from skyvern.forge.sdk.models import Step, StepStatus
@@ -35,6 +33,7 @@ from skyvern.forge.sdk.workflow.models.block import (
     FileDownloadBlock,
     FileParserBlock,
     FileUploadBlock,
+    ForLoopBlock,
     HttpRequestBlock,
     LoginBlock,
     SendEmailBlock,
@@ -52,6 +51,20 @@ LOG = structlog.get_logger()
 jinja_sandbox_env = SandboxedEnvironment()
 
 
+class SkyvernLoopItem:
+    def __init__(
+        self,
+        index: int,
+        value: Any,
+    ):
+        self.current_index = index
+        self.current_value = value
+        self.current_item = value
+
+    def __repr__(self) -> str:
+        return f"SkyvernLoopItem(current_value={self.current_value}, current_index={self.current_index})"
+
+
 async def build_file_tree(
     files: list[ScriptFileCreate],
     organization_id: str,
@@ -363,6 +376,7 @@ async def _create_workflow_block_run_and_task(
     prompt: str | None = None,
     schema: dict[str, Any] | list | str | None = None,
     url: str | None = None,
+    label: str | None = None,
 ) -> tuple[str | None, str | None, str | None]:
     """
     Create a workflow block run and optionally a task if workflow_run_id is available in context.
@@ -374,24 +388,34 @@ async def _create_workflow_block_run_and_task(
     workflow_run_id = context.workflow_run_id
     organization_id = context.organization_id
 
+    # if there's a parent_workflow_run_block_id and loop_metadata, update_block_metadata
+    if context.parent_workflow_run_block_id and context.loop_metadata and label:
+        workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
+        workflow_run_context.update_block_metadata(label, context.loop_metadata)
+
+    workflow_run_block = await app.DATABASE.create_workflow_run_block(
+        workflow_run_id=workflow_run_id,
+        parent_workflow_run_block_id=context.parent_workflow_run_block_id,
+        organization_id=organization_id,
+        block_type=block_type,
+        label=label,
+    )
+
+    workflow_run_block_id = workflow_run_block.workflow_run_block_id
+
     try:
         # Create workflow run block with appropriate parameters based on block type
         # TODO: support engine in the future
-        engine = None
-        workflow_run_block = await app.DATABASE.create_workflow_run_block(
-            workflow_run_id=workflow_run_id,
-            organization_id=organization_id,
-            block_type=block_type,
-            engine=engine,
-        )
-
-        workflow_run_block_id = workflow_run_block.workflow_run_block_id
         task_id = None
         step_id = None
 
         # Create task for task-based blocks
         if block_type in SCRIPT_TASK_BLOCKS:
             # Create task
+            if prompt:
+                prompt = _render_template_with_label(prompt, label)
+            if url:
+                url = _render_template_with_label(url, label)
             task = await app.DATABASE.create_task(
                 # fix HACK: changed the type of url to str | None to support None url. url is not used in the script right now.
                 url=url or "",
@@ -1107,6 +1131,7 @@ async def run_task(
             block_type=BlockType.TASK,
             prompt=prompt,
             url=url,
+            label=cache_key,
         )
         # set the prompt in the RunContext
         context = skyvern_context.ensure_context()
@@ -1155,6 +1180,7 @@ async def run_task(
         )
         await task_block.execute_safe(
             workflow_run_id=block_validation_output.workflow_run_id,
+            parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
             organization_id=block_validation_output.organization_id,
             browser_session_id=block_validation_output.browser_session_id,
         )
@@ -1180,6 +1206,7 @@ async def download(
             block_type=BlockType.FILE_DOWNLOAD,
             prompt=prompt,
             url=url,
+            label=cache_key,
         )
         # set the prompt in the RunContext
         context = skyvern_context.ensure_context()
@@ -1228,6 +1255,7 @@ async def download(
         )
         await file_download_block.execute_safe(
             workflow_run_id=block_validation_output.workflow_run_id,
+            parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
             organization_id=block_validation_output.organization_id,
             browser_session_id=block_validation_output.browser_session_id,
         )
@@ -1251,6 +1279,7 @@ async def action(
             block_type=BlockType.ACTION,
             prompt=prompt,
             url=url,
+            label=cache_key,
         )
         # set the prompt in the RunContext
         context = skyvern_context.ensure_context()
@@ -1297,6 +1326,7 @@ async def action(
         )
         await action_block.execute_safe(
             workflow_run_id=block_validation_output.workflow_run_id,
+            parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
             organization_id=block_validation_output.organization_id,
             browser_session_id=block_validation_output.browser_session_id,
         )
@@ -1320,6 +1350,7 @@ async def login(
             block_type=BlockType.LOGIN,
             prompt=prompt,
             url=url,
+            label=cache_key,
         )
         # set the prompt in the RunContext
         context = skyvern_context.ensure_context()
@@ -1365,6 +1396,7 @@ async def login(
         )
         await login_block.execute_safe(
             workflow_run_id=block_validation_output.workflow_run_id,
+            parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
             organization_id=block_validation_output.organization_id,
             browser_session_id=block_validation_output.browser_session_id,
         )
@@ -1390,6 +1422,7 @@ async def extract(
             prompt=prompt,
             schema=schema,
             url=url,
+            label=cache_key,
         )
         # set the prompt in the RunContext
         context = skyvern_context.ensure_context()
@@ -1437,15 +1470,16 @@ async def extract(
         )
         block_result = await extraction_block.execute_safe(
             workflow_run_id=block_validation_output.workflow_run_id,
+            parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
             organization_id=block_validation_output.organization_id,
             browser_session_id=block_validation_output.browser_session_id,
         )
         return block_result.output_parameter_value
 
 
-async def wait(seconds: int) -> None:
+async def wait(seconds: int, label: str | None = None) -> None:
     # Auto-create workflow block run if workflow_run_id is available (wait block doesn't create tasks)
-    workflow_run_block_id, _, _ = await _create_workflow_block_run_and_task(block_type=BlockType.WAIT)
+    workflow_run_block_id, _, _ = await _create_workflow_block_run_and_task(block_type=BlockType.WAIT, label=label)
 
     try:
         await asyncio.sleep(seconds)
@@ -1507,36 +1541,32 @@ async def run_script(
         raise Exception(f"No 'run_workflow' function found in {path}")
 
 
-async def generate_text(
+def _render_template_with_label(template: str, label: str | None = None) -> str:
-    text: str | None = None,
+    template_data = {}
-    intention: str | None = None,
+    context = skyvern_context.current()
-    data: dict[str, Any] | None = None,
+    if context and context.workflow_run_id and label:
-) -> str:
+        workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(context.workflow_run_id)
-    if text:
+        block_reference_data: dict[str, Any] = workflow_run_context.get_block_metadata(label)
-        return text
+        template_data = workflow_run_context.values.copy()
-    new_text = text or ""
+        if label in template_data:
-    if intention and data:
+            current_value = template_data[label]
-        try:
+            if isinstance(current_value, dict):
-            context = skyvern_context.ensure_context()
+                block_reference_data.update(current_value)
-            prompt = context.prompt
+            else:
-            # Build the element tree of the current page for the prompt
+                LOG.warning(
-            payload_str = json.dumps(data) if isinstance(data, (dict, list)) else (data or "")
+                    f"Script service: Parameter {label} has a registered reference value, going to overwrite it by block metadata"
-            script_generation_input_text_prompt = prompt_engine.load_prompt(
+                )
-                template="script-generation-input-text-generatiion",
+
-                intention=intention,
+        template_data[label] = block_reference_data
-                data=payload_str,
+
-                goal=prompt,
+        # inject the forloop metadata as global variables
-            )
+        if "current_index" in block_reference_data:
-            json_response = await app.SINGLE_INPUT_AGENT_LLM_API_HANDLER(
+            template_data["current_index"] = block_reference_data["current_index"]
-                prompt=script_generation_input_text_prompt,
+        if "current_item" in block_reference_data:
-                prompt_name="script-generation-input-text-generatiion",
+            template_data["current_item"] = block_reference_data["current_item"]
-                organization_id=context.organization_id,
+        if "current_value" in block_reference_data:
-            )
+            template_data["current_value"] = block_reference_data["current_value"]
-            new_text = json_response.get("answer", new_text)
+    return render_template(template, data=template_data)
-        except Exception:
-            LOG.exception("Failed to generate text for script")
-            raise
-    return new_text
 
 
 def render_template(template: str, data: dict[str, Any] | None = None) -> str:
@@ -1545,16 +1575,17 @@ def render_template(template: str, data: dict[str, Any] | None = None) -> str:
 
     TODO: complete this function so that block code shares the same template rendering logic
     """
-    template_data = data or {}
+    template_data = data.copy() if data else {}
     jinja_template = jinja_sandbox_env.from_string(template)
     context = skyvern_context.current()
-    if context and context.workflow_run_id:
+    if context:
-        workflow_run_id = context.workflow_run_id
+        template_data.update(context.script_run_parameters)
-        workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
+        if context.workflow_run_id:
-        template_data.update(workflow_run_context.values)
+            workflow_run_id = context.workflow_run_id
-        if template in template_data:
+            workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
-            return template_data[template]
+            template_data.update(workflow_run_context.values)
-
+            if template in template_data:
+                return template_data[template]
     return jinja_template.render(template_data)
 
 
@@ -1571,6 +1602,7 @@ def render_list(template: str, data: dict[str, Any] | None = None) -> list[str]:
 ## Non-task-based block helpers
 @dataclass
 class BlockValidationOutput:
+    context: skyvern_context.SkyvernContext
     label: str
     output_parameter: OutputParameter
     workflow: Workflow
@@ -1596,6 +1628,9 @@ async def _validate_and_get_output_parameter(label: str | None = None) -> BlockV
     if not workflow:
         raise Exception("Workflow not found")
     label = label or f"block_{uuid.uuid4()}"
+    if context.loop_metadata:
+        workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
+        workflow_run_context.update_block_metadata(label, context.loop_metadata)
     output_parameter = workflow.get_output_parameter(label)
     if not output_parameter:
         # NOT sure if this is legit hack to create output parameter like this
@@ -1608,6 +1643,7 @@ async def _validate_and_get_output_parameter(label: str | None = None) -> BlockV
             parameter_type=ParameterType.OUTPUT,
         )
     return BlockValidationOutput(
+        context=context,
         label=label,
         output_parameter=output_parameter,
         workflow=workflow,
@@ -1632,6 +1668,7 @@ async def run_code(
     )
     block_result = await code_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1652,6 +1689,22 @@ async def upload_file(
     path: str | None = None,
 ) -> None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    if s3_bucket:
+        s3_bucket = _render_template_with_label(s3_bucket, label)
+    if aws_access_key_id:
+        aws_access_key_id = _render_template_with_label(aws_access_key_id, label)
+    if aws_secret_access_key:
+        aws_secret_access_key = _render_template_with_label(aws_secret_access_key, label)
+    if region_name:
+        region_name = _render_template_with_label(region_name, label)
+    if azure_storage_account_name:
+        azure_storage_account_name = _render_template_with_label(azure_storage_account_name, label)
+    if azure_storage_account_key:
+        azure_storage_account_key = _render_template_with_label(azure_storage_account_key, label)
+    if azure_blob_container_name:
+        azure_blob_container_name = _render_template_with_label(azure_blob_container_name, label)
+    if path:
+        path = _render_template_with_label(path, label)
     file_upload_block = FileUploadBlock(
         label=block_validation_output.label,
         output_parameter=block_validation_output.output_parameter,
@@ -1668,6 +1721,7 @@ async def upload_file(
     )
     await file_upload_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1675,7 +1729,7 @@ async def upload_file(
 
 async def send_email(
     sender: str,
-    recipients: list[str],
+    recipients: list[str] | str,
     subject: str,
     body: str,
     file_attachments: list[str] = [],
@@ -1683,6 +1737,11 @@ async def send_email(
     parameters: list[PARAMETER_TYPE] | None = None,
 ) -> None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    sender = _render_template_with_label(sender, label)
+    if isinstance(recipients, str):
+        recipients = render_list(_render_template_with_label(recipients, label))
+    subject = _render_template_with_label(subject, label)
+    body = _render_template_with_label(body, label)
     workflow = block_validation_output.workflow
     smtp_host_parameter = workflow.get_parameter("smtp_host")
     smtp_port_parameter = workflow.get_parameter("smtp_port")
@@ -1706,6 +1765,7 @@ async def send_email(
     )
     await send_email_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1719,6 +1779,7 @@ async def parse_file(
     parameters: list[PARAMETER_TYPE] | None = None,
 ) -> None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    file_url = _render_template_with_label(file_url, label)
     file_parser_block = FileParserBlock(
         file_url=file_url,
         file_type=file_type,
@@ -1729,6 +1790,7 @@ async def parse_file(
     )
     await file_parser_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1745,6 +1807,8 @@ async def http_request(
     parameters: list[PARAMETER_TYPE] | None = None,
 ) -> None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    method = _render_template_with_label(method, label)
+    url = _render_template_with_label(url, label)
     http_request_block = HttpRequestBlock(
         method=method,
         url=url,
@@ -1758,6 +1822,7 @@ async def http_request(
     )
     await http_request_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1769,6 +1834,7 @@ async def goto(
     parameters: list[PARAMETER_TYPE] | None = None,
 ) -> None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    url = _render_template_with_label(url, label)
     goto_url_block = UrlBlock(
         url=url,
         label=block_validation_output.label,
@@ -1777,6 +1843,7 @@ async def goto(
     )
     await goto_url_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
@@ -1789,6 +1856,7 @@ async def prompt(
     parameters: list[PARAMETER_TYPE] | None = None,
 ) -> dict[str, Any] | list | str | None:
     block_validation_output = await _validate_and_get_output_parameter(label)
+    prompt = _render_template_with_label(prompt, label)
     prompt_block = TextPromptBlock(
         prompt=prompt,
         json_schema=schema,
@@ -1798,7 +1866,119 @@ async def prompt(
     )
     result = await prompt_block.execute_safe(
         workflow_run_id=block_validation_output.workflow_run_id,
+        parent_workflow_run_block_id=block_validation_output.context.parent_workflow_run_block_id,
         organization_id=block_validation_output.organization_id,
         browser_session_id=block_validation_output.browser_session_id,
     )
     return result.output_parameter_value
+
+
+async def loop(
+    values: Sequence[Any] | str,
+    complete_if_empty: bool = False,
+    label: str | None = None,
+) -> AsyncGenerator[SkyvernLoopItem, None]:
+    workflow_run_block_id, _, _ = await _create_workflow_block_run_and_task(block_type=BlockType.FOR_LOOP, label=label)
+    # process values:
+    loop_variable_reference = None
+    loop_values = None
+    if isinstance(values, list):
+        loop_values = values
+    elif isinstance(values, str):
+        loop_variable_reference = values
+    else:
+        raise ValueError(f"Invalid values type: {type(values)}")
+
+    # step. build the ForLoopBlock instance
+    block_validation_output = await _validate_and_get_output_parameter(label)
+    loop_block = ForLoopBlock(
+        label=block_validation_output.label,
+        output_parameter=block_validation_output.output_parameter,
+        loop_variable_reference=loop_variable_reference,
+        loop_blocks=[],
+        complete_if_empty=complete_if_empty,
+    )
+    workflow_run_id = block_validation_output.workflow_run_id
+    organization_id = block_validation_output.organization_id
+
+    if not loop_values:
+        workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
+        if workflow_run_block_id:
+            loop_values = await loop_block.get_values_from_loop_variable_reference(
+                workflow_run_context=workflow_run_context,
+                workflow_run_id=workflow_run_id,
+                workflow_run_block_id=workflow_run_block_id,
+                organization_id=organization_id,
+            )
+
+    if not loop_values:
+        # step 3. if loop_values is empty, record empty output parameter value
+        LOG.info(
+            "script service: No loop values found, terminating block",
+            block_type=BlockType.FOR_LOOP,
+            workflow_run_id=workflow_run_id,
+            complete_if_empty=complete_if_empty,
+        )
+        await loop_block.record_output_parameter_value(workflow_run_context, workflow_run_id, [])
+        # step 4. build response (success/failure) given the complete_if_empty value
+        if complete_if_empty:
+            await loop_block.build_block_result(
+                success=True,
+                failure_reason=None,
+                output_parameter_value=[],
+                status=BlockStatus.completed,
+                workflow_run_block_id=workflow_run_block_id,
+                organization_id=organization_id,
+            )
+            return
+        else:
+            await loop_block.build_block_result(
+                success=False,
+                failure_reason="No iterable value found for the loop block",
+                status=BlockStatus.terminated,
+                workflow_run_block_id=workflow_run_block_id,
+                organization_id=organization_id,
+            )
+            raise Exception("No iterable value found for the loop block")
+
+    # register the loop in the global context
+    block_validation_output.context.parent_workflow_run_block_id = workflow_run_block_id
+    block_validation_output.context.loop_output_values = []
+
+    # step 5. start the loop
+    try:
+        for index, value in enumerate(loop_values):
+            # register current_value, current_item and current_index in workflow run context
+            loop_metadata = {
+                "current_index": index,
+                "current_value": value,
+                "current_item": value,
+            }
+            block_validation_output.context.loop_metadata = loop_metadata
+            workflow_run_context.update_block_metadata(block_validation_output.label, loop_metadata)
+            # Build the SkyvernLoopItem for this loop
+            yield SkyvernLoopItem(index, value)
+
+        # build success output
+        if workflow_run_block_id:
+            await _update_workflow_block(
+                workflow_run_block_id,
+                BlockStatus.completed,
+                output=block_validation_output.context.loop_output_values,
+                label=label,
+            )
+    except Exception as e:
+        # build failure output
+        if workflow_run_block_id:
+            await _update_workflow_block(
+                workflow_run_block_id,
+                BlockStatus.failed,
+                failure_reason=str(e),
+                output=block_validation_output.context.loop_output_values,
+                label=label,
+            )
+        raise e
+    finally:
+        block_validation_output.context.parent_workflow_run_block_id = None
+        block_validation_output.context.loop_metadata = None
+        block_validation_output.context.loop_output_values = None