Build FlowForge with Gemini

Introduction

30 Second Summary

Repetitive knowledge work often means copying the same facts into the same prompts. Each run becomes hard to inspect or reuse.

In this project, you will build FlowForge, a browser-based workflow automation app powered by Gemini. Its four-block pipeline turns browser inputs into a grounded answer with a visible execution trace.

What You'll Build

Your finished app turns a browser question into a grounded Gemini answer while a four-row trace reveals the path from input to generation.

By the end of this project, you'll have:

• A visual four-block pipeline that turns browser inputs into a grounded Gemini answer.
• A dependency-aware engine that accepts blocks in any order. Its trace shows each completed block with its duration.
• Your template library reloads saved workflows after a server restart. The Recent Runs panel shows the latest in-memory executions.
• A Secret Mission: Add a gate block that skips an unnecessary Gemini call when the input is too short.

Are there any prerequisites?

You will need Python 3.10 or newer, Git, Visual Studio Code, and a Gemini API key. The project stays at $0 through the Gemini API Free Tier with billing disabled.

Before We Start

This first checkpoint locks in the scope of your compressed local MVP. You will define the people it serves through a saved workflow for standup-policy and escalation questions.

Set Up the FlowForge Workspace

FlowForge's backend depends on exact package versions. A reproducible Python environment keeps later API behavior consistent.

The Gemini API key belongs in the server environment because browser code can expose credentials. Visual Studio Code gives the project one local workspace while billing stays disabled.

In this step, get ready to:

• Create the local FlowForge workspace.
• Pin the Python dependencies.
• Activate the server environment with the Gemini key.

Create the workspace and source files

A VS Code workspace gives every file and terminal command the same project location. The flowforge-ai folder becomes that shared location.

• Use your operating system's application search.
• Enter Visual Studio Code.
• Select Visual Studio Code.
• Click File in the top menu.
• Select Open Folder....
• Select your Desktop.
• Select New Folder.
• Enter flowforge-ai as the folder name.

Opening this folder makes it the VS Code workspace. The Explorer and integrated terminal now use flowforge-ai as their project location.

• Select the new flowforge-ai folder.
• Click Select Folder.
• Select Yes, I trust the authors if the Workspace Trust dialog appears.

You should see flowforge-ai at the top of the Explorer sidebar. The empty folder is now your local workspace.

• Select the Explorer view in the left Activity Bar.
• Click New File... in the Explorer toolbar.
• Enter requirements.txt as the file name.
• Add the pinned dependency list to requirements.txt by pasting this code:

fastapi==0.141.1
google-genai==2.24.0
pydantic==2.13.5
uvicorn==0.53.0

What do these dependencies provide?

• FastAPI at 0.141.1 provides the backend API framework.
• The Google Gen AI SDK at 2.24.0 connects the backend to Gemini.
• Pydantic at 2.13.5 validates workflow data.
• Uvicorn at 0.53.0 runs the local API server.

• Save requirements.txt.
• Confirm requirements.txt appears directly under flowforge-ai in the Explorer.

Can't see the dependency file?

Check that the file is named requirements.txt. Confirm that it sits directly inside flowforge-ai.

Make sure the editor tab no longer shows an unsaved-change marker. Help me check my requirements.txt file.

✔️ Awesome, I've got everything!

Great. Double-check that you saved requirements.txt.

ⓧ I'd like to double check the full code

fastapi==0.141.1
google-genai==2.24.0
pydantic==2.13.5
uvicorn==0.53.0

• Click New File... in the Explorer toolbar.
• Enter .gitignore as the file name.
• Add the project exclusions to .gitignore by pasting this code:

tutorial-env/
__pycache__/
*.pyc
data/templates.json

What do these exclusions protect?

• The Git rule for tutorial-env/ keeps installed packages out of version control.
• The __pycache__/ rule ignores generated cache folders.
• The *.pyc rule ignores compiled Python files.
• The data/templates.json rule keeps locally saved demo templates out of commits.

• Save .gitignore.
• Confirm .gitignore appears beside requirements.txt in the Explorer.

Is the ignore file missing?

Check that the file name starts with a full stop. A file named gitignore will not apply these Git rules.

Confirm that .gitignore is inside flowforge-ai. Help me check my ignore file.

✔️ Awesome, I've got everything!

Your two project files now match the setup required for FlowForge.

ⓧ I'd like to double check the full code

tutorial-env/
__pycache__/
*.pyc
data/templates.json

The Google Gen AI SDK requires Python 3.10 or newer. Checking now prevents package compatibility failures during installation.

• Open the integrated terminal by clicking View in the top menu.
• Select Terminal.

Before you check, what Python version do you expect this terminal to use?

• Check the active Python version by running this command:

python --version

What does this command check?

The command asks the active python executable to report its version. That executable creates the virtual environment in the next substep.

✔️ I see version 3.10 or higher

You're set. Python 3.10 or newer supports the pinned Google Gen AI SDK.

ⓧ I see an older version

The installed Python cannot run this dependency set. Install Python 3.10 or newer before continuing.

• Open the official Python Unix setup guide.
• Follow the Linux guidance for installing a supported Python release.
• Return to the VS Code integrated terminal.
• Repeat the version check shown above.
• Continue when python reports 3.10 or newer.

ⓧ Command not found

The terminal cannot find a Python installation through this command. Install a supported Python release before creating the environment.

• Open the official Python Unix setup guide.
• Follow the Linux guidance for installing Python.
• Return to the VS Code integrated terminal.
• Repeat the version check shown above.
• Continue when python reports 3.10 or newer.

Create and install the virtual environment

A virtual environment isolates FlowForge's packages from the rest of your computer. The tutorial-env environment uses the Python version you just confirmed.

• Create the isolated environment inside flowforge-ai by running this command:

python -m venv tutorial-env

What does this command create?

• The venv module creates an isolated Python environment.
• The tutorial-env name matches the directory excluded by .gitignore.

• Confirm tutorial-env appears under flowforge-ai in the Explorer.

Was the environment folder not created?

Confirm that the terminal is using the flowforge-ai workspace. Check that your Python version passed the supported-version tab above.

Look for an error mentioning the environment module or permissions. Help me create tutorial-env.

• Activate tutorial-env in the current terminal by running this command:

source tutorial-env/bin/activate

What does activation change?

• The source command runs the activation script in the current shell.
• The activated environment directs python to FlowForge's isolated interpreter.
• The activated environment directs pip to FlowForge's isolated package directory.

• Confirm the terminal prompt begins with (tutorial-env).

The first installation can take a few minutes while pip downloads the pinned packages. A short pause here is expected.

• Install the dependencies from requirements.txt by running this command:

python -m pip install -r requirements.txt

What does this installation use?

The command reads every exact package pin from requirements.txt. The active environment keeps those installations inside tutorial-env.

• Wait for the command to return to the terminal prompt.
• Confirm no installation error remains in the final terminal lines.

That's the dependency environment locked down. FlowForge now has a repeatable Python package set.

Did the dependency installation fail?

Confirm the prompt still begins with (tutorial-env). Check that all four lines in requirements.txt match the double-check tab.

Help me diagnose this installation error. Share the non-secret error output with the NextWork community if the environment-specific issue continues.

Export the key and verify the packages

Credentials deserve extra care. Exporting the Gemini API key in the server terminal keeps it out of the browser files you build later.

The Gemini client detects GEMINI_API_KEY automatically. This project uses the free tier with billing left disabled.

• Keep billing disabled for this project.
• Export the key to the server environment by replacing your-api-key before running this command:

export GEMINI_API_KEY='your-api-key'

How does the server use this key?

The export makes GEMINI_API_KEY available to processes started from this terminal. The Gemini client can read it without placing the credential in frontend code.

• Confirm the terminal returns to the prompt without an error.

Concerned about the API key?

Do not add the key to requirements.txt or any browser file. Do not include the export command in screenshots.

If an older key later fails authentication, create a current authorization key in Google AI Studio. Help me check my key setup safely.

Before you run the final check, which four pinned packages do you expect to find?

• Display the installed packages by running this command:

python -m pip list

What should you see?

• The fastapi row shows 0.141.1.
• The google-genai row shows 2.24.0.
• The pydantic row shows 2.13.5.
• The uvicorn row shows 0.53.0.

Your setup is complete. The workspace now has its pinned files and an activated server environment.

Are the expected versions missing?

Confirm that the prompt begins with (tutorial-env). A different prompt can mean the package list came from another Python environment.

Repeat the installation command after activating tutorial-env. Help me compare my package versions.

• Frame your screenshot so only the dependency rows are visible.

Your FlowForge workspace is ready for backend code. Next, you will define the workflow contract and expose the first API endpoints.

Build the Workflow API

The pinned environment is ready for a working FastAPI backend. This step gives FlowForge an API that accepts workflow definitions.

A submitted-list runner can appear flexible while blocks arrive in dependency order. Moving a valid prompt block ahead of its input exposes the limitation this first engine needs to demonstrate.

In this step, get ready to:

• Define the workflow request and response contract.
• Build a sequential engine with block handlers and timing traces.
• Expose the workflow API and demonstrate its ordering limitation.

Model the workflow contract

A workflow contract defines the shape of every block sent to the API. Pydantic uses these models to validate nested workflow data before execution begins.

• Select the flowforge-ai folder in the VS Code Explorer sidebar.
• Click the New File icon in the Explorer toolbar.
• Enter models.py as the file name.
• Add the workflow definition models to models.py by pasting this code:

from typing import Any, Literal

from pydantic import BaseModel, Field

BlockType = Literal["input", "prompt"]


class BlockDefinition(BaseModel):
    id: str = Field(min_length=1)
    type: BlockType
    depends_on: list[str] = Field(default_factory=list)
    config: dict[str, Any] = Field(default_factory=dict)


class WorkflowDefinition(BaseModel):
    name: str = Field(min_length=1)
    blocks: list[BlockDefinition] = Field(min_length=1)

What does this contract define?

• The BlockType type limits this first engine to input and prompt blocks.
• The BlockDefinition model gives each block an ID, a type, dependencies, and configuration.
• The WorkflowDefinition model groups at least one block under a workflow name.

• Save models.py.
• Confirm the editor recognizes BlockDefinition and WorkflowDefinition as class definitions.

Seeing model syntax errors?

Check that both classes start at the left edge of models.py. Confirm each field inside a class uses four spaces of indentation.

Make sure the imports include both BaseModel and Field. Help me diagnose the model definitions.

The API also needs a consistent record of what happened during execution. Trace and result models make each block's status, duration, output, or error visible in one response.

• Append the trace and result models below WorkflowDefinition by pasting this code:

class TraceEntry(BaseModel):
    block_id: str
    block_type: str
    status: Literal["completed", "failed"]
    duration_ms: float
    output: Any = None
    error: str | None = None


class RunResult(BaseModel):
    run_id: str
    workflow_name: str
    status: Literal["completed"]
    final_output: Any
    outputs: dict[str, Any]
    trace: list[TraceEntry]

What do the result models capture?

• The TraceEntry model records one block's status and execution duration.
• The optional output and error fields support successful and failed trace entries.
• The RunResult model returns the final value alongside every intermediate output.

• Save models.py.
• Confirm the editor now recognizes five model classes in models.py.

Missing a result model?

Check that TraceEntry and RunResult sit below WorkflowDefinition. Keep both classes at the left edge of the file.

Confirm that Any and Literal remain in the first import. Help me check the complete model file.

✔️ Awesome, I've got everything!

Great. Double-check that you saved models.py.

ⓧ I'd like to double check the full code

from typing import Any, Literal

from pydantic import BaseModel, Field

BlockType = Literal["input", "prompt"]


class BlockDefinition(BaseModel):
    id: str = Field(min_length=1)
    type: BlockType
    depends_on: list[str] = Field(default_factory=list)
    config: dict[str, Any] = Field(default_factory=dict)


class WorkflowDefinition(BaseModel):
    name: str = Field(min_length=1)
    blocks: list[BlockDefinition] = Field(min_length=1)


class TraceEntry(BaseModel):
    block_id: str
    block_type: str
    status: Literal["completed", "failed"]
    duration_ms: float
    output: Any = None
    error: str | None = None


class RunResult(BaseModel):
    run_id: str
    workflow_name: str
    status: Literal["completed"]
    final_output: Any
    outputs: dict[str, Any]
    trace: list[TraceEntry]

Execute submitted blocks

The first engine uses a block registry to connect each block type to a Python handler. It executes blocks in the same order they appear in the request.

Every handler reads completed dependency outputs from a shared dictionary. A missing output becomes a workflow error with a clear dependency message.

• Select the flowforge-ai folder in the Explorer sidebar.
• Click the New File icon in the Explorer toolbar.
• Enter engine.py as the file name.
• Add the engine imports and dependency checks to engine.py by pasting this code:

import json
from time import perf_counter
from typing import Any, Callable
from uuid import uuid4

from models import BlockDefinition, RunResult, TraceEntry, WorkflowDefinition

BlockHandler = Callable[[BlockDefinition, dict[str, Any]], Any]


class WorkflowError(Exception):
    def __init__(self, message: str, trace: list[TraceEntry] | None = None):
        super().__init__(message)
        self.trace = trace or []


def dependency_values(
    block: BlockDefinition, outputs: dict[str, Any]
) -> list[Any]:
    missing = [dependency for dependency in block.depends_on if dependency not in outputs]
    if missing:
        raise WorkflowError(
            f"Block '{block.id}' is waiting for dependencies: {', '.join(missing)}"
        )
    return [outputs[dependency] for dependency in block.depends_on]

How are dependencies checked?

• The BlockHandler alias describes the shared input and output shape for handlers.
• The WorkflowError exception carries an optional execution trace.
• The dependency_values() function reports dependencies that do not have an output yet.

• Save engine.py.
• Confirm the editor recognizes WorkflowError and dependency_values().

Seeing unresolved model imports?

Confirm engine.py sits beside models.py in the flowforge-ai folder. Check that models.py is saved.

Verify that each imported class uses the same capitalization as its definition. Help me fix the engine imports.

Input blocks introduce raw values into a run. Prompt blocks combine dependency outputs with a string template.

• Append the value formatter and block handlers below dependency_values() by pasting this code:

def format_value(value: Any) -> str:
    if isinstance(value, list):
        return "\n".join(str(item) for item in value)
    if isinstance(value, dict):
        return json.dumps(value, indent=2)
    return str(value)


def input_block(block: BlockDefinition, outputs: dict[str, Any]) -> Any:
    if "value" not in block.config:
        raise WorkflowError(f"Input block '{block.id}' requires config.value")
    return block.config["value"]


def prompt_block(block: BlockDefinition, outputs: dict[str, Any]) -> str:
    dependency_values(block, outputs)
    template = block.config.get("template")
    if not isinstance(template, str):
        raise WorkflowError(f"Prompt block '{block.id}' requires config.template")

    variables = {
        dependency: format_value(outputs[dependency])
        for dependency in block.depends_on
    }
    try:
        return template.format(**variables)
    except KeyError as error:
        raise WorkflowError(
            f"Prompt block '{block.id}' references unknown placeholder {error}"
        ) from error

How do the first handlers work?

• The format_value() function converts lists and dictionaries into prompt-friendly text.
• The input_block() handler returns the configured input value.
• The prompt_block() handler maps dependency IDs to template placeholders.
• A missing placeholder becomes a WorkflowError instead of an unstructured formatting failure.

• Save engine.py.
• Confirm the editor recognizes format_value(), input_block(), and prompt_block().

Prompt handler showing an error?

Check the indentation inside the variables dictionary comprehension. Confirm the except KeyError block lines up with try.

Make sure json is imported at the top of engine.py. Help me debug the prompt handler.

The registry keeps dispatch separate from the execution loop. The loop stores each output before the next submitted block begins.

• Append the registry and sequential runner below prompt_block() by pasting this code:

BLOCK_HANDLERS: dict[str, BlockHandler] = {
    "input": input_block,
    "prompt": prompt_block,
}


def run_workflow(workflow: WorkflowDefinition) -> RunResult:
    outputs: dict[str, Any] = {}
    trace: list[TraceEntry] = []

    for block in workflow.blocks:
        started = perf_counter()
        try:
            output = BLOCK_HANDLERS[block.type](block, outputs)
            outputs[block.id] = output
            trace.append(
                TraceEntry(block_id=block.id, block_type=block.type, status="completed", duration_ms=round((perf_counter() - started) * 1000, 2), output=output)
            )
        except Exception as error:
            trace.append(
                TraceEntry(block_id=block.id, block_type=block.type, status="failed", duration_ms=round((perf_counter() - started) * 1000, 2), error=str(error))
            )
            if isinstance(error, WorkflowError):
                raise WorkflowError(str(error), trace) from error
            raise WorkflowError(f"Block '{block.id}' failed: {error}", trace) from error

    return RunResult(run_id=str(uuid4()), workflow_name=workflow.name, status="completed", final_output=outputs[workflow.blocks[-1].id], outputs=outputs, trace=trace)

How does the runner execute a workflow?

• The BLOCK_HANDLERS registry selects a function from each block's type.
• The loop processes workflow.blocks from the first submitted block to the last.
• The outputs dictionary makes completed values available to later handlers.
• The trace list records duration and outcome data for every attempted block.

• Save engine.py.
• Confirm the editor recognizes BLOCK_HANDLERS and run_workflow().

Runner structure showing errors?

Confirm the except block lines up with try inside the loop. Keep the final return RunResult line outside the loop.

Check that both registry keys match the values allowed by BlockType. Help me inspect the sequential runner.

✔️ Awesome, I've got everything!

Great. Double-check that you saved engine.py.

ⓧ I'd like to double check the full code

import json
from time import perf_counter
from typing import Any, Callable
from uuid import uuid4

from models import BlockDefinition, RunResult, TraceEntry, WorkflowDefinition

BlockHandler = Callable[[BlockDefinition, dict[str, Any]], Any]


class WorkflowError(Exception):
    def __init__(self, message: str, trace: list[TraceEntry] | None = None):
        super().__init__(message)
        self.trace = trace or []


def dependency_values(
    block: BlockDefinition, outputs: dict[str, Any]
) -> list[Any]:
    missing = [dependency for dependency in block.depends_on if dependency not in outputs]
    if missing:
        raise WorkflowError(
            f"Block '{block.id}' is waiting for dependencies: {', '.join(missing)}"
        )
    return [outputs[dependency] for dependency in block.depends_on]


def format_value(value: Any) -> str:
    if isinstance(value, list):
        return "\n".join(str(item) for item in value)
    if isinstance(value, dict):
        return json.dumps(value, indent=2)
    return str(value)


def input_block(block: BlockDefinition, outputs: dict[str, Any]) -> Any:
    if "value" not in block.config:
        raise WorkflowError(f"Input block '{block.id}' requires config.value")
    return block.config["value"]


def prompt_block(block: BlockDefinition, outputs: dict[str, Any]) -> str:
    dependency_values(block, outputs)
    template = block.config.get("template")
    if not isinstance(template, str):
        raise WorkflowError(f"Prompt block '{block.id}' requires config.template")

    variables = {
        dependency: format_value(outputs[dependency])
        for dependency in block.depends_on
    }
    try:
        return template.format(**variables)
    except KeyError as error:
        raise WorkflowError(
            f"Prompt block '{block.id}' references unknown placeholder {error}"
        ) from error


BLOCK_HANDLERS: dict[str, BlockHandler] = {
    "input": input_block,
    "prompt": prompt_block,
}


def run_workflow(workflow: WorkflowDefinition) -> RunResult:
    outputs: dict[str, Any] = {}
    trace: list[TraceEntry] = []

    for block in workflow.blocks:
        started = perf_counter()
        try:
            output = BLOCK_HANDLERS[block.type](block, outputs)
            outputs[block.id] = output
            trace.append(
                TraceEntry(block_id=block.id, block_type=block.type, status="completed", duration_ms=round((perf_counter() - started) * 1000, 2), output=output)
            )
        except Exception as error:
            trace.append(
                TraceEntry(block_id=block.id, block_type=block.type, status="failed", duration_ms=round((perf_counter() - started) * 1000, 2), error=str(error))
            )
            if isinstance(error, WorkflowError):
                raise WorkflowError(str(error), trace) from error
            raise WorkflowError(f"Block '{block.id}' failed: {error}", trace) from error

    return RunResult(run_id=str(uuid4()), workflow_name=workflow.name, status="completed", final_output=outputs[workflow.blocks[-1].id], outputs=outputs, trace=trace)

Expose and test the API

The application layer turns workflow results into HTTP responses. It also converts a WorkflowError into a structured client error with the trace collected so far.

• Select the flowforge-ai folder in the Explorer sidebar.
• Click the New File icon in the Explorer toolbar.
• Enter main.py as the file name.
• Add the health and workflow endpoints to main.py by pasting this code:

from fastapi import FastAPI, HTTPException

from engine import WorkflowError, run_workflow
from models import RunResult, WorkflowDefinition

app = FastAPI(title="FlowForge Gemini", version="1.0.0")


@app.get("/api/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/api/runs", response_model=RunResult)
def create_run(workflow: WorkflowDefinition) -> RunResult:
    try:
        return run_workflow(workflow)
    except WorkflowError as error:
        raise HTTPException(
            status_code=400,
            detail={
                "message": str(error),
                "trace": [entry.model_dump() for entry in error.trace],
            },
        ) from error

What does the API expose?

• The GET /api/health endpoint returns a small server health response.
• The POST /api/runs endpoint validates a WorkflowDefinition before running it.
• A failed workflow returns HTTP status 400 with its message and partial trace.

• Save main.py.
• Confirm main.py appears beside models.py and engine.py in the Explorer sidebar.

API imports not resolving?

Confirm all three Python files sit directly inside flowforge-ai. Save each file before starting the server.

Check that WorkflowError and run_workflow use the same spelling in both files. Help me diagnose the API imports.

✔️ Awesome, I've got everything!

Great. Double-check that you saved main.py.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException

from engine import WorkflowError, run_workflow
from models import RunResult, WorkflowDefinition

app = FastAPI(title="FlowForge Gemini", version="1.0.0")


@app.get("/api/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/api/runs", response_model=RunResult)
def create_run(workflow: WorkflowDefinition) -> RunResult:
    try:
        return run_workflow(workflow)
    except WorkflowError as error:
        raise HTTPException(
            status_code=400,
            detail={
                "message": str(error),
                "trace": [entry.model_dump() for entry in error.trace],
            },
        ) from error

Uvicorn loads the app object from main.py. The running process keeps the API available while you test requests through the generated documentation.

• Start the FlowForge API in the integrated terminal from earlier by running this command:

uvicorn main:app

• Confirm the terminal reports that the server is running at http://127.0.0.1:8000.

Server not starting?

Confirm the terminal still shows tutorial-env as active. Check that the terminal is inside the flowforge-ai folder.

Read the first traceback line that names one of your files. Help me troubleshoot the server startup.

• Open the FastAPI documentation in your browser.
• Expand POST /api/runs in the documentation.
• Enable editing for the request body.
• Replace the request body with this ordered workflow:

{
  "name": "ordered-demo",
  "blocks": [
    {
      "id": "question",
      "type": "input",
      "depends_on": [],
      "config": {
        "value": "What does FlowForge automate?"
      }
    },
    {
      "id": "prompt",
      "type": "prompt",
      "depends_on": ["question"],
      "config": {
        "template": "Question: {question}"
      }
    }
  ]
}

How is the ordered workflow connected?

• The question block stores the raw input value.
• The prompt block lists question as its dependency.
• The prompt template uses the dependency ID as its formatting placeholder.

Before you submit this workflow, what final value do you expect the prompt block to produce?

• Submit the ordered request.

You will see status set to completed. The final output reads Question: What does FlowForge automate?.

That first loop works. FlowForge can now validate a workflow, execute two blocks, and return their timing traces.

Ordered request not completing?

Check that the input block appears before the prompt block. Confirm the prompt dependency and template placeholder both use question.

Look at the response detail for the first failed block. Help me debug the ordered request.

A dependency declaration describes which output a block needs. This first runner still trusts the submitted array to place that dependency earlier.

• Replace the request body with the same blocks in reversed order by pasting this JSON:

{
  "name": "out-of-order-demo",
  "blocks": [
    {
      "id": "prompt",
      "type": "prompt",
      "depends_on": ["question"],
      "config": {
        "template": "Question: {question}"
      }
    },
    {
      "id": "question",
      "type": "input",
      "depends_on": [],
      "config": {
        "value": "What does FlowForge automate?"
      }
    }
  ]
}

What changed in this request?

• The prompt block now appears first in the submitted list.
• Its dependency still points to the valid question block.
• The workflow graph contains the same two blocks and the same dependency.

Before you submit the shuffled workflow, do you think the dependency declaration can overcome the submitted order?

• Submit the out-of-order request.

You will see an HTTP 400 response with Block 'prompt' is waiting for dependencies: question. The prompt ran before the input produced its value.

This failure is intentional. You have proved that dependency metadata alone does not make submitted-list execution dependency-aware.

Seeing a completed shuffled run?

Confirm the prompt object appears before the question object in the request body. Check that the running server uses the sequential run_workflow() from this step.

Restarting an older server process may be necessary after file changes. Help me reproduce the ordering limitation.

The workflow API now exposes the exact orchestration flaw the next build needs to solve. Next, you will make execution follow dependencies while adding retrieval and Gemini generation blocks.

Add Retrieval and Gemini Execution

Your ordered workflow proved that the backend can execute blocks. The shuffled version exposed how submitted-list execution breaks when a dependency appears later in the request.

This step turns that limitation into a dependency-aware directed acyclic graph executor. You will also add local retrieval plus Gemini generation as interchangeable workflow blocks.

In this step, get ready to:

• Add deterministic local retrieval and Gemini generation helpers.
• Replace list-order execution with validated topological scheduling.
• Prove a shuffled four-block workflow runs in dependency order.

Add retrieval and generation helpers

The Retrieve block uses lexical retrieval to score local facts by shared tokens. This keeps retrieval deterministic while preserving the Retrieve, Augment, Generate pipeline.

Why local lexical retrieval?

Token overlap selects facts without making a second remote request. It also gives you a scoring method that you can inspect directly.

This MVP treats retrieval as local ranking. The Gemini request only handles the final generation stage.

• In the VS Code Explorer sidebar, select models.py.
• Replace the existing BlockType declaration with this expanded declaration:

BlockType = Literal["input", "retrieve", "prompt", "llm"]

What does this declaration change?

• The retrieve type represents local context selection.
• The llm type represents the Gemini generation request.
• Pydantic rejects workflow blocks whose type falls outside this contract.

• Save models.py.
• Confirm the declaration now lists four supported block types in the editor.

• Create ai_blocks.py inside the flowforge-ai folder using the VS Code Explorer toolbar.
• Add the SDK client plus the local retrieval logic by pasting this first code chunk:

import re

from google import genai

GENERATION_MODEL = "gemini-3.8-flash"

client = genai.Client()


def token_set(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9]+", text.lower()))


def retrieve_context(query: str, documents: list[str], top_k: int = 2) -> list[str]:
    if not documents:
        return []

    query_terms = token_set(query)
    scored: list[tuple[float, int, str]] = []
    for index, document in enumerate(documents):
        document_terms = token_set(document)
        overlap = len(query_terms & document_terms)
        coverage = overlap / max(1, len(query_terms))
        scored.append((coverage, -index, document))

    scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
    return [document for _, _, document in scored[:top_k]]

How does retrieval work?

• The Google Gen AI SDK client reads the exported GEMINI_API_KEY from the server environment.
• The token_set() function normalizes text into unique lowercase tokens.
• The retrieve_context() function calculates how much of the query each document covers.
• The original document position resolves equal scores predictably.

• Add the generation helper below retrieve_context() by pasting this code:

def generate_text(prompt: str) -> str:
    response = client.models.generate_content(
        model=GENERATION_MODEL,
        contents=(
            "Answer only from the supplied workflow context. "
            "If the context is insufficient, say that the workflow lacks enough context.\n\n"
            f"{prompt}"
        ),
    )
    if response.text is None:
        raise RuntimeError("Gemini returned no text output")
    return response.text

What does generation do?

• The generate_text() function sends the composed prompt to Gemini.
• The instruction limits the answer to context supplied by the workflow.
• The helper raises a clear runtime error if the response contains no text.

• Save ai_blocks.py.
• Confirm ai_blocks.py appears beside engine.py in the Explorer sidebar.

Can't save the AI helpers?

Check that ai_blocks.py is inside the same flowforge-ai folder as engine.py. Confirm that the file name uses an underscore.

Compare the indentation inside both functions. Help me check ai_blocks.py for indentation or import problems.

✔️ Awesome, I've got everything!

Your model contract now accepts four block types. Double-check that you saved models.py plus ai_blocks.py.

ⓧ I'd like to double check the full code

from typing import Any, Literal

from pydantic import BaseModel, Field

BlockType = Literal["input", "retrieve", "prompt", "llm"]


class BlockDefinition(BaseModel):
    id: str = Field(min_length=1)
    type: BlockType
    depends_on: list[str] = Field(default_factory=list)
    config: dict[str, Any] = Field(default_factory=dict)


class WorkflowDefinition(BaseModel):
    name: str = Field(min_length=1)
    blocks: list[BlockDefinition] = Field(min_length=1)


class TraceEntry(BaseModel):
    block_id: str
    block_type: str
    status: Literal["completed", "failed"]
    duration_ms: float
    output: Any = None
    error: str | None = None


class RunResult(BaseModel):
    run_id: str
    workflow_name: str
    status: Literal["completed"]
    final_output: Any
    outputs: dict[str, Any]
    trace: list[TraceEntry]

import re

from google import genai

GENERATION_MODEL = "gemini-3.8-flash"

client = genai.Client()


def token_set(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9]+", text.lower()))


def retrieve_context(query: str, documents: list[str], top_k: int = 2) -> list[str]:
    if not documents:
        return []

    query_terms = token_set(query)
    scored: list[tuple[float, int, str]] = []
    for index, document in enumerate(documents):
        document_terms = token_set(document)
        overlap = len(query_terms & document_terms)
        coverage = overlap / max(1, len(query_terms))
        scored.append((coverage, -index, document))

    scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
    return [document for _, _, document in scored[:top_k]]


def generate_text(prompt: str) -> str:
    response = client.models.generate_content(
        model=GENERATION_MODEL,
        contents=(
            "Answer only from the supplied workflow context. "
            "If the context is insufficient, say that the workflow lacks enough context.\n\n"
            f"{prompt}"
        ),
    )
    if response.text is None:
        raise RuntimeError("Gemini returned no text output")
    return response.text

Replace list-order execution

A block registry maps each workflow type to its handler. A topological sort then schedules every block only after its dependencies are ready.

Why topological execution?

Workflow JSON describes dependency relationships through depends_on. The submitted array position carries no execution meaning.

Topological scheduling converts those relationships into a safe order. Validation rejects duplicate IDs plus missing references before any block runs.

• In the VS Code Explorer sidebar, select engine.py.
• Replace the existing sequential engine with the complete dependency-aware engine shown in the double-check tab below.
• Save engine.py after the replacement.

✔️ I've updated the engine

Confirm that engine.py now defines retrieve_block() plus llm_block(). Confirm that run_workflow() calls validate_and_sort().

ⓧ I'd like to double check the full code

import json
from collections import defaultdict, deque
from time import perf_counter
from typing import Any, Callable
from uuid import uuid4

from ai_blocks import generate_text, retrieve_context
from models import BlockDefinition, RunResult, TraceEntry, WorkflowDefinition

BlockHandler = Callable[[BlockDefinition, dict[str, Any]], Any]


class WorkflowError(Exception):
    def __init__(self, message: str, trace: list[TraceEntry] | None = None):
        super().__init__(message)
        self.trace = trace or []


def dependency_values(
    block: BlockDefinition, outputs: dict[str, Any]
) -> list[Any]:
    missing = [dependency for dependency in block.depends_on if dependency not in outputs]
    if missing:
        raise WorkflowError(
            f"Block '{block.id}' is waiting for dependencies: {', '.join(missing)}"
        )
    return [outputs[dependency] for dependency in block.depends_on]


def format_value(value: Any) -> str:
    if isinstance(value, list):
        return "\n".join(str(item) for item in value)
    if isinstance(value, dict):
        return json.dumps(value, indent=2)
    return str(value)


def input_block(block: BlockDefinition, outputs: dict[str, Any]) -> Any:
    if "value" not in block.config:
        raise WorkflowError(f"Input block '{block.id}' requires config.value")
    return block.config["value"]


def retrieve_block(block: BlockDefinition, outputs: dict[str, Any]) -> list[str]:
    values = dependency_values(block, outputs)
    if len(values) != 1:
        raise WorkflowError(f"Retrieve block '{block.id}' requires one dependency")

    documents = block.config.get("documents", [])
    if not isinstance(documents, list) or not all(
        isinstance(document, str) for document in documents
    ):
        raise WorkflowError(f"Retrieve block '{block.id}' requires string documents")

    top_k = max(1, int(block.config.get("top_k", 2)))
    return retrieve_context(str(values[0]), documents, top_k)


def prompt_block(block: BlockDefinition, outputs: dict[str, Any]) -> str:
    dependency_values(block, outputs)
    template = block.config.get("template")
    if not isinstance(template, str):
        raise WorkflowError(f"Prompt block '{block.id}' requires config.template")

    variables = {
        dependency: format_value(outputs[dependency])
        for dependency in block.depends_on
    }
    try:
        return template.format(**variables)
    except KeyError as error:
        raise WorkflowError(
            f"Prompt block '{block.id}' references unknown placeholder {error}"
        ) from error


def llm_block(block: BlockDefinition, outputs: dict[str, Any]) -> str:
    values = dependency_values(block, outputs)
    if len(values) != 1:
        raise WorkflowError(f"AI block '{block.id}' requires one prompt dependency")
    return generate_text(format_value(values[0]))


BLOCK_HANDLERS: dict[str, BlockHandler] = {
    "input": input_block,
    "retrieve": retrieve_block,
    "prompt": prompt_block,
    "llm": llm_block,
}


def validate_and_sort(workflow: WorkflowDefinition) -> list[BlockDefinition]:
    block_by_id: dict[str, BlockDefinition] = {}
    for block in workflow.blocks:
        if block.id in block_by_id:
            raise WorkflowError(f"Duplicate block id: '{block.id}'")
        if block.type not in BLOCK_HANDLERS:
            raise WorkflowError(f"Unsupported block type: '{block.type}'")
        block_by_id[block.id] = block

    indegree = {block.id: len(block.depends_on) for block in workflow.blocks}
    dependents: dict[str, list[str]] = defaultdict(list)

    for block in workflow.blocks:
        for dependency in block.depends_on:
            if dependency not in block_by_id:
                raise WorkflowError(
                    f"Block '{block.id}' references missing dependency '{dependency}'"
                )
            dependents[dependency].append(block.id)

    ready = deque(
        block.id for block in workflow.blocks if indegree[block.id] == 0
    )
    ordered_ids: list[str] = []

    while ready:
        block_id = ready.popleft()
        ordered_ids.append(block_id)
        for dependent in dependents[block_id]:
            indegree[dependent] -= 1
            if indegree[dependent] == 0:
                ready.append(dependent)

    if len(ordered_ids) != len(workflow.blocks):
        raise WorkflowError("Workflow contains a dependency cycle")

    return [block_by_id[block_id] for block_id in ordered_ids]


def run_workflow(workflow: WorkflowDefinition) -> RunResult:
    ordered_blocks = validate_and_sort(workflow)
    outputs: dict[str, Any] = {}
    trace: list[TraceEntry] = []

    for block in ordered_blocks:
        started = perf_counter()
        try:
            output = BLOCK_HANDLERS[block.type](block, outputs)
            outputs[block.id] = output
            trace.append(
                TraceEntry(
                    block_id=block.id,
                    block_type=block.type,
                    status="completed",
                    duration_ms=round((perf_counter() - started) * 1000, 2),
                    output=output,
                )
            )
        except Exception as error:
            trace.append(
                TraceEntry(
                    block_id=block.id,
                    block_type=block.type,
                    status="failed",
                    duration_ms=round((perf_counter() - started) * 1000, 2),
                    error=str(error),
                )
            )
            if isinstance(error, WorkflowError):
                raise WorkflowError(str(error), trace) from error
            raise WorkflowError(f"Block '{block.id}' failed: {error}", trace) from error

    return RunResult(
        run_id=str(uuid4()),
        workflow_name=workflow.name,
        status="completed",
        final_output=outputs[ordered_blocks[-1].id],
        outputs=outputs,
        trace=trace,
    )

How does the new engine run a graph?

• The BLOCK_HANDLERS registry gives each block type one execution function.
• The validate_and_sort() function builds dependency counts for every block.
• A queue preserves submitted order among blocks that are ready at the same time.
• The executor records each completed output plus its duration in the trace.

Engine import failing?

Confirm that ai_blocks.py sits beside engine.py. Check that both imported helper names match their definitions exactly.

Compare the closing brackets around BLOCK_HANDLERS plus validate_and_sort(). Help me diagnose an import or syntax error in the FlowForge engine.

Run a shuffled RAG workflow

The decisive test places the Gemini block first in the submitted JSON. Its dependencies still force the executor to run Input, Retrieve, Prompt, and Gemini in that order.

Use only non-sensitive demonstration facts in this request. Free-tier Gemini content may be used to improve Google's products.

• Return to the integrated terminal running Uvicorn from earlier.
• Stop the current server by pressing Ctrl+C.
• Restart the server with the updated modules by running this command:

uvicorn main:app

What does this command do?

Uvicorn imports the app object from main.py. The restarted process now loads the expanded models plus the dependency-aware engine.

The terminal stays occupied while the server accepts requests. Keep this process running during the workflow test.

• Switch back to the FastAPI documentation at http://127.0.0.1:8000/docs.
• Refresh the documentation page.
• Expand the POST /api/runs endpoint.
• Enable the endpoint's interactive request editor.
• Replace the request body with this shuffled workflow:

{"name":"standup-assistant","blocks":[{"id":"ai","type":"llm","depends_on":["prompt"],"config":{}},{"id":"prompt","type":"prompt","depends_on":["question","retrieval"],"config":{"template":"Question:\n{question}\n\nContext:\n{retrieval}\n\nGive a concise answer grounded in the context."}},{"id":"retrieval","type":"retrieve","depends_on":["question"],"config":{"documents":["Daily standups happen at 9:30 AM.","Each update should cover progress, plans, and blockers.","Blockers lasting more than one day must be escalated to the engineering manager.","Friday standups include a short retrospective."],"top_k":2}},{"id":"question","type":"input","depends_on":[],"config":{"value":"Summarize the standup policy and escalation rule."}}]}

What does this workflow test?

• The JSON deliberately submits ai before its prompt dependency.
• The retrieval block selects two local facts related to the question.
• The prompt block inserts the question plus retrieved context into its template.
• The final block sends that grounded prompt to Gemini.

Before you send the request, which block do you expect to appear first in the execution trace?

• Send the shuffled request from the interactive documentation.

You should see status set to completed. The generated final_output should summarize the standup policy plus the escalation rule.

The four trace entries should appear as question, retrieval, prompt, and ai. Dependency order now controls execution.

Workflow not completing?

Confirm that the restarted terminal still has tutorial-env activated. Confirm that GEMINI_API_KEY remains exported in that terminal session.

If the response mentions a rate limit, wait before retrying the request. Help me diagnose a failed shuffled FlowForge workflow.

That is the orchestration core working: shuffled JSON now becomes a validated retrieval and Gemini pipeline. Next, you will give this backend a browser interface that compiles user controls into the same workflow contract.

Build the Browser Workflow Builder

Your FastAPI backend can already schedule a shuffled workflow. The next goal is to make that engine accessible through a browser interface that produces a Gemini answer.

The form compiles visible controls into the same JSON contract you tested earlier. The Visual Studio Code Integrated Browser keeps the complete demonstration inside your workspace.

In this step, get ready to:

• Build the four-block workflow page with its configuration panels.
• Style the page as a responsive browser application.
• Connect the controls to the workflow API and render each execution trace.

Create the workflow page

The HTML page turns the backend contract into four visible blocks. Its form captures the question, knowledge base, prompt template, and workflow name.

• Switch back to the Explorer sidebar in VS Code.
• Select the static folder inside flowforge-ai.
• Click the New File icon in the Explorer toolbar.
• Enter index.html as the file name.
• Build the page shell in index.html by pasting this code:

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>FlowForge Gemini</title>
  <link rel="stylesheet" href="/static/styles.css">
</head>
<body>
  <main class="shell">
    <header>
      <p class="eyebrow">Gemini workflow automation</p>
      <h1>FlowForge</h1>
      <p>Configure a retrieval-augmented pipeline, save it, and inspect every block.</p>
    </header>

    <section class="pipeline" aria-label="Workflow blocks">
      <div class="block">Input</div><span>→</span>
      <div class="block">Retrieve</div><span>→</span>
      <div class="block">Prompt</div><span>→</span>
      <div class="block accent">Gemini</div>
    </section>

    <section class="grid">
    </section>
  </main>
  <script src="/static/app.js"></script>
</body>
</html>

What Does This Structure Do?

• The shell class provides one container for the complete interface.
• The pipeline section shows the four stages that execute on the backend.
• The grid section reserves space for the workflow form and output panel.
• The final script tag loads the browser logic from app.js.

• Save index.html.
• Press Cmd+Shift+P on macOS or Ctrl+Shift+P on Windows to open the Command Palette.
• Select Browser: Open Integrated Browser.
• Navigate to http://127.0.0.1:8000.

You will see the FlowForge heading followed by Input, Retrieve, Prompt, and Gemini blocks. The page is unstyled because its stylesheet is still empty.

Homepage Not Loading?

Confirm that index.html is inside the static folder. Check the terminal from earlier to confirm Uvicorn is still running.

A server restart may be required if the static folder was created after Uvicorn started. Help me diagnose why FastAPI is not serving static/index.html.

The next section captures the values required by the four backend block types. Each field has a stable element ID that the browser script uses later.

• In index.html, find the empty grid section.
• Add the workflow form inside that section by pasting this code:

      <form id="workflow-form" class="panel">
        <h2>Pipeline Builder</h2>
        <label>Workflow name
          <input id="workflow-name" value="standup-assistant" required>
        </label>
        <label>User question
          <textarea id="question" rows="3" required>Summarize the standup policy and escalation rule.</textarea>
        </label>
        <label>Knowledge base, one fact per line
          <textarea id="documents" rows="6" required>Daily standups happen at 9:30 AM.
Each update should cover progress, plans, and blockers.
Blockers lasting more than one day must be escalated to the engineering manager.
Friday standups include a short retrospective.</textarea>
        </label>
        <label>Prompt template
          <textarea id="prompt-template" rows="6" required>Question:
{question}

Context:
{retrieval}

Give a concise answer grounded in the context.</textarea>
        </label>
        <div class="actions">
          <button type="submit">Run Workflow</button>
          <button type="button" id="save-template" class="secondary">Save Template</button>
        </div>
      </form>

How Does the Form Map to the Workflow?

• The workflow name identifies the run and any saved template.
• The user question becomes the value of the input block.
• Each knowledge-base line becomes one document for local retrieval.
• The prompt template combines the question with the selected retrieval output.

• Save index.html.
• Refresh http://127.0.0.1:8000 in the Integrated Browser.

You will see the Pipeline Builder with four populated controls. You will also see the Run Workflow and Save Template buttons.

Form Fields Missing?

Confirm that the form sits between the opening and closing tags of the grid section. A form placed after </html> can render unpredictably.

Check that every textarea has its closing tag. Help me find a missing HTML closing tag in my FlowForge form.

The output side of the page makes execution observable. It separates the generated answer from the trace that explains how the graph ran.

• In index.html, find the closing </form> tag.
• Replace the closing section tag beneath the form with this panel structure:

      <section class="panel">
        <h2>Template Library</h2>
        <div class="template-row">
          <select id="template-list" aria-label="Saved templates"></select>
          <button id="load-template" class="secondary">Load</button>
        </div>
        <h2>Final Output</h2>
        <pre id="final-output">Run the workflow to see an answer.</pre>
        <p id="status" role="status"></p>
      </section>
    </section>

    <section class="panel">
      <h2>Execution Trace</h2>
      <table>
        <thead><tr><th>Block</th><th>Type</th><th>Status</th><th>Duration</th><th>Output</th></tr></thead>
        <tbody id="trace-body"></tbody>
      </table>
    </section>

    <section class="panel">
      <h2>Recent Runs</h2>
      <ul id="run-history"></ul>
    </section>

What Do These Panels Show?

• The Template Library contains controls for saved workflow definitions.
• The Final Output panel displays the last block's result.
• The Execution Trace table reserves one row for every completed block.
• The Recent Runs list displays the latest in-memory API results.

• Save index.html.
• Refresh the Integrated Browser.

You will see the Template Library, Final Output, Execution Trace, and Recent Runs panels beneath the workflow controls.

✔️ Awesome, I've got everything!

Great. Confirm that you saved static/index.html before moving on.

ⓧ I'd like to double check the full code

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>FlowForge Gemini</title>
  <link rel="stylesheet" href="/static/styles.css">
</head>
<body>
  <main class="shell">
    <header>
      <p class="eyebrow">Gemini workflow automation</p>
      <h1>FlowForge</h1>
      <p>Configure a retrieval-augmented pipeline, save it, and inspect every block.</p>
    </header>

    <section class="pipeline" aria-label="Workflow blocks">
      <div class="block">Input</div><span>→</span>
      <div class="block">Retrieve</div><span>→</span>
      <div class="block">Prompt</div><span>→</span>
      <div class="block accent">Gemini</div>
    </section>

    <section class="grid">
      <form id="workflow-form" class="panel">
        <h2>Pipeline Builder</h2>
        <label>Workflow name
          <input id="workflow-name" value="standup-assistant" required>
        </label>
        <label>User question
          <textarea id="question" rows="3" required>Summarize the standup policy and escalation rule.</textarea>
        </label>
        <label>Knowledge base, one fact per line
          <textarea id="documents" rows="6" required>Daily standups happen at 9:30 AM.
Each update should cover progress, plans, and blockers.
Blockers lasting more than one day must be escalated to the engineering manager.
Friday standups include a short retrospective.</textarea>
        </label>
        <label>Prompt template
          <textarea id="prompt-template" rows="6" required>Question:
{question}

Context:
{retrieval}

Give a concise answer grounded in the context.</textarea>
        </label>
        <div class="actions">
          <button type="submit">Run Workflow</button>
          <button type="button" id="save-template" class="secondary">Save Template</button>
        </div>
      </form>

      <section class="panel">
        <h2>Template Library</h2>
        <div class="template-row">
          <select id="template-list" aria-label="Saved templates"></select>
          <button id="load-template" class="secondary">Load</button>
        </div>
        <h2>Final Output</h2>
        <pre id="final-output">Run the workflow to see an answer.</pre>
        <p id="status" role="status"></p>
      </section>
    </section>

    <section class="panel">
      <h2>Execution Trace</h2>
      <table>
        <thead><tr><th>Block</th><th>Type</th><th>Status</th><th>Duration</th><th>Output</th></tr></thead>
        <tbody id="trace-body"></tbody>
      </table>
    </section>

    <section class="panel">
      <h2>Recent Runs</h2>
      <ul id="run-history"></ul>
    </section>
  </main>
  <script src="/static/app.js"></script>
</body>
</html>

Style the responsive interface

The CSS gives each workflow block a clear visual boundary. The dark panels keep the controls, answer, and trace readable within one screen.

• Select the static folder in the Explorer sidebar.
• Click the New File icon in the Explorer toolbar.
• Enter styles.css as the file name.
• Add the page theme and pipeline styles to styles.css by pasting this code:

:root {
  color-scheme: dark;
  font-family: Inter, ui-sans-serif, system-ui, sans-serif;
  background: #08111f;
  color: #e8eef8;
}

* { box-sizing: border-box; }
body { margin: 0; min-height: 100vh; background: radial-gradient(circle at top, #15325b, #08111f 42%); }
.shell { width: min(1100px, 92vw); margin: 0 auto; padding: 40px 0 64px; }
header { margin-bottom: 24px; }
.eyebrow { color: #78dce8; text-transform: uppercase; letter-spacing: .16em; font-size: .78rem; }
h1 { margin: 0; font-size: clamp(2.5rem, 7vw, 5rem); }
h2 { margin-top: 0; font-size: 1.05rem; }
.pipeline { display: flex; align-items: center; gap: 10px; margin: 28px 0; overflow-x: auto; }
.block { min-width: 130px; padding: 18px; border: 1px solid #365375; border-radius: 14px; background: #10213a; text-align: center; font-weight: 700; }
.block.accent { border-color: #78dce8; box-shadow: 0 0 26px #78dce833; }

How Does the Pipeline Styling Work?

• The root colors establish a shared dark palette for the page.
• The shell constrains the interface width while leaving space on smaller screens.
• The pipeline uses a horizontal flex layout with scrolling for narrow screens.
• The accent class gives the Gemini block a brighter border and glow.

• Save styles.css.
• Refresh the Integrated Browser.

You will see a dark blue page with four bordered workflow blocks. The Gemini block has a brighter cyan outline.

Styles Not Appearing?

Confirm that styles.css is inside static. Check that the stylesheet link still uses /static/styles.css.

Use a full browser refresh after saving the file. Help me diagnose why static/styles.css is not styling my FastAPI page.

The remaining styles turn the controls and result areas into reusable panels. The responsive rule collapses the two-column layout on smaller screens.

• Append the panel, form, table, and responsive styles to styles.css by pasting this code:

.grid { display: grid; grid-template-columns: 1.15fr .85fr; gap: 18px; }
.panel { margin-bottom: 18px; padding: 22px; border: 1px solid #294461; border-radius: 18px; background: #0d1a2dcc; backdrop-filter: blur(10px); }
label { display: grid; gap: 7px; margin-bottom: 16px; color: #b8c7d9; }
input, textarea, select { width: 100%; border: 1px solid #365375; border-radius: 10px; background: #071321; color: #f4f7fb; padding: 11px; font: inherit; }
textarea { resize: vertical; }
.actions, .template-row { display: flex; gap: 10px; }
button { border: 0; border-radius: 10px; padding: 11px 15px; background: #78dce8; color: #06101d; font-weight: 800; cursor: pointer; }
button.secondary { background: #263e5c; color: #e8eef8; }
pre { min-height: 130px; white-space: pre-wrap; overflow-wrap: anywhere; padding: 14px; border-radius: 10px; background: #071321; }
table { width: 100%; border-collapse: collapse; }
th, td { padding: 10px; border-bottom: 1px solid #294461; text-align: left; vertical-align: top; }
td:last-child { max-width: 420px; overflow-wrap: anywhere; }
#status { min-height: 1.4em; color: #78dce8; }
#run-history { padding-left: 20px; }
@media (max-width: 760px) { .grid { grid-template-columns: 1fr; } table { font-size: .8rem; } }

What Do These Styles Add?

• The grid places the form beside the output panel on wide screens.
• The panel class gives every major section the same border and background.
• The form rules keep inputs readable against the dark page.
• The media query changes the grid to one column below the configured width.

• Save styles.css.
• Refresh the Integrated Browser.
• Reduce the Integrated Browser panel width.

You will see the form and output panel side by side at a wide width. You will see them stack into one column at a narrow width.

✔️ Awesome, I've got everything!

Your FlowForge page now has its complete responsive dark interface.

ⓧ I'd like to double check the full code

:root {
  color-scheme: dark;
  font-family: Inter, ui-sans-serif, system-ui, sans-serif;
  background: #08111f;
  color: #e8eef8;
}

* { box-sizing: border-box; }
body { margin: 0; min-height: 100vh; background: radial-gradient(circle at top, #15325b, #08111f 42%); }
.shell { width: min(1100px, 92vw); margin: 0 auto; padding: 40px 0 64px; }
header { margin-bottom: 24px; }
.eyebrow { color: #78dce8; text-transform: uppercase; letter-spacing: .16em; font-size: .78rem; }
h1 { margin: 0; font-size: clamp(2.5rem, 7vw, 5rem); }
h2 { margin-top: 0; font-size: 1.05rem; }
.pipeline { display: flex; align-items: center; gap: 10px; margin: 28px 0; overflow-x: auto; }
.block { min-width: 130px; padding: 18px; border: 1px solid #365375; border-radius: 14px; background: #10213a; text-align: center; font-weight: 700; }
.block.accent { border-color: #78dce8; box-shadow: 0 0 26px #78dce833; }
.grid { display: grid; grid-template-columns: 1.15fr .85fr; gap: 18px; }
.panel { margin-bottom: 18px; padding: 22px; border: 1px solid #294461; border-radius: 18px; background: #0d1a2dcc; backdrop-filter: blur(10px); }
label { display: grid; gap: 7px; margin-bottom: 16px; color: #b8c7d9; }
input, textarea, select { width: 100%; border: 1px solid #365375; border-radius: 10px; background: #071321; color: #f4f7fb; padding: 11px; font: inherit; }
textarea { resize: vertical; }
.actions, .template-row { display: flex; gap: 10px; }
button { border: 0; border-radius: 10px; padding: 11px 15px; background: #78dce8; color: #06101d; font-weight: 800; cursor: pointer; }
button.secondary { background: #263e5c; color: #e8eef8; }
pre { min-height: 130px; white-space: pre-wrap; overflow-wrap: anywhere; padding: 14px; border-radius: 10px; background: #071321; }
table { width: 100%; border-collapse: collapse; }
th, td { padding: 10px; border-bottom: 1px solid #294461; text-align: left; vertical-align: top; }
td:last-child { max-width: 420px; overflow-wrap: anywhere; }
#status { min-height: 1.4em; color: #78dce8; }
#run-history { padding-left: 20px; }
@media (max-width: 760px) { .grid { grid-template-columns: 1fr; } table { font-size: .8rem; } }

Connect the page to the workflow API

The JavaScript layer translates the form into block definitions. It submits a shuffled graph so the browser demonstrates the dependency-aware executor you already built.

• Select the static folder in the Explorer sidebar.
• Click the New File icon in the Explorer toolbar.
• Enter app.js as the file name.
• Start the element helper and first three workflow blocks by pasting this code:

const element = (id) => document.getElementById(id);

function buildWorkflow() {
  const documents = element("documents").value
    .split("\n")
    .map((line) => line.trim())
    .filter(Boolean);

  return {
    name: element("workflow-name").value.trim(),
    blocks: [
      {
        id: "ai",
        type: "llm",
        depends_on: ["prompt"],
        config: {},
      },
      {
        id: "prompt",
        type: "prompt",
        depends_on: ["question", "retrieval"],
        config: { template: element("prompt-template").value },
      },
      {
        id: "retrieval",
        type: "retrieve",
        depends_on: ["question"],
        config: { documents, top_k: 2 },
      },

What Does This First Section Build?

• The element helper finds a control by its HTML ID.
• The document pipeline removes empty knowledge-base lines.
• The first three objects define the AI, prompt, and retrieval blocks.
• Their order is deliberately different from their dependency order.

• Complete buildWorkflow() by pasting this code directly beneath the retrieval block:

      {
        id: "question",
        type: "input",
        depends_on: [],
        config: { value: element("question").value },
      },
    ],
  };
}

Why Is the Input Block Last?

The input block has no dependencies even though it appears last in the array. This gives the backend scheduler a shuffled graph to reorder on every browser run.

• Save app.js.
• Refresh the Integrated Browser.
• Open the Integrated Browser developer console.
• Inspect the compiled workflow by entering this expression:

buildWorkflow()

What Does This Check Prove?

The expression reads the live form without contacting the API. Its result shows the block order as AI, prompt, retrieval, and question.

You will see a workflow object containing the form name and four block definitions. The question block appears last in the submitted array.

Workflow Object Not Appearing?

Check that the second snippet closes the block array, returned object, and function. Confirm that app.js is saved inside static.

Refresh the page before evaluating the function again. Help me find the syntax error in my buildWorkflow function.

• Append the API request and display helpers to app.js by pasting this code:

async function api(url, options = {}) {
  const response = await fetch(url, {
    headers: { "Content-Type": "application/json" },
    ...options,
  });
  const payload = await response.json();
  if (!response.ok) {
    throw new Error(
      typeof payload.detail === "string"
        ? payload.detail
        : JSON.stringify(payload.detail)
    );
  }
  return payload;
}

function display(value) {
  return typeof value === "string" ? value : JSON.stringify(value, null, 2);
}

How Does the API Helper Handle Responses?

• The request always tells FastAPI that its body uses JSON.
• The response body is parsed once for successful and failed requests.
• Failed responses become errors that the status area can display safely.
• The display helper formats objects as readable JSON text.

• Save app.js.
• Refresh the Integrated Browser.
• Check the run-history endpoint from the developer console by entering this expression:

api("/api/runs")

What Does This Request Check?

The helper sends a request to the existing run-history endpoint. The console shows a Promise that resolves with the current array of runs.

• Append the result renderers to app.js by pasting this code:

function renderRun(run) {
  element("final-output").textContent = display(run.final_output);
  const body = element("trace-body");
  body.replaceChildren();
  run.trace.forEach((entry) => {
    const row = document.createElement("tr");
    [
      entry.block_id,
      entry.block_type,
      entry.status,
      `${entry.duration_ms} ms`,
      display(entry.output ?? entry.error ?? ""),
    ].forEach((value) => {
      const cell = document.createElement("td");
      cell.textContent = value;
      row.appendChild(cell);
    });
    body.appendChild(row);
  });
}

function applyWorkflow(workflow) {
  const byId = Object.fromEntries(workflow.blocks.map((block) => [block.id, block]));
  element("workflow-name").value = workflow.name;
  element("question").value = byId.question?.config.value ?? "";
  element("documents").value = (byId.retrieval?.config.documents ?? []).join("\n");
  element("prompt-template").value = byId.prompt?.config.template ?? "";
}

How Are Results Rendered Safely?

• The final output is assigned through textContent so generated text cannot become page markup.
• Each trace entry creates one table row with five text cells.
• The workflow loader maps blocks by ID before restoring their values.

• Append the template and history refresh functions to app.js by pasting this code:

async function refreshTemplates() {
  const names = await api("/api/templates");
  const select = element("template-list");
  select.replaceChildren();
  names.forEach((name) => {
    const option = document.createElement("option");
    option.value = name;
    option.textContent = name;
    select.appendChild(option);
  });
}

async function refreshHistory() {
  const runs = await api("/api/runs");
  const list = element("run-history");
  list.replaceChildren();
  runs.forEach((run) => {
    const item = document.createElement("li");
    item.textContent = `${run.workflow_name}: ${run.status} (${run.run_id})`;
    list.appendChild(item);
  });
}

What Do the Refresh Functions Update?

• The template refresh replaces the select options with names returned by the API.
• The history refresh replaces the list with the latest in-memory runs.
• Both functions create text-only elements from API data.

• Save app.js.
• Refresh the Integrated Browser.
• Populate the Recent Runs panel from the developer console by entering this expression:

refreshHistory()

What Should This Refresh Show?

The function asks the API for its current in-memory history. A run from your earlier backend test appears if Uvicorn has remained active.

• Append the shared status wrapper and form submission handler to app.js by pasting this code:

async function withStatus(action) {
  const status = element("status");
  status.textContent = "Working...";
  try {
    await action();
    status.textContent = "Done";
  } catch (error) {
    status.textContent = error.message;
  }
}

element("workflow-form").addEventListener("submit", (event) => {
  event.preventDefault();
  withStatus(async () => {
    const run = await api("/api/runs", {
      method: "POST",
      body: JSON.stringify(buildWorkflow()),
    });
    renderRun(run);
    await refreshHistory();
  });
});

How Does a Browser Run Work?

• The status wrapper shows progress before the asynchronous action starts.
• The submit handler prevents the browser from reloading the page.
• The handler sends the compiled workflow to /api/runs.
• A successful response updates the answer, trace, and recent-run history.

• Save app.js.
• Refresh the Integrated Browser.
• Click Run Workflow.

The status changes while the request runs. You will then see a generated answer and four trace rows.

Run Not Completing?

Confirm that Uvicorn is still running in the terminal with GEMINI_API_KEY available. Check the status text for the backend error returned by the request.

If the page reloads when you submit the form, check the preventDefault() line. Help me diagnose why my FlowForge browser run is failing.

• Append the template controls and initial refresh to app.js by pasting this code:

element("save-template").addEventListener("click", () => {
  withStatus(async () => {
    await api("/api/templates", {
      method: "POST",
      body: JSON.stringify(buildWorkflow()),
    });
    await refreshTemplates();
  });
});

element("load-template").addEventListener("click", () => {
  withStatus(async () => {
    const name = element("template-list").value;
    if (!name) throw new Error("Save a template first");
    applyWorkflow(await api(`/api/templates/${encodeURIComponent(name)}`));
  });
});

Promise.all([refreshTemplates(), refreshHistory()]).catch((error) => {
  element("status").textContent = error.message;
});

How Is the Page Initialized?

• The save handler sends the current workflow contract to the template endpoint.
• The load handler restores the selected workflow through applyWorkflow().
• The final Promise refreshes template names and recent runs when the page loads.
• Any initialization error is written into the status area.

• Save app.js.
• Refresh the Integrated Browser.

✔️ Awesome, I've got everything!

Your browser script now compiles, submits, and renders the complete workflow.

ⓧ I'd like to double check the full code

const element = (id) => document.getElementById(id);

function buildWorkflow() {
  const documents = element("documents").value
    .split("\n")
    .map((line) => line.trim())
    .filter(Boolean);

  return {
    name: element("workflow-name").value.trim(),
    blocks: [
      {
        id: "ai",
        type: "llm",
        depends_on: ["prompt"],
        config: {},
      },
      {
        id: "prompt",
        type: "prompt",
        depends_on: ["question", "retrieval"],
        config: { template: element("prompt-template").value },
      },
      {
        id: "retrieval",
        type: "retrieve",
        depends_on: ["question"],
        config: { documents, top_k: 2 },
      },
      {
        id: "question",
        type: "input",
        depends_on: [],
        config: { value: element("question").value },
      },
    ],
  };
}

async function api(url, options = {}) {
  const response = await fetch(url, {
    headers: { "Content-Type": "application/json" },
    ...options,
  });
  const payload = await response.json();
  if (!response.ok) {
    throw new Error(
      typeof payload.detail === "string"
        ? payload.detail
        : JSON.stringify(payload.detail)
    );
  }
  return payload;
}

function display(value) {
  return typeof value === "string" ? value : JSON.stringify(value, null, 2);
}

function renderRun(run) {
  element("final-output").textContent = display(run.final_output);
  const body = element("trace-body");
  body.replaceChildren();
  run.trace.forEach((entry) => {
    const row = document.createElement("tr");
    [
      entry.block_id,
      entry.block_type,
      entry.status,
      `${entry.duration_ms} ms`,
      display(entry.output ?? entry.error ?? ""),
    ].forEach((value) => {
      const cell = document.createElement("td");
      cell.textContent = value;
      row.appendChild(cell);
    });
    body.appendChild(row);
  });
}

function applyWorkflow(workflow) {
  const byId = Object.fromEntries(workflow.blocks.map((block) => [block.id, block]));
  element("workflow-name").value = workflow.name;
  element("question").value = byId.question?.config.value ?? "";
  element("documents").value = (byId.retrieval?.config.documents ?? []).join("\n");
  element("prompt-template").value = byId.prompt?.config.template ?? "";
}

async function refreshTemplates() {
  const names = await api("/api/templates");
  const select = element("template-list");
  select.replaceChildren();
  names.forEach((name) => {
    const option = document.createElement("option");
    option.value = name;
    option.textContent = name;
    select.appendChild(option);
  });
}

async function refreshHistory() {
  const runs = await api("/api/runs");
  const list = element("run-history");
  list.replaceChildren();
  runs.forEach((run) => {
    const item = document.createElement("li");
    item.textContent = `${run.workflow_name}: ${run.status} (${run.run_id})`;
    list.appendChild(item);
  });
}

async function withStatus(action) {
  const status = element("status");
  status.textContent = "Working...";
  try {
    await action();
    status.textContent = "Done";
  } catch (error) {
    status.textContent = error.message;
  }
}

element("workflow-form").addEventListener("submit", (event) => {
  event.preventDefault();
  withStatus(async () => {
    const run = await api("/api/runs", {
      method: "POST",
      body: JSON.stringify(buildWorkflow()),
    });
    renderRun(run);
    await refreshHistory();
  });
});

element("save-template").addEventListener("click", () => {
  withStatus(async () => {
    await api("/api/templates", {
      method: "POST",
      body: JSON.stringify(buildWorkflow()),
    });
    await refreshTemplates();
  });
});

element("load-template").addEventListener("click", () => {
  withStatus(async () => {
    const name = element("template-list").value;
    if (!name) throw new Error("Save a template first");
    applyWorkflow(await api(`/api/templates/${encodeURIComponent(name)}`));
  });
});

Promise.all([refreshTemplates(), refreshHistory()]).catch((error) => {
  element("status").textContent = error.message;
});

Before you run the final check, which block do you expect the trace to execute first when the submitted array begins with the AI block?

• Click Run Workflow in the Pipeline Builder.

You will see a Gemini answer in Final Output. The trace contains four completed rows in question, retrieval, prompt, and AI dependency order.

That completes the browser loop. Your visible controls now compile into a shuffled graph that FastAPI validates and executes correctly.

Your full-stack workflow builder is working from browser input to generated output. Next, you will make workflows reusable with saved templates and recent-run history.

Save Templates and Track Runs

Your full browser flow now sends a shuffled graph through FastAPI. The page returns a Gemini answer with a four-row trace.

That workflow still feels disposable after a run. Local JSON template persistence makes its configuration reusable.

In-memory history makes recent results visible. A restart exposes exactly which state survives.

In this step, get ready to:

• Confirm how complete workflow definitions are stored.
• Reload a saved workflow after a browser refresh.
• Compare template persistence with process-only run history after a server restart.

Confirm the storage contract

The backend keeps two kinds of state in different places. data/templates.json persists named workflows on disk.

run_history keeps up to 20 completed results in memory. The contrast makes the restart test meaningful.

How does the state split work?

• save_template() stores workflow.model_dump(mode="json") under the workflow name.
• list_templates() returns the stored names in sorted order.
• get_template() validates a stored definition before returning it.
• create_run() inserts each completed RunResult at the front of run_history.
• del run_history[20:] discards entries beyond the newest 20.

• In the Visual Studio Code file sidebar, select main.py.
• Find TEMPLATE_FILE to confirm that saved workflows target data/templates.json.
• Find save_template() to confirm that validated workflow definitions are stored under their names.
• Find create_run() to confirm that each completed result enters the front of run_history.
• Find del run_history[20:] to confirm that older entries are discarded.

The code confirms that templates have disk-backed storage. It also confirms that history is capped inside the active process.

Save and reload a named template

Saving a template sends the complete graph to POST /api/templates. The backend creates the storage file during the first save.

• Switch back to the VS Code Integrated Browser from earlier.
• Replace standup-assistant in Workflow name with your workflow name.
• Click Save Template.

The status changes to Done. The template menu now includes your workflow name.

• Return to the Visual Studio Code workspace from earlier.
• Confirm data/templates.json appears inside the data folder.
• Select data/templates.json in the file sidebar.

The top-level JSON object contains your workflow name. Its value contains the complete serialized workflow definition.

• Switch back to the VS Code Integrated Browser.
• Refresh the page.
• Select your workflow name from the menu in Template Library.
• Click Load.

That reusable layer is now working. The Workflow name field returns to your workflow name after the page refresh.

Template missing after refresh?

• Confirm the status text changed to Done after the save.
• Save the template again if the name is absent.

Help me diagnose why my saved FlowForge template is missing after a browser refresh.

Compare persistence across a restart

Templates are disk-backed. A server restart leaves their JSON file in place.

run_history is process-backed. A fresh Uvicorn process starts with an empty list.

Why keep history temporary?

In-memory history makes each completed run immediately available to the browser. It avoids adding a database to this MVP.

run_history disappears with the process. This keeps the app's durability boundary clear.

• Switch back to the VS Code Integrated Browser.
• Click Run Workflow.

You'll see a Gemini answer in Final Output. The execution trace contains four completed rows.

• Confirm the newest item in Recent Runs starts with your workflow name.

Before you restart the server, which state do you expect to remain: the template list, the recent-run list, or both?

• Return to the integrated terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Start a fresh Uvicorn process by running this command:

uvicorn main:app

What does this command do?

• main identifies main.py as the application module.
• app selects the FastAPI application object.
• The new process initializes an empty run_history list.
• The existing data/templates.json file remains on disk.

The terminal shows the server startup messages. The server remains active while you complete the browser check.

Server not restarting?

• Confirm the previous server process stopped after Ctrl+C.
• Confirm the integrated terminal still has tutorial-env activated.

Help me restart the local FlowForge server.

• Switch back to the VS Code Integrated Browser.
• Refresh the page.

The template menu still contains your workflow name. The Recent Runs panel is empty because the server process restarted.

• Select your workflow name from the template menu.
• Click Load.

Before you run the loaded workflow, where do you expect its new history entry to appear?

• Click Run Workflow.

The final output shows a fresh Gemini answer. Recent Runs shows the reloaded workflow as the newest completed execution.

You have closed the gap between a one-off demo and a reusable local product. Saved templates now survive restarts while recent runs clearly reflect the active server session.

Secret mission

Add Conditional Routing

Add a gate block that checks input length before generation. Short inputs will finish with a skipped result without sending an unnecessary request to Gemini.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to continue demonstrating FlowForge, pause it for later, or remove it completely. The project uses the Gemini API Free Tier with billing disabled, so there are no paid cloud resources to leave running.

Resources you used:

• An active Uvicorn server that provides FlowForge at http://127.0.0.1:8000.
• A local flowforge-ai workspace containing the application source files.
• A Python virtual environment stored in tutorial-env.
• Saved workflow templates stored in data/templates.json.
• Up to 20 completed runs stored in the in-memory run_history list.
• A Gemini API key exported as GEMINI_API_KEY in the current shell.

Keep everything running

No action is needed while you continue demonstrating FlowForge. Future Gemini requests remain subject to free-tier limits.

• Leave Uvicorn active at http://127.0.0.1:8000.
• Keep the flowforge-ai workspace in Visual Studio Code.
• Keep billing disabled for this project.
• Use only non-sensitive demonstration data in future Gemini requests.

Pause - I'll come back to this later

Pausing FlowForge stops the local server without removing your source files. Your saved templates remain available for your return.

• Switch back to the VS Code terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Deactivate tutorial-env by running this command:

deactivate

What does this command do?

The deactivate command returns your shell to the system Python environment. Your project files remain untouched.

• Close the VS Code terminal to clear GEMINI_API_KEY from that shell.

FlowForge is safely paused. Stopping Uvicorn clears the in-memory run_history.

Your saved workflows remain in data/templates.json.

Delete - I don't want to use this again

Removing the workspace is permanent. Your saved templates disappear with it.

• Copy data/templates.json outside the flowforge-ai workspace if you want to preserve your templates.
• Switch back to the VS Code terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Remove the local workspace by running this cleanup sequence:

deactivate
cd ..
rm -rf flowforge-ai

What does this cleanup do?

• The deactivate command exits the virtual environment before cleanup.
• The cd command moves the shell to the directory containing flowforge-ai.
• The rm command permanently removes the workspace with its source files, virtual environment, and saved templates.

You should no longer see the flowforge-ai folder in its previous location.

Folder still present?

• Confirm Uvicorn is no longer using files from flowforge-ai.
• Check that you ran the removal sequence from the directory containing flowforge-ai.
• Help me diagnose why my FlowForge workspace was not removed.

Revoking the Gemini API key is permanent for that credential. Any other local project using the same key loses access too.

• Open Google's Gemini API key guidance.
• Follow the Google AI Studio link to its key-management page.
• Revoke the Gemini API key used for FlowForge if you no longer need it.
• Close the VS Code terminal to clear GEMINI_API_KEY from that shell.

Cleanup complete. The FlowForge workspace and its active credential are gone.

Nice Work!

Nice Work!

You did it! You built FlowForge as a full-stack workflow automation app with a FastAPI backend. It turns browser inputs into traced Gemini workflows.

You've learned how to:

• Build a low-code browser interface that compiles its controls into workflow JSON.
• Create a dependency-aware block registry that validates workflow graphs. Execute shuffled blocks in topological order with a trace for every result.
• Run a retrieval-augmented Gemini pipeline with local lexical context. Reuse workflows through saved templates. Inspect completed runs through recent history.
• Secret Mission: Extend the engine with a conditional gate that skips Gemini calls when short input fails the configured rule.

Ready to quiz yourself?