Build Stateful Agentic Apps with the Gemini Interactions API

57 min remaining
Step 6 of 12•7 min

6. Build the Agent Loop

agent.py holds everything that talks to the model: the configuration sent with every request, the orchestration loop and the structured incident report you will use later.

Add the code

Open agent.py, paste the following code and save the file:

agent.py(python)
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142
import json
from google import genai
from pydantic import BaseModel
from tools import TOOL_DECLARATIONS, execute_tool
 
client = genai.Client() # Reads GEMINI_API_KEY from the environment
MODEL = "gemini-3.8-flash"
MAX_TURNS = 25
 
SYSTEM_INSTRUCTION = """
You are the on-call Site Reliability Engineer for the Orbital 7 space station.
Bring the station to a safe, stable state and keep the crew alive.
 
Operating principles:
- Telemetry can be stale or wrong. Confirm anything important with a diagnostic before you rely on it.
- Every action takes time, and oxygen keeps falling while the scrubbers are offline. Do not waste actions.
- Coolant and power are limited. Plan before you spend them.
- Protect life support first, then the systems that keep the station observable, then everything else.
- When an action fails, read the error, work out why, and change your plan. Never repeat a failed action unchanged.
- Never declare the station stable until diagnostics confirm it.
- If there is no safe way forward, escalate to a human instead of guessing.
"""
 
# Interaction-scoped settings: these do not carry over with previous_interaction_id,
# so every request must send them again.
REQUEST_CONFIG = {
"model": MODEL,
"system_instruction": SYSTEM_INSTRUCTION,
"tools": TOOL_DECLARATIONS,
"generation_config": {"thinking_summaries": "auto"},
}
 
 
def print_thoughts(interaction):
"""Prints the model's thought summaries for this turn."""
for step in interaction.steps or []:
if step.type == "thought":
for block in step.summary or []:
if block.type == "text":
print(f" 💭 {block.text.strip()}")
 
 
def run_agent(station, interaction, session, payload_bytes=0):
"""The orchestration loop. Runs until the model completes or something stops it."""
for turn in range(1, MAX_TURNS + 1):
usage = interaction.usage
print(f"\n--- Turn {turn} | {interaction.id} | {interaction.status}")
print(f" 📦 sent {payload_bytes} bytes | model read "
f"{getattr(usage, 'total_input_tokens', '?')} input tokens "
f"({getattr(usage, 'total_cached_tokens', 0) or 0} cached)")
 
session["turns"].append({
"interaction_id": interaction.id, "status": interaction.status,
"payload_bytes": payload_bytes,
"input_tokens": getattr(usage, "total_input_tokens", None),
"cached_tokens": getattr(usage, "total_cached_tokens", None),
"calls": [],
})
print_thoughts(interaction)
 
if interaction.status == "completed":
print(f"\n🛰️ Agent: {interaction.output_text}")
return interaction
 
if interaction.status != "requires_action":
print(f"\n❌ Interaction ended with status '{interaction.status}'.")
return interaction
 
# The create response contains only this turn's model steps,
# so every function_call here is new and needs exactly one result.
results = []
for call in (s for s in interaction.steps if s.type == "function_call"):
args = call.arguments or {}
output, is_error = execute_tool(station, call.name, args)
icon = "⚠️ " if is_error else "✅"
print(f" 🤖 {call.name}({json.dumps(args)})")
print(f" {icon} {output}")
 
session["turns"][-1]["calls"].append(
{"call_id": call.id, "name": call.name, "arguments": args, "is_error": is_error})
results.append({
"type": "function_result",
"call_id": call.id,
"name": call.name,
"result": [{"type": "text", "text": output}],
"is_error": is_error,
})
 
payload_bytes = len(json.dumps(results).encode())
interaction = client.interactions.create(
previous_interaction_id=interaction.id,
input=results,
**REQUEST_CONFIG,
)
 
print(f"\n⚠️ Stopped after {MAX_TURNS} turns.")
return interaction
 
 
# ==========================================
# Structured incident report
# ==========================================
 
class TimelineEntry(BaseModel):
action: str
observation: str
reasoning: str
 
class FinalState(BaseModel):
core_temperature_celsius: float
power_bus_online: bool
life_support_online: bool
oxygen_percent: float
 
class IncidentReport(BaseModel):
title: str
root_cause: str
timeline: list[TimelineEntry]
final_state: FinalState
sectors_shed: list[str]
open_risks: list[str]
escalated_to_human: bool
 
 
def generate_report(previous_interaction_id):
"""Asks the same session for a machine-readable incident report."""
interaction = client.interactions.create(
model=MODEL,
previous_interaction_id=previous_interaction_id,
input="Write the incident report for this session. Report only what your tool results showed.",
response_format={
"type": "text",
"mime_type": "application/json",
"schema": IncidentReport.model_json_schema(),
},
)
return interaction, IncidentReport.model_validate_json(interaction.output_text)
 
 
def save_session(session, path="session.json"):
with open(path, "w") as f:
json.dump(session, f, indent=2)

How agent.py works

  • Principles, not steps (SYSTEM_INSTRUCTION): The system instruction tells the agent how to behave (verify, conserve, protect life support, escalate) but not what sequence of tools to call.
  • REQUEST_CONFIG on every call: Because model, tools, system_instruction and generation_config are interaction-scoped, REQUEST_CONFIG groups them in a single dictionary that is unpacked (**REQUEST_CONFIG) on every client.interactions.create() call. Only the conversation history comes from previous_interaction_id.
  • Thought summaries (print_thoughts): Setting "thinking_summaries": "auto" in generation_config includes thought steps in interaction.steps. print_thoughts() iterates over step.summary blocks and prints the model's reasoning before its tool calls.
  • One result per call (run_agent): The create response contains only this turn's newly generated model steps, so every function_call in interaction.steps is new. The loop executes each call via execute_tool(), records the call in session["turns"], and builds a function_result dict with a matching call_id and name. When the model requests multiple tool calls in a single turn, all of their results are sent back together in the next client.interactions.create() call.
  • Measurement and safety limits: Each turn logs payload_bytes (the UTF-8 byte length of the request payload your code sent) alongside interaction.usage.total_input_tokens and total_cached_tokens. MAX_TURNS stops any run that does not reach completed, and any status other than completed or requires_action exits the loop cleanly.
  • Structured incident report (IncidentReport and generate_report): Uses Pydantic's IncidentReport.model_json_schema() inside response_format to request a strictly typed JSON summary from the completed session, and parses interaction.output_text back into an IncidentReport instance with model_validate_json().