GenAI DisciplinesFind your levelMatch a JobBytesDocsAbout

Account

Free lesson · GenAI Application Engineering

Build an agentic loop executor with SSE-streamed steps

You will build an AgenticLoopExecutor in services/agentic_loop.py that runs multi-step agent execution via observe-think-act. The execute() method accepts a TaskRequest and loops: each iteration calls the Pydantic AI agent to observe state, think via LLM reasoning, and act by invoking tools. A StepRecord Pydantic model captures each iteration with fields: step_number, observation, reasoning, action (Optional[ToolCall]), result, duration_ms. The executor enforces max_iterations (default 10) and max_tool_errors (default 3), terminating gracefully when exceeded. Tool errors are fed back as error observations. stream_execution() yields SSE frames per step including reasoning and tool calls. FastAPI endpoint POST /api/v1/agent/execute returns streaming SSE. GET /api/v1/agent/executions/{id} retrieves full step history.

Course: Full-Stack GenAI Applications · Chapter 8 · MCP, Tool Execution & Agentic Backends

Free to read — no subscription required.

Subscribe to run the labs

Introduction

When you wire an LLM to a tool registry without a controlling loop, a single failed tool call or a malformed response can hang the request, silently truncate the agent's reasoning, or leak partial state to the client. Teams that ship agents without iteration budgets routinely burn model quotas on runaway loops that never converge and surface non-deterministic answers to end users. By the end of this lesson you'll be able to implement a multistep agentic loop executor that observes state, thinks via the LLM, acts on tools, recovers from failures, enforces a hard iteration cap, and streams each step as SSE so the client sees progress in real time.

Key Terminology

  • Observe-Think-Act cycle: the three-phase loop iteration where the executor collects state (observe), calls the LLM (think), and dispatches tool calls (act) before looping back.
  • Iteration budget: the max_iterations cap on TaskRequest that prevents runaway loops; when exhausted, the executor yields a safety-net FINAL_ANSWER event instead of hanging.
  • Tool history: the ordered list of ToolCallRecord entries accumulated during execution, capturing each tool name, arguments, result or error, and duration_ms for post-mortem analysis.
  • SSE step event: an AgenticStepEvent instance yielded by the async generator, serialized as a Server-Sent Event frame so the client sees TOOL_CALL, TOOL_RESULT, and FINAL_ANSWER progress in real time.
  • Tool registry: the injected dict[str, Awaitable] mapping tool names to async callables (Pydantic AI typed tools, MCP client wrappers, or ADK tool functions) that the executor dispatches in the act phase.

Concepts

The Observe-Think-Act Cycle

Every agentic loop iteration follows three discrete phases. The observe phase collects the current environment state — the accumulated message history, the most recent tool result, and any system-level constraints such as remaining iteration budget. The think phase sends this context to the LLM and receives either a final answer or one or more tool-call requests. The act phase dispatches each requested tool call through the MCP client or Pydantic AI tool registry, captures results or errors, appends them to history, and yields an SSE event before looping back to observe. When the LLM returns a response with no tool calls, or when the iteration limit is reached, the loop terminates and emits a final SSE event carrying the agent's answer.

Loading diagram...

The diagram above captures the full lifecycle. Notice that the SSE yield happens inside the act phase — this is deliberate. Emitting events after every tool execution means the client sees incremental progress even when a single iteration takes several seconds due to an expensive MCP tool call (see Code Walkthrough).

Error Recovery Strategy

Production agentic loops must handle three failure categories without crashing. Transient tool errors — such as a timeout from an MCP server — are caught by the try/except block inside the act phase, recorded in ToolCallRecord.error, and fed back to the LLM as a TOOL_ERROR message. Most capable models will retry with modified arguments or select an alternative tool on the next iteration. Malformed LLM output — such as invalid JSON in tool_calls[].function.arguments — should be caught by wrapping json.loads in its own try/except and returning a parse-error message to the model. Budget exhaustion occurs when iteration hits max_iterations; the executor yields the safety-net event rather than silently closing. In all three cases, the tool_history list preserves the full trace, enabling post-mortem analysis of why an agent failed to converge.

Code Walkthrough

This section demonstrates the observe-think-act cycle and error-recovery strategy from the previous section as concrete Python — first the Pydantic data models that flow through the loop, then the AgenticLoopExecutor that drives it.

Core Data Models

Before building the executor, you need Pydantic models that give the loop strong typing. The TaskRequest model carries the user prompt and configuration, ToolCallRecord captures each tool invocation with its result and timing, and AgenticStepEvent is the SSE payload sent to the client. These models enforce that every piece of data flowing through the loop is validated — a ValueError raised during deserialization surfaces immediately rather than corrupting downstream history. The AgenticLoopExecutor class references these models throughout its execute method, and the ToolCallRecord instances accumulate in a list that serves as the full tool-call history for observability and debugging.

Code snippet python 
1from pydantic import BaseModel, Field
2from enum import Enum
3from typing import Any
4import time
5
6class TaskRequest(BaseModel):
7    prompt: str
8    max_iterations: int = Field(default=10, ge=1, le=50)
9    model: str = "gpt-4o"
10    session_id: str | None = None
11
12class StepType(str, Enum):
13    TOOL_CALL = "tool_call"
14    TOOL_RESULT = "tool_result"
15    THINKING = "thinking"
16    FINAL_ANSWER = "final_answer"
17    ERROR = "error"
18
19class ToolCallRecord(BaseModel):
20    iteration: int
21    tool_name: str
22    arguments: dict[str, Any]
23    result: Any = None
24    error: str | None = None
25    duration_ms: float = 0.0
26    timestamp: float = Field(default_factory=time.time)
27
28class AgenticStepEvent(BaseModel):
29    step_type: StepType
30    iteration: int
31    content: str
32    tool_call: ToolCallRecord | None = None
33    is_final: bool = False
  • Lines 1-4: Import Pydantic's BaseModel and Field for validated models, Enum for the step-type enumeration, Any for flexible tool arguments and results, and time for recording durations and timestamps.
  • Lines 7-10: TaskRequest holds the user's prompt, a max_iterations budget clamped between 1 and 50 via Field constraints, the target LLM model string, and an optional session_id for correlating multi-turn conversations.
  • Lines 13-18: StepType enumerates the five SSE event categories — TOOL_CALL when a tool is dispatched, TOOL_RESULT when its output arrives, THINKING for the LLM's intermediate reasoning, FINAL_ANSWER when the loop terminates normally, and ERROR for recoverable failures.
  • Lines 21-27: ToolCallRecord captures a single tool invocation's metadata: which iteration triggered it, the tool_name and arguments sent, the result or error received, execution duration_ms, and a Unix timestamp defaulting to the current time.
  • Lines 30-34: AgenticStepEvent is the SSE payload model. It carries the step_type, current iteration count, a human-readable content string, an optional ToolCallRecord for tool-related events, and an is_final flag that tells the client to close the SSE connection.

The AgenticLoopExecutor

The executor is the central orchestration class. Its execute method is an async generator — it yields AgenticStepEvent instances that a FastAPI endpoint serializes into SSE frames. Internally, the method maintains a messages list that grows with each observe-think-act cycle, a tool_history list of ToolCallRecord entries, and an iteration counter checked against TaskRequest.max_iterations. Tool dispatch happens through an injected tool_registry dictionary that maps tool names to async callables — these callables can be Pydantic AI typed tools, direct MCP client invocations, or Google ADK tool wrappers. When a tool raises an exception, the executor does not crash; it catches the error, records it in the ToolCallRecord, appends an error message to history so the LLM can self-correct on the next iteration, and yields an ERROR-type SSE event. This error-recovery pattern is critical for production systems where MCP servers may be temporarily unreachable or return malformed responses.

Code snippet python 
1import json
2from typing import AsyncGenerator, Callable, Awaitable
3from litellm import acompletion
4
5class AgenticLoopExecutor:
6    def __init__(
7        self,
8        tool_registry: dict[str, Callable[..., Awaitable[Any]]],
9        tool_schemas: list[dict[str, Any]],
10        system_prompt: str = "You are a helpful assistant with tool access.",
11    ):
12        self.tool_registry = tool_registry
13        self.tool_schemas = tool_schemas
14        self.system_prompt = system_prompt
15
16    async def execute(
17        self, request: TaskRequest
18    ) -> AsyncGenerator[AgenticStepEvent, None]:
19        messages = [
20            {"role": "system", "content": self.system_prompt},
21            {"role": "user", "content": request.prompt},
22        ]
23        tool_history: list[ToolCallRecord] = []
24        iteration = 0
25
26        while iteration < request.max_iterations:
27            iteration += 1
28            response = await acompletion(
29                model=request.model,
30                messages=messages,
31                tools=self.tool_schemas,
32                tool_choice="auto",
33            )
34            choice = response.choices[0]
35            assistant_msg = choice.message
36
37            if not assistant_msg.tool_calls:
38                yield AgenticStepEvent(
39                    step_type=StepType.FINAL_ANSWER,
40                    iteration=iteration,
41                    content=assistant_msg.content or "",
42                    is_final=True,
43                )
44                return
45
46            messages.append(assistant_msg.model_dump())
47
48            for tc in assistant_msg.tool_calls:
49                fn_name = tc.function.name
50                fn_args = json.loads(tc.function.arguments)
51                yield AgenticStepEvent(
52                    step_type=StepType.TOOL_CALL,
53                    iteration=iteration,
54                    content=f"Calling {fn_name}({fn_args})",
55                )
56                record = ToolCallRecord(
57                    iteration=iteration,
58                    tool_name=fn_name,
59                    arguments=fn_args,
60                )
61                start = time.time()
62                try:
63                    handler = self.tool_registry[fn_name]
64                    result = await handler(**fn_args)
65                    record.result = result
66                    tool_msg_content = json.dumps(result, default=str)
67                except Exception as exc:
68                    record.error = str(exc)
69                    tool_msg_content = f"TOOL_ERROR: {exc}"
70                    yield AgenticStepEvent(
71                        step_type=StepType.ERROR,
72                        iteration=iteration,
73                        content=f"Tool {fn_name} failed: {exc}",
74                    )
75                record.duration_ms = (time.time() - start) * 1000
76                tool_history.append(record)
77                messages.append(
78                    {"role": "tool", "tool_call_id": tc.id,
79                     "content": tool_msg_content}
80                )
81                yield AgenticStepEvent(
82                    step_type=StepType.TOOL_RESULT,
83                    iteration=iteration,
84                    content=tool_msg_content,
85                    tool_call=record,
86                )
87
88        yield AgenticStepEvent(
89            step_type=StepType.FINAL_ANSWER,
90            iteration=iteration,
91            content="Iteration limit reached. Returning best partial answer.",
92            is_final=True,
93        )
  • Lines 1-3: Import json for serializing tool arguments and results, the typing constructs for the async generator signature, and acompletion from LiteLLM which provides a unified async completion interface across OpenAI, Anthropic, Gemini, and other providers.
  • Lines 6-15: The constructor accepts a tool_registry mapping tool names to async callables (these originate from Pydantic AI typed tools or MCP client wrappers), a tool_schemas list of OpenAI-format function definitions for the LLM, and an optional system_prompt.
  • Lines 17-25: The execute method initializes the messages list with the system prompt and user prompt, creates an empty tool_history accumulator, and sets iteration to zero.
  • Lines 27-34: The main while loop runs until the iteration budget is exhausted. Each iteration calls acompletion with the full message history, tool schemas, and tool_choice="auto" so the LLM decides whether to call a tool or produce a final answer.
  • Lines 35-44: If the assistant message contains no tool_calls, the loop has converged — the executor yields a FINAL_ANSWER event with is_final set to True and returns, closing the async generator.
  • Lines 46-47: When tool calls are present, the raw assistant message is appended to history so subsequent LLM calls see the full conversation thread.
  • Lines 49-55: For each tool call, the executor parses the function name and JSON arguments, then yields a TOOL_CALL event so the SSE client can display "Calling search_orders({...})" in real time.
  • Lines 56-60: A ToolCallRecord is initialized with the current iteration, tool name, and arguments. The start timestamp captures the moment before dispatch.
  • Lines 61-72: Inside the try/except block, the handler is looked up from tool_registry and invoked with the parsed arguments. On success, record.result stores the return value. On failure, record.error stores the exception message, the tool_msg_content is set to a TOOL_ERROR prefix that the LLM will interpret as a failure signal, and an ERROR SSE event is yielded.
  • Lines 73-83: The record's duration_ms is computed, the record is appended to tool_history, a tool-role message is added to the conversation history with the tool_call_id required by the OpenAI protocol, and a TOOL_RESULT SSE event is yielded with the full record attached.
  • Lines 85-90: If the while loop exits without the LLM producing a final answer, the executor yields a safety-net FINAL_ANSWER event indicating the iteration limit was reached — this prevents the client from hanging indefinitely on an open SSE connection.

You'll know it works when a /agent/run SSE consumer receives a TOOL_CALL event, then a TOOL_RESULT event with a populated ToolCallRecord, then a FINAL_ANSWER event with is_final=True — and tool_history reflects the same sequence with non-zero duration_ms values.

Do's and Don'ts

Do's

  1. ✓Do inject tool_registry as a dict[str, Callable[..., Awaitable[Any]]] into AgenticLoopExecutor — Decoupling the registry from the executor means the same loop driver works with Pydantic AI typed tools, direct MCP client invocations, or Google ADK wrappers without modifying execute; swapping a backend is a registry update, not a loop rewrite.
  2. ✓Do call messages.append(assistant_msg.model_dump()) before iterating over assistant_msg.tool_calls — Inserting the full assistant turn — including its tool_calls metadata — into the messages list before dispatching preserves the observe-think-act sequence; omitting it causes the next acompletion call to lack the LLM's own reasoning trace, which can produce repeated or contradictory tool invocations on the following iteration.
  3. ✓Do constrain TaskRequest.max_iterations with Field(ge=1, le=50) — Pydantic enforces the iteration budget at deserialization time, so a request that passes max_iterations=0 or max_iterations=500 raises a ValueError before the first acompletion call, preventing both a silent no-op loop and runaway quota consumption against the model.

Don'ts

  1. ✗Don't let a tool exception propagate out of the AgenticLoopExecutor.execute generator — Catching the error, recording it on ToolCallRecord.error, and appending a TOOL_ERROR message to messages converts a temporary MCP server timeout or malformed-response failure into a recoverable event the LLM can acknowledge on the next iteration rather than aborting the SSE stream entirely.
  2. ✗Don't call json.loads(tc.function.arguments) outside a try/except — LLMs occasionally emit syntactically invalid JSON in function arguments; an uncaught JSONDecodeError escapes the async generator and surfaces as a 500 to the FastAPI handler instead of a parse-error message that the LLM can self-correct from on the next cycle.
  3. ✗Don't emit the terminal AgenticStepEvent only on natural convergence — When the while loop exits because iteration reaches request.max_iterations without the LLM returning a tool_call-free response, the SSE consumer still requires an event with is_final=True; failing to yield one on budget exhaustion leaves the client connection open indefinitely with no termination signal.

This lesson is free to read. Its 3 hands-on labs — real code, in a cloud IDE — are part of the GenAI Application Engineering subscription.

Subscribe to run the labs Create a free account — return to this lesson →

From · cancel anytime · Already a subscriber? Sign in →

More free lessons in Full-Stack GenAI Applications

All free lessons in GenAI Application Engineering →