Telegram

This page documents the intended user-facing behavior for a Telegram distribution. The integration, trigger configuration, and framework helpers described here are design targets and work in progress rather than shipped functionality.

Telegram distributions connect bot chats, group mentions, replies, and other Telegram activity to Aion agent workflows. The goal is to let a developer configure which inbound Telegram events should count as triggers, normalize them into a generic A2A transport model, and then reply back into the same Telegram chat context by default.

Overview

The default Telegram loop should feel simple even though the transport model is generic:

Configure the Telegram trigger modes at the distribution layer.
Convert each selected Telegram update into a normalized A2A request.
Let the framework adapter resolve the response by normal precedence.
Send the response back into the same Telegram chat, thread, or reply chain by default.

This means the transport contract stays shared across distributions while the configuration layer remains platform-aware. For the lower-level contract, see Distribution and Event.

Default Request Loop

Configuration

WIP.

Message Mapping

Telegram distributions should map inbound and outbound messages through the same shared transport contracts used by other messaging integrations, while still preserving Telegram-specific chat and reply context. Inbound

Protocol-level request metadata and event identity are defined by Distribution, Event, and Distribution/Messaging.
Framework-level request mapping is described in LangGraph Message Mapping and Google ADK Message Mapping.

Outbound

Default response precedence is: SDK-managed response buffer first, explicit a2a_outbox second, and framework-native fallback third.
Structured outbound parts are defined by Distribution/Messaging and Distribution/Cards.
Framework-level response mapping is described in LangGraph Message Mapping and Google ADK Message Mapping.

Features

Mentions

Mentions in Telegram groups should be treated as normal inbound message events. The framework sees text plus normalized transport context, and the response flows back into the same group context.

LangGraph
Google ADK
A2A

from typing import Annotated, Optional, TypedDict
from langchain_core.messages import AIMessage, BaseMessage
from langgraph.graph import add_messages

from aion.shared.types import A2AInbox


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    a2a_inbox: Optional[A2AInbox]


def on_telegram_mention(state: AgentState) -> dict:
    inbound = state["messages"][-1]
    return {
        "messages": [
            AIMessage(content=f"Telegram mention received: {inbound.content}")
        ]
    }

The distribution should reply back into the same Telegram chat and keep the same reply context when one exists.

from google.adk.agents import BaseAgent
from google.adk.events import Event


class TelegramMentionAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        text = ctx.a2a_inbox.message.parts[0].text
        yield Event(
            author=self.name,
            content={"parts": [{"text": f"Telegram mention received: {text}"}]},
        )

{
  "message": {
    "role": "ROLE_USER",
    "extensions": [
      "https://docs.aion.to/a2a/extensions/aion/distribution/1.0.0",
      "https://docs.aion.to/a2a/extensions/aion/event/1.0.0"
    ],
    "parts": [
      { "text": "@bot summarize the last three messages" },
      {
        "data": {
          "userId": "tg-user-42",
          "messageId": "4002",
          "contextId": "tg-chat-9988",
          "parentContextId": "3998",
          "trajectory": "conversation"
        },
        "mediaType": "application/json",
        "metadata": {
          "https://docs.aion.to/a2a/extensions/aion/event/1.0.0": {
            "schema": "https://docs.aion.to/a2a/extensions/aion/distribution/messaging/1.0.0#MessageEventPayload"
          }
        }
      },
      {
        "data": {
          "provider": "telegram",
          "event": {
            "update_id": 9001,
            "message": {
              "message_id": 4002,
              "chat": { "id": -9988, "type": "supergroup" },
              "from": { "id": 42, "username": "tg_user_42" },
              "text": "@bot summarize the last three messages"
            }
          }
        },
        "mediaType": "application/json",
        "metadata": {
          "https://docs.aion.to/a2a/extensions/aion/event/1.0.0": {
            "schema": "https://docs.aion.to/a2a/extensions/aion/distribution/messaging/1.0.0#SourceSystemEventPayload"
          }
        }
      }
    ],
    "metadata": {
      "https://docs.aion.to/a2a/extensions/aion/event/1.0.0": {
        "type": "to.aion.distribution.message.1.0.0",
        "source": "aion://distribution/telegram-dist-123",
        "id": "evt-telegram-mention-001"
      }
    }
  }
}

Reactions

Telegram reactions should usually arrive as activity events, not as conversational messages. They should still be available to the framework through the same generic envelope.

LangGraph
Google ADK
A2A

def on_telegram_reaction(state: AgentState) -> dict:
    inbox = state.get("a2a_inbox")
    event_type = inbox.message.metadata[
        "https://docs.aion.to/a2a/extensions/aion/event/1.0.0"
    ]["type"]

    if event_type == "to.aion.distribution.activity.1.0.0":
        return {"messages": [AIMessage(content="Reaction observed.")]}

    return {}

class TelegramReactionAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        yield Event(
            author=self.name,
            content={"parts": [{"text": "Reaction observed."}]},
        )

{
  "message": {
    "role": "ROLE_USER",
    "metadata": {
      "https://docs.aion.to/a2a/extensions/aion/event/1.0.0": {
        "type": "to.aion.distribution.activity.1.0.0",
        "source": "aion://distribution/telegram-dist-123",
        "id": "evt-telegram-reaction-001"
      }
    },
    "parts": [
      {
        "data": {
          "provider": "telegram",
          "event": {
            "update_type": "message_reaction",
            "emoji": "🔥"
          }
        },
        "mediaType": "application/json",
        "metadata": {
          "https://docs.aion.to/a2a/extensions/aion/event/1.0.0": {
            "schema": "https://docs.aion.to/a2a/extensions/aion/distribution/messaging/1.0.0#SourceSystemEventPayload"
          }
        }
      }
    ]
  }
}

Cards

Telegram does not use the same visual model as Slack Block Kit, but it still benefits from a generic card document. The distribution should interpret that document into Telegram-native layouts such as inline keyboards, media captions, or structured reply content.

LangGraph
Google ADK
A2A

from a2a.types import Message, Part, Role
from aion.shared.types import A2AOutbox


def send_telegram_card(state: AgentState) -> dict:
    card_document = """
    <Card title="Digest ready">
      <Text>Choose how you want to continue.</Text>
      <Actions>
        <Button url="https://example.com/digest/42">Open digest</Button>
      </Actions>
    </Card>
    """.strip()

    return {
        "a2a_outbox": A2AOutbox(
            message=Message(
                role=Role.ROLE_AGENT,
                parts=[
                    Part(text="Your digest is ready"),
                    Part(
                        raw=card_document,
                        filename="digest-ready.card.jsx",
                        mediaType="application/vnd.aion.card+jsx",
                    ),
                ],
            )
        )
    }

from a2a.types import Message, Part, Role
from aion.shared.types import A2AOutbox
from google.adk.events import Event, EventActions


class TelegramCardAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        card_document = """
        <Card title="Choose an option">
          <Text>Telegram should render this as a native interactive reply.</Text>
        </Card>
        """.strip()

        yield Event(
            author=self.name,
            actions=EventActions(
                state_delta={
                    "a2a_outbox": A2AOutbox(
                        message=Message(
                            role=Role.ROLE_AGENT,
                            parts=[
                                Part(text="Choose an option"),
                                Part(
                                    raw=card_document,
                                    filename="choose-an-option.card.jsx",
                                    mediaType="application/vnd.aion.card+jsx",
                                ),
                            ],
                        )
                    )
                }
            ),
        )

{
  "message": {
    "role": "ROLE_AGENT",
    "extensions": [
      "https://docs.aion.to/a2a/extensions/aion/distribution/cards/1.0.0"
    ],
    "parts": [
      { "text": "Choose an option" },
      {
        "raw": "<Card title=\"Choose an option\">\n  <Text>Telegram should render this as a native interactive reply.</Text>\n</Card>",
        "filename": "choose-an-option.card.jsx",
        "mediaType": "application/vnd.aion.card+jsx",
        "metadata": {
          "https://docs.aion.to/a2a/extensions/aion/distribution/cards/1.0.0": {
            "schema": "https://docs.aion.to/a2a/extensions/aion/distribution/cards/1.0.0#CardPayload"
          }
        }
      }
    ]
  }
}

Streaming

Telegram should preserve the same target message context while streaming. The intent is to post the reply once and then edit that message as new text arrives, similar to how streaming should work in other conversational distributions.

LangGraph
Google ADK
A2A

from langchain_core.messages import AIMessageChunk
from langgraph.types import StreamWriter
from aion.langgraph import emit_message


def stream_telegram_reply(state: AgentState, writer: StreamWriter):
    emit_message(writer, AIMessageChunk(content="Here"))
    emit_message(writer, AIMessageChunk(content=" is the first part "))
    emit_message(writer, AIMessageChunk(content="of the answer."))
    return state

class TelegramStreamingAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        yield Event(
            author=self.name,
            content={"parts": [{"text": "Here"}]},
            partial=True,
        )
        yield Event(
            author=self.name,
            content={"parts": [{"text": " is the first part "}]},
            partial=True,
        )
        yield Event(
            author=self.name,
            content={"parts": [{"text": "of the answer."}]},
        )

{
  "task": {
    "id": "task-telegram-stream-42",
    "contextId": "ctx-telegram-stream-42",
    "status": {
      "state": "working"
    }
  }
}

DMs

Telegram private chats are the cleanest default request loop. The inbound message maps to trajectory = "direct-message" and the default outbound response goes right back to that chat.

LangGraph
Google ADK
A2A

def on_telegram_dm(state: AgentState) -> dict:
    inbound = state["messages"][-1]
    return {
        "messages": [
            AIMessage(content=f"Telegram DM received: {inbound.content}")
        ]
    }

class TelegramDMAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        text = ctx.a2a_inbox.message.parts[0].text
        yield Event(
            author=self.name,
            content={"parts": [{"text": f"Telegram DM received: {text}"}]},
        )

{
  "data": {
    "userId": "tg-user-42",
    "messageId": "4002",
    "contextId": "tg-private-chat-42",
    "parentContextId": null,
    "trajectory": "direct-message"
  }
}

Usage

Core concepts

Distributions

Tools

Overview

Default Request Loop

Configuration

Message Mapping

Features

Mentions

Reactions

Cards

Streaming

DMs

Usage

Core concepts

Distributions

Tools

​Overview

​Default Request Loop

​Configuration

​Message Mapping

​Features

​Mentions

​Reactions

​Cards

​Streaming

​DMs

Overview

Default Request Loop

Configuration

Message Mapping

Features

Mentions

Reactions

Cards

Streaming

DMs