Human-in-the-Loop AI Agents with Google ADK and Telegram

Source: Dev.to

Source: Dev.to

Overview

Google’s Agent Development Kit (ADK) includes a human_in_loop sample that demonstrates how to pause an agent for human approval. The sample is an excellent starting point for understanding the mechanics of long‑running tools.

This post extends that concept with a chat UI that:

1. Accepts user requests.
2. Lets the agent decide whether human review is required.
3. Sends a Telegram message to a manager’s phone with Approve / Reject buttons.

The result feels much closer to a real‑world usage pattern.

ADK Long‑Running Tools

• A normal ADK tool returns a result immediately.
• A LongRunningFunctionTool returns a preliminary response (status: pending) and expects the application to feed back an updated response later—once the external process (human, webhook, queue, etc.) completes.

The original sample implements 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)],
)

• ask_for_approval simply 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.

My version keeps the same agent definition but replaces the simulation with a full async pipeline that you can run in a browser and hand to someone.

Architecture Diagram

Below is a cleaned‑up representation of the flow.
You can keep the ASCII art in a fenced code block, or use the Mermaid diagram for a visual rendering (if your markdown viewer supports it).

Option 1 – ASCII diagram (plain text)

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

Option 2 – Mermaid flowchart (visual)

flowchart TD
    A[User (Browser)] -->|POST /chat| B[FastAPI]
    B -->|run_async| C[ADK Agent]
    C -->|detect pending| D[Telegram Bot]
    D -->|ask_for_approval() returns ticket_id| E[Approve / Reject buttons]
    E -->|manager taps button| F[Background polling loop]
    F -->|feeds updated FunctionResponse| C
    C -->|calls reimburse() or informs rejection| G[Agent outcome]
    G -->|Browser polls /status/{ticket_id}| H[UI update]

Feel free to use whichever format best fits your documentation workflow.

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:
        # Detect the pending long‑running function call
        if part.function_call and part.function_call.id in (event.long_running_tool_ids or []):
            long_running_fc = part.function_call

        # Capture the initial response (contains ticketId)
        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:

1. Sends a Telegram message.
2. Stores the ticket in an in‑memory dict.
3. Returns status: pending_approval to the browser immediately.

The agent session stays alive—it’s just waiting for the next message.

Sending the Telegram 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": (
            "*Reimbursement Approval Required*\n\n"
            f"*Purpose:* {purpose}\n"
            f"*Amount:* ${amount:.2f}"
        ),
        "parse_mode": "Markdown",
        "reply_markup": keyboard,
    },
)

The manager sees a nicely formatted message and taps Approve or Reject on their phone.

Telegram Long‑Polling Loop

Instead of a public HTTPS webhook, I use Telegram’s long‑polling API in an asyncio background task started via FastAPI’s lifespan:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start the polling loop when the app starts
    task = asyncio.create_task(_telegram_polling_loop())
    yield
    # Cancel the loop when the app shuts down
    task.cancel()

Handling the Callback Query

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 the agent's final text
    ...

Key point: The function_call_id must match the original call. ADK tracks in‑flight tool calls by ID; if the IDs differ, the agent never knows the approval happened.

Front‑End Polling

The front‑end sends the initial chat request, receives a pending_approval response with a ticket_id, and then polls /status/{ticket_id} every 2 seconds:

async function pollStatus(ticketId, pendingMsgEl) {
  while (true) {
    // wait 2 seconds before the next request
    await new Promise(r => setTimeout(r, 2000));

    const res  = await fetch(`/status/${ticketId}`);
    const data = await res.json();

    // stop polling once the manager has decided
    if (data.status === 'approved' || data.status === 'rejected') {
      pendingMsgEl.remove();                     // remove the “pending” bubble
      addMessage(data.result, data.status);      // show the final result
      break;
    }
  }
}

When the manager makes a decision on Telegram, the background process updates the ticket’s status and result. The next poll picks up the change and renders a green (approved) or red (rejected) message bubble.

Lessons Learned

• Exact ID matching is crucial.

  • ADK tracks in‑flight tool calls by function_call_id.
  • The updated FunctionResponse must carry the same id (and name) as the original FunctionCall.

• Using Telegram’s long‑polling API avoids the need for a publicly reachable webhook, simplifying local development.

• The pattern demonstrated here—detect → notify → await → feed back—is reusable for any human‑in‑the‑loop workflow (e.g., Slack, email, custom UI).

Happy hacking!

Handling Long‑Running Function Calls

When a FunctionResponse is returned with an incorrect ID, the agent either ignores the response or raises an error.
The correct ID is taken from event.long_running_tool_ids—a set of IDs that the ADK marks as long‑running on the event where the function‑call part appears.

1. Capture the long‑running function call

# During the initial `run_async` pass
if part.function_call.id in (event.long_running_tool_ids or []):
    # Save the entire FunctionCall object, not just its ID
    long_running_fc = part.function_call

2. Return the response later

types.FunctionResponse(
    id=long_running_fc.id,      # ← critical: use the saved ID
    name=long_running_fc.name,
    response={ ... },         # your function’s result here
)

Note: Always store the whole FunctionCall object (or at least both its id and name) while the agent processes its initial turn. This ensures the subsequent FunctionResponse can be matched correctly.

Production‑Ready Considerations

Before exposing this to real users, address the following items:

Quick Start Guide

1. Clone / copy the files

2. Set up environment variables

   cp .env.example .env
   # Fill in GOOGLE_API_KEY, TELEGRAM_API_KEY, TELEGRAM_CHAT_ID

3. Install dependencies

   pip install google-api fastapi uvicorn httpx python-dotenv

4. Run the service

   uvicorn api:app --reload

5. Test – Open the Swagger UI (e.g., http://localhost:8000/docs) and try the following commands:

   • "Reimburse $50 for lunch" → auto‑approved, no Telegram message.
   • "Reimburse $200 for conference travel" → Telegram message appears; wait for approval.

Why This Matters

Google ADK’s LongRunningFunctionTool is the ideal abstraction for any agent workflow that must pause and wait for external input—such as human approval, a webhook, a queue message, etc.

The sample below demonstrates the core pattern, and this post explains how to integrate it into a real asynchronous web service with a live notification channel.

Full Code

# Insert your full implementation here.
# Example:
# from google_adk import LongRunningFunctionTool
# ...

Feel free to ask questions in the comments!