agentconnect.agents package

Independent agent implementations for the AgentConnect decentralized framework.

This module provides various autonomous agent implementations that operate as independent entities in a decentralized network. Each agent maintains its own identity, capabilities, and can optionally implement its own internal multi-agent system while communicating with other agents through capability-based discovery.

Key components:

  • AIAgent: Independent AI-powered agent with potential for internal multi-agent structures

  • HumanAgent: Human-in-the-loop agent that can interact securely with the decentralized network

  • TelegramAIAgent: AI agent that integrates with Telegram for user interactions

  • MemoryType: Enum for different types of agent memory

Each agent operates autonomously and can discover and communicate with other agents based on capabilities rather than pre-defined connections, enabling a truly decentralized architecture.

class AIAgent(agent_id, name, provider_type, model_name, api_key, identity, capabilities=None, personality='helpful and professional', organization_id=None, interaction_modes=[InteractionMode.HUMAN_TO_AGENT, InteractionMode.AGENT_TO_AGENT], max_tokens_per_minute=5500, max_tokens_per_hour=100000, is_ui_mode=False, memory_type=MemoryType.BUFFER, prompt_tools=None, prompt_templates=None, custom_tools=None, agent_type='ai')

Bases: BaseAgent

Independent AI Agent implementation that operates autonomously in a decentralized network.

This agent uses language models to generate responses, can discover and communicate with other agents based on their capabilities (not pre-defined connections), and can implement its own internal multi-agent structure if needed. It operates as a peer in a decentralized system rather than as part of a centrally controlled hierarchy.

Key features:

  • Autonomous operation with independent decision-making

  • Capability-based discovery of other agents

  • Secure identity verification and communication

  • Potential for internal multi-agent structures

  • Dynamic request routing based on capabilities

Parameters:
property hub

Get the hub property.

async process_message(message)

Process an incoming message autonomously and generate a response.

This method represents the agent’s autonomous decision loop, where it: :rtype: Optional[Message]

  • Verifies message security independently

  • Makes decisions on how to respond based on capabilities

  • Can dynamically discover and collaborate with other agents as needed

  • Maintains its own internal state and conversation tracking

  • Operates without central coordination or control

The agent can leverage its internal workflow (which may include its own multi-agent system) to generate appropriate responses and handle complex tasks that may require collaboration with other independent agents in the decentralized network.

Parameters:

message (Message)

Return type:

Message | None

property prompt_tools

Get the prompt_tools property.

property registry

Get the registry property.

reset_interaction_state()

Reset the interaction state of the agent.

This resets both the cooldown state and the turn counter.

Return type:

None

set_cooldown(duration)

Set a cooldown period for the agent.

Parameters:

duration (int) – Cooldown duration in seconds

Return type:

None

class HumanAgent(agent_id, name, identity, organization_id=None)

Bases: BaseAgent

Human agent implementation for interactive communication with AI agents.

This agent handles:

  • Real-time text input/output

  • Message verification and security

  • Graceful conversation management

  • Error handling and recovery

Parameters:
async process_message(message)

Process incoming messages from other agents

Return type:

Optional[Message]

Parameters:

message (Message)

async start_interaction(target_agent)

Start an interactive session with an AI agent

Return type:

None

Parameters:

target_agent (BaseAgent)

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 on_shutdown()

Handler for bot shutdown.

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 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

class MemoryType(*values)

Bases: str, Enum

Types of memory storage backends.

BUFFER = 'buffer'

Subpackages

Submodules