Step 4: Make it durable

Right now your agent lives inside one HTTP request. Refresh the page halfway through a long walk, or push a new deploy, and the turn is gone, along with wherever your character was headed. If you're on Python, use Vercel Workflows to make the agent durable. Every model call and every move gets written to a log as it happens, failed steps retry on their own, and an interrupted run picks up from the last step it finished. Close the tab, and your character keeps exploring on the big screen.

1. Ask your agent

   Nothing to provision: Workflows are built into Vercel. You only need a couple of decorators and a new entry in pyproject.toml.

   Make my agent in `backend/` durable with Vercel Workflows. Follow https://ai-python.dev/docs/basics/durable-execution and the example app at https://github.com/vercel-labs/ai-python/tree/main/examples/apps/durable_agent_workflows. Add the `vercel` package and register a workflow in `pyproject.toml`. Run each chat turn as a workflow where every model call and every game action is its own step, and have `/api/agent/chat` start the run and stream its events back to the UI. Moves and chat messages aren't safe to repeat, so don't retry those steps.

   Copy

   What your agent is going to implement

   1. Register the workflow

      The vercel package has the Workflow SDK for Python. Vercel finds your workflows through pyproject.toml.

      [project]
      dependencies = ["ai[vercel]", "fastapi", "httpx", "vercel"]
      
      [tool.vercel]
      entrypoint = "app.server:app"
      
      # New: tells Vercel where your workflows live.
      [[tool.vercel.workflows]]
      entrypoint = "app.worker:workflow"

   2. Model calls and game actions become steps

      A step takes JSON in and gives JSON back, so its result can be saved and replayed. On Python, AI SDK for Python models, messages, and tools all round-trip through JSON as is.

      import ai
      import vercel.workflow
      
      workflow = vercel.workflow.Workflows()
      
      @workflow.step  # retried if it fails, then saved to the run's log
      async def llm_step(model: dict, messages: list[dict], tools: list[dict]) -> dict:
          async with ai.stream(
              ai.Model.model_validate(model),
              [ai.messages.Message.model_validate(m) for m in messages],
              tools=[ai.Tool.model_validate(t) for t in tools],
          ) as stream:
              async for _ in stream:
                  pass
          return stream.message.model_dump(mode="json")
      
      @workflow.step(max_retries=0)  # two moves up aren't the same as one
      async def act(session: dict, path: str, body: dict) -> dict:
          return await GameClient(session).act(path, body)
      
      @ai.tool
      async def move(direction: Literal["up", "down", "left", "right"]) -> dict:
          """Move one tile. Returns the game's receipt with your new x, y and xp."""
          return await act(current_session.get(), "/game/controls/move", {"direction": direction})

   3. A loop that runs as a workflow

      The usual agent loop, rewritten so each model call goes through the step. The chat route starts a run and sends its events to the UI. The example app has the whole thing.

      class DurableAgent(ai.Agent):
          TOOLS = [move, say]
      
          async def loop(self, context: ai.Context):
              while context.keep_running():
                  # One step per model call. On replay, the saved answer comes back.
                  reply = ai.messages.Message.model_validate(await llm_step(
                      context.model.model_dump(mode="json"),
                      [m.model_dump(mode="json") for m in context.messages],
                      [t.model_dump(mode="json") for t in context.tools],
                  ))
                  context.add(reply)
                  async with ai.Stream.replay_message(reply) as replay:
                      async for event in replay:
                          yield event  # so the UI still sees the text
                  async with ai.ToolRunner() as runner:
                      for call in reply.tool_calls:
                          runner.schedule(context.resolve(call))
                      async for event in runner.events():
                          yield event
                      if (tool_message := runner.get_tool_message()) is not None:
                          context.add(tool_message)
      
      @workflow.workflow
      @ai.messages.use_random(vercel.workflow.random)  # same message IDs on every replay
      async def run_turn(turn: dict) -> dict:
          current_session.set(turn["session"])
          ...  # run DurableAgent, collect its events, return them with the messages
      
      @app.post("/api/agent/chat")
      async def chat(request: ChatRequest, session: Session = Depends(get_session)):
          run = await vercel.workflow.start(run_turn, {"session": ..., "messages": ...})
          output = await run.return_value()
          return StreamingResponse(to_sse(output["events"]), headers=ai_sdk.UI_MESSAGE_STREAM_HEADERS)

2. Deploy and try to break it

   vercel deploy --prod

   Copy

   Ask your agent to go visit every location on the map, then refresh the page while it's walking. Your character keeps going. To watch each step land, open your project in the Vercel dashboard and go to Observability, then Workflows.