Google's Agent Development Kit (ADK) ships with a human_in_loop sample that demonstrates the concept of pausing an agent for human approval. It's a great starting point for understanding the mechanics.
This post builds on that concept with a chat UI where users submit requests, the agent decides if it needs human review, and a Telegram message lands in a manager's phone with Approve / Reject buttons — closer to how this pattern would feel in real usage. Here's how I built it.
The Original Sample — What It Shows
The original sample teaches a critical ADK concept: long-running tools.
A normal ADK tool returns a result immediately. A LongRunningFunctionTool returns a preliminary response (status: pending) and then expects your application to feed back an updated response later—once the external process (a human, a webhook, a queue) completes.
The sample agent is a reimbursement bot:
root_agent = Agent(
model='gemini-2.5-flash',
name='reimbursement_agent',
instruction="""
If the amount is less than $100, automatically approve.
If greater than $100, ask for approval from the manager.
If approved, call reimburse(). If rejected, inform the employee.
""",
tools=[reimburse, LongRunningFunctionTool(func=ask_for_approval)],
)
The ask_for_approval tool just returns a ticket ID and status: pending. The sample then simulates approval by immediately feeding back status: approved in the same script. No real human ever sees it.
A More Intuitive Implementation
My version keeps the same agent definition but replaces the simulation with a full async pipeline — something you can actually open in a browser and hand to someone.
Architecture at a Glance
User (Browser) ──POST /chat──► FastAPI ──run_async──► ADK Agent
│ │
detect pending ask_for_approval()
│ returns ticket_id
▼
Telegram Bot
(Approve / Reject buttons)
│
──────────────┘ (manager taps button)
│
▼
Background polling loop
feeds updated FunctionResponse
back into runner.run_async()
│
▼
Agent calls reimburse() or informs rejection
│
▼
Browser polls /status/{ticket_id}
and updates the UI
Key Pieces
1. The FastAPI POST /chat Endpoint
The endpoint runs the agent and watches the event stream for a long-running tool call:
async for event in runner.run_async(
session_id=req.session_id,
user_id=req.user_id,
new_message=content,
):
for part in event.content.parts:
if part.function_call and part.function_call.id in (event.long_running_tool_ids or []):
long_running_fc = part.function_call
if part.function_response and long_running_fc and part.function_response.id == long_running_fc.id:
initial_response = part.function_response
ticket_id = initial_response.response.get("ticketId")
If a pending ticket is found, the API sends a Telegram message and stores the ticket in an in-memory dict, then returns status: pending_approval to the browser immediately. The agent session stays alive—it's just waiting for the next message.
2. Telegram Approval Message
keyboard = {
"inline_keyboard": [[
{"text": "✅ Approve", "callback_data": f"approve:{ticket_id}"},
{"text": "❌ Reject", "callback_data": f"reject:{ticket_id}"},
]]
}
await _telegram_post("sendMessage", {
"chat_id": TELEGRAM_CHAT_ID,
"text": f"*Reimbursement Approval Required*\n\n*Purpose:* {purpose}\n*Amount:* ${amount:.2f}",
"parse_mode": "Markdown",
"reply_markup": keyboard,
})
The manager sees a formatted message and taps Approve or Reject on their phone.
3. Background Long-Polling Loop
Instead of webhooks (which require a public HTTPS endpoint), I use Telegram's long-polling API in an asyncio background task started via FastAPI's lifespan:
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(_telegram_polling_loop())
yield
task.cancel()
The polling loop receives the callback query, looks up the pending ticket, and feeds the decision back into the runner:
updated_part = types.Part(
function_response=types.FunctionResponse(
id=ticket["function_call_id"], # must match the original call
name=ticket["function_call_name"],
response={
"status": decision, # "approved" or "rejected"
"ticketId": ticket_id,
"approver_feedback": f"{decision.capitalize()} via Telegram",
},
)
)
async for event in runner.run_async(
session_id=ticket["session_id"],
user_id=ticket["user_id"],
new_message=types.Content(parts=[updated_part], role="user"),
):
... # collect agent's final text
This is the exact pattern from the README: same id and name as the original FunctionCall, sent back with role="user". Without this, the agent has no idea the approval happened.
4. Browser Polls for Resolution
The frontend sends the initial chat request, gets back pending_approval with a ticket_id, and starts polling /status/{ticket_id} every 2 seconds:
async function pollStatus(ticketId, pendingMsgEl) {
while (true) {
await new Promise(r => setTimeout(r, 2000));
const res = await fetch(`/status/${ticketId}`);
const data = await res.json();
if (data.status === 'approved' || data.status === 'rejected') {
pendingMsgEl.remove();
addMessage(data.result, data.status);
break;
}
}
}
When the manager decides on Telegram, the background loop updates the ticket status and result. The browser picks it up on the next poll and renders a green (approved) or red (rejected) message bubble.
The Crucial ADK Detail: Long-Running Tool IDs
The most important thing I learned building this: you must match the function_call_id exactly when sending back the updated response.
ADK tracks in-flight tool calls by ID. If you send back a FunctionResponse with the wrong ID, the agent ignores it or throws an error. The ID comes from event.long_running_tool_ids—a set of IDs that ADK marks as long-running on the event where the function call part appears.
# Capture during initial run_async
if part.function_call.id in (event.long_running_tool_ids or []):
long_running_fc = part.function_call # save this entire object
# Use later when feeding back the result
types.FunctionResponse(
id=long_running_fc.id, # ← critical
name=long_running_fc.name,
response={...},
)
Store the whole FunctionCall object—not just the ticket ID—while the agent is running its initial turn.
Taking It Further
This implementation is already closer to real life than a script, but a few things you'd want before putting it in front of real users:
Persistent storage: The in-memory_pendingdict is gone on restart. Swap it for Redis or Postgres — the ticket state and session data need to survive deploys.
Telegram webhooks: Long-polling works great for self-hosted or local setups. If you have a public HTTPS endpoint, switch to webhooks — no background thread, lower latency.
ADK session persistence:InMemorySessionServicealso dies on restart. ADK ships database-backed session services for exactly this; swap it in and your conversation history survives.
Auth: Tie/chatto your existing auth system so users can only access their own sessions and tickets.
Running It Yourself
# 1. Clone / copy the files
# 2. Set up .env
cp .env.example .env
# Fill in GOOGLE_API_KEY, TELEGRAM_API_KEY, TELEGRAM_CHAT_ID
# 3. Install deps
pip install google-adk fastapi uvicorn httpx python-dotenv
# 4. Run
uvicorn api:app --reload
Open http://localhost:8000 and try:
- "Reimburse $50 for lunch" → auto-approved, no Telegram message
- "Reimburse $200 for conference travel" → Telegram message appears, wait for approval
Takeaway
Google ADK's LongRunningFunctionTool is the right abstraction for any agent workflow that needs to pause and wait for external input—human approval, a webhook, a queue message. The sample shows you what to do; this post shows you how to wire it into a real async web service with a live notification channel.
The full code is on GitHub: jameszh/adk_human_in_loop. Happy to answer questions in the comments.
SOCIAL SHARE CARD GENERATOR