Course ES
← back to chapter

Planner agent

An agent that keeps a task list: `add_step` to decompose a goal, `mark_done` to work through it, then a summary. The list is real per-request state the tools mutate — the simplest form of agent memory.

Showcase — Planner agent

An agent that keeps a task list: add_step to decompose a goal, mark_done to work through it, then a summary. The list is real per-request state the tools mutate — the simplest form of agent memory.

Run

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Open http://localhost:3000. Gemini: PROVIDER=gemini docker compose up --build.

What's where

  • backend/ai_openai.py / backend/ai_gemini.py — stateful add_step/mark_done tools + loop.
  • frontend/app/page.tsx — goal box, plan checklist + summary.

Stop

docker compose down

Run locally

Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.

Download planner-agent.zip

unzip planner-agent.zip
cd planner-agent
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

Type some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.


  

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

backend/ai_openai.py

"""Showcase 3 (OpenAI): a planning agent with state.

An agent that keeps a task list. Given a goal it calls `add_step` to break the
goal down, then `mark_done` as it works through them, and finally reports. The
list is real per-request state the tools mutate — the simplest form of agent
memory, and the thing that turns a one-shot answer into a process you can watch.
"""
import json

from openai import OpenAI

_client = OpenAI()

_MODEL = "gpt-5.4-nano"

_TOOLS = [
    {"type": "function", "name": "add_step",
     "description": "Add a concrete step to the plan.",
     "parameters": {"type": "object", "properties": {"description": {"type": "string"}}, "required": ["description"]}},
    {"type": "function", "name": "mark_done",
     "description": "Mark the step at the given 0-based index as done.",
     "parameters": {"type": "object", "properties": {"index": {"type": "integer"}}, "required": ["index"]}},
]

_GOAL = ("You are a planning agent. Break the user's goal into concrete steps using "
         "`add_step`, then simulate doing them by calling `mark_done` for each in order. "
         "When every step is done, give a short summary of the plan and the outcome.")


def run(goal: str) -> str:
    g = goal.strip()
    if not g:
        return "Give the agent a goal to plan (e.g. 'plan a launch for a new mobile app')."
    tasks: list[dict] = []

    def add_step(description: str) -> dict:
        tasks.append({"step": description, "done": False})
        return {"steps": tasks}

    def mark_done(index: int) -> dict:
        if 0 <= index < len(tasks):
            tasks[index]["done"] = True
        return {"steps": tasks}

    impl = {"add_step": add_step, "mark_done": mark_done}

    input_list = [{"role": "user", "content": g}]
    response = None
    for _ in range(12):
        response = _client.responses.create(model=_MODEL, instructions=_GOAL, input=input_list, tools=_TOOLS)
        calls = [i for i in response.output if i.type == "function_call"]
        if not calls:
            break
        input_list += response.output
        for c in calls:
            result = impl[c.name](**json.loads(c.arguments))
            input_list.append({"type": "function_call_output", "call_id": c.call_id, "output": json.dumps(result)})
    checklist = "\n".join(f"[{'x' if t['done'] else ' '}] {t['step']}" for t in tasks)
    return f"{response.output_text if response else ''}\n\n--- plan ---\n{checklist}"

backend/ai_gemini.py

"""Showcase 3 (Gemini): a planning agent with state.

Same task-list agent and per-request state, Gemini's tool-calling.
"""
import os

from google import genai
from google.genai import types

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

_MODEL = "gemini-3.1-flash-lite"

_CONFIG = types.GenerateContentConfig(
    system_instruction=("You are a planning agent. Break the user's goal into concrete steps "
                        "using `add_step`, then simulate doing them by calling `mark_done` for each "
                        "in order. When every step is done, summarize the plan and outcome."),
    tools=[types.Tool(function_declarations=[
        types.FunctionDeclaration(name="add_step", description="Add a concrete step to the plan.",
                                  parameters=types.Schema(type=types.Type.OBJECT,
                                      properties={"description": types.Schema(type=types.Type.STRING)}, required=["description"])),
        types.FunctionDeclaration(name="mark_done", description="Mark the step at the given 0-based index as done.",
                                  parameters=types.Schema(type=types.Type.OBJECT,
                                      properties={"index": types.Schema(type=types.Type.INTEGER)}, required=["index"])),
    ])],
)


def run(goal: str) -> str:
    g = goal.strip()
    if not g:
        return "Give the agent a goal to plan (e.g. 'plan a launch for a new mobile app')."
    tasks: list[dict] = []

    def add_step(description: str) -> dict:
        tasks.append({"step": description, "done": False})
        return {"steps": tasks}

    def mark_done(index: int) -> dict:
        if 0 <= index < len(tasks):
            tasks[index]["done"] = True
        return {"steps": tasks}

    impl = {"add_step": add_step, "mark_done": mark_done}

    contents = [types.Content(role="user", parts=[types.Part(text=g)])]
    response = None
    for _ in range(12):
        response = _client.models.generate_content(model=_MODEL, contents=contents, config=_CONFIG)
        if not response.function_calls:
            break
        contents.append(response.candidates[0].content)
        parts = [types.Part.from_function_response(name=fc.name, response=impl[fc.name](**dict(fc.args)))
                 for fc in response.function_calls]
        contents.append(types.Content(role="user", parts=parts))
    checklist = "\n".join(f"[{'x' if t['done'] else ' '}] {t['step']}" for t in tasks)
    return f"{(response.text or '') if response else ''}\n\n--- plan ---\n{checklist}"

Project files

  • .gitignore
  • README.es.md
  • README.md
  • backend/Dockerfile
  • backend/ai_gemini.py
  • backend/ai_openai.py
  • backend/main.py
  • backend/requirements.txt
  • bootstrap-secrets.sh
  • docker-compose.yml
  • frontend/Dockerfile
  • frontend/app/layout.tsx
  • frontend/app/page.tsx
  • frontend/next.config.ts
  • frontend/package.json
  • frontend/tsconfig.json