Skip to content

Telegram Gateway Integration

The HLD Telegram Gateway is a high-performance, asynchronous bridge designed to connect the Telegram Bot API with HLD Agent Runtimes (ADK). It manages stateful sessions, filters agent reasoning for cleaner user UX, and ensures compliant message delivery through automated chunking and HTML translation.

Architecture Overview

The gateway acts as a stateless proxy that maintains long-lived SSE connections to the agent runtime.

  1. Ingress: Telegram Webhook/Polling receives an Update.
  2. Authentication: The gateway validates the chat_id against an internal allowlist.
  3. Session Mapping: The Telegram User ID is mapped to a persistent Agent Session ID.
  4. Execution: The message is forwarded to the Agent Runtime's /run_sse endpoint.
  5. Stream Processing: The bridge parses the incoming Server-Sent Events (SSE), stripping internal reasoning markers.
  6. Egress: The final response is translated to Telegram-compliant HTML and delivered.

Core Implementation

1. Stateful Session Management

To maintain conversation context, every Telegram user is mapped to a unique agent session. If no active session exists, the gateway auto-provisions one.

async def _ensure_session(user_id: str) -> str:
    \"\"\"Get or create an Agent session for a Telegram user.\"\"\"
    async with httpx.AsyncClient(timeout=15) as client:
        # Check for existing sessions
        resp = await client.get(
            f"{BASE_URL}/apps/{APP_NAME}/users/{user_id}/sessions",
            headers=headers,
        )
        if resp.status_code == 200:
            sessions = resp.json()
            if sessions:
                return sessions[-1]["id"]

        # Create new session if none found
        resp = await client.post(
            f"{BASE_URL}/apps/{APP_NAME}/users/{user_id}/sessions",
            headers=headers,
        )
        return resp.json()["id"]

2. Stream Handling and Reasoning Filtering

The HLD Runtimes emit detailed reasoning steps that are vital for debugging but should not be shown to end-users. The gateway filters these out in real-time.

async for line in resp.aiter_lines():
    if not line.startswith("data: "):
        continue

    event = json.loads(line[6:])
    parts = event.get("content", {}).get("parts", [])

    for part in parts:
        # Filter reasoning/thought markers
        if part.get("thought", False):
            continue

        text_chunk = part.get("text", "")
        if text_chunk:
            full_response += text_chunk

3. Telegram HTML Compliance

Telegram's message limit is 4096 characters, and its HTML parser is strict. The gateway uses a custom translator and chunking logic to ensure long responses are delivered reliably.

def deliver_response(text: str):
    # Convert internal markdown to Telegram-safe HTML
    html_response = to_telegram_html(text)

    if len(html_response) <= 4096:
        send_message(html_response)
    else:
        # Split on newline boundaries to avoid breaking HTML tags
        chunks = chunk_text(html_response, limit=4096)
        for chunk in chunks:
            send_message(chunk)

Security and Access Control

Access is strictly controlled via an environment-level allowlist. Requests from unauthorized chat_ids are dropped before reaching the agent runtime.

  • ALLOWED_CHAT_IDS: A comma-separated list of authorized IDs.
  • API_KEY_AUTH: The bridge injects a secure X-API-Key into all requests to the agent runtime.

Deployment

The gateway is typically deployed as a systemd service to ensure high availability.

[Unit]
Description=HLD Telegram Gateway
After=network.target

[Service]
ExecStart=/usr/bin/python3 telegram_bridge.py
WorkingDirectory=/opt/hld-gateway
Restart=always
User=hld-service

[Install]
WantedBy=multi-user.target