agentconnect.agents.telegram.telegram_agent module

AgentConnect Telegram AI Agent implementation.

This module provides a Telegram bot interface for AgentConnect, allowing users to interact with an AI agent through Telegram.

class TelegramAIAgent(agent_id, name, provider_type, model_name, api_key, identity, capabilities=None, personality='helpful and friendly', organization_id=None, interaction_modes=[InteractionMode.HUMAN_TO_AGENT, InteractionMode.AGENT_TO_AGENT], groups_file='groups.txt', max_tokens_per_minute=5500, max_tokens_per_hour=100000, telegram_token=None)

Bases: AIAgent

An AgentConnect agent that interacts with users through Telegram.

This agent extends AIAgent to provide Telegram integration, enabling:

  • Natural language conversations with users via Telegram private chats

  • Group chat interactions through bot mentions

  • Media message handling (photos, documents, voice, etc.)

  • Announcements to registered groups

  • Integration with other AgentConnect agents via collaboration requests

The agent connects to the Telegram API and processes messages concurrently with AgentConnect inter-agent communications, allowing it to serve as both a user interface and a collaborative agent within the AgentConnect ecosystem.

Parameters:

  • agent_id (str) – Unique identifier for the agent

  • name (str) – Human-readable name for the agent (appears in Telegram)

  • provider_type (ModelProvider) – Type of LLM provider (e.g., GOOGLE, OPENAI)

  • model_name (ModelName) – Specific LLM to use (e.g., GEMINI2_FLASH, GPT4)

  • api_key (str) – API key for the LLM provider

  • identity (AgentIdentity) – Identity information for the agent

  • capabilities (List[Capability], optional) – Additional capabilities beyond Telegram-specific ones

  • personality (str, optional) – Description of the agent’s personality

  • organization_id (str, optional) – ID of the organization the agent belongs to

  • interaction_modes (List[InteractionMode], optional) – Supported interaction modes

  • groups_file (str, optional) – File path to store registered group IDs

  • max_tokens_per_minute (int, optional) – Rate limiting for token usage per minute

  • max_tokens_per_hour (int, optional) – Rate limiting for token usage per hour

  • telegram_token (str, optional) – Telegram Bot API token (can also use TELEGRAM_BOT_TOKEN env var)

Note

When running the agent, both the Telegram bot polling and AgentConnect message processing loops run concurrently, allowing the agent to respond to both Telegram users and other agents in the AgentConnect ecosystem.

Example

from agentconnect.agents.telegram import TelegramAIAgent
from agentconnect.core.types import AgentIdentity, ModelProvider, ModelName

# Initialize the agent
agent = TelegramAIAgent(
    agent_id="telegram_bot",
    name="My Assistant",
    provider_type=ModelProvider.GOOGLE,
    model_name=ModelName.GEMINI2_FLASH,
    api_key="your_google_api_key",
    identity=AgentIdentity.create_key_based(),
    telegram_token="your_telegram_token"
)

# Register with communication hub
await hub.register_agent(agent)

# Start the agent
await agent.run()
HELP_TEXT = "I'm an AgentConnect-powered conversational Telegram bot. Here's what I can do:\n\n• Chat with you about any topic (just type normally)\n• Create and send announcements to groups\n• Process your messages using AI capabilities\n• Collaborate with other agents when needed\n\n<b>Commands:</b>\n/start - Restart the bot or get welcome message\n/help - Show this help message\n\nYou can also use the buttons below to access specific features."
get_custom_tools()

Get custom tools for the agent workflow.

Return type:

List[BaseTool]

Returns:

List of BaseTool instances

async start_telegram_bot()

Start the Telegram bot polling.

This method initializes the bot’s connection to the Telegram API and registers all message handlers.

Returns:

None

Raises:

RuntimeError – If the Telegram bot cannot be started

async stop_telegram_bot()

Stop the Telegram bot polling.

This method gracefully shuts down the Telegram bot, closing the connection to the Telegram API and saving any persistent data like registered group IDs. It should be called before shutting down the application to ensure clean termination.

Returns:

None

async process_message(message)

Process an incoming AgentConnect message.

This overrides the AIAgent.process_message method to handle Telegram-specific message processing, including both direct Telegram messages and collaboration requests from other agents.

Return type:

Message | None

Parameters:

message (Message)

async run()

Start the Telegram bot and the agent’s message processing loop.

This method starts two concurrent processes: 1. The Telegram bot polling loop that listens for messages from Telegram users 2. The parent AIAgent’s message processing loop that handles inter-agent communications

Both processes run concurrently, allowing the agent to serve as both a Telegram bot and an AgentConnect collaborative agent simultaneously.

Returns:

None

Raises:

  • RuntimeError – If the Telegram bot cannot be started

  • ConnectionError – If there are network issues with the Telegram API

  • Exception – Any unhandled exceptions from either processing loop

async on_shutdown()

Handler for bot shutdown.

message_history: List[Message]
pending_requests: Dict[str, Dict[str, Any]]