Human-in-the-Loop demo using Chainlit (frontend) and a FastAPI + ADK-style backend. The agent reasons over a task, proposes a plan, waits for human approval, then executes.
Chainlit (thin client)
│
│ POST /api/run { message, session_id }
▼
FastAPI (/api/run)
│ owns all phase logic — returns typed responses
│
▼
Postgres (agent_sessions table)
phase | plan | history_digest
The frontend renders purely on response.type. Migrating to a different frontend
(React, CLI, etc.) is a rendering change only — no business logic to re-implement.
idle
└─ any message ────────────────► awaiting_approval
│
┌───────────────────┼──────────────────┐
"proceed" "tweak: …" "cancel"
│ │ │
executing awaiting_approval idle
│
executed
│
follow-up message
│
┌────────────┴────────────┐
new task question
│ │
awaiting_approval executed (stays, answers directly)
chainlit-gemini-demo/
├── backend/
│ ├── main.py # FastAPI app — single /api/run endpoint
│ ├── db.py # Session state (mock dict or Postgres)
│ ├── models.py # Pydantic RunRequest / RunResponse
│ └── agent/
│ ├── planner.py # Reasons over request, produces Plan
│ ├── executor.py # Executes approved plan, writes history_digest
│ └── responder.py # Classifies follow-ups: new task vs direct answer
└── frontend/
└── app.py # Chainlit thin client
- uv installed
cd backend
# Default: mock DB (no Postgres required)
uv run uvicorn main:app --reload --port 8000cd frontend
uv run chainlit run app.py --port 8080Open http://localhost:8080 and send a task like "investigate the recent spike in API errors".
- Create the session table in your existing Postgres instance:
CREATE TABLE agent_sessions (
session_id TEXT PRIMARY KEY,
phase TEXT NOT NULL DEFAULT 'idle',
plan JSONB,
history_digest TEXT,
updated_at TIMESTAMPTZ DEFAULT now()
);- Set environment variables before starting the backend:
export USE_MOCK_DB=false
export DATABASE_URL=postgresql://user:password@host:5432/dbname
uv run uvicorn main:app --reload --port 8000No other code changes are required.
Each agent file contains a comment marking where the ADK runner call goes. The overall pattern is the same — replace the mock return values with:
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
runner = Runner(agent=LlmAgent(model="gemini-2.0-flash", ...), ...)
async for event in runner.run_async(user_id=..., session_id=..., new_message=...):
if event.is_final_response():
return parse_response(event.content)Request
{ "message": "string", "session_id": "string" }Response — type: plan
{
"type": "plan",
"plan_id": "uuid",
"reasoning": "string",
"steps": [{ "id": "step-1", "description": "…", "tool": "…", "args": {} }],
"summary": "string"
}Response — type: result | answer | cancelled
{ "type": "result", "message": "string" }While type == "plan", the frontend shows Proceed / Tweak / Cancel buttons and
sends the chosen action back as the next message:
"proceed"— execute the plan"cancel"— abort"tweak: <instruction>"— revise and re-plan